Python数据清洗实战：建模思维下的缺失值、异常值处理与特征工程

1. 项目概述：从“脏数据”到“干净模型”的必经之路

搞数学建模的朋友，尤其是用Python的，肯定都经历过这个阶段：拿到数据，满心欢喜准备大干一场，结果一打开CSV或者Excel，血压瞬间就上来了。缺失值、异常值、重复记录、格式不统一、量纲差异巨大……这些“脏数据”就像横在建模路上的第一道坎，处理不好，后面再高级的算法、再精巧的模型，效果都得大打折扣，甚至得出完全错误的结论。

这个项目，就是我结合自己多年参加国赛、美赛以及实际科研项目中的无数次“踩坑”经历，总结、提炼并封装的一套Python数据清洗专用代码库。它不是一个简单的函数集合，而是一套基于建模思维的清洗流程和工具箱。建模中的数据清洗，和普通数据分析有一个核心区别：我们的目标不是得到一个“好看”的数据集，而是为后续的模型构建一个稳定、可靠、符合算法假设的“输入源”。这意味着，每一步清洗决策，都需要考虑对模型可能产生的影响。

这套代码的价值在于，它把那些散落在各个教程、需要反复调试的“脏活累活”，变成了清晰、可复现的标准化操作。无论你是刚接触建模的新手，还是有一定经验但希望提升效率的老手，这套代码都能帮你节省大量在数据预处理阶段反复试错的时间，让你把精力更集中在特征工程和模型调优这些更有创造性的环节上。接下来，我就把这套“压箱底”的实践心得，从设计思路到代码细节，毫无保留地拆解给你看。

2. 核心清洗流程设计与建模思维

数据清洗不是漫无目的地修修补补，必须有一个清晰的流程和明确的目标。在数学建模中，我通常遵循“诊断 -> 规划 -> 执行 -> 验证”的闭环流程，这套代码也是围绕这个逻辑构建的。

2.1 诊断阶段：全面“体检”数据

清洗的第一步是了解你的数据有多“脏”。很多新手会直接上手处理缺失值，这是大忌。必须先做全面的描述性统计和可视化诊断。

import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns def data_diagnosis(df, save_report=False): """ 对数据框进行全方位诊断，生成报告。 参数: df: pandas DataFrame，待诊断的数据。 save_report: bool，是否将报告保存为文本文件。 返回: dict，包含各类诊断信息的字典。 """ diagnosis = {} # 1. 基础信息 diagnosis['shape'] = df.shape diagnosis['dtypes'] = df.dtypes.to_dict() diagnosis['columns'] = df.columns.tolist() # 2. 缺失值统计（建模中非常关键！） missing_stats = df.isnull().sum() missing_percent = (missing_stats / len(df)) * 100 diagnosis['missing_count'] = missing_stats[missing_stats > 0].to_dict() diagnosis['missing_percent'] = missing_percent[missing_percent > 0].to_dict() # 3. 唯一值统计（识别潜在的分类变量或常量列） unique_stats = df.nunique() diagnosis['unique_values'] = unique_stats.to_dict() # 4. 数值型数据描述统计（识别量纲和异常） numeric_cols = df.select_dtypes(include=[np.number]).columns if len(numeric_cols) > 0: diagnosis['numeric_describe'] = df[numeric_cols].describe().to_dict() # 计算偏度和峰度，初步判断分布 from scipy import stats skewness = df[numeric_cols].apply(lambda x: stats.skew(x.dropna())) kurtosis = df[numeric_cols].apply(lambda x: stats.kurtosis(x.dropna())) diagnosis['skewness'] = skewness.to_dict() diagnosis['kurtosis'] = kurtosis.to_dict() # 5. 生成文本报告 report_lines = [] report_lines.append("="*50) report_lines.append("数据诊断报告") report_lines.append("="*50) report_lines.append(f"数据形状: {diagnosis['shape']}") report_lines.append(f"列数据类型:\n{pd.Series(diagnosis['dtypes']).to_string()}") report_lines.append("\n--- 缺失值分析 ---") for col, percent in diagnosis.get('missing_percent', {}).items(): report_lines.append(f" {col}: {diagnosis['missing_count'][col]} 条缺失, 占比 {percent:.2f}%") report_lines.append("\n--- 唯一值分析 ---") for col, count in diagnosis['unique_values'].items(): report_lines.append(f" {col}: {count} 个唯一值") if 'numeric_describe' in diagnosis: report_lines.append("\n--- 数值列统计摘要 (部分) ---") # 示例展示前3列 for col in list(numeric_cols)[:3]: stats = diagnosis['numeric_describe'][col] report_lines.append(f" {col}: 均值={stats['mean']:.2f}, 标准差={stats['std']:.2f}, 范围=[{stats['min']:.2f}, {stats['max']:.2f}]") report = "\n".join(report_lines) print(report) if save_report: with open('data_diagnosis_report.txt', 'w', encoding='utf-8') as f: f.write(report) print("诊断报告已保存至 'data_diagnosis_report.txt'") return diagnosis

