3. 数据漂移检测基础：统计检验方法

数据漂移检测的核心在于量化模型上线后输入特征分布与训练期分布之间的差异。本章聚焦三种工业界最常用的统计检验方法：KS 检验（连续特征）、卡方检验（分类特征）和PSI 群体稳定性指标（通用量化指标）。掌握这些方法，是搭建漂移监控系统的第一步。

3.1 三种方法的适用场景与核心差异

3.2 KS 检验（Kolmogorov-Smirnov Test）

KS 检验通过计算训练集与生产集经验累积分布函数（ECDF）之间的最大垂直距离 D 统计量，判断两个样本是否来自同一分布。

计算步骤

1. 将训练集与生产集的特征值合并排序。
2. 在每个数据点计算训练集累积比例 F_train(x) 与生产集累积比例 F_prod(x)。
3. 取绝对值差的最大值：D = max|F_train(x) - F_prod(x)|。
4. 根据样本量查表或计算 p 值，判断是否拒绝原假设。

import numpy as np
from scipy.stats import ks_2samp

# 模拟训练集与生产集特征
train_feature = np.random.normal(0, 1, 10000)
prod_feature  = np.random.normal(0.15, 1.05, 10000)

stat, p_value = ks_2samp(train_feature, prod_feature)
print(f"KS 统计量 D = {stat:.4f}, p 值 = {p_value:.4f}")
# 输出示例: D = 0.0521, p = 0.0000 → 分布存在显著差异

3.3 卡方检验（Chi-Square Test）

卡方检验适用于分类特征，通过比较每个类别在训练集与生产集中的实际频数与期望频数，计算卡方统计量：

χ² = Σ (O_i - E_i)² / E_i

其中：
O_i = 生产集中第 i 类的观测频数
E_i = 训练集中第 i 类的比例 × 生产集总样本数

计算步骤

1. 统计训练集各分类的占比 p_train_i。
2. 计算生产集各分类的期望频数 E_i = p_train_i × N_prod。
3. 代入公式计算 χ² 值，自由度 = 类别数 - 1。
4. 查卡方分布表得到 p 值。

from scipy.stats import chisquare

# 训练集各分类占比
train_pct = [0.30, 0.45, 0.25]  # A/B/C 三类
# 生产集观测频数
prod_obs  = [2800, 4200, 3000]  # 总样本 10000
# 计算期望频数
expected = [p * 10000 for p in train_pct]

stat, p_value = chisquare(prod_obs, f_exp=expected)
print(f"卡方统计量 = {stat:.2f}, p 值 = {p_value:.4f}")

3.4 PSI 群体稳定性指标

PSI 是金融风控领域最广泛使用的分布漂移量化指标，不依赖分布假设，通过分箱后比较各箱占比变化来评估整体偏移程度。

PSI 计算公式

PSI = Σ (P_prod_i - P_train_i) × ln(P_prod_i / P_train_i)

其中：
P_train_i = 训练集在第 i 个箱中的占比
P_prod_i  = 生产集在第 i 个箱中的占比

分箱策略详解

分箱策略直接影响 PSI 的敏感度和稳定性。工业界常用以下三种策略：

PSI 计算完整示例

import numpy as np
import pandas as pd

def calculate_psi(train, prod, bins=10, bucket_type='quantile'):
    """
    计算 PSI 群体稳定性指标
    :param train: 训练集特征数组
    :param prod:  生产集特征数组
    :param bins:  分箱数
    :param bucket_type: 'quantile'(等频) 或 'uniform'(等距)
    :return: PSI 值
    """
    # 1. 基于训练集确定分箱边界
    if bucket_type == 'quantile':
        # 等频分箱：使用分位数
        percentiles = np.linspace(0, 100, bins + 1)
        bin_edges = np.percentile(train, percentiles)
        # 确保边界唯一
        bin_edges = np.unique(bin_edges)
    else:
        # 等距分箱
        bin_edges = np.linspace(train.min(), train.max(), bins + 1)

    # 2. 计算训练集各箱占比
    train_counts, _ = np.histogram(train, bins=bin_edges)
    train_pct = train_counts / len(train)

    # 3. 计算生产集各箱占比（使用相同边界）
    prod_counts, _ = np.histogram(prod, bins=bin_edges)
    prod_pct = prod_counts / len(prod)

    # 4. 处理零占比：用极小值替代避免 ln(0)
    train_pct = np.where(train_pct == 0, 1e-6, train_pct)
    prod_pct  = np.where(prod_pct == 0, 1e-6, prod_pct)

    # 5. 计算 PSI
    psi = np.sum((prod_pct - train_pct) * np.log(prod_pct / train_pct))
    return psi

# 示例调用
train_data = np.random.normal(0, 1, 50000)
prod_data  = np.random.normal(0.2, 1.1, 50000)
psi_value  = calculate_psi(train_data, prod_data, bins=10)
print(f"PSI = {psi_value:.4f}")
# 输出示例: PSI = 0.0872 → 轻度漂移

3.5 PSI 分箱策略对比流程图

3.6 PSI 阈值与告警策略

3.7 三种方法在监控系统中的协作关系