4. 数据采集与清洗：内部ERP数据、外部市场数据、供应商报价数据的标准化处理

Should-cost 模型的精度高度依赖输入数据的质量。在智能寻源算法中，数据来源分散、格式各异、口径不统一是常态。本章聚焦于三类核心数据源——内部 ERP、外部市场、供应商报价——的采集策略与标准化清洗流程，确保算法模型能够基于一致、干净、可对齐的数据进行成本拆解与预测。

4.1 三类数据源的特征与采集策略

不同数据源在时效性、结构化程度、可信度上存在显著差异。下表对比了三类数据源的核心特征及推荐采集方式：

4.2 数据标准化处理流程

标准化处理的核心目标是消除数据间的“语义鸿沟”与“格式差异”，使不同来源的数据能够在统一的维度上对齐。整个流程分为四个阶段：

1. 字段映射与对齐： 将不同系统中的同类字段映射到统一的数据字典。例如，ERP 中的“物料编码”、供应商报价中的“零件号”、市场数据中的“商品代码”均映射为统一的 material_id。
2. 单位与量纲归一化： 强制转换为标准单位。重量统一为 kg，长度统一为 mm，货币统一为 USD（或按项目设定基准货币），数量级统一为单件成本。
3. 时间戳标准化： 所有数据记录的时间戳统一为 UTC+8 的 YYYY-MM-DD HH:mm:ss 格式，并标注数据有效期（valid_from / valid_to），便于时序对齐。
4. 异常值与缺失值处理： 对超出 3σ 范围的价格异常点进行标记并隔离；对缺失的关键字段（如材料牌号、加工工艺）采用规则填充或模型预测填充。

4.3 数据清洗的典型挑战与应对

在实际项目中，以下三类问题最为常见，需要设计专门的清洗规则：

• 供应商报价中的“水分”识别： 部分供应商会虚报材料用量或工时。清洗规则应对比报价中的材料重量与 BOM 理论重量，偏差超过 15% 时触发人工复核。
• 外部市场数据的滞后性修正： 大宗商品指数通常有 1-2 天的发布延迟。清洗流程需引入时间戳对齐算法，将报价日期与最近可用的市场指数进行匹配，而非简单使用发布日期。
• ERP 数据中的“幽灵物料”： 已停产但未标记失效的物料编码会干扰成本模型。清洗阶段需联合 ERP 中的物料状态字段与最近采购日期，自动标记超过 180 天无交易的物料为“待确认”。

4.4 数据标准化架构图

下图展示了从多源数据采集到标准化输出的整体架构，重点体现数据流经清洗、映射、归一化三个核心模块的处理路径：

4.5 代码示例：数据清洗管道核心逻辑

以下 Python 代码片段展示了标准化管道中字段映射与单位归一化的核心实现，采用链式调用风格：

import pandas as pd
import numpy as np

class DataStandardizer:
    """数据标准化管道"""
    
    def __init__(self, unit_map: dict, field_map: dict):
        self.unit_map = unit_map      # 单位转换映射表
        self.field_map = field_map    # 字段名映射表
    
    def rename_fields(self, df: pd.DataFrame) -> pd.DataFrame:
        """字段映射对齐"""
        return df.rename(columns=self.field_map)
    
    def normalize_units(self, df: pd.DataFrame, 
                        unit_col: str, value_col: str) -> pd.DataFrame:
        """单位与量纲归一化"""
        df['base_unit'] = df[unit_col].map(self.unit_map)
        # 转换系数查找，默认系数为 1.0
        df['conversion_factor'] = df[unit_col].apply(
            lambda u: self.unit_map.get(u, {}).get('factor', 1.0)
        )
        df['standard_value'] = df[value_col] * df['conversion_factor']
        df['standard_unit'] = 'kg'  # 统一为 kg
        return df
    
    def flag_outliers(self, df: pd.DataFrame, 
                      value_col: str, n_sigma: float = 3.0) -> pd.DataFrame:
        """基于 3σ 原则标记异常值"""
        mean = df[value_col].mean()
        std = df[value_col].std()
        df['is_outlier'] = np.abs(df[value_col] - mean) > n_sigma * std
        return df
    
    def run_pipeline(self, raw_df: pd.DataFrame) -> pd.DataFrame:
        """执行完整标准化管道"""
        df = raw_df.copy()
        df = self.rename_fields(df)
        df = self.normalize_units(df, 'unit', 'price')
        df = self.flag_outliers(df, 'standard_value')
        return df

# 使用示例
standardizer = DataStandardizer(
    unit_map={
        'g': {'factor': 0.001},
        'kg': {'factor': 1.0},
        'lb': {'factor': 0.4536}
    },
    field_map={
        '物料编码': 'material_id',
        '供应商名称': 'supplier_name',
        '报价金额': 'price',
        '计量单位': 'unit'
    }
)

cleaned_df = standardizer.run_pipeline(raw_quote_df)

4.6 数据质量校验清单

标准化完成后，建议逐项执行以下校验，确保数据满足 Should-cost 模型的输入要求：

通过上述标准化处理，三类异构数据源将被整合为一份结构统一、语义对齐、质量可控的标准化数据集，为后续的 Should-cost 模型训练与实时寻源算法提供可靠的数据底座。