失败案例复盘：模型为什么“突然不行了“：安全视角下的根因分析与修复策略

import pandas as pd
import numpy as np
from scipy import stats
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.ensemble import IsolationForest

class DataDriftDetector:
    def __init__(self, alpha=0.05):
        """
        数据漂移检测器
        alpha: 显著性水平，默认0.05
        """
        self.alpha = alpha
        self.scaler = StandardScaler()
        self.pca = PCA(n_components=2)
        self.isolation_forest = IsolationForest(contamination=0.1, random_state=42)
        self.ref_features = None
        self.ref_data_pca = None
        self.is_trained = False
    
    def fit(self, ref_data):
        """
        使用参考数据（如训练数据）训练漂移检测器
        ref_data: 参考数据，DataFrame格式
        """
        # 标准化数据
        self.ref_features = ref_data.columns
        scaled_ref = self.scaler.fit_transform(ref_data)
        # 降维
        self.ref_data_pca = self.pca.fit_transform(scaled_ref)
        # 训练隔离森林模型
        self.isolation_forest.fit(scaled_ref)
        self.is_trained = True
        print("数据漂移检测器训练完成")
    
    def detect_drift(self, current_data):
        """
        检测当前数据是否存在漂移
        current_data: 当前数据，DataFrame格式
        返回: 漂移检测结果，包含统计检验结果和异常检测结果
        """
        if not self.is_trained:
            raise ValueError("检测器未训练")
        
        # 确保当前数据的特征与参考数据一致
        if set(current_data.columns) != set(self.ref_features):
            raise ValueError("当前数据的特征与参考数据不一致")
        
        # 标准化当前数据
        scaled_current = self.scaler.transform(current_data)
        # 降维
        current_data_pca = self.pca.transform(scaled_current)
        
        # 1. 统计检验：KS检验
        ks_results = {}
        for i, feature in enumerate(self.ref_features):
            ks_stat, p_value = stats.kstest(current_data[feature], self.ref_data[feature])
            ks_results[feature] = {
                'ks_stat': ks_stat,
                'p_value': p_value,
                'is_drift': p_value < self.alpha
            }
        
        # 2. 异常检测：隔离森林
        anomaly_scores = self.isolation_forest.predict(scaled_current)
        anomaly_rate = np.sum(anomaly_scores == -1) / len(anomaly_scores)
        
        # 3. 距离度量：PCA空间中的平均距离
        avg_distance = np.mean(np.sqrt(np.sum((current_data_pca - self.ref_data_pca.mean(axis=0))**2, axis=1)))
        
        # 综合判断
        drift_detected = any(result['is_drift'] for result in ks_results.values()) or anomaly_rate > 0.2
        
        return {
            'drift_detected': drift_detected,
            'ks_test_results': ks_results,
            'anomaly_rate': anomaly_rate,
            'avg_distance': avg_distance,
            'anomaly_scores': anomaly_scores
        }

# 示例用法
if __name__ == "__main__":
    # 生成参考数据（正态分布）
    np.random.seed(42)
    ref_data = pd.DataFrame({
        'feature1': np.random.normal(0, 1, 1000),
        'feature2': np.random.normal(0, 1, 1000),
        'feature3': np.random.normal(0, 1, 1000)
    })
    
    # 生成当前数据（存在漂移）
    current_data = pd.DataFrame({
        'feature1': np.random.normal(1, 1, 1000),  # 均值漂移
        'feature2': np.random.normal(0, 2, 1000),  # 方差漂移
        'feature3': np.random.normal(0, 1, 1000)   # 无漂移
    })
    
    # 创建漂移检测器
    detector = DataDriftDetector()
    # 训练检测器
    detector.fit(ref_data)
    # 检测漂移
    drift_result = detector.detect_drift(current_data)
    
    print(f"是否检测到漂移: {drift_result['drift_detected']}")
    print(f"异常率: {drift_result['anomaly_rate']:.2f}")
    print(f"PCA空间平均距离: {drift_result['avg_distance']:.2f}")
    print("KS检验结果:")
    for feature, result in drift_result['ks_test_results'].items():
        print(f"- {feature}: KS统计量={result['ks_stat']:.4f}, p值={result['p_value']:.4f}, 是否漂移={result['is_drift']}")

import os
import json
import logging
import pandas as pd
import numpy as np
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("ModelFailureAnalyzer")

