3. 数据采集与清洗实战：从MES/ERP系统提取数据、处理缺失值与异常值、数据标准化与归一化

各位同学，欢迎来到第三章。说实话，这一章才是真正动手的开始。前两章我们聊了半导体供应链的宏观逻辑，讲了物料需求预测的数学框架。但到了实战环节，你会发现一个残酷的现实：你拿到的数据，永远是脏的。

我在台积电做数据项目时，有个深刻的体会——数据清洗占了我60%以上的时间。MES系统吐出来的数据，ERP系统导出的报表，格式不统一、字段缺失、时间戳错乱……你想想看，如果输入是垃圾，再牛的模型也白搭。所以这一章，我带你走一遍数据采集与清洗的完整流程。

3.1 从MES/ERP系统提取数据

半导体工厂里，MES（制造执行系统）和ERP（企业资源计划系统）是两个核心数据源。MES管的是产线上的实时数据，比如每台机台的运行状态、每批晶圆的加工进度。ERP管的是资源层面的数据，比如物料库存、采购订单、客户需求。

我个人习惯的做法是：先拉MES的工单数据，再补ERP的库存数据。因为物料需求预测的核心是“未来要生产什么”，而MES的工单数据直接告诉你生产计划。

下面是一个典型的MES数据提取脚本。假设我们通过API接口获取数据：

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

def fetch_mes_workorders(start_date, end_date):
    """
    从MES系统获取工单数据
    我在项目中遇到过，MES的API返回格式经常变，所以加了异常处理
    """
    url = "http://mes.internal.semi/api/workorders"
    params = {
        "start": start_date.strftime("%Y-%m-%d"),
        "end": end_date.strftime("%Y-%m-%d"),
        "status": "active"  # 只取活跃工单
    }
    
    try:
        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        df = pd.DataFrame(data)
        return df
    except requests.exceptions.RequestException as e:
        print(f"API请求失败: {e}")
        return None

# 拉取最近30天的工单
today = datetime.now()
start = today - timedelta(days=30)
df_mes = fetch_mes_workorders(start, today)
print(f"获取到 {len(df_mes)} 条工单记录")

嗯，这里要注意：MES系统的数据量通常很大，一次拉取全量数据会卡死。我建议分批次拉取，比如按天或按周。另外，ERP系统的数据提取方式类似，但通常需要走数据库直连：

import pymysql

def fetch_erp_inventory():
    """
    从ERP系统获取物料库存数据
    我曾经因为没加索引，查询跑了20分钟……后来加了物料编码索引
    """
    conn = pymysql.connect(
        host="erp-db.internal",
        user="readonly_user",
        password="******",
        database="erp_inventory"
    )
    
    query = """
    SELECT material_code, warehouse, quantity, last_update_time
    FROM inventory_snapshot
    WHERE last_update_time >= DATE_SUB(NOW(), INTERVAL 7 DAY)
    """
    
    df = pd.read_sql(query, conn)
    conn.close()
    return df

df_erp = fetch_erp_inventory()
print(f"ERP库存数据: {df_erp.shape}")

3.2 处理缺失值与异常值

数据到手了，接下来就是清洗。说白了，就是处理那些“不该出现”和“该出现却没出现”的数据。

3.2.1 缺失值处理

半导体数据中，缺失值很常见。比如MES系统里某个机台的加工时间没记录，或者ERP系统里某个物料的库存数量为空。为什么会这样？可能是传感器故障、人工录入遗漏、系统间同步延迟。

我一般按这个优先级处理：

1. 删除：如果缺失比例小于5%，直接删掉对应行。简单粗暴，但有效。
2. 填充：如果缺失比例在5%-30%之间，用均值、中位数或前向填充。
3. 标记：如果缺失比例超过30%，说明这个字段本身有问题，我会单独标记出来，作为特征送入模型。

来看代码：

def handle_missing_values(df):
    """
    处理缺失值
    我在项目中遇到过，直接填充均值有时会引入偏差，所以加了分组逻辑
    """
    # 先看缺失情况
    missing_ratio = df.isnull().mean()
    print("各字段缺失比例:")
    print(missing_ratio[missing_ratio > 0])
    
    # 处理数值型字段
    numeric_cols = df.select_dtypes(include=['float64', 'int64']).columns
    for col in numeric_cols:
        if missing_ratio[col] < 0.05:
            # 缺失少，直接删除
            df.dropna(subset=[col], inplace=True)
        elif missing_ratio[col] < 0.3:
            # 用该物料类型的中位数填充
            # 你想想看，不同物料的数值差异很大，不能统一填充
            df[col] = df.groupby('material_type')[col].transform(
                lambda x: x.fillna(x.median())
            )
        else:
            # 缺失多，标记为-1，保留信息
            df[col + '_missing'] = df[col].isnull().astype(int)
            df[col].fillna(-1, inplace=True)
    
    return df