实操心得：诊断报告一定要看缺失值比例。如果某列缺失超过30%，在建模中就需要慎重考虑是否直接删除该特征，因为填充可能会引入巨大噪声。对于唯一值数量等于或接近样本数的列（如ID），通常直接删除，除非有特殊用途（如时间序列ID）。偏度（Skewness）绝对值大于1通常认为分布有偏，可能需要做对数变换等处理，以满足后续线性模型的假设。

2.2 规划阶段：制定清洗策略

拿到诊断报告后，不要立刻写代码。拿出一张纸或打开注释，针对每一类问题制定策略。这是区分“脚本小子”和“建模者”的关键一步。

1. 缺失值处理策略：

   • 删除：整行缺失率极高（如>50%）或对模型无关紧要的列。
   • 填充：
     • 数值型：中位数（抗异常值）、均值、众数、插值法（时间序列）、预测模型填充（如KNN）。
     • 分类型：众数、自定义“未知”类别。
   • 不处理：某些树模型（如XGBoost, LightGBM）可以原生处理缺失值，将其视为一种特殊的分支。

2. 异常值处理策略：

   • 识别方法：箱线图（IQR法则）、Z-score（标准差法）、基于模型（如Isolation Forest）。
   • 处理方式：删除、截断（Winsorization）、视为缺失值后填充。

3. 数据类型与格式统一：

   • 日期时间列解析。
   • 分类变量的编码（Label Encoding, One-Hot Encoding）。
   • 字符串字段的规整（去除首尾空格、大小写统一）。

4. 重复值处理：根据业务逻辑判断是删除还是保留。

注意：所有策略都需要记录原因。例如：“age列使用中位数填充，因为其分布有偏，均值易受极端年龄影响。”这在你后期回溯模型结果或撰写论文时至关重要。

3. 核心清洗模块代码实现与解析

有了策略，我们就可以用代码将其实现。我将清洗过程模块化，每个函数负责一个特定任务，并包含详细的参数说明和逻辑注释。

3.1 缺失值处理模块

这是最核心的模块之一。我提供了多种填充方法，并强调了建模中的注意事项。

