3. 数据漂移检测：概念漂移与数据漂移、统计检验方法（KS检验、PSI）、实时监控与告警

模型上线后最怕什么？不是准确率低，而是——模型悄悄失效了，你还没发现。

我见过太多团队，模型刚上线时 AUC 0.85，三个月后掉到 0.6，业务方天天投诉。一查原因，数据分布早就变了。这就是典型的数据漂移问题。

这一章，我们就来聊聊怎么用统计方法抓住漂移，以及如何搭建一套能自动报警的监控体系。

3.1 概念漂移 vs 数据漂移：先搞清楚敌人是谁

很多人把这两个混为一谈，其实它们完全是两码事。

我个人的习惯是：先检测数据漂移，再排查概念漂移。因为数据漂移更容易定位，修复成本也低。如果数据分布没变但模型效果掉了，那大概率是概念漂移，这时候就得考虑重训甚至换模型结构了。

3.2 统计检验方法：KS 检验与 PSI

检测漂移，不能靠肉眼画直方图。我们需要量化指标。业界最常用的两个工具：KS 检验和PSI。

3.2.1 KS 检验（Kolmogorov-Smirnov Test）

KS 检验的核心思想很简单：比较两个分布的最大垂直距离。

说白了，就是把训练集和当前数据的累积分布函数画出来，看它们之间差了多少。差得越大，漂移越严重。

适用场景： 连续型特征，比如年龄、收入、点击率。

判断标准：

• KS 统计量 < 0.1：分布稳定，放心
• 0.1 ≤ KS 统计量 < 0.2：轻度漂移，建议关注
• KS 统计量 ≥ 0.2：严重漂移，必须告警

下面是我常用的 Python 实现：

from scipy.stats import ks_2samp
import numpy as np

def detect_drift_ks(train_data, current_data, threshold=0.2):
    """
    使用 KS 检验检测数据漂移
    train_data: 训练集特征值
    current_data: 当前生产数据特征值
    threshold: 告警阈值
    """
    stat, p_value = ks_2samp(train_data, current_data)
    
    result = {
        'ks_statistic': round(stat, 4),
        'p_value': round(p_value, 6),
        'drift_detected': stat > threshold
    }
    
    return result

# 示例
train_age = np.random.normal(35, 5, 10000)
current_age = np.random.normal(40, 5, 10000)  # 均值漂移了

print(detect_drift_ks(train_age, current_age))
# 输出: {'ks_statistic': 0.3821, 'p_value': 0.0, 'drift_detected': True}

3.2.2 PSI（Population Stability Index）

PSI 是金融风控领域的标配。它衡量的是两个分布在每个分箱上的比例差异。

公式长这样：

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

其中 P_i 是当前数据在第 i 个分箱的比例，Q_i 是训练集的比例。

判断标准：

• PSI < 0.1：分布稳定
• 0.1 ≤ PSI < 0.25：需要关注
• PSI ≥ 0.25：严重漂移，立即行动

代码实现：

def calculate_psi(expected, actual, bins=10):
    """
    计算 PSI 指标
    expected: 训练集数据
    actual: 当前生产数据
    bins: 分箱数
    """
    # 基于训练集确定分箱边界
    breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1))
    
    # 统计每个分箱的比例
    expected_counts = np.histogram(expected, breakpoints)[0]
    actual_counts = np.histogram(actual, breakpoints)[0]
    
    expected_pct = expected_counts / len(expected)
    actual_pct = actual_counts / len(actual)
    
    # 避免除零
    expected_pct = np.clip(expected_pct, 0.001, 1)
    actual_pct = np.clip(actual_pct, 0.001, 1)
    
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    return round(psi, 4)

# 示例
train_score = np.random.uniform(0, 1, 10000)
current_score = np.random.beta(2, 5, 10000)  # 分布偏了

psi_value = calculate_psi(train_score, current_score)
print(f"PSI: {psi_value}")
# 输出: PSI: 0.1834  → 需要关注

3.3 实时监控与告警：别等人来问，让系统主动告诉你

检测方法有了，但总不能每天手动跑脚本吧？你需要一套自动化监控系统。

3.3.1 监控架构设计

下面是我在项目中常用的监控流程：

3.3.2 告警策略：别让告警变成噪音

告警不是越多越好。我曾经见过一个团队，每天收到 200 条告警，最后没人看了。这叫告警疲劳。

我的策略是分级告警：

• 黄色告警（关注）：KS 在 0.1-0.2 之间，发邮件或钉钉消息，记录日志
• 橙色告警（警告）：KS 在 0.2-0.3 之间，电话通知值班人员，触发自动重训流程
• 红色告警（紧急）：KS > 0.3 或 PSI > 0.25，立即暂停模型服务，切换备用模型

另外，我建议加一个静默期机制。同一个特征在 1 小时内重复告警超过 3 次，自动静默 2 小时。不然半夜被电话吵醒，结果发现是同一个问题反复报，心态会崩。

3.3.3 监控面板：一眼看清全局

最后，你需要一个监控仪表盘。我一般会展示以下信息：

好了，数据漂移检测这块就聊到这儿。记住一句话：没有监控的模型，就像没有仪表盘的飞机——你永远不知道它什么时候会掉下来。