4. 数据采集与清洗

在价格偏差分析驱动的库存动态优化中，数据质量直接决定模型输出与现场执行效果。本章聚焦于如何从现场环境中识别可靠数据源、选择适配的采集方法，并通过系统化的清洗与预处理流程，将原始数据转化为可用于偏差计算与优化决策的高质量数据集。

4.1 现场库存数据源识别

现场库存数据通常分散在多个异构系统中，识别并归类这些数据源是采集工作的第一步。根据数据产生与存储方式，可将数据源分为以下三类：

4.2 数据采集方法

根据数据源的接口能力与现场网络条件，选择以下一种或组合采集方法：

• API/Web Service 直连：适用于ERP、WMS等现代系统，通过RESTful API或SOAP接口定时拉取增量数据，延迟低、结构稳定。
• 数据库中间表/视图：在源系统侧建立只读视图或中间表，由采集程序按调度周期（如每5分钟）轮询读取，适合对源系统改造受限的场景。
• 消息队列/流式采集：通过Kafka、RabbitMQ等消息中间件订阅实时库存变动事件，适用于IoT设备与高频变动场景。
• 文件导入/ETL批处理：对于遗留系统或第三方平台，通过SFTP接收CSV、XML或EDI文件，按固定批次（如每日凌晨）加载至数据湖。
• 边缘网关采集：在仓库现场部署边缘计算节点，直接读取RFID读写器、智能货架等设备数据，经初步聚合后上传至中心平台。

4.3 数据清洗与预处理流程

原始采集数据通常包含重复、缺失、异常值及格式不一致等问题，必须经过系统化的清洗与预处理才能用于价格偏差计算。标准流程包含以下六个步骤：

1. 数据去重：基于业务主键（如物料编码+库位+批次+时间戳）识别并移除完全重复记录，保留最新一条。
2. 缺失值处理：对于库存数量、库位编码等关键字段缺失的记录，优先通过关联单据补全；无法补全的标记为脏数据并移入异常库。
3. 异常值检测与修正：使用IQR（四分位距）或Z-score方法识别库存数量、价格等字段的离群点，结合业务规则（如库存数量不能为负）进行修正或剔除。
4. 格式标准化：统一日期格式（如YYYY-MM-DD HH:mm:ss）、货币单位（统一为元）、计量单位（统一为千克/个/箱），并对文本字段进行大小写与空格清理。
5. 数据关联与补全：将库存数据与物料主数据、供应商数据、价格表进行左连接，补全物料类别、供应商编码、标准成本等衍生字段。
6. 衍生字段计算：基于清洗后的数据计算库龄（当前时间-入库时间）、周转率（出库量/平均库存）、价格偏差率（(实际价格-标准价格)/标准价格）等分析指标。

4.4 数据质量评估指标

在清洗与预处理完成后，需对数据集进行质量评估，确保其满足价格偏差分析的要求。建议从以下五个维度进行量化评估：

4.5 数据清洗代码示例（Python）

以下代码片段展示了使用Pandas进行库存数据清洗与预处理的典型实现：

import pandas as pd
import numpy as np

# 加载原始库存数据
df = pd.read_csv('raw_inventory.csv', parse_dates=['record_time'])

# 步骤1：基于复合主键去重
df.drop_duplicates(subset=['material_code', 'location', 'batch_no', 'record_time'],
                   keep='last', inplace=True)

# 步骤2：缺失值处理 - 库位编码缺失则标记为脏数据
df['location'].fillna('UNKNOWN', inplace=True)
df['quantity'] = df.groupby('material_code')['quantity'].transform(
    lambda x: x.fillna(x.median()))

# 步骤3：异常值检测 - 库存数量不能为负，且不超过均值+3倍标准差
df = df[(df['quantity'] >= 0) & 
        (df['quantity'] <= df['quantity'].mean() + 3 * df['quantity'].std())]

# 步骤4：格式标准化 - 统一日期与货币单位
df['record_time'] = pd.to_datetime(df['record_time'], format='%Y-%m-%d %H:%M:%S')
df['unit_price'] = df['unit_price'].round(2)

# 步骤5：数据关联 - 关联物料主数据补全类别
master = pd.read_csv('material_master.csv')
df = df.merge(master[['material_code', 'category', 'standard_cost']],
              on='material_code', how='left')

# 步骤6：衍生字段计算
df['stock_age_days'] = (pd.Timestamp.now() - df['record_time']).dt.days
df['price_deviation_rate'] = (df['unit_price'] - df['standard_cost']) / df['standard_cost']

# 输出清洗后数据集
df.to_parquet('cleaned_inventory.parquet', index=False)
print(f"清洗完成，共保留 {len(df)} 条有效记录")