📚 课程目录第4章数据清洗与预处理

4. 数据清洗与预处理：缺失值填充、异常值检测与数据标准化

数据清洗与预处理是现金流预测模型与发票对账流程中决定成败的基础环节。原始财务数据通常存在三个核心问题：缺失值（如未录入的银行流水）、异常值（如误记的发票金额）以及量纲不一致（如“元”与“万元”混用）。本章将系统讲解这三类问题的处理策略，确保后续模型训练与对账逻辑的准确性。

实战原则： 清洗策略必须与业务场景绑定。例如，银行流水缺失应优先使用前向填充（时间序列特性），而发票金额异常则需结合 IQR 与 Z-score 双重验证。

4.1 缺失值填充策略

缺失值处理没有万能公式，需根据数据分布与业务含义选择策略。下表对比了三种最常用的填充方法及其适用场景：

填充策略	适用场景	优点	缺点	现金流/发票典型用例
均值填充	数据近似正态分布，缺失率低（<5%）	计算简单，保持整体均值	降低方差，扭曲分布形态	月度水电费均值填充
中位数填充	数据偏态分布，存在离群值	抗离群值，稳健性强	忽略尾部信息	发票金额（长尾分布）
前向填充（ffill）	时间序列数据，缺失值前后相关	保留时间趋势，符合业务逻辑	无法处理连续缺失	银行日终余额、应收账款账龄

  代码实战： 以下 Python 代码演示了在 Pandas DataFrame 中应用三种填充策略，并对比填充前后的统计量变化。

import pandas as pd
import numpy as np

# 模拟现金流数据（含缺失值）
np.random.seed(42)
data = {
    'date': pd.date_range('2024-01-01', periods=10, freq='D'),
    'daily_balance': [120.5, 118.3, np.nan, 122.1, 119.8, np.nan, 125.0, 124.2, 121.7, np.nan],
    'invoice_amount': [1500, 2300, 1800, np.nan, 2100, 1950, np.nan, 2200, 1750, 2000]
}
df = pd.DataFrame(data)

# 策略1：均值填充（仅用于演示，实际需按列处理）
df['balance_mean'] = df['daily_balance'].fillna(df['daily_balance'].mean())

# 策略2：中位数填充
df['balance_median'] = df['daily_balance'].fillna(df['daily_balance'].median())

# 策略3：前向填充（时间序列推荐）
df['balance_ffill'] = df['daily_balance'].fillna(method='ffill')

print("原始数据与填充结果对比：")
print(df[['date', 'daily_balance', 'balance_mean', 'balance_median', 'balance_ffill']])

4.2 异常值检测：IQR 与 Z-score 双引擎

发票对账中最常见的异常包括：金额录入多一位零、重复记账、跨月冲销未标记。我们采用两种互补的统计方法进行检测。

4.2.1 IQR（四分位距法）

IQR 不依赖数据分布假设，适合偏态明显的发票金额数据。异常值定义为低于 Q1 - 1.5×IQR 或高于 Q3 + 1.5×IQR 的值。

4.2.2 Z-score（标准分数法）

Z-score 假设数据近似正态分布，通常以 |Z| > 3 作为异常阈值。适用于银行流水等相对对称的指标。

检测方法	分布假设	阈值规则	对离群值敏感度	推荐场景
IQR	无（非参数）	Q1 - 1.5×IQR / Q3 + 1.5×IQR	中等	发票金额、应收账款
Z-score	正态分布	\|Z\| > 3	高（易受极端值影响）	银行流水、日交易量

避坑提示： 不要直接删除所有异常值！在发票对账中，异常值可能是“大额真实交易”或“跨月冲销”。建议先标记，再由人工复核。

from scipy import stats

