4、模型漂移检测实战：预测分布对比、误差分布分析

好，咱们进入实战环节。前面聊了那么多概念和理论，说白了都是为了这一刻——用代码真正把漂移检测跑起来。

我个人习惯把模型漂移检测分成两条线：一条盯着预测分布，一条盯着误差分布。两条线互相印证，才能下结论说模型到底有没有衰减。

4.1 预测分布对比：你猜模型在想什么？

先问个问题：模型输出的概率分布变了，意味着什么？

举个例子，一个二分类模型上线时对正类的平均预测概率是0.7，三个月后变成了0.55。这说明模型对正类的信心在下降，背后可能是特征分布变了，也可能是数据标注标准变了。

我常用的方法是PSI（Population Stability Index，群体稳定性指标）。它衡量两个分布之间的差异，公式长这样：

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

其中P_i是参考分布（通常是训练集）在第i个分箱的比例，Q_i是当前分布的比例。

嗯，这里要注意：PSI对分箱数量很敏感。我一般分10箱，每箱样本量不少于50。

下面是我常用的PSI计算代码：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10):
    """
    计算群体稳定性指标(PSI)
    
    Parameters:
    -----------
    expected : array-like, 参考分布（训练集预测概率）
    actual : array-like, 当前分布（生产环境预测概率）
    bins : int, 分箱数量
    
    Returns:
    --------
    psi : float, PSI值
    """
    # 基于参考分布确定分箱边界
    breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1))
    breakpoints[0] = breakpoints[0] - 0.001  # 确保最小值落入第一箱
    breakpoints[-1] = breakpoints[-1] + 0.001  # 确保最大值落入最后一箱
    
    # 分箱
    expected_bins = np.digitize(expected, breakpoints, right=True)
    actual_bins = np.digitize(actual, breakpoints, right=True)
    
    psi = 0.0
    for i in range(1, bins + 1):
        expected_pct = np.sum(expected_bins == i) / len(expected)
        actual_pct = np.sum(actual_bins == i) / len(actual)
        
        # 避免除零和取对数零
        expected_pct = max(expected_pct, 0.0001)
        actual_pct = max(actual_pct, 0.0001)
        
        psi += (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    
    return psi

# 使用示例
train_preds = np.random.beta(2, 5, 10000)  # 模拟训练集预测概率
prod_preds = np.random.beta(1.5, 6, 10000)  # 模拟生产环境预测概率

psi_value = calculate_psi(train_preds, prod_preds)
print(f"PSI值: {psi_value:.4f}")

if psi_value < 0.1:
    print("✅ 分布稳定")
elif psi_value < 0.25:
    print("⚠️ 轻度漂移，建议监控")
else:
    print("🚨 严重漂移，必须告警")

4.2 误差分布分析：模型错在哪？

预测分布变了，不一定代表模型变差了。有时候分布变了但准确率没降，说明模型在适应新数据。反过来，分布没变但误差变大了，那才是真问题。

所以我习惯两条线一起看。误差分布分析主要关注三个指标：

指标	含义	漂移信号
MAE	平均误差	持续上升说明模型整体在退化
RMSE	均方根误差	上升幅度大于MAE，说明存在大误差样本
误差分布偏度	误差是否对称	偏度变化说明模型在系统性偏向某类样本

来看代码实现：

def error_distribution_analysis(y_true, y_pred, window_size=1000):
    """
    误差分布分析，检测模型性能衰减
    
    Parameters:
    -----------
    y_true : array-like, 真实值
    y_pred : array-like, 预测值
    window_size : int, 滑动窗口大小
    
    Returns:
    --------
    metrics : dict, 包含MAE、RMSE、偏度等指标
    """
    errors = y_true - y_pred
    
    # 整体指标
    mae = np.mean(np.abs(errors))
    rmse = np.sqrt(np.mean(errors ** 2))
    skewness = np.mean(((errors - np.mean(errors)) / np.std(errors)) ** 3)
    
    # 滑动窗口指标（用于趋势检测）
    n_windows = len(errors) // window_size
    window_mae = []
    window_rmse = []
    
    for i in range(n_windows):
        start = i * window_size
        end = start + window_size
        window_errors = errors[start:end]
        window_mae.append(np.mean(np.abs(window_errors)))
        window_rmse.append(np.sqrt(np.mean(window_errors ** 2)))
    
    # 检测趋势：如果最近窗口的MAE比第一个窗口高20%，认为存在漂移
    drift_detected = False
    if len(window_mae) >= 2:
        baseline_mae = window_mae[0]
        current_mae = window_mae[-1]
        if current_mae > baseline_mae * 1.2:
            drift_detected = True
    
    return {
        'mae': mae,
        'rmse': rmse,
        'skewness': skewness,
        'window_mae': window_mae,
        'window_rmse': window_rmse,
        'drift_detected': drift_detected
    }

# 模拟数据
np.random.seed(42)
y_true = np.random.randn(10000)
y_pred = y_true + np.random.randn(10000) * 0.1  # 初始误差小

# 模拟性能衰减：从第5000个样本开始误差增大
y_pred[5000:] = y_true[5000:] + np.random.randn(5000) * 0.3

result = error_distribution_analysis(y_true, y_pred)
print(f"MAE: {result['mae']:.4f}")
print(f"RMSE: {result['rmse']:.4f}")
print(f"误差偏度: {result['skewness']:.4f}")
print(f"漂移检测: {'检测到漂移' if result['drift_detected'] else '未检测到漂移'}")

4.3 综合监控框架：把两条线串起来

单独看预测分布或误差分布都有盲区。我设计了一个综合监控框架，把两条线合并成一个决策矩阵：

def comprehensive_drift_monitor(train_preds, prod_preds, y_true, y_pred):
    """
    综合漂移监控：预测分布 + 误差分布
    
    Returns:
    --------
    status : str, 模型状态
    details : dict, 详细指标
    """
    # 1. 预测分布对比
    psi = calculate_psi(train_preds, prod_preds)
    
    # 2. 误差分布分析
    error_result = error_distribution_analysis(y_true, y_pred)
    
    # 3. 综合判断
    psi_drift = psi >= 0.1
    error_drift = error_result['drift_detected']
    
    if not psi_drift and not error_drift:
        status = "✅ 模型健康"
    elif psi_drift and not error_drift:
        status = "⚠️ 数据分布变化，但模型仍有效"
    elif not psi_drift and error_drift:
        status = "⚠️ 数据分布未变，但模型性能下降（可能是模型老化）"
    else:
        status = "🚨 双重漂移：数据分布和模型性能均恶化"
    
    return status, {
        'psi': psi,
        'mae': error_result['mae'],
        'rmse': error_result['rmse'],
        'skewness': error_result['skewness'],
        'window_mae_trend': error_result['window_mae'][-5:] if len(error_result['window_mae']) >= 5 else error_result['window_mae']
    }

# 使用示例
status, details = comprehensive_drift_monitor(train_preds, prod_preds, y_true, y_pred)
print(f"模型状态: {status}")
print(f"详细指标: PSI={details['psi']:.4f}, MAE={details['mae']:.4f}")

这个框架我用了两年多，在三个项目里验证过。说白了，它解决了一个核心问题：当告警响起时，你能立刻知道问题出在哪。

4.4 可视化：一张图看懂漂移

光有数字不够，人眼对图形的敏感度远高于数字。我每次做漂移检测报告，都会附上下面这张图：

这张图的核心逻辑是：两条检测线独立运行，结果汇聚到决策矩阵。你不需要记住所有公式，只要看懂这个流程，就能快速定位问题。

4.5 避坑指南：我踩过的那些坑

最后分享几个实战中容易翻车的地方：

1. 数据对齐问题：预测分布对比时，训练集和生产集的时间窗口要对齐。我曾经拿一个月的训练集和一天的生产集对比，PSI直接爆表。后来我规定：参考窗口至少是检测窗口的3倍长度。
2. 冷启动问题：新模型上线时没有足够的历史数据做参考。我的做法是先收集两周数据作为基线，之后再开启漂移检测。
3. 告警疲劳：阈值设得太低，每天收到几十条告警，团队很快就麻木了。我建议告警阈值分三级：黄色（关注）、橙色（调查）、红色（立即行动）。只有红色才发即时通知。
4. 忽略业务上下文：有一次PSI飙到0.3，我正准备发告警，业务同事说「哦，我们昨天改了定价策略」。所以告警一定要关联业务事件日历，避免误报。

嗯，实战部分就到这里。代码可以直接复制到你的项目里跑，有问题随时调。记住：漂移检测不是一次性工作，它是个持续的过程。你跑得越久，对模型的行为就越了解。