模型输入数据异常检测告警系统

模型输入数据异常检测告警系统

核心监控指标

• 数据分布偏移率：使用Kolmogorov-Smirnov检验检测输入特征分布变化，阈值设置为0.05
• 缺失值率：监控关键字段缺失比例，超过5%触发告警
• 异常值检测：基于IQR方法识别数值型特征异常值，异常值占比超过3%时告警
• 数据量波动：每小时数据量变化幅度超过20%进行预警

告警配置方案

import pandas as pd
from scipy import stats
import numpy as np

class InputDataMonitor:
    def __init__(self):
        self.thresholds = {
            'ks_test': 0.05,
            'missing_rate': 0.05,
            'outlier_rate': 0.03,
            'volume_change': 0.20
        }
    
    def detect_anomalies(self, current_data, reference_data):
        # K-S检验
        ks_stat, p_value = stats.ks_2samp(current_data['feature1'], reference_data['feature1'])
        if p_value < self.thresholds['ks_test']:
            self.send_alert('Data distribution shift detected')
        
        # 缺失值检测
        missing_rate = current_data.isnull().sum() / len(current_data)
        if missing_rate.max() > self.thresholds['missing_rate']:
            self.send_alert('High missing rate detected')
        
        # 异常值检测
        Q1 = current_data.quantile(0.25)
        Q3 = current_data.quantile(0.75)
        IQR = Q3 - Q1
        outliers = ((current_data < (Q1 - 1.5 * IQR)) | 
                  (current_data > (Q3 + 1.5 * IQR))).sum() / len(current_data)
        if outliers.max() > self.thresholds['outlier_rate']:
            self.send_alert('High outlier rate detected')

    def send_alert(self, message):
        # Slack告警集成
        print(f"ALERT: {message}")
        # 实际部署时集成Slack/钉钉/Webhook等通知方式

部署步骤

1. 配置监控服务定时任务（每小时）
2. 设置数据采集接口
3. 集成告警通知机制
4. 验证异常检测准确性