当前位置：首页 > news >正文

【AI Study】第四天，Pandas（10）- 实用技巧

news 来源：原创 2025/6/20 9:08:49

文章概要

本文详细介绍 Pandas 的实用技巧，包括：

常用操作
调试技巧
工具集成
实际应用示例

常用操作

数据检查

# 基本信息检查
def check_dataframe_info(df):"""检查 DataFrame 基本信息Args:df (pd.DataFrame): 输入数据Returns:dict: 数据信息"""info = {'shape': df.shape,'dtypes': df.dtypes,'memory_usage': df.memory_usage(deep=True).sum(),'null_counts': df.isnull().sum(),'unique_counts': df.nunique()}return info# 数据质量检查
def check_data_quality(df):"""检查数据质量Args:df (pd.DataFrame): 输入数据Returns:dict: 质量检查结果"""quality = {'duplicates': df.duplicated().sum(),'null_percentage': (df.isnull().sum() / len(df) * 100).round(2),'unique_percentage': (df.nunique() / len(df) * 100).round(2)}return quality# 数据分布检查
def check_distribution(df, columns=None):"""检查数据分布Args:df (pd.DataFrame): 输入数据columns (list): 要检查的列Returns:dict: 分布信息"""if columns is None:columns = df.select_dtypes(include=[np.number]).columnsdistribution = {}for col in columns:distribution[col] = {'mean': df[col].mean(),'std': df[col].std(),'min': df[col].min(),'max': df[col].max(),'skew': df[col].skew(),'kurtosis': df[col].kurtosis()}return distribution

数据验证

# 数据类型验证
def validate_dtypes(df, expected_dtypes):"""验证数据类型Args:df (pd.DataFrame): 输入数据expected_dtypes (dict): 期望的数据类型Returns:bool: 验证结果"""for col, dtype in expected_dtypes.items():if col not in df.columns:raise ValueError(f"列 {col} 不存在")if df[col].dtype != dtype:raise TypeError(f"列 {col} 的数据类型不匹配")return True# 数值范围验证
def validate_range(df, column, min_val=None, max_val=None):"""验证数值范围Args:df (pd.DataFrame): 输入数据column (str): 列名min_val (float): 最小值max_val (float): 最大值Returns:bool: 验证结果"""if min_val is not None and (df[column] < min_val).any():raise ValueError(f"列 {column} 存在小于 {min_val} 的值")if max_val is not None and (df[column] > max_val).any():raise ValueError(f"列 {column} 存在大于 {max_val} 的值")return True# 唯一性验证
def validate_unique(df, columns):"""验证唯一性Args:df (pd.DataFrame): 输入数据columns (list): 列名列表Returns:bool: 验证结果"""if df[columns].duplicated().any():raise ValueError(f"列 {columns} 存在重复值")return True

数据转换

# 日期时间转换
def convert_datetime(df, columns, format=None):"""转换日期时间Args:df (pd.DataFrame): 输入数据columns (list): 列名列表format (str): 日期格式Returns:pd.DataFrame: 转换后的数据"""for col in columns:df[col] = pd.to_datetime(df[col], format=format)return df# 分类数据转换
def convert_categorical(df, columns):"""转换分类数据Args:df (pd.DataFrame): 输入数据columns (list): 列名列表Returns:pd.DataFrame: 转换后的数据"""for col in columns:df[col] = df[col].astype('category')return df# 数值数据转换
def convert_numeric(df, columns, errors='coerce'):"""转换数值数据Args:df (pd.DataFrame): 输入数据columns (list): 列名列表errors (str): 错误处理方式Returns:pd.DataFrame: 转换后的数据"""for col in columns:df[col] = pd.to_numeric(df[col], errors=errors)return df

调试技巧

问题诊断

