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

3、数据清洗自动化：缺失值处理策略、异常值检测与处理、重复数据删除、自动化清洗Pipeline设计

数据清洗这事儿，说白了就是跟脏数据打架。我做了这么多年特征工程，最深的体会就是：数据质量决定了模型的天花板。你算法再牛，喂进去一堆垃圾，出来的还是垃圾。

这一章，咱们就聊聊怎么把数据清洗这件事自动化。别小看它，一个成熟的自动化清洗Pipeline，能帮你省下至少60%的预处理时间。

3.1 缺失值处理策略

缺失值是最常见的问题。我见过最夸张的项目，某个特征缺失率高达80%。怎么办？直接删掉？太浪费了。得看情况。

3.1.1 缺失值分类

先搞清楚缺失的类型，这决定了你的处理策略。

缺失类型	含义	典型场景	处理建议
MCAR（完全随机缺失）	缺失与数据本身无关	传感器偶尔故障	可直接删除或简单填充
MAR（随机缺失）	缺失与其他特征有关	高收入人群不愿填收入	用其他特征建模预测
MNAR（非随机缺失）	缺失与缺失值本身有关	温度计超过量程不显示	需领域知识，谨慎处理

3.1.2 自动化填充策略

我个人习惯把填充策略分成三档：

简单填充：均值、中位数、众数。速度快，适合MCAR场景。
模型填充：用KNN、随机森林预测缺失值。精度高，但慢。
插值填充：时间序列数据用线性插值或样条插值。

核心原则：缺失率 < 5% 用简单填充；5%-20% 用模型填充；> 20% 考虑是否保留该特征。

来看一段自动化填充的代码：

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer, KNNImputer

def auto_impute(df, strategy='auto'):
    """自动化缺失值填充"""
    df_clean = df.copy()
    
    for col in df_clean.columns:
        missing_rate = df_clean[col].isnull().mean()
        
        if missing_rate == 0:
            continue
        
        if strategy == 'auto':
            if missing_rate < 0.05:
                # 低缺失率，用中位数
                if df_clean[col].dtype in ['int64', 'float64']:
                    imputer = SimpleImputer(strategy='median')
                else:
                    imputer = SimpleImputer(strategy='most_frequent')
            elif missing_rate < 0.2:
                # 中等缺失率，用KNN
                imputer = KNNImputer(n_neighbors=5)
            else:
                # 高缺失率，标记后填充
                df_clean[col + '_missing'] = df_clean[col].isnull().astype(int)
                imputer = SimpleImputer(strategy='median')
        
        df_clean[col] = imputer.fit_transform(df_clean[[col]])
    
    return df_clean

我的经验：千万别一股脑全用均值填充。我遇到过连续特征用均值填充后，模型直接崩了——因为那个特征分布是双峰的，均值落在谷底，反而引入了错误信息。

3.2 异常值检测与处理

异常值比缺失值更隐蔽。你想想看，一个数值看起来正常，但实际上是传感器漂移产生的。怎么发现？

3.2.1 常用检测方法

我一般按数据分布来选方法：

Z-Score：适合正态分布。|Z| > 3 视为异常。
IQR（四分位距）：适合偏态分布。Q1 - 1.5*IQR 到 Q3 + 1.5*IQR 之外算异常。
孤立森林：适合高维数据。不需要假设分布。
DBSCAN：适合密度不均匀的数据。

举个例子，用IQR检测异常：

def detect_outliers_iqr(df, column):
    """用IQR检测异常值"""
    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.index.tolist(), lower_bound, upper_bound

3.2.2 异常值处理策略

检测出来之后怎么处理？我总结了三种方案：

直接删除：异常点很少（< 1%）且确认是噪声。
截断处理：用边界值替换。比如把超出3σ的值拉到3σ。
视为缺失：标记为NaN，走缺失值处理流程。

