📚 课程目录第3章数据清洗

3. 数据清洗：缺失值处理、异常值检测、数据去重

数据清洗这步，说白了就是给原始数据「洗澡」。脏数据进模型，结果一定脏。我见过太多团队在特征工程上花了大把时间，结果因为缺失值没处理好，模型上线就崩。嗯，咱们今天就把这块彻底讲透。

  核心观点：数据清洗不是一次性工作，而是贯穿整个特征生产流水线的持续校验过程。

3.1 缺失值处理：别让「空」毁了你的模型

缺失值是最常见的数据质量问题。我个人习惯，拿到数据第一件事就是跑 df.isnull().sum()，看看哪些列有空洞。

3.1.1 缺失值类型与处理策略

缺失类型	含义	推荐处理方式
完全随机缺失 (MCAR)	缺失与任何变量无关	直接删除或简单填充
随机缺失 (MAR)	缺失与其他变量相关	模型预测填充或多重插补
非随机缺失 (MNAR)	缺失与自身值相关	需业务判断，谨慎处理

为什么区分这个？我在项目中遇到过，用户收入字段缺失往往是因为高收入人群不愿填写，这就是典型的 MNAR。如果直接填充均值，模型会严重低估高收入群体的特征。

3.1.2 常用填充方法

删除法：缺失比例 < 5% 且样本充足时可用。但别盲目删，我见过有人删掉 30% 的数据，模型直接废了。
统计填充：均值/中位数/众数。连续变量用中位数更稳健，离散变量用众数。
前向/后向填充：时间序列数据常用。比如传感器每隔 5 秒上报一次，中间丢了几秒，用前值填充很合理。
模型预测填充：用其他特征训练模型预测缺失值。KNN、随机森林都行，但计算量大。

# 我个人常用的缺失值处理流水线
import pandas as pd
from sklearn.impute import SimpleImputer

# 数值列：用中位数填充
num_imputer = SimpleImputer(strategy='median')
df['age'] = num_imputer.fit_transform(df[['age']])

# 类别列：用众数填充
cat_imputer = SimpleImputer(strategy='most_frequent')
df['gender'] = cat_imputer.fit_transform(df[['gender']])

# 时间序列：前向填充
df['sensor_value'] = df['sensor_value'].ffill()

小技巧：填充后建议加一列「是否缺失」的标识特征。有时缺失本身就是一个信号，比如用户没填手机号可能意味着活跃度低。

3.2 异常值检测：揪出数据里的「害群之马」

异常值不一定是错误，但必须被识别。我记得有一次做用户行为分析，发现某个用户的点击量是平均值的 1000 倍，结果是爬虫在刷数据。如果不处理，模型会学歪。

3.2.1 常用检测方法

方法	适用场景	优点	缺点
Z-Score	近似正态分布	简单快速	对极端值敏感
IQR 四分位距	偏态分布	稳健，不受极端值影响	阈值需人工设定
孤立森林	高维数据	线性时间复杂度	对局部异常不敏感
DBSCAN 聚类	密度不均匀数据	可发现任意形状异常	参数敏感

# IQR 方法实战
def detect_outliers_iqr(df, column):
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
    return outliers

# 使用示例
outliers_df = detect_outliers_iqr(df, 'price')
print(f"发现 {len(outliers_df)} 个异常值")

避坑指南：我曾经直接删掉所有异常值，结果模型在线上遇到真实极端情况时完全失效。正确的做法是：先标记异常值，分析其来源，再决定是修正、删除还是保留。

3.3 数据去重：别让重复样本「刷存在感」

重复数据会导致模型过拟合，还会让评估指标虚高。你想想看，如果训练集里同一个样本出现 10 次，模型就会拼命记住它，泛化能力自然差。

3.3.1 去重策略

完全去重：所有字段完全一致。用 df.drop_duplicates() 搞定。
部分去重：关键字段一致即可。比如用户 ID 相同但注册时间不同，保留最新的一条。
模糊去重：文本相似度去重。比如「北京市朝阳区」和「北京朝阳区」其实是同一个地址。

# 完全去重
df_clean = df.drop_duplicates()

# 按用户 ID 去重，保留最新记录
df_dedup = df.sort_values('timestamp').drop_duplicates(subset='user_id', keep='last')

# 模糊去重（使用 recordlinkage 库）
import recordlinkage
indexer = recordlinkage.Index()
indexer.block('city')
pairs = indexer.index(df)

3.3.2 去重注意事项

嗯，这里要注意：去重前一定要确认「重复」的定义。我遇到过业务方说「用户 ID 重复就是重复」，结果发现同一个用户在不同设备上注册了多个账号，这些其实是独立用户。所以去重逻辑必须和业务对齐。

经验之谈：在离线流水线中，建议在数据接入层就做一次轻量去重，在特征计算层再做一次深度去重。两层防护，避免重复数据污染下游。

3.4 数据清洗流水线架构

下面这张图是我在实际项目中沉淀下来的清洗流水线，你可以直接复用。

3.5 实战：构建可复用的清洗模块

在实际的离线流水线中，我不会每次都手写清洗逻辑。我会封装成可配置的模块，通过 YAML 文件控制参数。

# config/cleaning_config.yaml
missing_value:
  strategy: "median"  # mean, median, mode, constant
  columns: ["age", "income", "score"]
  
outlier_detection:
  method: "iqr"  # zscore, iqr, isolation_forest
  threshold: 1.5
  action: "cap"  # remove, cap, mark
  
deduplication:
  subset: ["user_id", "event_id"]
  keep: "last"
  sort_by: "timestamp"

# 清洗模块核心代码
class DataCleaner:
    def __init__(self, config_path):
        self.config = self._load_config(config_path)
    
    def clean(self, df):
        df = self._handle_missing(df)
        df = self._detect_outliers(df)
        df = self._deduplicate(df)
        return df
    
    def _handle_missing(self, df):
        # 根据配置处理缺失值
        pass
    
    def _detect_outliers(self, df):
        # 根据配置检测异常值
        pass
    
    def _deduplicate(self, df):
        # 根据配置去重
        pass

  核心原则：数据清洗模块必须可复现、可配置、可监控。每次清洗都要记录日志，包括处理了多少缺失值、删除了多少异常值、去重了多少条记录。这样出了问题才能追溯。

好了，数据清洗这块就讲到这里。记住，清洗不是一次性的，每次数据更新都要跑一遍。把清洗逻辑固化到流水线里，让机器替你做这些脏活累活。