# 内存使用诊断
def diagnose_memory(df):"""诊断内存使用Args:df (pd.DataFrame): 输入数据Returns:pd.DataFrame: 内存使用信息"""memory_usage = df.memory_usage(deep=True)memory_usage = memory_usage.to_frame('memory_usage')memory_usage['percentage'] = memory_usage['memory_usage'] / memory_usage['memory_usage'].sum() * 100return memory_usage# 性能诊断
def diagnose_performance(func, *args, **kwargs):"""诊断函数性能Args:func (function): 要诊断的函数*args: 位置参数**kwargs: 关键字参数Returns:dict: 性能信息"""import timeimport memory_profiler# 时间性能start_time = time.time()result = func(*args, **kwargs)end_time = time.time()# 内存性能memory_usage = memory_profiler.memory_usage((func, args, kwargs))return {'execution_time': end_time - start_time,'memory_usage': max(memory_usage) - min(memory_usage)}# 数据问题诊断
def diagnose_data_issues(df):"""诊断数据问题Args:df (pd.DataFrame): 输入数据Returns:dict: 问题信息"""issues = {'missing_values': df.isnull().sum(),'duplicates': df.duplicated().sum(),'outliers': {},'inconsistencies': {}}# 检测异常值for col in df.select_dtypes(include=[np.number]).columns:q1 = df[col].quantile(0.25)q3 = df[col].quantile(0.75)iqr = q3 - q1outliers = df[(df[col] < q1 - 1.5 * iqr) | (df[col] > q3 + 1.5 * iqr)][col]issues['outliers'][col] = len(outliers)return issues

性能分析

# 代码性能分析
def profile_code(func, *args, **kwargs):"""分析代码性能Args:func (function): 要分析的函数*args: 位置参数**kwargs: 关键字参数Returns:dict: 性能分析结果"""import cProfileimport pstats# 创建性能分析器profiler = cProfile.Profile()# 运行函数profiler.enable()result = func(*args, **kwargs)profiler.disable()# 获取统计信息stats = pstats.Stats(profiler)return {'result': result,'stats': stats}# 内存分析
def profile_memory(func, *args, **kwargs):"""分析内存使用Args:func (function): 要分析的函数*args: 位置参数**kwargs: 关键字参数Returns:dict: 内存分析结果"""import memory_profiler# 运行函数并分析内存memory_usage = memory_profiler.memory_usage((func, args, kwargs))return {'peak_memory': max(memory_usage),'memory_increase': max(memory_usage) - min(memory_usage)}

内存分析

# 内存使用分析
def analyze_memory_usage(df):"""分析内存使用Args:df (pd.DataFrame): 输入数据Returns:pd.DataFrame: 内存使用分析结果"""# 计算每列的内存使用memory_usage = df.memory_usage(deep=True)# 计算每列的数据类型dtypes = df.dtypes# 计算每列的唯一值数量unique_counts = df.nunique()# 合并结果analysis = pd.DataFrame({'memory_usage': memory_usage,'dtype': dtypes,'unique_count': unique_counts})return analysis# 内存优化建议
def suggest_memory_optimization(df):"""提供内存优化建议Args:df (pd.DataFrame): 输入数据Returns:dict: 优化建议"""suggestions = {}# 分析每列for col in df.columns:col_suggestions = []# 检查数据类型if df[col].dtype == 'object':# 如果唯一值比例小于 50%，建议转换为 categoryif df[col].nunique() / len(df) < 0.5:col_suggestions.append('考虑转换为 category 类型')# 检查数值类型elif pd.api.types.is_numeric_dtype(df[col]):# 检查是否可以转换为更小的整数类型if df[col].dtype == 'int64':if df[col].min() >= -32768 and df[col].max() <= 32767:col_suggestions.append('考虑转换为 int16 类型')elif df[col].min() >= -2147483648 and df[col].max() <= 2147483647:col_suggestions.append('考虑转换为 int32 类型')if col_suggestions:suggestions[col] = col_suggestionsreturn suggestions

工具集成

与其他库的集成

