4. 数据采集与清洗：成本历史数据的来源、数据质量检查、缺失值处理、异常值检测与处理

在智能成本估算接口的调试与模型训练中，数据质量直接决定估算精度。本章聚焦于成本历史数据的全生命周期处理：从多源采集、质量检查，到缺失值与异常值的系统性修复。只有经过严格清洗的数据，才能为参数化估算模型提供可靠的训练基础。

4.1 成本历史数据的来源与采集策略

成本历史数据通常分散在多个异构系统中，采集时需兼顾覆盖率与一致性。常见来源包括：

4.2 数据质量检查：六维评估模型

在清洗之前，必须对原始数据进行系统性质量评估。推荐使用以下六维检查框架：

1. 完整性：检查必填字段（如项目ID、成本类型、金额）是否存在空值。
2. 准确性：抽样验证数值是否在合理范围内（如人月单价不应为负数）。
3. 一致性：同一项目在不同系统中的成本总额是否匹配。
4. 时效性：数据时间戳是否在预期窗口内，是否存在过期数据。
5. 唯一性：检查是否存在重复记录（如同一笔成本被多次录入）。
6. 规范性：字段格式是否符合约定标准（如日期格式统一为 YYYY-MM-DD）。

# 数据质量检查示例（Python 伪代码）
def quality_check(df):
    report = {}
    report['completeness'] = df[['project_id', 'cost_amount']].isnull().sum()
    report['accuracy'] = (df['cost_amount'] < 0).sum()
    report['uniqueness'] = df.duplicated(subset=['project_id', 'cost_date']).sum()
    return report

4.3 缺失值处理策略

成本数据中的缺失值通常由系统对接遗漏或人工录入疏忽导致。处理策略需根据缺失率与字段重要性分级：

# 缺失值处理示例
import pandas as pd
import numpy as np

def handle_missing(df, strategy='median'):
    # 数值型字段：中位数填充
    num_cols = df.select_dtypes(include=[np.number]).columns
    for col in num_cols:
        if df[col].isnull().mean() < 0.2:
            df[col].fillna(df[col].median(), inplace=True)
    # 类别型字段：众数填充
    cat_cols = df.select_dtypes(include=['object']).columns
    for col in cat_cols:
        if df[col].isnull().mean() < 0.15:
            df[col].fillna(df[col].mode()[0], inplace=True)
    return df

4.4 异常值检测与处理

成本数据中的异常值可能源于录入错误、系统故障或真实极端事件（如突发采购）。检测方法需结合统计与业务规则：

4.4.1 常用检测方法对比

4.4.2 处理流程

1. 标记：使用至少两种方法交叉验证，标记为“疑似异常”。
2. 核实：回源系统确认是否为真实异常（如单据错误）。
3. 处理：确认异常则修正或删除；真实极端值则保留并记录。
4. 记录：将异常处理日志写入元数据表，供模型审计使用。

# 异常值检测示例（IQR 方法）
def detect_outliers_iqr(df, column):
    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)]
    return outliers, lower_bound, upper_bound

4.5 清洗后的数据验证与版本管理

清洗完成后，必须进行数据验证以确保清洗逻辑未引入系统性偏差：

• 分布对比：绘制清洗前后的成本分布直方图，检查均值与标准差变化是否在可接受范围（通常 < 5%）。
• 业务规则校验：随机抽取 50 条记录，人工核对清洗结果是否符合业务逻辑。
• 版本标记：每次清洗生成一个数据版本号（如 v1.2.0），并记录清洗规则快照。

# 数据版本管理示例
data_version = {
    "version": "1.2.0",
    "timestamp": "2025-03-21T14:30:00Z",
    "source_count": 12500,
    "after_clean_count": 11820,
    "rules_applied": ["median_fill_cost", "iqr_outlier_removal", "business_rule_unit_price"]
}