4. 特征分布漂移检测：基于KS检验、PSI、JS散度的特征分布差异量化方法

在金融风控场景中，模型上线后，特征分布随时间或客群变化而产生的漂移是导致模型失效的核心原因。本节系统介绍三种最常用的分布差异量化方法：KS检验（Kolmogorov-Smirnov Test）、PSI（Population Stability Index，群体稳定性指标）和JS散度（Jensen-Shannon Divergence）。三者分别从统计假设检验、分箱离散化、信息论距离三个视角衡量分布偏移程度。

4.1 三种方法的原理与适用场景

4.2 KS检验：连续分布的最大差异度量

KS统计量定义为参考分布（训练集）与当前分布（生产集）的ECDF之间的最大绝对差：

D = max |F_ref(x) - F_cur(x)|

其中：
- F_ref(x) = 训练样本中特征值 ≤ x 的比例
- F_cur(x) = 生产样本中特征值 ≤ x 的比例
- D ∈ [0, 1]，D 越大表示分布差异越大

在金融特征平台中，通常设定阈值：D < 0.1 视为无漂移；0.1 ≤ D < 0.2 为轻度漂移，需关注；D ≥ 0.2 为严重漂移，触发告警。

4.3 PSI：金融行业标准稳定性指标

PSI的计算分为三步：分箱、计算占比、加权求和。公式如下：

PSI = Σ (p_cur_i - p_ref_i) * ln(p_cur_i / p_ref_i)

其中：
- p_ref_i = 参考样本在第 i 个分箱中的占比
- p_cur_i = 当前样本在第 i 个分箱中的占比
- 分箱数通常取 10（等频分箱）或按业务规则划分

PSI的行业判读标准：

• PSI < 0.1：无显著漂移，特征稳定
• 0.1 ≤ PSI < 0.25：轻度漂移，需持续监控
• PSI ≥ 0.25：严重漂移，需立即排查并考虑特征重构或模型重训

4.4 JS散度：对称且有界的信息论距离

JS散度基于KL散度改进，解决了KL散度非对称且无上界的问题：

JS(P || Q) = 0.5 * KL(P || M) + 0.5 * KL(Q || M)

其中 M = 0.5 * (P + Q)，KL(P || Q) = Σ p_i * ln(p_i / q_i)

JS散度取值范围：[0, ln2]（对数底为e时）或 [0, 1]（对数底为2时）

JS散度天然对称，且当两个分布完全相同时为0，完全不相交时为ln2。在金融特征平台中，JS散度常用于：

• 多模态特征（如收入分布存在多个峰值）的漂移检测
• 类别不平衡特征（如违约率极低的二值特征）的分布对比
• 需要跨特征统一阈值管理的场景

4.5 三种方法对比总结

4.6 特征平台中的工程化实现要点

在金融特征平台中落地分布漂移检测，需关注以下工程细节：

1. 参考窗口与检测窗口：参考窗口通常取模型训练集最近3~6个月数据；检测窗口取最近1周或1个月的生产数据，按日/周滚动计算。
2. 分箱策略统一：PSI和JS散度依赖分箱，需在特征上线时固化分箱边界（基于训练集），避免每次计算时重新分箱导致结果不可比。
3. 缺失值处理：缺失值应单独作为一个分箱参与计算，或明确排除并在文档中标注。
4. 阈值动态调整：固定阈值（如PSI=0.25）可能不适应所有特征，建议对每个特征基于历史数据计算基线均值和标准差，采用“均值+3σ”作为动态阈值。

# Python伪代码：特征平台分布漂移检测流水线
def compute_drift(feature_name, ref_data, cur_data, method='psi', bins=10):
    if method == 'ks':
        from scipy.stats import ks_2samp
        stat, p_value = ks_2samp(ref_data, cur_data)
        return {'statistic': stat, 'p_value': p_value}
    elif method == 'psi':
        # 使用预定义分箱边界
        bin_edges = get_predefined_bins(feature_name)
        ref_hist = np.histogram(ref_data, bins=bin_edges)[0] / len(ref_data)
        cur_hist = np.histogram(cur_data, bins=bin_edges)[0] / len(cur_data)
        psi = np.sum((cur_hist - ref_hist) * np.log(cur_hist / ref_hist + 1e-10))
        return {'psi': psi}
    elif method == 'js':
        # 使用核密度估计或分箱后计算
        p = np.histogram(ref_data, bins=bins, density=True)[0]
        q = np.histogram(cur_data, bins=bins, density=True)[0]
        m = 0.5 * (p + q)
        js = 0.5 * (np.sum(p * np.log(p / m + 1e-10)) + np.sum(q * np.log(q / m + 1e-10)))
        return {'js_divergence': js}