3. 数据源识别与采集：供应商报价单、历史采购订单、合同价格表的数据来源与采集方法

价格方差清洗的第一步，是精准识别并采集三个核心数据源。这三个数据源分别承载着供应商的“期望价格”、企业的“实际支付价格”以及双方的“契约价格”。任何数据源的缺失或采集偏差，都会导致后续方差计算失真。

3.1 三大核心数据源定义与特征

3.2 数据采集方法与优先级策略

不同数据源的采集难度和时效性不同，建议按照以下优先级和采集方法执行：

1. 历史采购订单（最高优先级）：直接从ERP系统导出最近12-24个月的所有采购订单明细。这是最真实、最完整的交易数据源。
2. 合同价格表（次高优先级）：从合同管理系统或采购部门归档文件中提取当前有效的合同价格清单。注意区分“已过期合同”和“当前生效合同”。
3. 供应商报价单（辅助验证）：从询价记录或供应商门户中获取最新一轮的报价单。主要用于验证合同价格是否与市场行情一致。

3.3 数据源采集流程（SVG流程图）

下图展示了从三个数据源到价格方差清洗前的标准化采集流程。每个数据源经过独立的格式校验和字段映射后，进入统一的“价格数据湖”，为后续的方差计算做准备。

3.4 数据采集常见问题与应对

3.5 数据采集自动化脚本示例（Python伪代码）

以下是一个简化的数据采集脚本框架，用于从不同格式的文件中提取并标准化价格数据：

import pandas as pd
from datetime import datetime

def collect_price_data(source_type, file_path):
    """
    根据数据源类型采集并标准化价格数据
    """
    if source_type == 'quotation':
        # 读取供应商报价单（假设为Excel格式）
        df = pd.read_excel(file_path, sheet_name='报价明细')
        df['数据源'] = '报价单'
        df['采集时间'] = datetime.now()
        # 标准化字段名称
        df.rename(columns={
            '物料编码': 'material_code',
            '供应商编码': 'supplier_code',
            '报价单价': 'unit_price',
            '报价日期': 'price_date'
        }, inplace=True)
        
    elif source_type == 'purchase_order':
        # 读取历史采购订单（假设为CSV格式）
        df = pd.read_csv(file_path, encoding='utf-8')
        df['数据源'] = '采购订单'
        df['采集时间'] = datetime.now()
        df.rename(columns={
            'PO号': 'po_number',
            '物料编码': 'material_code',
            '供应商编码': 'supplier_code',
            '单价': 'unit_price',
            '下单日期': 'price_date'
        }, inplace=True)
        
    elif source_type == 'contract':
        # 读取合同价格表（假设为PDF解析后的结构化数据）
        df = pd.read_excel(file_path, sheet_name='合同价格')
        df['数据源'] = '合同'
        df['采集时间'] = datetime.now()
        df.rename(columns={
            '合同编号': 'contract_number',
            '物料编码': 'material_code',
            '供应商编码': 'supplier_code',
            '合同单价': 'unit_price',
            '生效日期': 'price_date'
        }, inplace=True)
    
    # 统一价格单位（假设所有价格已转换为“元/基本单位”）
    df['unit_price'] = df['unit_price'].astype(float)
    
    return df[['material_code', 'supplier_code', 'unit_price', 'price_date', '数据源', '采集时间']]

# 示例调用
quotation_df = collect_price_data('quotation', './data/供应商报价单_2024.xlsx')
po_df = collect_price_data('purchase_order', './data/历史采购订单_2024.csv')
contract_df = collect_price_data('contract', './data/合同价格表_2024.xlsx')

# 合并到统一数据湖
price_data_lake = pd.concat([quotation_df, po_df, contract_df], ignore_index=True)
print(f"数据湖总记录数：{len(price_data_lake)}")