AI模型监控：这次怎么落地的

import numpy as np
from scipy.stats import ks_2samp

# 模拟训练数据和线上数据的特征分布
train_data = np.random.normal(0, 1, 10000)  # 训练数据
online_data = np.random.normal(0.2, 1, 10000)  # 线上数据，均值轻微偏移

# 使用KS检验检测分布差异
statistic, p_value = ks_2samp(train_data, online_data)

print(f"KS统计量: {statistic:.4f}")
print(f"P值: {p_value:.6f}")

if p_value < 0.05:
    print("警告：检测到显著的分布漂移")
else:
    print("分布没有显著漂移")

def monitor_feature_drift(train_features, online_features, threshold=0.1):
    """
    监控特征漂移，返回漂移程度和建议

    Args:
        train_features: 训练数据的特征字典 {feature_name: array}
        online_features: 线上数据的特征字典
        threshold: PSI阈值，超过则告警

    Returns:
        drift_report: 包含每个特征的漂移状态和建议
    """
    from scipy.stats import entropy

    drift_report = {}

    for feature in train_features.keys():
        train_vals = train_features[feature]
        online_vals = online_features[feature]

        # 计算分布（这里简化为直方图）
        train_hist, _ = np.histogram(train_vals, bins=20, density=True)
        online_hist, _ = np.histogram(online_vals, bins=20, density=True)

        # 计算PSI
        psi = np.sum((online_hist - train_hist) * np.log(online_hist / (train_hist + 1e-10) + 1e-10))

        # 判断漂移等级
        if psi < threshold:
            level = "stable"
            suggestion = "正常，无需干预"
        elif psi < threshold * 2.5:
            level = "warning"
            suggestion = "轻微漂移，建议关注"
        else:
            level = "critical"
            suggestion = "严重漂移，考虑重新训练"

        drift_report[feature] = {
            "psi": psi,
            "level": level,
            "suggestion": suggestion
        }

    return drift_report

def monitor_confidence_drift(predictions_history, window=100):
    """
    监控预测置信度的漂移

    Args:
        predictions_history: 历史预测置信度列表
        window: 滑动窗口大小

    Returns:
        drift_signal: 漂移信号强度 (0-1)
    """
    if len(predictions_history) < window * 2:
        return 0  # 数据不足，无法判断

    # 计算最近窗口和历史窗口的统计差异
    recent = predictions_history[-window:]
    historical = predictions_history[-(window*2):-window]

    # 计算均值差异
    mean_diff = abs(np.mean(recent) - np.mean(historical))

    # 计算方差差异
    var_diff = abs(np.var(recent) - np.var(historical))

    # 归一化漂移信号
    drift_signal = min(1.0, (mean_diff + var_diff) / 2.0)

    return drift_signal

class ModelPerformanceMonitor:
    def __init__(self, metrics_window=1000):
        self.metrics_window = metrics_window
        self.predictions = []
        self.labels = []
        self.business_metrics = []

    def record_prediction(self, prediction, label, business_value):
        """
        记录单次预测

        Args:
            prediction: 模型预测值
            label: 真实标签（如果有）
            business_value: 业务指标（点击/转化/订单金额等）
        """
        self.predictions.append(prediction)
        if label is not None:
            self.labels.append(label)
        self.business_metrics.append(business_value)

        # 保持窗口大小
        if len(self.predictions) > self.metrics_window:
            self.predictions.pop(0)
            if self.labels:
                self.labels.pop(0)
            self.business_metrics.pop(0)

    def evaluate(self):
        """
        评估当前窗口的模型性能
        """
        results = {}

        # 离线指标（如果有标签）
        if self.labels:
            from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

            binary_preds = [1 if p > 0.5 else 0 for p in self.predictions]

            results['accuracy'] = accuracy_score(self.labels, binary_preds)
            results['precision'] = precision_score(self.labels, binary_preds, zero_division=0)
            results['recall'] = recall_score(self.labels, binary_preds, zero_division=0)
            results['f1'] = f1_score(self.labels, binary_preds, zero_division=0)

        # 在线业务指标
        if self.business_metrics:
            results['business_conversion'] = np.mean(self.business_metrics)
            results['business_std'] = np.std(self.business_metrics)

        # 预测置信度分布
        results['prediction_mean'] = np.mean(self.predictions)
        results['prediction_std'] = np.std(self.predictions)

        return results