避坑指南：我曾经在一个金融风控项目里，直接用Z-Score删掉了所有异常值。结果模型上线后，把大额交易全判成了欺诈——因为训练集里根本没有大额样本。记住：异常不一定是错误，可能是你还没理解的数据模式。

3.3 重复数据删除

重复数据看起来简单，但坑也不少。完全重复好办，drop_duplicates() 一行搞定。麻烦的是近似重复。

比如两条用户记录，姓名差一个字，地址一样。这算不算重复？

3.3.1 近似重复检测

我常用的方法：

编辑距离：适合短文本，比如姓名、地址。
SimHash：适合长文本，比如评论、描述。
基于聚类的去重：先聚类，再在簇内找重复。

from fuzzywuzzy import fuzz

def find_duplicate_pairs(df, col, threshold=90):
    """用模糊匹配找近似重复"""
    duplicates = []
    for i in range(len(df)):
        for j in range(i+1, len(df)):
            score = fuzz.ratio(df.iloc[i][col], df.iloc[j][col])
            if score >= threshold:
                duplicates.append((i, j, score))
    return duplicates

小技巧：处理近似重复时，别急着删。先看看重复的原因。我遇到过因为系统编码问题导致的重复，修好源头比事后清洗更有效。

3.4 自动化清洗Pipeline设计

好了，前面都是单个模块。现在把它们串起来，做成一个自动化的Pipeline。

3.4.1 Pipeline架构

这是我常用的架构：

3.4.2 实现一个完整的Pipeline

来看看代码实现：

class AutoCleaningPipeline:
    def __init__(self, config=None):
        self.config = config or {
            'missing_threshold': 0.8,  # 缺失率超过80%直接删除
            'outlier_method': 'iqr',
            'duplicate_threshold': 90
        }
        self.logs = []
    
    def fit_transform(self, df):
        """执行自动化清洗"""
        df_clean = df.copy()
        
        # 1. 删除高缺失率特征
        high_missing = [col for col in df_clean.columns 
                       if df_clean[col].isnull().mean() > self.config['missing_threshold']]
        df_clean.drop(columns=high_missing, inplace=True)
        self.logs.append(f"删除高缺失率特征: {high_missing}")
        
        # 2. 缺失值填充
        df_clean = auto_impute(df_clean)
        self.logs.append("缺失值填充完成")
        
        # 3. 异常值处理
        for col in df_clean.select_dtypes(include=[np.number]).columns:
            outliers, lb, ub = detect_outliers_iqr(df_clean, col)
            if len(outliers) > 0:
                # 截断处理
                df_clean[col] = df_clean[col].clip(lb, ub)
                self.logs.append(f"{col}: 处理 {len(outliers)} 个异常值")
        
        # 4. 删除完全重复
        before = len(df_clean)
        df_clean.drop_duplicates(inplace=True)
        after = len(df_clean)
        self.logs.append(f"删除完全重复: {before - after} 条")
        
        return df_clean
    
    def get_report(self):
        """生成清洗报告"""
        return '\n'.join(self.logs)

关键设计思路：

每个步骤都记录日志，方便回溯
策略可配置，不同场景用不同参数
异常处理不中断流程，用try-except兜底

3.4.3 生产环境注意事项

嗯，这里要注意。本地跑通和线上稳定是两码事。我踩过的坑：

数据漂移：昨天的清洗策略，今天可能就不适用了。建议定期重算统计量。
性能问题：KNN填充在百万级数据上慢得离谱。大数据量用Spark或Dask。
可复现性：每次清洗结果要一致。设置随机种子，记录参数版本。

血的教训：我曾经在生产Pipeline里忘了设置随机种子，结果每次跑出来的训练集都不一样。模型评估指标忽高忽低，排查了整整两天才发现是这个问题。从那以后，我所有Pipeline的第一行代码都是 np.random.seed(42)。

数据清洗自动化，说白了就是把经验固化成代码。你踩过的坑越多，Pipeline就越健壮。别指望一蹴而就，迭代优化才是常态。