class ModelFailureAnalyzer:
    def __init__(self, log_dir, model_dir):
        """
        模型失败分析器
        log_dir: 日志目录
        model_dir: 模型目录
        """
        self.log_dir = log_dir
        self.model_dir = model_dir
        self.failure_cases = []
        self.failure_patterns = {
            'data_drift': '数据分布与训练数据不一致',
            'concept_drift': '业务概念发生变化',
            'model_aging': '模型长时间未更新',
            'adversarial_attack': '遭受对抗攻击',
            'system_failure': '系统或基础设施故障',
            'code_bug': '模型或推理代码存在bug',
            'dependency_issue': '依赖库版本不兼容'
        }
    
    def collect_data(self, failure_time, time_window=3600):
        """
        收集失败前后的数据
        failure_time: 失败发生时间，格式为"YYYY-MM-DD HH:MM:SS"
        time_window: 时间窗口，单位为秒，默认3600秒（1小时）
        返回: 收集的数据，包含日志数据、模型数据和业务数据
        """
        logger.info(f"开始收集失败前后的数据，失败时间: {failure_time}，时间窗口: {time_window}秒")
        
        # 1. 解析失败时间
        failure_dt = datetime.strptime(failure_time, "%Y-%m-%d %H:%M:%S")
        start_dt = failure_dt - pd.Timedelta(seconds=time_window)
        end_dt = failure_dt + pd.Timedelta(seconds=time_window)
        
        # 2. 收集日志数据
        log_data = self._collect_logs(start_dt, end_dt)
        
        # 3. 收集模型数据
        model_data = self._collect_model_data()
        
        # 4. 收集业务数据
        business_data = self._collect_business_data(start_dt, end_dt)
        
        logger.info("数据收集完成")
        return {
            'log_data': log_data,
            'model_data': model_data,
            'business_data': business_data,
            'time_range': {
                'start_time': start_dt.strftime("%Y-%m-%d %H:%M:%S"),
                'end_time': end_dt.strftime("%Y-%m-%d %H:%M:%S"),
                'failure_time': failure_time
            }
        }
    
    def _collect_logs(self, start_dt, end_dt):
        """
        收集日志数据
        """
        logs = []
        # 遍历日志目录，收集指定时间范围内的日志
        for filename in os.listdir(self.log_dir):
            if filename.endswith('.log'):
                filepath = os.path.join(self.log_dir, filename)
                try:
                    with open(filepath, 'r') as f:
                        for line in f:
                            # 假设日志格式为：2026-01-09 10:00:00 - INFO - 日志内容
                            try:
                                log_time_str = line.split(' - ')[0]
                                log_dt = datetime.strptime(log_time_str, "%Y-%m-%d %H:%M:%S")
                                if start_dt <= log_dt <= end_dt:
                                    logs.append(line.strip())
                            except:
                                continue
                except Exception as e:
                    logger.error(f"读取日志文件 {filename} 失败: {e}")
        return logs
    
    def _collect_model_data(self):
        """
        收集模型数据
        """
        model_data = {
            'model_files': [],
            'model_metadata': {}
        }
        
        # 收集模型文件信息
        for filename in os.listdir(self.model_dir):
            filepath = os.path.join(self.model_dir, filename)
            if os.path.isfile(filepath):
                model_data['model_files'].append({
                    'filename': filename,
                    'size': os.path.getsize(filepath),
                    'modified_time': datetime.fromtimestamp(os.path.getmtime(filepath)).strftime("%Y-%m-%d %H:%M:%S")
                })
        
        # 收集模型元数据
        metadata_file = os.path.join(self.model_dir, 'metadata.json')
        if os.path.exists(metadata_file):
            with open(metadata_file, 'r') as f:
                model_data['model_metadata'] = json.load(f)
        
        return model_data
    
    def _collect_business_data(self, start_dt, end_dt):
        """
        收集业务数据
        """
        # 这里简化处理，实际项目中需要从数据库或业务系统中收集数据
        business_data = {
            'transaction_count': np.random.randint(1000, 5000),
            'fraud_count': np.random.randint(10, 100),
            'avg_transaction_amount': np.random.uniform(100, 1000),
            'user_count': np.random.randint(500, 2000)
        }
        return business_data
    
    def analyze_root_cause(self, collected_data):
        """
        分析失败的根因
        collected_data: 收集的数据
        返回: 根因分析结果
        """
        logger.info("开始根因分析")
        
        root_causes = []
        
        # 1. 分析日志数据，查找错误和异常
        log_data = collected_data['log_data']
        error_logs = [log for log in log_data if 'ERROR' in log or 'Exception' in log]
        warning_logs = [log for log in log_data if 'WARNING' in log]
        
        if error_logs:
            root_causes.append({
                'type': 'system_failure',
                'description': f"发现{len(error_logs)}条错误日志，可能是系统故障导致模型失效",
                'evidence': error_logs[:5]  # 只显示前5条错误日志
            })
        
        if warning_logs:
            root_causes.append({
                'type': 'system_failure',
                'description': f"发现{len(warning_logs)}条警告日志，可能是系统异常导致模型性能下降",
                'evidence': warning_logs[:5]  # 只显示前5条警告日志
            })
        
        # 2. 分析模型数据，查找模型老化和版本问题
        model_data = collected_data['model_data']
        model_metadata = model_data['model_metadata']
        
        if 'training_time' in model_metadata:
            training_dt = datetime.strptime(model_metadata['training_time'], "%Y-%m-%d %H:%M:%S")
            current_dt = datetime.now()
            days_since_training = (current_dt - training_dt).days
            
            if days_since_training > 30:  # 模型超过30天未更新
                root_causes.append({
                    'type': 'model_aging',
                    'description': f"模型已超过{days_since_training}天未更新，可能存在模型老化问题",
                    'evidence': {
                        'training_time': model_metadata['training_time'],
                        'days_since_training': days_since_training
                    }
                })
        
        # 3. 分析业务数据，查找业务变化
        business_data = collected_data['business_data']
        # 这里简化处理，实际项目中需要与历史业务数据进行比较
        if business_data['fraud_count'] > 50:  # 欺诈交易数量异常增加
            root_causes.append({
                'type': 'concept_drift',
                'description': f"欺诈交易数量异常增加，当前欺诈交易数: {business_data['fraud_count']}，可能存在概念漂移",
                'evidence': business_data
            })
        
        # 4. 生成根因分析报告
        report = {
            'root_causes': root_causes,
            'confidence': self._calculate_confidence(root_causes),
            'suggestions': self._generate_suggestions(root_causes)
        }
        
        logger.info("根因分析完成")
        return report
    
    def _calculate_confidence(self, root_causes):
        """
        计算根因分析的置信度
        """
        if not root_causes:
            return 0.0
        
        # 简单的置信度计算，根据根因类型的权重和证据的充分性
        confidence = 0.0
        for cause in root_causes:
            if cause['type'] in ['code_bug', 'system_failure']:
                confidence += 0.3
            elif cause['type'] in ['data_drift', 'concept_drift']:
                confidence += 0.2
            elif cause['type'] in ['model_aging', 'adversarial_attack']:
                confidence += 0.1
        
        # 限制置信度在0-1之间
        return min(confidence, 1.0)
    
    def _generate_suggestions(self, root_causes):
        """
        根据根因分析结果，生成修复建议
        """
        suggestions = []
        
        for cause in root_causes:
            if cause['type'] == 'data_drift':
                suggestions.append("更新训练数据集，包含最新的数据分布")
                suggestions.append("采用数据增强技术，提高模型的泛化能力")
            elif cause['type'] == 'concept_drift':
                suggestions.append("重新定义业务概念，更新模型的目标函数")
                suggestions.append("采用在线学习或增量学习技术，实时更新模型")
            elif cause['type'] == 'model_aging':
                suggestions.append("定期重新训练模型，保持模型的时效性")
                suggestions.append("建立模型自动更新机制，根据性能指标触发更新")
            elif cause['type'] == 'system_failure':
                suggestions.append("检查系统基础设施，修复故障")
                suggestions.append("优化系统资源配置，提高系统的可靠性")
            elif cause['type'] == 'code_bug':
                suggestions.append("修复模型或推理代码中的bug")
                suggestions.append("增加代码测试覆盖率，防止类似问题再次发生")
        
        # 去重建议
        return list(set(suggestions))
    
    def generate_report(self, failure_time, collected_data, root_cause_report):
        """
        生成失败复盘报告
        """
        logger.info("生成失败复盘报告")
        
        report = {
            'report_id': f"FR_{datetime.now().strftime('%Y%m%d%H%M%S')}",
            'failure_time': failure_time,
            'collection_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            'collected_data_summary': {
                'log_count': len(collected_data['log_data']),
                'model_files_count': len(collected_data['model_data']['model_files']),
                'business_metrics': collected_data['business_data']
            },
            'root_cause_analysis': root_cause_report,
            'action_items': root_cause_report['suggestions'],
            'status': 'draft'
        }
        
        # 保存报告
        report_file = os.path.join(self.log_dir, f"failure_report_{report['report_id']}.json")
        with open(report_file, 'w') as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        
        logger.info(f"失败复盘报告已生成，文件路径: {report_file}")
        return report

