📚 课程目录第2章数据清洗实战

2. 数据清洗实战

拿到一批量产测试数据，别急着分析。我见过太多人上来就画图、算良率，结果折腾半天发现数据本身就有问题。数据清洗，说白了就是给数据「洗澡」。脏数据洗不干净，后面所有分析都是白搭。

这一章，我带你过一遍量产测试数据清洗的四个核心动作：缺失值处理、异常值检测、重复数据去重、测试项名称标准化。每个动作我都会给出实战代码和避坑经验。

核心原则：清洗操作必须可追溯、可复现。每次修改数据都要留日志，别直接覆盖原始文件。我在项目里吃过这个亏——清洗完发现某个过滤条件设错了，原始数据又没备份，只能重新从产线拉数据，耽误了两天。

数据清洗整体流程

先看一张图，把清洗流程串起来。你按这个顺序走，基本不会漏东西。

缺失值处理

量产数据里，缺失值最常见的原因就几个：测试机台掉电、探针接触不良、某个测试项被跳过。我遇到过最离谱的一次，整批数据里某个电压测试项缺失了60%，后来发现是测试程序里那个测试项的条件判断写错了。

缺失值检测

先用 pandas 快速扫一眼缺失情况：

import pandas as pd
import numpy as np

# 读取量产测试数据
df = pd.read_csv('production_test_data.csv')

# 查看每列的缺失数量和比例
missing_count = df.isnull().sum()
missing_ratio = df.isnull().mean() * 100

# 汇总成表格
missing_df = pd.DataFrame({
    '缺失数量': missing_count,
    '缺失比例(%)': missing_ratio.round(2)
})
print(missing_df[missing_df['缺失数量'] > 0])

缺失值处理策略

处理缺失值，没有万能公式。我一般按这个优先级来：

缺失比例	处理方式	适用场景
< 5%	删除该行	数据量大，少量缺失不影响统计
5% - 30%	填充（均值/中位数/前值）	测试项有物理上下界，填充值合理
> 30%	删除该列	缺失太多，填充会引入偏差
关键测试项	标记为FAIL，单独分析	良率敏感项，不能随便填充

我的习惯：对于电流、电压这类有物理意义的测试项，我倾向用中位数填充。均值容易被极端值拉偏，中位数更稳健。有一次我处理VDDmin数据，均值填充后良率曲线出现异常凸起，换成中位数就正常了。

# 按列缺失比例处理
for col in df.columns:
    if col in ['Lot', 'Wafer', 'Site', 'TestTime']:
        continue  # 关键信息列不处理
    
    miss_ratio = df[col].isnull().mean()
    
    if miss_ratio < 0.05:
        # 缺失少，直接删除
        df.dropna(subset=[col], inplace=True)
    elif miss_ratio < 0.30:
        # 用中位数填充
        median_val = df[col].median()
        df[col].fillna(median_val, inplace=True)
        print(f'{col}: 填充中位数 {median_val:.4f}')
    else:
        # 缺失太多，删除该列
        df.drop(columns=[col], inplace=True)
        print(f'{col}: 缺失率 {miss_ratio:.2%}，删除该列')

异常值检测

异常值在量产数据里太常见了。探针没扎好、测试机台通道故障、甚至操作员不小心碰了线缆，都会产生异常值。我见过一个项目，某颗芯片的IDD电流测出来是正常值的100倍，一看日志——探针扎偏了。

IQR方法（最常用）

IQR方法对量产数据特别友好，因为它不假设数据分布。你想想看，测试数据经常不是正态分布，用3σ方法容易误判。

def detect_outliers_iqr(data, column, factor=1.5):
    """
    使用IQR方法检测异常值
    factor=1.5 为常规阈值，factor=3.0 为严格阈值
    """
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].quantile(0.75)
    IQR = Q3 - Q1
    
    lower_bound = Q1 - factor * IQR
    upper_bound = Q3 + factor * IQR
    
    outliers = data[(data[column] < lower_bound) | 
                    (data[column] > upper_bound)]
    
    return outliers, lower_bound, upper_bound

