3. 数据采集与清洗

良率分析的起点是数据。在半导体制造中，数据分散在多个异构系统中，格式、粒度、质量参差不齐。本章将系统梳理良率数据的三大核心来源——MES、EAP、测试机，并给出标准化的数据清洗流程与Python实现。

3.1 良率数据来源全景

半导体产线数据流可抽象为“设备层 → 控制层 → 执行层 → 分析层”四层架构。良率相关的原始数据主要来自以下三个系统：

3.2 各系统数据特征与采集要点

3.2.1 MES 数据

MES 记录每一批晶圆（Lot）在产线上的完整流动轨迹。典型数据包括：

• 批次信息： Lot ID、产品型号、工艺流程（Flow）、当前步骤（Step）
• 事件记录： 进站时间、出站时间、操作员、设备分配
• 良率相关： 站点良率、累计良率、缺陷代码

采集方式：通过 MES 的 API 或数据库直连，按批次或时间区间拉取。建议使用增量采集策略，避免全量重复加载。

3.2.2 EAP 数据

EAP 负责设备与 MES 之间的自动化通信，记录设备运行参数。关键特征：

• 高频时序数据： 温度、压力、气体流量、RF功率等，采样频率可达 1Hz~100Hz
• 设备状态： Chamber ID、Recipe Step、Alarm 代码
• 数据量巨大： 单台设备单日可产生数 GB 数据

采集策略：采用时间窗口 + 设备ID 分片拉取，配合数据压缩（如 Parquet 格式）存储。

3.2.3 测试机数据

测试机（Tester）在 CP（晶圆测试）和 FT（终测）阶段产生 Die 级数据：

• 坐标信息： Die 在晶圆上的 X/Y 坐标
• 测试结果： Bin 分类（Pass/Fail/各失效类别）、测试项数值
• 映射文件： Wafer Map 数据，用于可视化良率分布

采集方式：测试机通常输出标准格式文件（如 STDF、CSV、自定义格式），通过文件解析或专用接口入库。

3.3 数据清洗方法与Python实现

原始数据通常存在以下问题：缺失值、重复记录、异常值、格式不一致、时间戳漂移。下面给出标准清洗流程及 Python 代码实现。

3.3.1 清洗流程总览

1. 格式统一： 将各系统字段名映射为标准命名（如 lot_id、equipment_id、timestamp）
2. 缺失值处理： 根据业务规则填充或删除
3. 重复记录去重： 基于主键（Lot ID + Step + Timestamp）去除完全重复行
4. 异常值过滤： 使用 IQR 或 Z-score 方法剔除物理不可达值
5. 时间戳对齐： 统一时区、格式，处理漂移
6. 数据关联： 基于关联键合并 MES、EAP、测试机数据

3.3.2 Python 实现：数据清洗核心函数

以下代码演示了良率数据清洗的典型操作。假设已从 MES、EAP、测试机采集到三个 DataFrame：df_mes、df_eap、df_tester。

import pandas as pd
import numpy as np
from datetime import datetime

# ========== 1. 格式统一 ==========
def standardize_columns(df, mapping_dict):
    """
    将各系统字段名映射为标准命名
    mapping_dict 示例: {'LOT_ID': 'lot_id', 'EQP_ID': 'equipment_id'}
    """
    return df.rename(columns=mapping_dict)

# ========== 2. 缺失值处理 ==========
def handle_missing_values(df, threshold=0.5, fill_strategy='median'):
    """
    删除缺失率超过 threshold 的列
    对数值型列使用 median 或 mean 填充
    """
    # 删除高缺失率列
    missing_ratio = df.isnull().mean()
    cols_to_drop = missing_ratio[missing_ratio > threshold].index
    df_clean = df.drop(columns=cols_to_drop)

    # 填充数值列
    numeric_cols = df_clean.select_dtypes(include=[np.number]).columns
    for col in numeric_cols:
        if df_clean[col].isnull().sum() > 0:
            if fill_strategy == 'median':
                df_clean[col].fillna(df_clean[col].median(), inplace=True)
            elif fill_strategy == 'mean':
                df_clean[col].fillna(df_clean[col].mean(), inplace=True)

    # 填充类别列（使用众数）
    cat_cols = df_clean.select_dtypes(include=['object']).columns
    for col in cat_cols:
        if df_clean[col].isnull().sum() > 0:
            df_clean[col].fillna(df_clean[col].mode()[0], inplace=True)

    return df_clean

