3. 数据采集与清洗

在半导体制造中，制造执行系统（MES）是数据中枢，实时记录每一片晶圆在各工艺站点的量测数据。本章将系统讲解如何从MES抽取SPC数据，并完成缺失值填充与异常值检测，为后续良率建模奠定干净、可靠的数据基础。

3.1 MES系统数据接口与SPC数据抽取

MES系统通常提供REST API、数据库直连（JDBC/ODBC）或消息队列（Kafka/MQ）三种数据获取方式。对于批量建模场景，推荐使用数据库直连+定时批处理架构。

3.2 数据字段定义与宽表构建

从MES抽取的原始数据通常为长表格式（每行一个量测值），需要透视（Pivot）为宽表，使每个参数成为独立列。标准SPC宽表字段如下：

3.3 缺失值处理策略

半导体量测数据缺失原因包括：设备未量测、数据传输出错、站点跳过量测。缺失率超过50%的参数应直接剔除；低于5%的缺失可采用填充策略。

3.3.1 缺失值处理方法对比

import pandas as pd
import numpy as np

# 加载SPC宽表
df = pd.read_parquet('spc_wide.parquet')

# 1. 统计缺失率
missing_rate = df.isnull().mean()
print("各列缺失率:\n", missing_rate[missing_rate > 0])

# 2. 剔除缺失率 > 50% 的列
cols_to_drop = missing_rate[missing_rate > 0.5].index.tolist()
df.drop(columns=cols_to_drop, inplace=True)
print(f"剔除高缺失列: {cols_to_drop}")

# 3. 按批次分组，使用前向填充+线性插值
df.sort_values(['lot_id', 'measure_time'], inplace=True)
df['param_CD'] = df.groupby('lot_id')['param_CD'].transform(
    lambda x: x.ffill().interpolate(method='linear', limit_direction='both')
)
df['param_THK'] = df.groupby('lot_id')['param_THK'].transform(
    lambda x: x.fillna(x.median())
)

print("缺失值处理完成，剩余缺失数:", df.isnull().sum().sum())

3.4 异常值检测：3σ原则

3σ原则基于正态分布假设：数据落在均值±3σ区间外的概率仅为0.27%，可视为异常。适用于膜厚、关键尺寸等近似正态分布的参数。

def detect_3sigma(df, column, k=3):
    """
    3σ异常值检测
    Parameters:
        df: DataFrame
        column: 列名
        k: 标准差倍数，默认3
    Returns:
        df: 带异常标记的DataFrame
    """
    mean = df[column].mean()
    std = df[column].std()
    lower = mean - k * std
    upper = mean + k * std
    
    df[f'{column}_outlier_3sigma'] = ~df[column].between(lower, upper)
    print(f"{column}: 均值={mean:.2f}, 标准差={std:.2f}")
    print(f"下限={lower:.2f}, 上限={upper:.2f}")
    print(f"异常数量: {df[f'{column}_outlier_3sigma'].sum()}")
    
    return df

# 应用3σ检测
df = detect_3sigma(df, 'param_CD', k=3)
df = detect_3sigma(df, 'param_THK', k=3)

3.5 异常值检测：IQR方法

IQR（四分位距）方法不依赖正态分布假设，对偏态分布更鲁棒。定义：IQR = Q3 - Q1，异常阈值为 [Q1 - 1.5×IQR, Q3 + 1.5×IQR]。

def detect_iqr(df, column, multiplier=1.5):
    """
    IQR异常值检测
    Parameters:
        df: DataFrame
        column: 列名
        multiplier: IQR倍数，默认1.5
    Returns:
        df: 带异常标记的DataFrame
    """
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - multiplier * IQR
    upper = Q3 + multiplier * IQR
    
    df[f'{column}_outlier_iqr'] = ~df[column].between(lower, upper)
    print(f"{column}: Q1={Q1:.2f}, Q3={Q3:.2f}, IQR={IQR:.2f}")
    print(f"下限={lower:.2f}, 上限={upper:.2f}")
    print(f"异常数量: {df[f'{column}_outlier_iqr'].sum()}")
    
    return df

# 应用IQR检测
df = detect_iqr(df, 'param_OVL', multiplier=1.5)

# 合并两种方法的异常标记（取并集）
df['is_anomaly'] = (
    df['param_CD_outlier_3sigma'] | 
    df['param_THK_outlier_3sigma'] | 
    df['param_OVL_outlier_iqr']
)
print(f"总异常样本数: {df['is_anomaly'].sum()}")

3.6 异常值处理策略

检测到异常值后，根据业务场景选择以下处理方式：

1. 删除异常样本：适用于明显量测错误（如CD为负值），且异常比例<1%。
2. 替换为边界值：将异常值截断至阈值上限或下限，保留样本量。
3. 标记为缺失后填充：先标记异常为NaN，再使用3.3节的缺失值填充方法处理。
4. 保留并标记：在建模时增加“是否异常”特征列，让模型自行学习异常模式。

def handle_outliers(df, column, method='clip'):
    """
    异常值处理
    method: 'drop' - 删除, 'clip' - 截断, 'fill_nan' - 标记为缺失
    """
    outlier_col = f'{column}_outlier_iqr'
    if outlier_col not in df.columns:
        df = detect_iqr(df, column)
    
    if method == 'drop':
        df_clean = df[~df[outlier_col]].copy()
        print(f"删除异常: {df[outlier_col].sum()} 行")
    elif method == 'clip':
        Q1 = df[column].quantile(0.25)
        Q3 = df[column].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
        df[column] = df[column].clip(lower, upper)
        df_clean = df.copy()
        print(f"截断异常至 [{lower:.2f}, {upper:.2f}]")
    elif method == 'fill_nan':
        df.loc[df[outlier_col], column] = np.nan
        df_clean = df.copy()
        print(f"标记 {df[outlier_col].sum()} 个异常为缺失")
    
    return df_clean

# 对CD使用截断，对OVL使用标记为缺失
df = handle_outliers(df, 'param_CD', method='clip')
df = handle_outliers(df, 'param_OVL', method='fill_nan')

# 最终检查
print("\n清洗后数据概况:")
print(f"总样本数: {len(df)}")
print(f"缺失值总数: {df.isnull().sum().sum()}")
print(f"异常标记总数: {df['is_anomaly'].sum()}")