3. 数据采集与清洗

价格偏差预防的起点是高质量的数据。采购计划动态优化依赖历史采购数据的完整性与准确性。本节聚焦数据采集范围、价格标准化方法、异常值检测与处理策略，以及数据质量评估框架，确保后续建模与优化建立在可靠的数据基础之上。

3.1 采购历史数据源与采集范围

采购数据源通常分散在 ERP、SRM、WMS 及外部价格指数平台中。数据采集需覆盖以下核心维度：

3.2 价格数据标准化

原始价格数据存在单位不统一、币种混杂、含税/不含税口径差异等问题。标准化是数据清洗的核心环节，需执行以下步骤：

1. 单位统一：将所有价格转换为标准计量单位（如元/千克、元/个）。对于按“吨”计价但采购单位为“千克”的数据，需做换算。
2. 币种转换：统一转换为人民币（CNY），使用订单日期的央行中间汇率进行换算。
3. 含税/不含税归一：统一转换为不含税价格。若原始数据为含税价，按适用税率（如 13%）做价税分离。
4. 价格类型对齐：区分“订单价”、“收货价”、“结算价”，优先使用结算价作为实际采购成本。
5. 时间戳标准化：统一为 YYYY-MM-DD 格式，并标注价格对应的业务日期（订单日期 vs 收货日期）。

# 价格标准化示例（Python 伪代码）
def standardize_price(row):
    # 1. 单位换算：若采购单位是"吨"，转换为"千克"
    if row['unit'] == '吨':
        row['price_per_kg'] = row['price'] / 1000
    else:
        row['price_per_kg'] = row['price']

    # 2. 币种转换：非 CNY 按当日汇率换算
    if row['currency'] != 'CNY':
        rate = get_exchange_rate(row['currency'], 'CNY', row['date'])
        row['price_cny'] = row['price_per_kg'] * rate
    else:
        row['price_cny'] = row['price_per_kg']

    # 3. 含税转不含税
    if row['tax_flag'] == '含税':
        row['price_ex_tax'] = row['price_cny'] / (1 + row['tax_rate'])
    else:
        row['price_ex_tax'] = row['price_cny']

    return row['price_ex_tax']

3.3 异常值检测与处理

异常值会严重扭曲价格基线，导致偏差预警失效。常用的检测方法包括统计法、四分位距法（IQR）和基于业务规则的检测。

3.3.1 异常值检测方法对比

3.3.2 异常值处理策略

• 删除：对于明显录入错误（如价格多一位小数、数量为负），直接删除该记录。
• 修正：若能通过关联单据（如合同价、历史均价）确认正确值，则进行修正。
• 标记：对于无法确认但疑似异常的数据，保留并添加异常标记，供后续建模时加权或排除。
• 插值：对于时间序列中的孤立异常点，可使用前后相邻值的均值或中位数替换。

3.4 数据质量评估

数据清洗完成后，需从四个维度评估数据质量，确保可用于价格偏差建模：

3.5 数据清洗工作流总结

将上述步骤整合为可重复执行的清洗流水线，建议按以下顺序编排：

1. 数据接入：从各源系统抽取原始数据，统一存储到数据湖或数据仓库。
2. 字段映射：将不同系统的字段名映射到统一数据模型。
3. 价格标准化：执行单位、币种、含税状态、时间戳的转换。
4. 异常值检测：依次使用 IQR 法和业务规则法进行检测，生成异常标记。
5. 异常处理：根据业务确认结果，执行删除、修正或保留标记。
6. 质量评估：计算完整性、准确性、一致性、时效性指标，生成质量报告。
7. 数据发布：将清洗后的数据集写入分析库，供价格基线计算和偏差预警使用。