# 示例：检测VDD电压异常值
outliers, low, high = detect_outliers_iqr(df, 'VDD', factor=1.5)
print(f'检测到 {len(outliers)} 个异常值')
print(f'正常范围: [{low:.4f}, {high:.4f}]')

3σ方法（适合正态分布数据）

如果测试项数据接近正态分布，比如某些经过校准的电压基准，可以用3σ方法：

def detect_outliers_3sigma(data, column):
    mean = data[column].mean()
    std = data[column].std()
    
    lower_bound = mean - 3 * std
    upper_bound = mean + 3 * std
    
    outliers = data[(data[column] < lower_bound) | 
                    (data[column] > upper_bound)]
    return outliers, lower_bound, upper_bound

注意：3σ方法对异常值本身很敏感。如果数据里已经有大量异常值，均值和标准差会被拉偏，导致正常数据被误判为异常。我曾经在一个项目里用3σ方法检测漏电流，结果因为几个极端值把阈值拉得太宽，反而漏掉了真正的异常。后来我改用IQR方法，效果好了很多。

异常值处理原则

先标记，后处理：不要直接删除异常值。先加一列标记，保留原始数据。
区分物理异常和统计异常：有些值虽然统计上异常，但物理上可能合理。比如大功率芯片的电流就是比小功率芯片高很多。
按批次/机台分组检测：不同机台、不同批次的测试条件可能不同，统一阈值会误判。

# 按机台分组检测异常值
def detect_outliers_by_group(data, group_col, value_col):
    results = []
    for group, group_data in data.groupby(group_col):
        outliers, low, high = detect_outliers_iqr(group_data, value_col)
        if len(outliers) > 0:
            results.append({
                'group': group,
                'outlier_count': len(outliers),
                'lower_bound': low,
                'upper_bound': high
            })
    return pd.DataFrame(results)

# 按测试机台检测VDD异常
outlier_summary = detect_outliers_by_group(df, 'TesterID', 'VDD')
print(outlier_summary)

重复数据去重

重复数据在量产数据里很隐蔽。我遇到过的情况：产线操作员误操作导致同一颗芯片测了两次、数据合并时文件重复导入、测试程序循环逻辑bug导致重复记录。

重复数据检测

量产数据里，判断重复的关键字段通常是 Lot + Wafer + Site + TestTime 的组合。同一颗芯片在同一时间不应该有两条记录。

# 检测完全重复的行
duplicate_rows = df[df.duplicated()]
print(f'完全重复的行数: {len(duplicate_rows)}')

# 按关键字段检测重复
key_cols = ['Lot', 'Wafer', 'Site', 'TestTime']
duplicate_keys = df[df.duplicated(subset=key_cols, keep=False)]
print(f'关键字段重复的行数: {len(duplicate_keys)}')

# 查看重复的具体情况
duplicate_summary = duplicate_keys.groupby(key_cols).size().reset_index(name='count')
print(duplicate_summary[duplicate_summary['count'] > 1].head())

去重策略

重复类型	保留规则	说明
完全重复	保留第一条	数据完全一样，保留任意一条即可
关键字段重复，测试值不同	保留最后一次测试	可能是重测，最后一次结果更可靠
关键字段重复，部分字段缺失	保留完整记录	合并两条记录的信息

# 去重：保留最后一次测试结果
df_deduplicated = df.sort_values('TestTime').drop_duplicates(
    subset=['Lot', 'Wafer', 'Site'], 
    keep='last'
)

print(f'去重前数据量: {len(df)}')
print(f'去重后数据量: {len(df_deduplicated)}')
print(f'去除重复数: {len(df) - len(df_deduplicated)}')

避坑指南：我曾经遇到一个项目，去重后发现良率下降了2%。查了半天，发现是同一颗芯片在良率测试和特性测试阶段各测了一次，两次结果都有效。去重时把特性测试的数据删了，导致良率统计偏低。所以去重前一定要搞清楚业务逻辑，别一刀切。

测试项名称标准化

这个坑我踩过太多次了。不同工程师写的测试程序，命名风格五花八门：VDD、vdd、Vdd_1、VDD_MAIN……数据合并时，同一个测试项被当成不同列，分析结果一塌糊涂。