class MultiWindowMonitor:
    def __init__(self):
        self.realtime_data = []  # 1小时窗口
        self.short_term_data = []  # 1天窗口
        self.long_term_data = []   # 7天窗口

    def add_data_point(self, value, timestamp):
        """
        添加数据点，自动分配到不同窗口
        """
        now = datetime.now()

        self.long_term_data.append((value, timestamp))
        if timestamp > now - timedelta(days=1):
            self.short_term_data.append((value, timestamp))
        if timestamp > now - timedelta(hours=1):
            self.realtime_data.append((value, timestamp))

        # 清理过期数据
        self.realtime_data = [(v, t) for v, t in self.realtime_data if t > now - timedelta(hours=1)]
        self.short_term_data = [(v, t) for v, t in self.short_term_data if t > now - timedelta(days=1)]
        self.long_term_data = [(v, t) for v, t in self.long_term_data if t > now - timedelta(days=7)]

    def evaluate(self):
        """
        评估不同窗口的性能
        """
        results = {}

        for name, data in [('realtime', self.realtime_data),
                          ('short_term', self.short_term_data),
                          ('long_term', self.long_term_data)]:
            if not data:
                continue

            values = [v for v, t in data]

            results[f'{name}_mean'] = np.mean(values)
            results[f'{name}_std'] = np.std(values)
            results[f'{name}_min'] = np.min(values)
            results[f'{name}_max'] = np.max(values)

        return results

class ContextAwareAlert:
    def __init__(self, history_window=7):
        self.history_window = history_window
        self.history_data = []

    def add_data_point(self, value, timestamp):
        """
        添加历史数据点
        """
        self.history_data.append((value, timestamp))

        # 清理过期数据
        now = datetime.now()
        self.history_data = [(v, t) for v, t in self.history_data
                            if t > now - timedelta(days=self.history_window)]

    def should_alert(self, current_value, timestamp):
        """
        判断是否应该告警

        Args:
            current_value: 当前指标值
            timestamp: 当前时间

        Returns:
            (should_alert, reason)
        """
        if len(self.history_data) < 24:  # 至少需要1天数据
            return False, "历史数据不足"

        # 计算同一时间段的历史均值和标准差
        hour_of_day = timestamp.hour
        day_of_week = timestamp.weekday()

        historical_values = []
        for v, t in self.history_data:
            if t.hour == hour_of_day and t.weekday() == day_of_week:
                historical_values.append(v)

        if len(historical_values) < 3:
            return False, "同一时段历史样本不足"

        mean = np.mean(historical_values)
        std = np.std(historical_values)

        # 判断当前值是否偏离历史均值超过2个标准差
        z_score = abs(current_value - mean) / (std + 1e-10)

        if z_score > 2:
            return True, f"当前值{current_value:.2f}偏离历史均值{mean:.2f}超过2个标准差"
        else:
            return False, f"当前值{current_value:.2f}在正常范围内"

