3、数据漂移检测实战：基于特征分布的单变量检测方法

说实话，很多团队做模型监控，第一步就卡在「到底该用什么指标」上。我见过不少项目，一上来就上深度学习做分布对比，结果数据量不够、解释性又差，最后连业务方都看不懂报告。

其实，单变量检测方法才是工业界的基石。KS检验和PSI，这两个东西你搞懂了，80%的漂移场景都能覆盖。今天我就带你手撸一遍，顺便聊聊我踩过的坑。

3.1 为什么先做单变量？

多变量检测听起来高大上，但落地时有个致命问题：可解释性差。你告诉业务方「模型特征空间发生了漂移」，人家问「哪个特征？漂了多少？」，你答不上来，那就尴尬了。

单变量检测就不一样。每个特征单独算分，哪个特征漂了、漂了多少、什么时候开始的，一目了然。我个人的习惯是：先单变量扫一遍，发现问题再上多变量。这样效率最高。

3.2 KS检验：两样本分布是否一致？

KS检验，全称Kolmogorov-Smirnov检验。说白了，就是比较两个样本的累积分布函数（CDF）的最大垂直距离。距离越大，分布差异越大。

公式很简单：

D = max |F1(x) - F2(x)|

其中F1是参考分布的CDF，F2是当前分布的CDF。D值在0到1之间，越接近1说明漂移越严重。

3.2.1 什么时候用KS？

• 连续型特征：比如年龄、收入、点击率等数值型变量
• 样本量较大：一般要求每组样本量>50，否则统计功效不足
• 分布形状敏感：KS对分布的位置、离散度、形状变化都敏感

3.2.2 Python实现

直接用scipy，几行代码搞定：

from scipy.stats import ks_2samp
import numpy as np

def ks_detection(reference, current, threshold=0.1):
    """
    单特征KS检验漂移检测
    :param reference: 参考分布（训练集）
    :param current: 当前分布（生产数据）
    :param threshold: D值阈值
    :return: (是否漂移, D值, p值)
    """
    stat, p_value = ks_2samp(reference, current)
    is_drift = stat > threshold
    return is_drift, stat, p_value

# 示例
ref_data = np.random.normal(0, 1, 10000)
cur_data = np.random.normal(0.2, 1.1, 10000)

drifted, d_val, p_val = ks_detection(ref_data, cur_data)
print(f"漂移状态: {drifted}, D值: {d_val:.4f}, p值: {p_val:.4f}")

你看，就这么简单。但实际落地时，有个坑我必须提醒你：

3.3 PSI：群体稳定性指标

PSI（Population Stability Index）是金融风控领域的标配。它衡量的是两个分布在各分箱上的占比差异。公式如下：

PSI = Σ (P_i - Q_i) * ln(P_i / Q_i)

其中P_i是当前分布在第i个分箱的占比，Q_i是参考分布在第i个分箱的占比。

3.3.1 PSI的解读标准

PSI值	含义	建议操作
< 0.1	无显著漂移	无需处理
0.1 - 0.25	轻度漂移	需关注，建议排查
> 0.25	严重漂移	立即告警，触发模型重训

这个标准是行业共识，但别死板套用。我在做信贷风控时，PSI阈值设到0.2就告警了，因为金融场景对稳定性要求极高。但在推荐系统里，0.3以内我都觉得正常——用户行为本来就在变。

3.3.2 Python实现

PSI的实现核心是分箱。我习惯用等频分箱，因为等距分箱容易导致某些箱子样本太少：

import numpy as np
import pandas as pd

def psi_detection(reference, current, bins=10, threshold=0.25):
    """
    单特征PSI漂移检测
    :param reference: 参考分布
    :param current: 当前分布
    :param bins: 分箱数
    :param threshold: PSI阈值
    :return: (是否漂移, PSI值)
    """
    # 基于参考分布确定分箱边界（等频分箱）
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(reference, percentiles)
    bin_edges[0] = -np.inf
    bin_edges[-1] = np.inf
    
    # 计算各箱占比
    ref_counts = np.histogram(reference, bins=bin_edges)[0]
    cur_counts = np.histogram(current, bins=bin_edges)[0]
    
    ref_pct = ref_counts / len(reference)
    cur_pct = cur_counts / len(current)
    
    # 处理零值（加极小值避免除零）
    ref_pct = np.where(ref_pct == 0, 1e-6, ref_pct)
    cur_pct = np.where(cur_pct == 0, 1e-6, cur_pct)
    
    # 计算PSI
    psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
    
    is_drift = psi > threshold
    return is_drift, psi

# 示例
ref_data = np.random.normal(0, 1, 10000)
cur_data = np.random.normal(0.5, 1.2, 10000)

drifted, psi_val = psi_detection(ref_data, cur_data)
print(f"漂移状态: {drifted}, PSI值: {psi_val:.4f}")

3.4 KS vs PSI：怎么选？

很多初学者会问：到底用KS还是PSI？我的回答是：两个都用，但看场景。

对比维度	KS检验	PSI
统计原理	基于CDF最大距离	基于分箱占比差异
对分布形状敏感度	高	中
对样本量敏感度	高（p值受样本量影响大）	低（占比计算相对稳定）
可解释性	中（需要理解CDF）	高（业务方容易理解占比变化）
适用特征类型	连续型	连续型、离散型均可
工业界普及度	高（学术界常用）	极高（金融风控标配）

我个人习惯是：连续特征两个都算，取告警更严格的那个。离散特征只用PSI。这样既不会漏报，也不会因为单一指标的缺陷而误报。

3.5 知识体系总览

下面这张图，是我做漂移检测系统时的核心逻辑。你看一遍，基本就知道整个单变量检测的流程了：

3.6 落地时的三个关键点

代码写完了，理论也懂了，但落地时还有几个坑。我直接给你列出来：

1. 参考分布怎么选？ 我建议用最近一周或一个月的生产数据作为参考，而不是训练集。因为训练集可能已经过时了。每周更新一次参考分布，这样能捕捉到缓慢漂移。
2. 阈值怎么调？ 别用默认值。我一般会先跑一周的历史数据，统计PSI和KS的百分位数，然后取P95或P99作为阈值。这样能自适应你的业务场景。
3. 告警后怎么办？ 漂移检测只是第一步。告警后要自动触发根因分析：哪个特征漂了？漂的方向是什么？对模型输出有多大影响？这些都要串联起来。