# 示例用法
if __name__ == "__main__":
    # 创建失败分析器
    analyzer = ModelFailureAnalyzer(
        log_dir="./logs",
        model_dir="./models"
    )
    
    # 模拟失败时间
    failure_time = "2026-01-09 10:00:00"
    
    # 1. 收集数据
    collected_data = analyzer.collect_data(failure_time)
    
    # 2. 分析根因
    root_cause_report = analyzer.analyze_root_cause(collected_data)
    
    # 3. 生成报告
    report = analyzer.generate_report(failure_time, collected_data, root_cause_report)
    
    # 打印报告摘要
    print(f"报告ID: {report['report_id']}")
    print(f"失败时间: {report['failure_time']}")
    print(f"置信度: {report['root_cause_analysis']['confidence']:.2f}")
    print(f"根因数量: {len(report['root_cause_analysis']['root_causes'])}")
    print("根因:")
    for i, cause in enumerate(report['root_cause_analysis']['root_causes']):
        print(f"  {i+1}. {analyzer.failure_patterns[cause['type']]}: {cause['description']}")
    print("建议措施:")
    for i, suggestion in enumerate(report['action_items']):
        print(f"  {i+1}. {suggestion}")

# 安装必要的依赖
pip install pandas numpy scipy scikit-learn matplotlib seaborn

# 安装日志分析工具
pip install loguru

# 安装模型管理工具
pip install mlflow