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

3. 数据采集与清洗：从MES/SPC系统获取原始数据

在半导体制造与高端电子组装产线中，缺陷密度监控的起点并非看板本身，而是底层的数据管道。MES（制造执行系统）记录每一片晶圆或每一块基板的加工历史、设备参数与检验结果；SPC（统计过程控制）系统则按时间序列采集关键质量指标。本章将聚焦如何从这两个源头获取原始数据，并完成缺失值、异常值与重复数据的清洗，为后续的实时看板与报警阈值计算提供可靠输入。

  核心原则： 数据清洗的目标不是“删除所有异常”，而是区分“真实工艺波动”与“传感器/录入错误”。错误的清洗策略会直接导致误报警或漏报警。

3.1 数据源识别与接口规范

MES与SPC系统的数据输出格式通常分为三类：

数据源	典型输出格式	采集频率	关键字段
MES 检验模块	CSV / JSON / 数据库表	每批次/每片	LotID, WaferID, Step, DefectCount, DieSize
SPC 系统	时间序列数据库 / REST API	每5分钟/每小时	Timestamp, Parameter, Value, UCL, LCL
设备自动化日志	SECS/GEM 消息	事件触发	EquipmentID, EventType, AlarmCode

以下是一个从MES系统获取原始缺陷数据的Python示例（模拟API调用）：

import pandas as pd
import requests
from datetime import datetime, timedelta

# 模拟MES API端点
MES_API_URL = "http://mes-fab01.internal/api/v1/defects"

def fetch_mes_defects(lot_id: str, start_time: datetime, end_time: datetime) -> pd.DataFrame:
    params = {
        "lot": lot_id,
        "from": start_time.isoformat(),
        "to": end_time.isoformat(),
        "format": "json"
    }
    response = requests.get(MES_API_URL, params=params, timeout=30)
    response.raise_for_status()
    data = response.json()
    
    # 转换为DataFrame
    df = pd.DataFrame(data["records"])
    df["timestamp"] = pd.to_datetime(df["timestamp"])
    return df

# 示例调用
df_raw = fetch_mes_defects("LOT-2025-003", 
                           datetime(2025, 3, 1, 0, 0, 0),
                           datetime(2025, 3, 1, 23, 59, 59))
print(f"获取到 {len(df_raw)} 条原始记录")
print(df_raw.head())

工程提示： 实际产线中MES接口往往有速率限制（如每秒最多10次请求）。建议使用批量拉取+本地缓存策略，避免因超时导致数据断层。

3.2 缺失值处理策略

原始数据中的缺失值通常由传感器离线、通信中断或检验跳过引起。不同缺失模式需要不同处理方式：

完全随机缺失（MCAR）： 如某台设备临时断网导致5分钟数据丢失。可采用前向填充（ffill）或线性插值。
随机缺失（MAR）： 如缺陷密度值缺失但批次号存在。可使用同批次均值填充。
非随机缺失（MNAR）： 如设备报警时故意不记录缺陷数。此类缺失必须标记为“特殊事件”，不可简单填充。

以下为缺失值处理的标准流程：

def handle_missing_values(df: pd.DataFrame, strategy: str = "interpolate") -> pd.DataFrame:
    """
    处理缺陷密度数据中的缺失值
    策略: 'drop' - 删除含缺失的行
          'ffill' - 前向填充
          'interpolate' - 线性插值（默认）
          'mean_by_lot' - 按批次均值填充
    """
    df_clean = df.copy()
    
    if strategy == "drop":
        df_clean = df_clean.dropna(subset=["defect_density"])
    elif strategy == "ffill":
        df_clean["defect_density"] = df_clean["defect_density"].ffill()
    elif strategy == "interpolate":
        df_clean["defect_density"] = df_clean["defect_density"].interpolate(method="linear")
    elif strategy == "mean_by_lot":
        lot_mean = df_clean.groupby("lot_id")["defect_density"].transform("mean")
        df_clean["defect_density"] = df_clean["defect_density"].fillna(lot_mean)
    else:
        raise ValueError(f"未知策略: {strategy}")
    
    missing_before = df["defect_density"].isna().sum()
    missing_after = df_clean["defect_density"].isna().sum()
    print(f"缺失值: {missing_before} -> {missing_after}")
    return df_clean

注意： 当缺失比例超过30%时，任何填充策略都会引入显著偏差。此时应优先排查数据源链路，而非依赖算法补全。

3.3 异常值检测与处理

缺陷密度数据中的异常值可能来自：晶圆边缘效应、设备突发故障、或人为录入错误。常用的检测方法包括：