class PriorityAlert:
    def __init__(self):
        self.alert_handlers = {
            'P0': self.handle_p0_alert,
            'P1': self.handle_p1_alert,
            'P2': self.handle_p2_alert
        }

    def evaluate_alert_priority(self, metrics):
        """
        评估告警优先级

        Args:
            metrics: 包含各种监控指标的字典

        Returns:
            priority: 'P0', 'P1', 'P2' 或 None
            reason: 告警原因
        """
        # P0条件：完全不可用
        if (metrics.get('error_rate', 0) > 0.1 or
            metrics.get('latency_p99', 0) > 5000 or
            metrics.get('accuracy', 1) < 0.6):
            return 'P0', '模型服务不可用或性能严重下降'

        # P1条件：性能下降显著
        if (metrics.get('accuracy', 1) < metrics.get('baseline_accuracy', 1) * 0.9 or
            metrics.get('conversion_rate', 1) < metrics.get('baseline_conversion', 1) * 0.9):
            return 'P1', '模型性能下降超过10%'

        # P2条件：轻微异常
        if (metrics.get('data_drift', 0) > 0.1 or
            metrics.get('confidence_drift', 0) > 0.1):
            return 'P2', '检测到轻微数据或置信度漂移'

        return None, '正常'

    def handle_p0_alert(self, reason):
        """处理P0告警"""
        # 立即通知on-call人员
        self.send_immediate_notification(reason)
        # 尝试自动回滚到上一个稳定版本
        self.attempt_auto_rollback()

    def handle_p1_alert(self, reason):
        """处理P1告警"""
        # 发送邮件和Slack通知
        self.send_email_notification(reason)
        self.send_slack_notification(reason)
        # 记录到监控看板
        self.log_to_dashboard(reason)

    def handle_p2_alert(self, reason):
        """处理P2告警"""
        # 只记录到监控看板，不主动通知
        self.log_to_dashboard(reason)

    def send_immediate_notification(self, reason):
        """立即通知on-call人员"""
        # 实现细节取决于你的通知系统
        pass

    def attempt_auto_rollback(self):
        """尝试自动回滚"""
        # 实现细节取决于你的部署系统
        pass

    # ... 其他通知方法的实现

class DataQualityMonitor:
    def __init__(self, feature_specs):
        """
        Args:
            feature_specs: 特征规格字典，包含每个特征的预期范围和类型
            例如:
            {
                'age': {'min': 0, 'max': 120, 'type': 'numeric'},
                'gender': {'values': ['M', 'F'], 'type': 'categorical'}
            }
        """
        self.feature_specs = feature_specs
        self.history = {}

    def check_data_quality(self, data):
        """
        检查数据质量

        Args:
            data: 输入数据字典

        Returns:
            quality_report: 质量报告，包含问题和建议
        """
        issues = []

        for feature, spec in self.feature_specs.items():
            if feature not in data:
                issues.append(f"特征缺失: {feature}")
                continue

            value = data[feature]

            # 检查数值范围
            if spec.get('type') == 'numeric':
                if 'min' in spec and value < spec['min']:
                    issues.append(f"{feature}值{value}低于最小值{spec['min']}")
                if 'max' in spec and value > spec['max']:
                    issues.append(f"{feature}值{value}高于最大值{spec['max']}")

            # 检查分类值
            elif spec.get('type') == 'categorical':
                if value not in spec.get('values', []):
                    issues.append(f"{feature}值{value}不在允许的值列表中")

            # 记录历史
            if feature not in self.history:
                self.history[feature] = []
            self.history[feature].append(value)

            # 检测分布异常（简化版）
            if len(self.history[feature]) > 100:
                recent_values = self.history[feature][-100:]
                if value not in recent_values[-20:]:
                    issues.append(f"{feature}值{value}与最近20个样本差异较大")

        return {
            'passed': len(issues) == 0,
            'issues': issues
        }

class TrendMonitor:
    def __init__(self, window=7):
        self.window = window
        self.history = []

    def add_data_point(self, value, timestamp):
        self.history.append((value, timestamp))

    def calculate_trend(self):
        """
        计算趋势

        Returns:
            trend: 'increasing', 'decreasing', 'stable'
            slope: 趋势斜率
            confidence: 趋势置信度
        """
        if len(self.history) < self.window * 2:
            return 'stable', 0, 0

        # 提取最近window天的数据
        now = datetime.now()
        recent = [(v, t) for v, t in self.history
                 if t > now - timedelta(days=self.window)]

        if len(recent) < 3:
            return 'stable', 0, 0

        # 简单线性回归
        times = [(t - recent[0][1]).total_seconds() for v, t in recent]
        values = [v for v, t in recent]

        slope, intercept = np.polyfit(times, values, 1)

        # 计算R方（拟合优度）
        predictions = [slope * t + intercept for t in times]
        ss_res = sum((values[i] - predictions[i])**2 for i in range(len(values)))
        ss_tot = sum((values[i] - np.mean(values))**2 for i in range(len(values)))
        r_squared = 1 - (ss_res / ss_tot)

        # 判断趋势
        if r_squared < 0.5:
            return 'stable', slope, r_squared
        elif slope > 0.01:
            return 'increasing', slope, r_squared
        elif slope < -0.01:
            return 'decreasing', slope, r_squared
        else:
            return 'stable', slope, r_squared

    def should_alert_on_trend(self):
        """
        基于趋势判断是否需要告警
        """
        trend, slope, confidence = self.calculate_trend()

        # 如果趋势是下降且置信度较高
        if trend == 'decreasing' and confidence > 0.6:
            return True, f"检测到下降趋势（斜率{slope:.4f}，置信度{confidence:.2f}）"

        return False, f"趋势{trend}，无需告警"

