4. 数据清洗实战：缺失值处理、重复值删除、异常值识别与过滤、数据类型转换

数据清洗是良率报告自动化中最耗时但最关键的一步。原始测试数据通常包含测试机台未采集、多站点重复记录、探针卡异常导致的离群值以及字段类型不匹配等问题。本章将围绕四个核心清洗任务，结合 Python 与 Pandas 给出可直接复用的实战方案。

核心原则：清洗操作必须记录日志（删除行数、填充均值、类型变更字段），以便后续审计与回溯。切勿在原文件上直接修改，始终维护清洗副本。

4.1 缺失值处理

良率数据中缺失值常见于：测试项未执行（如 CP1 未测某参数）、机台通讯超时（Die 坐标存在但电性值为空）、手动录入遗漏。处理策略取决于缺失率与业务含义。

缺失率范围 推荐策略 适用场景
< 5% 删除含缺失的行（dropna） 随机缺失，样本量充足
5% – 20% 均值/中位数填充 连续型电性参数（如 Iddq）
> 20% 标记为独立类别或建模预测 分类字段（如 Bin Code） 
import pandas as pd
import numpy as np

# 读取原始良率数据
df = pd.read_csv("wafer_test_raw.csv")

# 1. 查看缺失概况
print(df.isnull().sum())
print(f"缺失率: {df.isnull().mean().round(4)}")

# 2. 删除缺失率 < 5% 的列中的缺失行
threshold = 0.05
cols_low_missing = df.columns[df.isnull().mean() < threshold]
df_clean = df.dropna(subset=cols_low_missing)

# 3. 对连续型参数用中位数填充
fill_cols = ['Vt_avg', 'Idsat', 'Ioff']
for col in fill_cols:
    median_val = df_clean[col].median()
    df_clean[col].fillna(median_val, inplace=True)
    print(f"{col} 填充中位数: {median_val:.4f}")
实战提示：对于良率 Bin 映射表，缺失的 Bin Code 应填充为 "Unknown" 而非数值 0，避免与良品 Bin 1 混淆。
缺失值处理决策流程 原始数据缺失检测 缺失率 < 5%? 删除缺失行 连续型? 中位数填充 标记为 Unknown

4.2 重复值删除

重复数据主要来源于：多站点合并时重复读取、同一 Die 被多次测试（Rerun）、数据库 ETL 重复推送。删除前必须明确“唯一键”定义。

  • 完全重复：所有列值相同，直接删除。
  • 部分重复：基于关键列（Wafer_ID + Die_X + Die_Y + Test_Time）去重，保留最新记录。
# 查看重复行数量
print(f"完全重复行数: {df_clean.duplicated().sum()}")

# 删除完全重复
df_clean.drop_duplicates(inplace=True)

# 基于关键列去重，保留最后一条（最新测试结果）
key_cols = ['Wafer_ID', 'Die_X', 'Die_Y', 'Test_Item']
df_clean.drop_duplicates(subset=key_cols, keep='last', inplace=True)
print(f"去重后行数: {len(df_clean)}")
注意：对于良率分析，同一 Die 的多次测试应保留最后一次（Rerun 结果），而非第一次。使用 keep='last' 参数。

4.3 异常值识别与过滤

异常值在良率数据中通常对应：探针卡接触不良（开路/短路）、测试机台通道损坏（固定值）、工艺缺陷（真实但极端的电性参数）。识别方法需结合统计与业务规则。

方法 原理 适用参数
IQR 四分位距 Q1 - 1.5*IQR ~ Q3 + 1.5*IQR Vt、Idsat 等对称分布参数
Z-Score |Z| > 3 视为异常 正态分布参数（如 Ioff）
固定阈值 Spec Limit 上下限 所有测试项（业务强制） 
# IQR 方法过滤
Q1 = df_clean['Vt_avg'].quantile(0.25)
Q3 = df_clean['Vt_avg'].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR

mask_iqr = (df_clean['Vt_avg'] >= lower) & (df_clean['Vt_avg'] <= upper)
print(f"IQR 过滤前: {len(df_clean)}, 过滤后: {mask_iqr.sum()}")

# Z-Score 方法
from scipy import stats
z_scores = np.abs(stats.zscore(df_clean['Idsat']))
mask_z = z_scores < 3
print(f"Z-Score 过滤前: {len(df_clean)}, 过滤后: {mask_z.sum()}")

# 固定 Spec Limit 过滤（业务规则优先）
spec_lower, spec_upper = 0.1, 2.0  # 示例 Iddq 规格
mask_spec = (df_clean['Iddq'] >= spec_lower) & (df_clean['Iddq'] <= spec_upper)