def handle_missing_values(df, strategy='median', numeric_cols=None, categorical_cols=None, custom_fill_dict=None, drop_threshold=0.7): """ 智能处理缺失值。 参数: df: 待处理的DataFrame。 strategy: 填充策略，可选 'median', 'mean', 'mode', 'knn', 'interpolate'。 numeric_cols: 指定需要处理的数值列列表，为None则自动识别。 categorical_cols: 指定需要处理的分类列列表，为None则自动识别。 custom_fill_dict: 字典，指定特定列的填充值，例如 {'age': 25, 'city': 'Unknown'}。 drop_threshold: 某列缺失率超过此阈值，则直接删除该列。默认0.7（70%）。 返回: 处理后的DataFrame和删除的列列表。 """ df_clean = df.copy() dropped_columns = [] # 1. 自动识别列类型（如果未指定） if numeric_cols is None: numeric_cols = df_clean.select_dtypes(include=[np.number]).columns.tolist() if categorical_cols is None: # 通常认为 object, category, bool 是分类/文本 cat_dtypes = ['object', 'category', 'bool'] categorical_cols = df_clean.select_dtypes(include=cat_dtypes).columns.tolist() all_cols = numeric_cols + categorical_cols all_cols = [col for col in all_cols if col in df_clean.columns] # 去重并确保存在 # 2. 高缺失率列删除 missing_ratio = df_clean[all_cols].isnull().sum() / len(df_clean) cols_to_drop = missing_ratio[missing_ratio > drop_threshold].index.tolist() if cols_to_drop: print(f"[信息] 删除缺失率 > {drop_threshold*100:.0f}% 的列: {cols_to_drop}") df_clean = df_clean.drop(columns=cols_to_drop) dropped_columns.extend(cols_to_drop) # 更新待处理的列列表 numeric_cols = [c for c in numeric_cols if c not in cols_to_drop] categorical_cols = [c for c in categorical_cols if c not in cols_to_drop] all_cols = [c for c in all_cols if c not in cols_to_drop] # 3. 应用自定义填充（优先级最高） if custom_fill_dict: for col, fill_value in custom_fill_dict.items(): if col in df_clean.columns: df_clean[col] = df_clean[col].fillna(fill_value) print(f"[信息] 列 '{col}' 使用自定义值 '{fill_value}' 填充。") # 4. 处理数值型缺失 num_missing_cols = [col for col in numeric_cols if col in df_clean.columns and df_clean[col].isnull().any()] if num_missing_cols: if strategy == 'median': fill_values = df_clean[num_missing_cols].median() elif strategy == 'mean': fill_values = df_clean[num_missing_cols].mean() elif strategy == 'mode': fill_values = df_clean[num_missing_cols].mode().iloc[0] # 取第一个众数 elif strategy == 'interpolate': # 插值法，对时间序列数据友好 for col in num_missing_cols: df_clean[col] = df_clean[col].interpolate(method='linear', limit_direction='both') print(f"[信息] 对数值列 {num_missing_cols} 使用线性插值填充。") fill_values = None elif strategy == 'knn': # KNN填充，需要安装 scikit-learn try: from sklearn.impute import KNNImputer imputer = KNNImputer(n_neighbors=5) df_clean[num_missing_cols] = imputer.fit_transform(df_clean[num_missing_cols]) print(f"[信息] 对数值列 {num_missing_cols} 使用KNN（n=5）填充。") fill_values = None except ImportError: print("[警告] 未找到scikit-learn，KNN填充失败，回退到中位数填充。") fill_values = df_clean[num_missing_cols].median() strategy = 'median' else: print(f"[警告] 不支持的策略 '{strategy}'，对数值列使用中位数填充。") fill_values = df_clean[num_missing_cols].median() strategy = 'median' if fill_values is not None: df_clean[num_missing_cols] = df_clean[num_missing_cols].fillna(fill_values) print(f"[信息] 数值列使用'{strategy}'填充: {fill_values.to_dict()}") # 5. 处理分类型缺失（通常用众数或‘Unknown’） cat_missing_cols = [col for col in categorical_cols if col in df_clean.columns and df_clean[col].isnull().any()] if cat_missing_cols: for col in cat_missing_cols: # 检查是否已在自定义填充中处理过 if not (custom_fill_dict and col in custom_fill_dict): mode_val = df_clean[col].mode() fill_val = mode_val.iloc[0] if not mode_val.empty else 'Unknown' df_clean[col] = df_clean[col].fillna(fill_val) print(f"[信息] 分类列 '{col}' 使用众数 '{fill_val}' 填充。") return df_clean, dropped_columns

关键点解析：

• 策略选择：strategy='median'是默认且最稳健的选择，因为中位数对异常值不敏感。mean在数据对称时可用。knn更智能但计算成本高，适合特征间相关性强的场景。
• 高缺失率删除：drop_threshold参数是关键。在建模中，如果一个特征大部分数据都缺失，其信息量极低，强行填充反而会引入大量噪声，直接删除是更明智的选择。这个阈值可以根据数据集大小和领域知识调整。
• 自定义填充优先：custom_fill_dict参数非常实用。比如，你知道数据中age的缺失可能是因为新生儿，可以指定填充为0。这体现了业务逻辑融入清洗过程。