class ActionableAlert:
    def generate_alert_message(self, alert_type, metrics, context):
        """
        生成可行动的告警信息

        Args:
            alert_type: 告警类型 ('P0', 'P1', 'P2')
            metrics: 监控指标
            context: 上下文信息（数据版本、模型版本等）

        Returns:
            alert_message: 结构化的告警信息
        """
        message = {
            'priority': alert_type,
            'timestamp': datetime.now().isoformat(),
            'summary': '',
            'details': {},
            'impact': '',
            'possible_causes': [],
            'recommended_actions': [],
            'context': context
        }

        if alert_type == 'P0':
            if metrics.get('error_rate', 0) > 0.1:
                message['summary'] = '模型服务错误率过高'
                message['impact'] = '用户请求大量失败，严重影响用户体验'
                message['possible_causes'] = [
                    '模型服务崩溃',
                    '依赖服务不可用',
                    '资源耗尽'
                ]
                message['recommended_actions'] = [
                    '立即检查模型服务状态',
                    '尝试自动回滚到上一个稳定版本',
                    '检查依赖服务（数据库、缓存等）',
                    '增加资源分配'
                ]

            elif metrics.get('accuracy', 1) < 0.6:
                message['summary'] = '模型预测准确率严重下降'
                message['impact'] = '模型预测结果几乎不可用'
                message['possible_causes'] = [
                    '输入数据异常',
                    '模型文件损坏',
                    '业务场景发生重大变化'
                ]
                message['recommended_actions'] = [
                    '检查输入数据质量',
                    '尝试回滚到上一个模型版本',
                    '检查数据管道是否正常',
                    '评估是否需要紧急重新训练'
                ]

        elif alert_type == 'P1':
            message['summary'] = '模型性能下降'
            message['impact'] = '用户体验受到一定影响，但不完全不可用'
            message['possible_causes'] = [
                '数据分布发生漂移',
                '模型性能自然衰减',
                '用户行为模式变化'
            ]
            message['recommended_actions'] = [
                '分析数据漂移情况',
                '评估是否需要重新训练',
                '考虑A/B测试新模型',
                '与业务团队确认是否有产品变更'
            ]

        elif alert_type == 'P2':
            message['summary'] = '模型轻微异常'
            message['impact'] = '当前影响较小，但需要关注趋势'
            message['possible_causes'] = [
                '正常的数据波动',
                '轻微的数据漂移',
                '暂时的环境因素'
            ]
            message['recommended_actions'] = [
                '持续监控趋势',
                '记录异常现象',
                '定期评估模型性能'
            ]

        return message

from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, ClassificationPerformancePreset

# 配置列映射
column_mapping = ColumnMapping(
    target='target',
    prediction='prediction',
    numerical_features=['feature1', 'feature2'],
    categorical_features=['feature3']
)

# 生成数据漂移报告
drift_report = Report(metrics=[DataDriftPreset()])
drift_report.run(reference_data=train_data, current_data=online_data, column_mapping=column_mapping)

# 生成模型性能报告
performance_report = Report(metrics=[ClassificationPerformancePreset()])
performance_report.run(reference_data=train_data, current_data=online_data, column_mapping=column_mapping)

# 导出报告
drift_report.save_html('drift_report.html')
performance_report.save_html('performance_report.html')