# 综合过滤：取交集
df_final = df_clean[mask_iqr & mask_z & mask_spec].copy()
print(f"最终有效数据: {len(df_final)}")
异常值识别与过滤流程 原始参数数据 IQR 四分位距 Z-Score 阈值 Spec Limit 规格 综合过滤（交集）

4.4 数据类型转换

原始 CSV 或数据库导出常导致类型错乱：数值列被读为字符串（如 Bin Code 带引号）、时间列未解析、分类列未转为 category 类型。类型转换直接影响后续聚合与绘图效率。

原始类型 目标类型 转换方法 常见问题
object（数字字符串） float64 / int64 pd.to_numeric() 含逗号或单位符号
object（日期字符串） datetime64 pd.to_datetime() 格式不统一（MM/DD vs DD-MM）
int64（分类编码） category astype('category') 编码无实际顺序意义 
# 1. 数值列转换（处理含逗号的数字）
df_final['Yield_Pct'] = pd.to_numeric(df_final['Yield_Pct'], errors='coerce')
# errors='coerce' 将无法转换的值变为 NaN，便于后续处理

# 2. 时间列统一解析
df_final['Test_Date'] = pd.to_datetime(df_final['Test_Date'], format='%Y-%m-%d')
# 若格式不统一，可先尝试 infer_datetime_format=True

# 3. 分类列优化内存
df_final['Bin_Code'] = df_final['Bin_Code'].astype('category')
df_final['Test_Item'] = df_final['Test_Item'].astype('category')

# 4. 验证转换结果
print(df_final.dtypes)
print(f"内存占用: {df_final.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
性能建议：将重复值高的文本列（如 Bin_Code、Test_Item）转为 category 类型，可减少 50%–70% 内存占用，加速 groupby 与绘图操作。

4.5 清洗流水线封装

将上述四个步骤封装为可复用的清洗函数，支持参数化配置，便于集成到自动化报表流程中。

def clean_yield_data(df, config):
    """
    良率数据清洗流水线
    config: dict, 包含缺失阈值、异常方法、类型映射等
    """
    df_clean = df.copy()

    # Step 1: 缺失值处理
    for col, strategy in config['missing'].items():
        if strategy == 'drop':
            df_clean.dropna(subset=[col], inplace=True)
        elif strategy == 'median':
            df_clean[col].fillna(df_clean[col].median(), inplace=True)
        elif strategy == 'unknown':
            df_clean[col].fillna('Unknown', inplace=True)

    # Step 2: 重复值删除
    if config.get('dedup_keys'):
        df_clean.drop_duplicates(subset=config['dedup_keys'], keep='last', inplace=True)
    else:
        df_clean.drop_duplicates(inplace=True)

    # Step 3: 异常值过滤
    for col, method in config['outlier'].items():
        if method == 'iqr':
            Q1, Q3 = df_clean[col].quantile([0.25, 0.75])
            IQR = Q3 - Q1
            mask = (df_clean[col] >= Q1 - 1.5*IQR) & (df_clean[col] <= Q3 + 1.5*IQR)
            df_clean = df_clean[mask]
        elif method == 'zscore':
            from scipy import stats
            df_clean = df_clean[np.abs(stats.zscore(df_clean[col])) < 3]

    # Step 4: 类型转换
    for col, dtype in config['dtype_map'].items():
        if dtype == 'datetime':
            df_clean[col] = pd.to_datetime(df_clean[col])
        elif dtype == 'category':
            df_clean[col] = df_clean[col].astype('category')
        else:
            df_clean[col] = pd.to_numeric(df_clean[col], errors='coerce')

    return df_clean

# 使用示例
config = {
    'missing': {'Vt_avg': 'median', 'Bin_Code': 'unknown'},
    'dedup_keys': ['Wafer_ID', 'Die_X', 'Die_Y'],
    'outlier': {'Idsat': 'iqr', 'Ioff': 'zscore'},
    'dtype_map': {'Test_Date': 'datetime', 'Bin_Code': 'category'}
}
df_final = clean_yield_data(df, config)
print(f"清洗完成，最终行数: {len(df_final)}")
清洗日志示例：
缺失填充: Vt_avg 填充 0.4521 (中位数), Bin_Code 填充 'Unknown' (12 条)
重复删除: 基于 Wafer_ID+Die_X+Die_Y 删除 34 条重复
异常过滤: Idsq IQR 过滤 8 条, Ioff Z-Score 过滤 3 条
类型转换: Test_Date → datetime, Bin_Code → category

通过以上四个步骤的实战清洗，良率数据从原始杂乱状态变为结构化、可分析的高质量数据集，为后续趋势图绘制与 Pareto 分析奠定坚实基础。

← 上一章 📖 返回目录 下一章 →