3.2 异常值检测与处理模块

异常值不一定是错误，但会严重影响许多模型（特别是线性回归、SVM、K-Means）的性能。

def detect_and_treat_outliers(df, columns=None, method='iqr', treat_method='cap', multiplier=1.5, z_threshold=3): """ 检测并处理异常值。 参数: df: 待处理的DataFrame。 columns: 需要处理的数值列列表，为None则处理所有数值列。 method: 检测方法，'iqr'（箱线图）或 'zscore'。 treat_method: 处理方法，'remove'（删除行），'cap'（截断），'nullify'（设为NaN）。 multiplier: IQR方法的乘数，通常1.5（温和）或3（严格）。 z_threshold: Z-score方法的阈值，通常取2或3。 返回: 处理后的DataFrame，以及被标记为异常值的索引字典。 """ df_clean = df.copy() if columns is None: columns = df_clean.select_dtypes(include=[np.number]).columns.tolist() outlier_indices_dict = {} for col in columns: if col not in df_clean.columns: continue series = df_clean[col].dropna() if len(series) < 2: # 数据太少无法计算 continue outliers_mask = pd.Series(False, index=df_clean.index) if method == 'iqr': Q1 = series.quantile(0.25) Q3 = series.quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - multiplier * IQR upper_bound = Q3 + multiplier * IQR # 标记异常值 outliers_mask = (df_clean[col] < lower_bound) | (df_clean[col] > upper_bound) elif method == 'zscore': from scipy import stats # 计算Z-score，忽略NaN z_scores = np.abs(stats.zscore(series)) # 将Z-score映射回原DataFrame的索引 z_series = pd.Series(z_scores, index=series.index) outliers_mask = z_series > z_threshold # 对齐索引 outliers_mask = outliers_mask.reindex(df_clean.index, fill_value=False) else: print(f"[警告] 未知的异常值检测方法: {method}，跳过列 {col}。") continue outlier_indices = df_clean.index[outliers_mask].tolist() outlier_indices_dict[col] = outlier_indices if outlier_indices: print(f"[信息] 列 '{col}' 检测到 {len(outlier_indices)} 个异常值。") if treat_method == 'remove': # 注意：这里先记录，最后统一删除，避免循环中修改索引 pass # 将在循环外统一处理 elif treat_method == 'cap': # Winsorization: 将异常值截断到边界 df_clean.loc[df_clean[col] < lower_bound, col] = lower_bound if method=='iqr' else series.mean() - z_threshold * series.std() df_clean.loc[df_clean[col] > upper_bound, col] = upper_bound if method=='iqr' else series.mean() + z_threshold * series.std() print(f" 已执行截断处理。") elif treat_method == 'nullify': df_clean.loc[outliers_mask, col] = np.nan print(f" 已将异常值设为NaN。") else: print(f" 未执行处理，仅检测。") # 统一执行删除操作（删除在任何列中被标记为异常值的行） if treat_method == 'remove' and outlier_indices_dict: all_outlier_indices = set() for indices in outlier_indices_dict.values(): all_outlier_indices.update(indices) rows_to_drop = list(all_outlier_indices) if rows_to_drop: print(f"[信息] 删除包含异常值的行，共 {len(rows_to_drop)} 行。") df_clean = df_clean.drop(index=rows_to_drop) return df_clean, outlier_indices_dict

实操心得：异常值处理要谨慎。method='iqr'更通用，multiplier=1.5是标准，对于小数据集或想保留更多信息时可以用3。treat_method='cap'（截断）是我在建模中最常用的方法，因为它保留了样本量，只是限制了极端值的影响，这对后续的回归类模型尤其友好。'remove'在异常值比例很低（如<1%）且样本量充足时可以考虑。处理完后，一定要重新检查数据的分布情况。

3.3 数据类型转换与编码模块

模型只能处理数值。因此，日期、分类文本都必须转换成数字。