# ========== 3. 去重 ==========
def remove_duplicates(df, subset_cols=None):
    """
    基于指定列去重，保留第一条记录
    """
    if subset_cols is None:
        subset_cols = df.columns.tolist()
    return df.drop_duplicates(subset=subset_cols, keep='first')

# ========== 4. 异常值过滤（IQR方法） ==========
def filter_outliers_iqr(df, col, multiplier=1.5):
    """
    使用 IQR 方法过滤单列异常值
    返回过滤后的 DataFrame
    """
    Q1 = df[col].quantile(0.25)
    Q3 = df[col].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - multiplier * IQR
    upper_bound = Q3 + multiplier * IQR
    return df[(df[col] >= lower_bound) & (df[col] <= upper_bound)]

# ========== 5. 时间戳对齐 ==========
def align_timestamps(df, time_col='timestamp', target_tz='Asia/Shanghai'):
    """
    统一时间戳格式和时区
    """
    df[time_col] = pd.to_datetime(df[time_col], errors='coerce')
    # 如果有时区信息则转换，否则添加时区
    if df[time_col].dt.tz is None:
        df[time_col] = df[time_col].dt.tz_localize(target_tz)
    else:
        df[time_col] = df[time_col].dt.tz_convert(target_tz)
    return df

# ========== 6. 数据关联（合并MES、EAP、测试机） ==========
def merge_yield_data(df_mes, df_eap, df_tester, merge_keys):
    """
    基于关联键合并三源数据
    merge_keys 示例: ['lot_id', 'equipment_id', 'timestamp_window']
    """
    # 先合并 MES 和 EAP
    df_merged = pd.merge(df_mes, df_eap, on=merge_keys, how='left')
    # 再合并测试机数据
    df_merged = pd.merge(df_merged, df_tester, on=merge_keys, how='left')
    return df_merged

# ========== 完整清洗流水线示例 ==========
def full_cleaning_pipeline(df_mes_raw, df_eap_raw, df_tester_raw):
    # 格式统一
    mes_mapping = {'LOT_ID': 'lot_id', 'STEP': 'step', 'QTY': 'quantity'}
    eap_mapping = {'EQP': 'equipment_id', 'PARAM': 'parameter', 'VAL': 'value'}
    tester_mapping = {'DIE_X': 'x', 'DIE_Y': 'y', 'BIN': 'bin_code'}

    df_mes = standardize_columns(df_mes_raw, mes_mapping)
    df_eap = standardize_columns(df_eap_raw, eap_mapping)
    df_tester = standardize_columns(df_tester_raw, tester_mapping)

    # 缺失值处理
    df_mes = handle_missing_values(df_mes, threshold=0.5)
    df_eap = handle_missing_values(df_eap, threshold=0.5)
    df_tester = handle_missing_values(df_tester, threshold=0.5)

    # 去重
    df_mes = remove_duplicates(df_mes, subset_cols=['lot_id', 'step', 'timestamp'])
    df_eap = remove_duplicates(df_eap, subset_cols=['equipment_id', 'parameter', 'timestamp'])
    df_tester = remove_duplicates(df_tester, subset_cols=['lot_id', 'x', 'y'])

    # 异常值过滤（示例：过滤测试机中异常电流值）
    if 'current' in df_tester.columns:
        df_tester = filter_outliers_iqr(df_tester, 'current', multiplier=3.0)

    # 时间戳对齐
    df_mes = align_timestamps(df_mes, 'timestamp')
    df_eap = align_timestamps(df_eap, 'timestamp')
    df_tester = align_timestamps(df_tester, 'test_time')

    # 数据关联
    merge_keys = ['lot_id', 'equipment_id']
    df_final = merge_yield_data(df_mes, df_eap, df_tester, merge_keys)

    return df_final

3.4 数据质量监控指标

清洗完成后，需要建立数据质量监控体系，确保后续分析基于可靠数据。常用监控指标如下：

3.5 本章小结

• 良率数据三大来源：MES（批次事件）、EAP（设备参数）、测试机（Die级结果）
• 数据清洗六大步骤：格式统一 → 缺失值处理 → 去重 → 异常值过滤 → 时间戳对齐 → 数据关联
• Python 实现需注意：保留原始数据副本、使用 IQR 方法过滤异常值、统一时区处理
• 建立数据质量监控体系，设置合理阈值，确保分析数据可靠