方法	适用场景	阈值设定	优缺点
IQR（四分位距）	非正态分布、有离群点	Q1 - 1.5×IQR 或 Q3 + 1.5×IQR	稳健，不受极端值影响；但对小样本敏感
Z-Score	近似正态分布	\|Z\| > 3	计算简单；但受异常值本身影响均值与标准差
修正Z-Score（MAD）	含离群点的非正态分布	\|M\| > 3.5	基于中位数，稳健性优于Z-Score
移动窗口法	时间序列实时检测	均值 ± 3×滚动标准差	适应局部趋势；窗口大小需调参

以下为基于IQR与修正Z-Score的异常值检测实现：

import numpy as np

def detect_outliers_iqr(series: pd.Series, multiplier: float = 1.5) -> pd.Series:
    """使用IQR方法检测异常值，返回布尔掩码"""
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - multiplier * IQR
    upper_bound = Q3 + multiplier * IQR
    return (series < lower_bound) | (series > upper_bound)

def detect_outliers_modified_zscore(series: pd.Series, threshold: float = 3.5) -> pd.Series:
    """使用修正Z-Score（基于MAD）检测异常值"""
    median = series.median()
    mad = np.median(np.abs(series - median))
    if mad == 0:
        return pd.Series([False] * len(series), index=series.index)
    modified_z = 0.6745 * (series - median) / mad
    return np.abs(modified_z) > threshold

# 应用示例
df_raw["is_outlier_iqr"] = detect_outliers_iqr(df_raw["defect_density"])
df_raw["is_outlier_mad"] = detect_outliers_modified_zscore(df_raw["defect_density"])

# 异常值处理：标记但不删除，保留给人工审核
df_raw["defect_density_clean"] = df_raw["defect_density"].copy()
df_raw.loc[df_raw["is_outlier_mad"], "defect_density_clean"] = np.nan
print(f"IQR检测到异常: {df_raw['is_outlier_iqr'].sum()} 条")
print(f"MAD检测到异常: {df_raw['is_outlier_mad'].sum()} 条")

  最佳实践： 在良率监控场景中，异常值不应直接删除，而应标记为“待确认”状态。许多工艺异常本身就是报警信号，过早清洗会削弱监控灵敏度。

3.4 重复数据识别与去重

重复数据通常由以下原因引起：

数据重传： MES系统因网络抖动重复发送同一检验记录。
多系统冗余采集： SPC与MES同时记录了同一测量点的数据。
人工补录： 操作员手动录入时误提交两次。

去重策略需基于业务主键：

def deduplicate_defect_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    去除重复的缺陷记录
    主键: lot_id + wafer_id + step_name + timestamp（精确到秒）
    """
    # 标记完全重复的行
    dup_mask = df.duplicated(subset=["lot_id", "wafer_id", "step_name", "timestamp"], keep="first")
    print(f"发现 {dup_mask.sum()} 条重复记录")
    
    # 保留第一条，删除后续重复
    df_dedup = df[~dup_mask].copy()
    
    # 对于时间戳接近但非完全相同的记录（如1秒内重复），合并为单条
    df_dedup = df_dedup.sort_values("timestamp")
    df_dedup["time_diff"] = df_dedup.groupby(["lot_id", "wafer_id", "step_name"])["timestamp"].diff()
    
    # 时间差小于1秒且缺陷密度相同的视为重复
    near_dup = (df_dedup["time_diff"] < pd.Timedelta(seconds=1)) & \
               (df_dedup["defect_density"] == df_dedup["defect_density"].shift(1))
    print(f"近重复记录（1秒内）: {near_dup.sum()} 条")
    
    df_final = df_dedup[~near_dup].drop(columns=["time_diff"])
    return df_final

df_clean = deduplicate_defect_data(df_raw)

3.5 数据清洗全流程架构

以下SVG图展示了从MES/SPC原始数据到清洗后标准数据集的完整管道：

3.6 清洗效果验证与监控

数据清洗完成后，必须通过以下指标验证清洗效果：

验证指标	计算公式	可接受范围	监控频率
缺失率	缺失值数 / 总记录数	< 5%	每批次
异常率	标记异常数 / 总记录数	< 2%	每日
重复率	去重记录数 / 原始记录数	< 1%	每批次
数据完整性	必填字段非空比例	100%	实时

自动化建议： 将上述验证指标集成到数据管道的监控看板中。当缺失率或异常率超过阈值时，自动触发数据源链路检查告警，而非等待下游看板出现异常。

通过本章的数据采集与清洗流程，原始MES/SPC数据被转化为高质量、标准化的时序数据集，为第4章的缺陷密度计算与实时看板渲染奠定了可靠的数据基础。