def standardize_data_types(df, datetime_cols=None, categorical_cols=None, encoding='onehot', drop_first=True): """ 标准化数据类型：解析日期时间，编码分类变量。 参数: df: 待处理的DataFrame。 datetime_cols: 需要解析为日期时间的列名列表。支持自动推断。 categorical_cols: 需要编码的分类列列表。为None则自动识别object类型。 encoding: 编码方式，'onehot'（独热编码）或 'label'（标签编码）。 drop_first: 独热编码时是否丢弃第一列以避免共线性（对于线性模型重要）。 返回: 处理后的DataFrame和编码器字典（用于后续预测数据的转换）。 """ df_clean = df.copy() encoders = {} # 保存编码器，用于后续数据 # 1. 处理日期时间列 if datetime_cols is not None: for col in datetime_cols: if col in df_clean.columns: try: df_clean[col] = pd.to_datetime(df_clean[col], errors='coerce') print(f"[信息] 已将列 '{col}' 转换为日期时间类型。") # 可扩展：提取年、月、日、星期等特征，这对时间序列建模非常有用 # df_clean[f'{col}_year'] = df_clean[col].dt.year # df_clean[f'{col}_month'] = df_clean[col].dt.month # ... except Exception as e: print(f"[警告] 列 '{col}' 日期时间转换失败: {e}") # 2. 处理分类变量编码 if categorical_cols is None: # 自动识别：通常是非数值型且唯一值数量适中的列 potential_cat_cols = df_clean.select_dtypes(include=['object', 'category']).columns # 排除唯一值过多（如ID）或过少（如全一样）的列 categorical_cols = [] for col in potential_cat_cols: n_unique = df_clean[col].nunique() if 1 < n_unique <= 50: # 阈值可根据数据调整 categorical_cols.append(col) else: print(f"[信息] 列 '{col}' 有 {n_unique} 个唯一值，暂不自动进行编码处理。") for col in categorical_cols: if col not in df_clean.columns: continue if encoding == 'label': from sklearn.preprocessing import LabelEncoder le = LabelEncoder() # 处理可能存在的NaN，先填充一个特殊值如‘Unknown’ col_data = df_clean[col].fillna('Unknown').astype(str) df_clean[col] = le.fit_transform(col_data) encoders[col] = le print(f"[信息] 列 '{col}' 已进行标签编码。") elif encoding == 'onehot': # 使用pandas的get_dummies更直观 dummies = pd.get_dummies(df_clean[col], prefix=col, prefix_sep='_', dummy_na=False, drop_first=drop_first) # 删除原列，添加新列 df_clean = df_clean.drop(columns=[col]) df_clean = pd.concat([df_clean, dummies], axis=1) # 记录编码信息（这里简化，实际可保存categories_） encoders[col] = {'type': 'onehot', 'drop_first': drop_first, 'categories': df_clean[col].unique() if col in df_clean.columns else None} print(f"[信息] 列 '{col}' 已进行独热编码，生成 {dummies.shape[1]} 个新特征。") else: print(f"[警告] 未知的编码方式: {encoding}，跳过列 {col}。") return df_clean, encoders

关键点解析：

• 独热编码 vs 标签编码：encoding='onehot'是更安全的选择，它不会引入错误的序关系（比如把‘高’、‘中’、‘低’编码成2,1,0，模型会误以为‘高’比‘中’大2倍）。缺点是如果类别很多，会产生大量稀疏特征，增加维度。label编码适用于树模型，且类别具有内在顺序（如学历‘小学’，‘中学’，‘大学’）。
• drop_first=True：在独热编码时，对于有k个类别的特征，只需要k-1个哑变量即可完全表示，丢弃的一列可以作为“基准”。这有助于避免线性回归中的多重共线性问题，是一个重要的建模技巧。
• 日期特征：代码中注释了提取年、月、日等衍生特征的部分。在实际建模中，这往往是提升模型效果的关键步骤，因为模型很难直接从datetime对象中学习周期性模式。

3.4 数据标准化与规范化模块

很多模型（如KNN、SVM、神经网络、主成分分析）都基于距离计算，因此不同特征的量纲（单位）差异会严重影响结果。标准化/规范化就是解决这个问题的。

