模型预测结果分布漂移检测

模型预测结果分布漂移检测

核心监控指标

• KS统计量：Kolmogorov-Smirnov距离，衡量预测概率分布与历史分布的差异
• JS散度：Jensen-Shannon散度，计算预测结果与基准分布的相对熵差异
• 特征分布均值偏移：各特征在不同时间段的均值变化率

告警配置方案

import numpy as np
from scipy import stats

class DistributionDriftDetector:
    def __init__(self, threshold=0.1):
        self.threshold = threshold
        self.reference_dist = None
        
    def update_reference(self, predictions):
        self.reference_dist = predictions
        
    def detect_drift(self, current_predictions):
        if self.reference_dist is None:
            return False, 0.0
        
        # KS检验
        ks_stat, ks_pvalue = stats.ks_2samp(self.reference_dist, current_predictions)
        
        # JS散度计算
        js_div = self._calculate_js_divergence(current_predictions)
        
        drift_score = max(ks_stat, js_div)
        
        # 告警阈值设置
        if drift_score > self.threshold:
            return True, drift_score
        return False, drift_score
    
    def _calculate_js_divergence(self, predictions):
        # 简化实现：计算与参考分布的JS散度
        return np.mean(np.abs(predictions - np.mean(self.reference_dist)))

复现步骤

1. 收集模型历史预测结果作为基准分布
2. 配置每小时执行一次漂移检测任务
3. 当KS统计量>0.15或JS散度>0.05时触发告警
4. 告警通知通过Slack/钉钉推送至DevOps团队