标准化规则

统一大写：所有测试项名称转大写
统一分隔符：用下划线 _，不用空格、连字符、点号
去除特殊字符：去掉 @、#、$、% 等符号
统一缩写：比如 VDD 和 VDDCORE 要区分，但 VDD 和 VDD_MAIN 要统一
添加单位后缀：电压加 _V，电流加 _A，频率加 _HZ

def standardize_column_name(name):
    """
    标准化测试项名称
    """
    # 转大写
    name = name.upper()
    
    # 替换常见分隔符为下划线
    name = name.replace(' ', '_')
    name = name.replace('-', '_')
    name = name.replace('.', '_')
    
    # 去除特殊字符
    import re
    name = re.sub(r'[^A-Z0-9_]', '', name)
    
    # 去除首尾下划线
    name = name.strip('_')
    
    # 合并连续下划线
    while '__' in name:
        name = name.replace('__', '_')
    
    return name

# 应用标准化
name_mapping = {}
for col in df.columns:
    if col in ['Lot', 'Wafer', 'Site', 'TestTime', 'TesterID']:
        continue  # 关键信息列不处理
    new_name = standardize_column_name(col)
    name_mapping[col] = new_name
    
# 输出映射关系
print('名称标准化映射:')
for old, new in name_mapping.items():
    if old != new:
        print(f'  {old} → {new}')

# 重命名列
df.rename(columns=name_mapping, inplace=True)

名称冲突处理

标准化后可能出现名称冲突——两个不同的原始名称变成了同一个标准化名称。比如 VDD_1 和 VDD1 都变成了 VDD1。这时候需要人工介入，确认是否真的是同一个测试项。

# 检测名称冲突
from collections import Counter

new_names = [standardize_column_name(col) for col in df.columns 
             if col not in ['Lot', 'Wafer', 'Site', 'TestTime', 'TesterID']]
name_counts = Counter(new_names)

conflicts = {name: count for name, count in name_counts.items() if count > 1}
if conflicts:
    print('名称冲突检测:')
    for name, count in conflicts.items():
        original_names = [col for col in df.columns 
                         if standardize_column_name(col) == name]
        print(f'  {name}: {count}个原始名称 → {original_names}')
else:
    print('未检测到名称冲突')

最佳实践：我建议在测试程序开发阶段就制定命名规范，并写入团队文档。如果项目已经跑起来了，就用上面的脚本做一次批量标准化，同时输出映射表，方便后续追溯。标准化后的名称最好和测试规范文档里的名称一致，这样工程师看数据时不用来回翻文档。

清洗后的数据验证

清洗完别急着往下走。我习惯做三件事验证：

数据量对比：清洗前后行数、列数变化是否合理
关键统计量对比：均值、标准差、分位数有没有异常跳变
良率对比：清洗前后的良率差异应该在1%以内，超过这个范围要仔细排查

# 清洗前后对比
print('=== 清洗前后数据对比 ===')
print(f'原始数据: {len(df_original)} 行, {len(df_original.columns)} 列')
print(f'清洗后数据: {len(df_cleaned)} 行, {len(df_cleaned.columns)} 列')
print(f'删除行数: {len(df_original) - len(df_cleaned)}')
print(f'删除列数: {len(df_original.columns) - len(df_cleaned.columns)}')

# 关键测试项统计对比
key_tests = ['VDD', 'IDD', 'FREQ']
for test in key_tests:
    if test in df_original.columns and test in df_cleaned.columns:
        orig_mean = df_original[test].mean()
        clean_mean = df_cleaned[test].mean()
        diff_pct = abs(orig_mean - clean_mean) / orig_mean * 100
        print(f'{test}: 均值变化 {diff_pct:.2f}%')

数据清洗这一步，说白了就是给分析打地基。地基不稳，上面盖的楼再漂亮也得塌。我做了这么多年芯片测试，见过太多分析报告因为数据没洗干净，结论完全跑偏。花30%的时间做清洗，能省后面70%的返工时间——这笔账，你算算值不值。