def scale_features(df, columns=None, method='standard', scaler_dict=None): """ 对数值特征进行标准化或规范化。 参数: df: 待处理的DataFrame。 columns: 需要缩放的列列表，为None则缩放所有数值列。 method: 缩放方法，'standard'（标准化，Z-score），'minmax'（归一化到[0,1]），'robust'（抗异常值标准化）。 scaler_dict: 可传入已拟合好的scaler字典，用于在测试集上做相同变换。 返回: 缩放后的DataFrame，以及拟合好的scaler字典（用于后续数据转换）。 """ from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler df_scaled = df.copy() if columns is None: columns = df_scaled.select_dtypes(include=[np.number]).columns.tolist() columns = [col for col in columns if col in df_scaled.columns] if not columns: print("[警告] 未找到需要缩放的数值列。") return df_scaled, {} if scaler_dict is None: scaler_dict = {} for col in columns: # 确保是二维数据，sklearn scaler需要 col_data = df_scaled[[col]].values if method == 'standard': scaler = StandardScaler() elif method == 'minmax': scaler = MinMaxScaler() elif method == 'robust': scaler = RobustScaler() # 使用中位数和IQR，对异常值不敏感 else: print(f"[警告] 未知的缩放方法: {method}，对列 {col} 使用标准化。") scaler = StandardScaler() # 如果提供了已拟合的scaler，则直接使用，否则拟合新的 if col in scaler_dict: fitted_scaler = scaler_dict[col] else: fitted_scaler = scaler.fit(col_data) scaler_dict[col] = fitted_scaler df_scaled[col] = fitted_scaler.transform(col_data).flatten() print(f"[信息] 已使用 '{method}' 方法对 {len(columns)} 个数值列进行缩放。") return df_scaled, scaler_dict

方法选择指南：

• method='standard'（标准化）：最常用。将数据转换为均值为0，标准差为1的分布。适用于数据大致符合正态分布的情况。
• method='minmax'（归一化）：将数据缩放到[0, 1]区间。对异常值非常敏感，如果数据中存在极端值，整个缩放区间会被扭曲。
• method='robust'：使用中位数和四分位距（IQR）进行缩放，对异常值不敏感。当你的数据中有异常值且不想（或不能）删除/截断时，这是最佳选择。

重要提示：scaler_dict参数是关键。在建模中，你必须用训练集拟合scaler，然后用这个scaler去转换验证集和测试集。绝对不能用测试集的数据去重新拟合scaler，这会引入数据泄露（Data Leakage），导致模型评估结果过于乐观。这个函数的设计考虑到了这一点。

4. 完整清洗管道与实战示例

将上述模块组合起来，形成一个完整的、可配置的数据清洗管道（Pipeline）。这是项目代码的精华，让你一键完成从原始数据到建模可用数据的转换。

class DataCleaningPipeline: """ 数据清洗管道，按顺序执行清洗步骤。 """ def __init__(self): self.steps = [] self.diagnosis_report = None self.encoders = {} self.scalers = {} self.dropped_cols = [] def add_step(self, name, function, **kwargs): """添加一个清洗步骤。""" self.steps.append({'name': name, 'func': function, 'kwargs': kwargs}) def run(self, df, verbose=True): """执行清洗管道。""" df_clean = df.copy() if verbose: print("="*60) print("开始执行数据清洗管道") print("="*60) for i, step in enumerate(self.steps): step_name = step['name'] step_func = step['func'] step_kwargs = step['kwargs'] if verbose: print(f"\n步骤 {i+1}: {step_name}") print(f" 参数: {step_kwargs}") # 特殊处理需要保存中间结果的步骤 if step_func.__name__ == 'handle_missing_values': df_clean, dropped = step_func(df_clean, **step_kwargs) self.dropped_cols.extend(dropped) elif step_func.__name__ == 'standardize_data_types': df_clean, encoders = step_func(df_clean, **step_kwargs) self.encoders.update(encoders) elif step_func.__name__ == 'scale_features': df_clean, scalers = step_func(df_clean, **step_kwargs) self.scalers.update(scalers) else: df_clean = step_func(df_clean, **step_kwargs) if verbose: print(f" 完成。数据形状: {df_clean.shape}") if verbose: print("\n" + "="*60) print("数据清洗管道执行完毕！") print(f"原始数据形状: {df.shape}") print(f"清洗后数据形状: {df_clean.shape}") if self.dropped_cols: print(f"删除的列: {self.dropped_cols}") print("="*60) return df_clean def transform_new_data(self, df_new): """ 使用已拟合的管道（编码器、缩放器）转换新数据（如测试集）。 注意：新数据应进行相同的缺失值处理、异常值处理等，但使用训练集学到的参数。 """ df_transformed = df_new.copy() # 这里需要根据实际保存的中间状态（如填充值、异常值边界）来转换新数据 # 这是一个简化示例，实际应用中需要更严谨地保存和加载所有转换状态 print("[信息] 新数据转换功能需要根据具体保存的管道状态实现。") return df_transformed