df_clean = handle_missing_values(df_mes)
print(f"清洗后数据量: {len(df_clean)}")

3.2.2 异常值处理

异常值在半导体数据里太常见了。比如某个机台的加工时间突然变成0，或者物料消耗量突然暴增100倍。这些异常值如果不处理，会严重扭曲预测结果。

我常用的方法是3σ原则和IQR（四分位距）法。对于半导体数据，我更推荐IQR法，因为数据分布往往不是正态的。

def detect_outliers_iqr(df, column):
    """
    用IQR法检测异常值
    我记得有一次，某个物料的数据分布特别偏，IQR法反而把正常值误判了
    所以加了skewness判断
    """
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    
    outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
    
    # 检查数据偏度
    skewness = df[column].skew()
    if abs(skewness) > 1:
        print(f"警告: {column} 偏度 {skewness:.2f}，建议使用对数变换后再检测")
    
    return outliers, lower_bound, upper_bound

# 检测加工时间的异常值
outliers, lb, ub = detect_outliers_iqr(df_clean, 'processing_time')
print(f"检测到 {len(outliers)} 条异常值")
print(f"正常范围: [{lb:.2f}, {ub:.2f}]")

处理异常值，我一般不是直接删除，而是先分析原因。如果是传感器故障导致的，我会标记并填充；如果是真实的生产波动（比如紧急加单），我会保留但单独标记。

3.3 数据标准化与归一化

数据清洗完了，但不同字段的量纲差异很大。比如物料消耗量可能是几千，而加工时间只有几十。如果不做标准化，模型会天然偏向数值大的特征。

标准化和归一化，说白了就是让数据“站在同一个起跑线上”。

3.3.1 标准化（Z-score）

标准化是把数据变成均值为0、标准差为1的分布。适合数据本身近似正态分布的场景。我在做线性回归模型时，特别喜欢用标准化。

from sklearn.preprocessing import StandardScaler

def standardize_features(df, columns):
    """
    标准化处理
    我建议先做异常值处理，再做标准化，否则异常值会拉偏均值和标准差
    """
    scaler = StandardScaler()
    df_scaled = df.copy()
    df_scaled[columns] = scaler.fit_transform(df[columns])
    
    print("标准化后各字段统计:")
    print(df_scaled[columns].describe().loc[['mean', 'std']])
    
    return df_scaled, scaler

# 对数值型特征做标准化
numeric_features = ['processing_time', 'quantity', 'cycle_time']
df_standardized, scaler = standardize_features(df_clean, numeric_features)

3.3.2 归一化（Min-Max）

归一化是把数据缩放到[0,1]区间。适合数据分布没有明显边界、或者需要保留原始相对关系的场景。比如神经网络模型，我一般用归一化。

from sklearn.preprocessing import MinMaxScaler

def normalize_features(df, columns):
    """
    归一化处理
    你想想看，如果数据里有极端值，归一化会把正常值压缩到很小的区间
    所以一定要先处理异常值
    """
    scaler = MinMaxScaler(feature_range=(0, 1))
    df_normalized = df.copy()
    df_normalized[columns] = scaler.fit_transform(df[columns])
    
    print("归一化后各字段范围:")
    for col in columns:
        print(f"{col}: [{df_normalized[col].min():.3f}, {df_normalized[col].max():.3f}]")
    
    return df_normalized, scaler

df_normalized, minmax_scaler = normalize_features(df_clean, numeric_features)

3.4 本章知识体系

为了让你更直观地理解本章的逻辑，我画了一张流程图。它展示了从数据提取到清洗、再到标准化的完整链路：

这张图把本章的核心逻辑串起来了。你从MES和ERP系统拿到原始数据，先合并对齐，然后处理缺失值和异常值，最后做标准化或归一化。每一步都有坑，但每一步也都有对应的解法。

好了，这一章的内容就到这里。数据清洗是个体力活，但也是最能体现工程师功底的地方。下一章我们会用清洗好的数据，开始真正的特征工程和模型训练。到时候你会发现，前面这些“脏活累活”有多值得。

---

公众号:蓝海资料掘金营，微信deep3321