# 与 NumPy 集成
def numpy_integration(df):"""与 NumPy 集成Args:df (pd.DataFrame): 输入数据Returns:np.ndarray: NumPy 数组"""# 转换为 NumPy 数组array = df.values# 使用 NumPy 函数mean = np.mean(array, axis=0)std = np.std(array, axis=0)return array# 与 Matplotlib 集成
def matplotlib_integration(df):"""与 Matplotlib 集成Args:df (pd.DataFrame): 输入数据"""import matplotlib.pyplot as plt# 创建图表fig, axes = plt.subplots(2, 2, figsize=(12, 8))# 绘制不同类型的图表df.plot(kind='line', ax=axes[0, 0])df.plot(kind='bar', ax=axes[0, 1])df.plot(kind='scatter', x=df.columns[0], y=df.columns[1], ax=axes[1, 0])df.plot(kind='box', ax=axes[1, 1])plt.tight_layout()plt.show()# 与 Seaborn 集成
def seaborn_integration(df):"""与 Seaborn 集成Args:df (pd.DataFrame): 输入数据"""import seaborn as sns# 设置样式sns.set_style('whitegrid')# 创建图表fig, axes = plt.subplots(2, 2, figsize=(12, 8))# 绘制不同类型的图表sns.lineplot(data=df, ax=axes[0, 0])sns.barplot(data=df, ax=axes[0, 1])sns.scatterplot(data=df, x=df.columns[0], y=df.columns[1], ax=axes[1, 0])sns.boxplot(data=df, ax=axes[1, 1])plt.tight_layout()plt.show()

工作流自动化

# 数据处理工作流
def data_processing_workflow(df):"""数据处理工作流Args:df (pd.DataFrame): 输入数据Returns:pd.DataFrame: 处理后的数据"""# 1. 数据清洗df = df.dropna()df = df.drop_duplicates()# 2. 数据转换df = convert_datetime(df, ['date'])df = convert_categorical(df, ['category'])df = convert_numeric(df, ['value'])# 3. 数据验证validate_dtypes(df, {'date': 'datetime64[ns]','category': 'category','value': 'float64'})# 4. 数据聚合result = df.groupby('category').agg({'value': ['mean', 'std', 'min', 'max']})return result# 报告生成工作流
def report_generation_workflow(df, output_file):"""报告生成工作流Args:df (pd.DataFrame): 输入数据output_file (str): 输出文件路径"""# 1. 数据检查info = check_dataframe_info(df)quality = check_data_quality(df)distribution = check_distribution(df)# 2. 数据可视化matplotlib_integration(df)seaborn_integration(df)# 3. 生成报告with pd.ExcelWriter(output_file) as writer:# 基本信息pd.DataFrame(info).to_excel(writer, sheet_name='基本信息')# 质量检查pd.DataFrame(quality).to_excel(writer, sheet_name='质量检查')# 分布信息pd.DataFrame(distribution).to_excel(writer, sheet_name='分布信息')# 原始数据df.to_excel(writer, sheet_name='原始数据')

环境配置

# 环境检查
def check_environment():"""检查环境配置Returns:dict: 环境信息"""import sysimport platformenvironment = {'python_version': sys.version,'platform': platform.platform(),'pandas_version': pd.__version__,'numpy_version': np.__version__,'matplotlib_version': matplotlib.__version__,'seaborn_version': sns.__version__}return environment# 环境配置
def configure_environment():"""配置环境Returns:dict: 配置信息"""# 设置 Pandas 显示选项pd.set_option('display.max_rows', 100)pd.set_option('display.max_columns', 100)pd.set_option('display.width', 1000)# 设置 Matplotlib 样式plt.style.use('seaborn')# 设置 Seaborn 样式sns.set_style('whitegrid')return {'pandas_options': pd.get_option('display.max_rows'),'matplotlib_style': plt.style.available[0],'seaborn_style': sns.axes_style()}