实战示例：清洗一个模拟的客户数据集

# 1. 创建模拟脏数据 np.random.seed(42) n_samples = 1000 data = { 'customer_id': range(n_samples), 'age': np.random.randint(18, 70, n_samples).astype(float), 'income': np.random.lognormal(mean=10, sigma=0.5, size=n_samples), 'city': np.random.choice(['北京', '上海', '广州', '深圳', '杭州', np.nan], n_samples, p=[0.2,0.2,0.15,0.15,0.1,0.2]), 'member_level': np.random.choice(['青铜', '白银', '黄金', '钻石', np.nan], n_samples, p=[0.4,0.3,0.2,0.05,0.05]), 'last_purchase_amount': np.random.exponential(scale=500, size=n_samples), 'days_since_last_visit': np.random.randint(0, 365, n_samples), } # 人为制造一些缺失和异常 df = pd.DataFrame(data) df.loc[np.random.choice(df.index, size=50), 'age'] = np.nan df.loc[np.random.choice(df.index, size=30), 'income'] = np.nan # 制造异常值 df.loc[0, 'income'] = 1000000 # 一个异常高的收入 df.loc[1, 'last_purchase_amount'] = -100 # 一个负的消费金额（异常） df.loc[2, 'days_since_last_visit'] = 1000 # 一个异常大的天数 print("原始数据预览:") print(df.head()) print(f"\n原始数据形状: {df.shape}") print("\n缺失值统计:") print(df.isnull().sum()) # 2. 实例化管道并配置步骤 pipeline = DataCleaningPipeline() # 步骤1: 数据诊断 pipeline.add_step('数据诊断', data_diagnosis, save_report=True) # 步骤2: 处理缺失值 (对收入使用中位数，城市使用众数，会员等级自定义填充) custom_fill = {'member_level': '青铜'} # 假设缺失的会员等级默认为青铜 pipeline.add_step('处理缺失值', handle_missing_values, strategy='median', custom_fill_dict=custom_fill, drop_threshold=0.5) # 步骤3: 处理异常值 (对收入、消费金额使用IQR法截断) pipeline.add_step('处理异常值', detect_and_treat_outliers, columns=['income', 'last_purchase_amount', 'days_since_last_visit'], method='iqr', treat_method='cap', multiplier=1.5) # 步骤4: 数据类型与编码 (城市、会员等级进行独热编码) pipeline.add_step('分类变量编码', standardize_data_types, categorical_cols=['city', 'member_level'], encoding='onehot', drop_first=True) # 步骤5: 特征缩放 (对数值型特征进行标准化) pipeline.add_step('特征标准化', scale_features, method='standard') # 3. 运行管道 df_cleaned = pipeline.run(df, verbose=True) print("\n清洗后数据预览 (前5行):") print(df_cleaned.head()) print(f"\n清洗后数据形状: {df_cleaned.shape}") print(f"\n新生成的列: {[col for col in df_cleaned.columns if col not in df.columns]}")

运行这段代码，你将看到一个完整的、自动化的清洗过程在终端打印出来，从原始凌乱的数据，一步步得到干净、规整、可直接用于建模的数值型数据表。

5. 常见问题、避坑指南与进阶技巧