# IQR 检测
Q1 = df['invoice_amount'].quantile(0.25)
Q3 = df['invoice_amount'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
iqr_outliers = (df['invoice_amount'] < lower_bound) | (df['invoice_amount'] > upper_bound)

# Z-score 检测
z_scores = np.abs(stats.zscore(df['daily_balance'].dropna()))
z_outliers = z_scores > 3

print(f"IQR 检测到的异常索引: {df[iqr_outliers].index.tolist()}")
print(f"Z-score 检测到的异常索引: {df[z_outliers].index.tolist() if len(z_outliers) > 0 else '无'}")

4.3 数据标准化：消除量纲影响

现金流预测模型（如 LSTM、XGBoost）对输入特征的尺度敏感。当“银行余额”以元为单位（数值在 10⁵~10⁷ 之间）而“利率”以百分比表示（0.01~0.05）时，模型会天然偏向大数值特征。标准化是解决这一问题的关键步骤。

4.3.1 Z-score 标准化（StandardScaler）

将数据转换为均值为 0、标准差为 1 的分布。公式：z = (x - μ) / σ。适用于数据近似正态分布的场景。

4.3.2 Min-Max 归一化（MinMaxScaler）

将数据缩放到 [0, 1] 区间。公式：x_norm = (x - min) / (max - min)。适用于有明确边界的数据（如百分比、评分）。

标准化方法	输出范围	对离群值敏感度	保留分布形状	推荐模型场景
Z-score	无固定范围（通常 -3~3）	高	是	线性回归、SVM、PCA
Min-Max	[0, 1]	极高（受 max/min 影响）	否（压缩尾部）	神经网络、KNN、深度学习

from sklearn.preprocessing import StandardScaler, MinMaxScaler

# 准备数据（假设已填充缺失值）
balance_data = df['balance_ffill'].values.reshape(-1, 1)

# Z-score 标准化
scaler_std = StandardScaler()
balance_std = scaler_std.fit_transform(balance_data)

# Min-Max 归一化
scaler_mm = MinMaxScaler()
balance_mm = scaler_mm.fit_transform(balance_data)

print("原始数据前5个值:", balance_data[:5].flatten())
print("Z-score 标准化后:", balance_std[:5].flatten().round(3))
print("Min-Max 归一化后:", balance_mm[:5].flatten().round(3))

4.4 清洗流水线：从原始数据到建模就绪

在实际项目中，上述步骤需要串联成自动化流水线。以下是一个完整的 Python 函数，整合了缺失值填充、异常值标记与标准化：

def clean_cashflow_pipeline(df, fill_strategy='ffill', outlier_method='iqr', scale_method='standard'):
    """
    现金流数据清洗流水线
    参数:
        df: 包含 'daily_balance' 和 'invoice_amount' 列的 DataFrame
        fill_strategy: 'mean' / 'median' / 'ffill'
        outlier_method: 'iqr' / 'zscore'
        scale_method: 'standard' / 'minmax'
    返回:
        清洗后的 DataFrame 与异常标记列
    """
    df_clean = df.copy()

    # 步骤1：缺失值填充
    if fill_strategy == 'mean':
        df_clean['daily_balance'] = df_clean['daily_balance'].fillna(df_clean['daily_balance'].mean())
        df_clean['invoice_amount'] = df_clean['invoice_amount'].fillna(df_clean['invoice_amount'].mean())
    elif fill_strategy == 'median':
        df_clean['daily_balance'] = df_clean['daily_balance'].fillna(df_clean['daily_balance'].median())
        df_clean['invoice_amount'] = df_clean['invoice_amount'].fillna(df_clean['invoice_amount'].median())
    else:  # ffill
        df_clean['daily_balance'] = df_clean['daily_balance'].fillna(method='ffill')
        df_clean['invoice_amount'] = df_clean['invoice_amount'].fillna(method='ffill')

    # 步骤2：异常值检测（标记，不删除）
    df_clean['is_outlier'] = False
    if outlier_method == 'iqr':
        for col in ['daily_balance', 'invoice_amount']:
            Q1 = df_clean[col].quantile(0.25)
            Q3 = df_clean[col].quantile(0.75)
            IQR = Q3 - Q1
            lower = Q1 - 1.5 * IQR
            upper = Q3 + 1.5 * IQR
            df_clean['is_outlier'] |= (df_clean[col] < lower) | (df_clean[col] > upper)
    else:  # zscore
        for col in ['daily_balance', 'invoice_amount']:
            z = np.abs(stats.zscore(df_clean[col]))
            df_clean['is_outlier'] |= (z > 3)

    # 步骤3：标准化（仅对非异常值计算参数，避免污染）
    from sklearn.preprocessing import StandardScaler, MinMaxScaler
    scaler = StandardScaler() if scale_method == 'standard' else MinMaxScaler()
    for col in ['daily_balance', 'invoice_amount']:
        # 用非异常值拟合 scaler
        mask = ~df_clean['is_outlier']
        scaler.fit(df_clean.loc[mask, [col]])
        df_clean[f'{col}_scaled'] = scaler.transform(df_clean[[col]])

    return df_clean

# 执行流水线
df_cleaned = clean_cashflow_pipeline(df, fill_strategy='ffill', outlier_method='iqr', scale_method='standard')
print("清洗完成，数据形状:", df_cleaned.shape)
print("异常值数量:", df_cleaned['is_outlier'].sum())

  关键要点回顾：
  缺失值填充：时间序列用前向填充，偏态分布用中位数，正态分布用均值。
异常值检测：IQR 适合发票金额（偏态），Z-score 适合银行流水（对称）。
数据标准化：Z-score 保留分布形状，Min-Max 适合有界数据；标准化参数应基于非异常值计算。
流水线设计：将清洗步骤封装为可复用函数，确保模型训练与预测时使用相同的参数。