3. 数据准备与预处理：数据采集规范、数据清洗、缺失值处理、异常值检测与处理

数据准备与预处理是额度模型批次偏移根因分析的基石。原始数据通常来自多个业务系统，存在格式不统一、字段缺失、异常跳变等问题。若未经过严格规范的数据采集与清洗，后续的偏移检测与归因分析将失去可信度。本章围绕数据采集规范、数据清洗、缺失值处理、异常值检测与处理四个核心环节，构建标准化的数据预处理流水线。

3.1 数据采集规范

数据采集规范决定了分析数据的质量上限。在额度模型批次偏移场景中，需重点关注以下采集要点：

3.2 数据清洗

数据清洗的目标是消除格式错误、重复记录、逻辑矛盾等脏数据。针对额度模型批次偏移分析，清洗流程应包含以下步骤：

1. 格式标准化：统一日期格式（YYYY-MM-DD HH:mm:ss）、金额单位（分/元）、分类编码（如性别统一为 0/1）。
2. 重复记录去重：基于业务主键（如申请单号+模型版本）检测完全重复与近似重复，保留最新时间戳的记录。
3. 逻辑校验：检查字段间逻辑一致性，如“年龄 < 18 且职业为退休”属于矛盾数据，需标记或剔除。
4. 范围约束：对数值型字段设定业务合理范围（如额度申请金额 0~10,000,000），超出范围的数据需人工复核。

3.3 缺失值处理

缺失值在批次偏移分析中具有双重影响：一方面，缺失率本身的时序变化可能是批次偏移的信号；另一方面，缺失值填充方式会影响特征分布统计量。处理策略需区分缺失机制：

3.4 异常值检测与处理

异常值检测在批次偏移场景中需区分两类异常：全局异常（偏离整体分布）与局部异常（偏离近期批次分布）。后者往往是批次偏移的早期信号。

下图展示了异常值检测与处理的完整流程：

异常值处理策略需结合业务场景：

• 业务规则优先：如额度申请金额为负数，直接判定为异常并剔除。
• 统计阈值法：使用 Z-Score（|Z| > 3）或 IQR（Q1-1.5IQR, Q3+1.5IQR）标记极端值。
• 模型集成法：孤立森林（Isolation Forest）与局部异常因子（LOF）结合，降低单一方法误判率。
• 时序滑动窗口：以近 7 天或 30 天为窗口，计算当前批次各特征分位数，若超出历史窗口的 3 倍 IQR 范围，标记为批次偏移候选异常。

3.5 数据预处理流水线代码示例

以下代码展示了使用 Python 实现的数据预处理流水线核心逻辑：

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest

def preprocess_batch(df, batch_id, config):
    """
    额度模型批次数据预处理流水线
    :param df: 原始批次 DataFrame
    :param batch_id: 批次标识
    :param config: 配置字典，包含字段映射、阈值等
    :return: 清洗后的 DataFrame 与质量报告
    """
    report = {'batch_id': batch_id, 'total_rows': len(df)}
    
    # 1. 格式标准化
    df['apply_date'] = pd.to_datetime(df['apply_date'], format='%Y/%m/%d')
    df['income'] = df['income'].str.replace(',', '').astype(float) * 100  # 元转分
    
    # 2. 重复记录去重
    before_dedup = len(df)
    df = df.drop_duplicates(subset=['apply_id', 'model_version'], keep='last')
    report['dup_removed'] = before_dedup - len(df)
    
    # 3. 逻辑校验
    mask = (df['age'] < 18) & (df['occupation'] == '退休')
    df = df[~mask]
    report['logic_error'] = mask.sum()
    
    # 4. 缺失值处理（保留缺失指示变量）
    for col in ['income', 'credit_score']:
        df[f'{col}_missing'] = df[col].isna().astype(int)
        df[col] = df[col].fillna(df[col].median())
    
    # 5. 异常值检测（孤立森林）
    numeric_cols = ['income', 'credit_score', 'age', 'loan_amount']
    iso_forest = IsolationForest(contamination=0.05, random_state=42)
    outliers = iso_forest.fit_predict(df[numeric_cols])
    df['is_outlier'] = (outliers == -1).astype(int)
    report['outlier_ratio'] = df['is_outlier'].mean()
    
    # 6. 批次偏移预警
    if report['outlier_ratio'] > 0.1:
        report['shift_warning'] = True
        report['action'] = '标记为疑似偏移批次，进入根因分析'
    else:
        report['shift_warning'] = False
        report['action'] = '正常批次，进入建模流程'
    
    return df, report