在实际使用中，你肯定会遇到各种问题。下面是我总结的“血泪教训”和进阶建议。

5.1 缺失值处理中的陷阱

• 陷阱1：盲目使用均值填充。如果数据分布有偏（比如收入），均值会被少数极高值拉高，用均值填充会系统性高估缺失值。务必先看分布，有偏分布用中位数。
• 陷阱2：用全局统计量填充时间序列数据。时间序列数据（如每日销售额）的缺失值，用前后时间的值插值（method='interpolate'）远比用整个序列的均值/中位数合理。
• 陷阱3：忽略缺失机制。数据为什么缺失？如果是随机缺失（MCAR），上述方法尚可。如果是非随机缺失（如高收入人群不愿透露收入），填充会引入偏差。此时，考虑使用多重插补（Multiple Imputation）或将“是否缺失”作为一个新的布尔特征加入模型。

5.2 异常值处理的误区

• 误区：把异常值一律当错误删除。在欺诈检测、故障诊断等场景，异常值本身就是我们要找的“信号”。清洗前要明确建模目标。如果是预测普通客户行为，可以处理异常值；如果是识别欺诈，则需要重点研究异常值。
• 技巧：可视化确认。处理前后，一定要用箱线图或散点图对比查看。seaborn.boxplot或plt.scatter是你的好朋友。
• 进阶方法：模型法检测。对于高维数据，可以使用sklearn.ensemble.IsolationForest或sklearn.neighbors.LocalOutlierFactor来检测异常点，这些方法能考虑特征间的相互关系。

5.3 分类编码的维度灾难与稀疏性

• 问题：一个类别有上百种取值（如“商品品牌”）。独热编码会产生上百个新特征，导致数据极度稀疏，增加计算负担和过拟合风险。
• 解决方案：
  1. 频率编码：用该类别的出现频率（或对数频率）来代替独热编码。df['city_freq'] = df.groupby('city')['city'].transform('count') / len(df)。
  2. 目标编码：用该类别下目标变量的均值（如果是回归）或正例比例（如果是分类）来编码。但要极其小心数据泄露，必须只在训练集上计算编码，再应用到验证/测试集，通常需要配合交叉验证进行。
  3. 嵌入层：对于深度学习模型，可以使用嵌入层（Embedding Layer）将高维类别映射到低维稠密向量。

5.4 管道化与可复现性

• 核心原则：清洗代码必须封装成函数或类，并且所有参数（如填充值、异常值边界、编码映射、缩放器）都要从训练集中学习并保存下来。
• 为什么：当你用训练集训练好模型后，拿到新的预测数据（测试集或生产数据）时，必须用完全相同的转换去处理它。你不能用新数据重新计算中位数来填充，因为那和模型训练时看到的数据分布不一致了。
• 实现：这就是上面DataCleaningPipeline类中encoders和scalers字典的作用，以及transform_new_data方法的意图。在实际项目中，你需要用pickle或joblib库将这些转换器（fitted scalers,fitted encoders）和模型一起保存。

5.5 效率与大数据处理

• 当数据量很大（GB级别）时，Pandas可能内存不足。
• 解决方案：
  • 分块处理：使用pandas.read_csv(chunksize=50000)分批读入和处理。
  • 使用Dask或Modin：这些库提供了类似Pandas的API，但可以并行化和分布式处理。
  • 使用数据库：对于超大规模数据，在SQL数据库中进行初步的过滤、聚合和类型转换，再将结果集导入Pandas进行精细清洗。

数据清洗是数学建模中既繁琐又至关重要的一步，它没有唯一的正确答案，但有一套经过实践检验的最佳实践和需要规避的陷阱。这套代码是我多年经验的结晶，它提供的不是死板的规则，而是一个灵活、可扩展的框架。希望你能理解每个步骤背后的“为什么”，并根据自己遇到的具体数据和问题，调整策略和参数。记住，清洗的最终目的是让数据更好地“讲述故事”，让模型更准确地捕捉规律。磨刀不误砍柴工，在数据清洗上多花的一份心思，最终都会体现在你模型性能的提升上。