模型监控与漂移检测

# 模型健康度监控的基础框架
from typing import Dict, List
from dataclasses import dataclass, field
from datetime import datetime
import numpy as np

@dataclass
class ModelHealthReport:
    model_id: str
    timestamp: datetime
    data_drift_score: float        # 数据漂移分数 0-1
    prediction_drift_score: float  # 预测分布漂移分数
    performance_score: float       # 如果有真实标签
    alert_level: str = "green"     # green / yellow / red
    details: Dict = field(default_factory=dict)

class ModelMonitor:
    """模型健康度监控器"""
    
    def __init__(self, model_id: str, drift_threshold: float = 0.15):
        self.model_id = model_id
        self.drift_threshold = drift_threshold
        self.reference_stats: Dict = {}
        self.alerts: List[str] = []
    
    def set_reference(self, train_data: np.ndarray):
        self.reference_stats = {
            "mean": np.mean(train_data, axis=0),
            "std": np.std(train_data, axis=0),
            "min": np.min(train_data, axis=0),
            "max": np.max(train_data, axis=0),
        }
    
    def evaluate(self, production_data: np.ndarray) -> ModelHealthReport:
        report = ModelHealthReport(
            model_id=self.model_id,
            timestamp=datetime.now(),
            data_drift_score=0.0,
            prediction_drift_score=0.0,
            performance_score=1.0,
        )
        # 计算标准化均值差异
        mean_diff = np.abs(np.mean(production_data, axis=0) - self.reference_stats["mean"])
        std_safe = np.where(self.reference_stats["std"] > 0, self.reference_stats["std"], 1.0)
        normalized_diff = mean_diff / std_safe
        report.data_drift_score = float(np.mean(normalized_diff))
        
        if report.data_drift_score > self.drift_threshold:
            report.alert_level = "red"
            self.alerts.append(f"数据漂移超标: {report.data_drift_score:.3f}")
        elif report.data_drift_score > self.drift_threshold * 0.7:
            report.alert_level = "yellow"
        return report

# 生产环境中的模型监控指标采集
from collections import defaultdict
import time

class MetricsCollector:
    """收集模型推理指标用于监控"""
    
    def __init__(self, window_seconds: int = 3600):
        self.window = window_seconds
        self.latencies: List[float] = []
        self.prediction_counts: Dict[str, int] = defaultdict(int)
        self.error_counts: Dict[str, int] = defaultdict(int)
        self.prediction_values: List[float] = []
    
    def record_prediction(self, prediction: float, latency_ms: float,
                          model_version: str, success: bool = True):
        self.latencies.append(latency_ms)
        self.prediction_values.append(prediction)
        self.prediction_counts[model_version] += 1
        if not success:
            self.error_counts[model_version] += 1
    
    def get_summary(self) -> Dict:
        now = time.time()
        # 清理过期数据（简化版）
        n = len(self.latencies)
        if n == 0:
            return {"status": "no_data"}
        
        p50 = sorted(self.latencies)[n // 2]
        p99 = sorted(self.latencies)[int(n * 0.99)]
        error_rate = sum(self.error_counts.values()) / max(sum(self.prediction_counts.values()), 1)
        
        return {
            "total_predictions": n,
            "latency_p50_ms": round(p50, 2),
            "latency_p99_ms": round(p99, 2),
            "error_rate": round(error_rate, 4),
            "pred_mean": round(np.mean(self.prediction_values), 4),
            "pred_std": round(np.std(self.prediction_values), 4),
            "versions": dict(self.prediction_counts),
        }

# 协变量漂移检测：PSI (Population Stability Index)
import numpy as np
from typing import Tuple

def compute_psi(expected: np.ndarray, actual: np.ndarray,
                buckets: int = 10) -> float:
    """计算 PSI 分数，衡量两个分布的差异"""
    def _bin_proportions(data: np.ndarray) -> np.ndarray:
        # 使用 expected 的分位数作为分桶边界
        percentiles = np.linspace(0, 100, buckets + 1)
        bin_edges = np.percentile(expected, percentiles)
        bin_edges[0] -= 1e-10  # 避免边界值问题
        bin_edges[-1] += 1e-10
        counts = np.histogram(data, bins=bin_edges)[0]
        proportions = (counts + 1e-6) / (len(data) + buckets * 1e-6)  # 平滑
        return proportions
    
    exp_props = _bin_proportions(expected)
    act_props = _bin_proportions(actual)
    
    # PSI = sum((actual - expected) * ln(actual / expected))
    psi = np.sum((act_props - exp_props) * np.log(act_props / exp_props))
    return float(psi)

# 使用示例
train_ages = np.random.normal(32, 8, 5000)        # 训练数据年龄
prod_ages_new = np.random.normal(45, 10, 3000)    # 生产数据年龄（漂移了）
prod_ages_old = np.random.normal(33, 8, 3000)     # 生产数据年龄（未漂移）

psi_drifted = compute_psi(train_ages, prod_ages_new)
psi_stable = compute_psi(train_ages, prod_ages_old)
print(f"漂移场景 PSI: {psi_drifted:.4f}")   # 预期 > 0.25
print(f"稳定场景 PSI: {psi_stable:.4f}")    # 预期 < 0.1

# PSI 解读: < 0.1 无漂移, 0.1-0.25 中等漂移, > 0.25 严重漂移

# 概念漂移检测：使用滑动窗口的性能监控
from collections import deque
import numpy as np

class ConceptDriftDetector:
    """基于滑动窗口准确率的概念漂移检测"""
    
    def __init__(self, window_size: int = 500, threshold: float = 0.05):
        self.window_size = window_size
        self.threshold = threshold
        self.recent_acc = deque(maxlen=window_size)
        self.reference_acc = None
        self.drift_detected = False
    
    def set_reference_accuracy(self, acc: float):
        self.reference_acc = acc
    
    def update(self, prediction: int, ground_truth: int) -> bool:
        """输入单个预测和真实标签，返回是否检测到漂移"""
        correct = int(prediction == ground_truth)
        self.recent_acc.append(correct)
        
        if len(self.recent_acc) < self.window_size:
            return False
        
        current_acc = np.mean(self.recent_acc)
        if self.reference_acc is None:
            self.reference_acc = current_acc
            return False
        
        acc_drop = self.reference_acc - current_acc
        self.drift_detected = acc_drop > self.threshold
        return self.drift_detected
    
    def get_status(self) -> Dict:
        if not self.recent_acc:
            return {"status": "no_data"}
        current = np.mean(self.recent_acc)
        return {
            "reference_accuracy": round(self.reference_acc, 4) if self.reference_acc else None,
            "current_accuracy": round(current, 4),
            "accuracy_drop": round(self.reference_acc - current, 4) if self.reference_acc else None,
            "drift_detected": self.drift_detected,
            "window_size": len(self.recent_acc),
        }

# 多特征批量漂移检测管道
from scipy import stats
from typing import Dict, List, Tuple
import numpy as np

class DriftDetectionPipeline:
    """多特征批量漂移检测"""
    
    def __init__(self, significance: float = 0.05, psi_threshold: float = 0.1):
        self.significance = significance
        self.psi_threshold = psi_threshold
    
    def ks_test(self, ref: np.ndarray, prod: np.ndarray) -> Dict:
        statistic, p_value = stats.ks_2samp(ref, prod)
        return {"statistic": round(statistic, 4), "p_value": round(p_value, 6),
                "drift": p_value < self.significance}
    
    def psi_score(self, ref: np.ndarray, prod: np.ndarray, buckets: int = 10) -> float:
        edges = np.percentile(ref, np.linspace(0, 100, buckets + 1))
        edges[0] -= 1e-10; edges[-1] += 1e-10
        ref_p = (np.histogram(ref, bins=edges)[0] + 1e-6) / (len(ref) + buckets * 1e-6)
        prod_p = (np.histogram(prod, bins=edges)[0] + 1e-6) / (len(prod) + buckets * 1e-6)
        return float(np.sum((prod_p - ref_p) * np.log(prod_p / ref_p)))
    
    def detect_all(self, ref_data: Dict[str, np.ndarray],
                   prod_data: Dict[str, np.ndarray]) -> Dict:
        """对所有特征执行漂移检测"""
        results = {}
        for feature in ref_data:
            if feature not in prod_data:
                results[feature] = {"status": "missing"}
                continue
            ref, prod = ref_data[feature], prod_data[feature]
            if np.issubdtype(ref.dtype, np.number) and len(np.unique(ref)) > 10:
                ks_result = self.ks_test(ref, prod)
                psi = self.psi_score(ref, prod)
                results[feature] = {
                    "method": "KS + PSI", "ks": ks_result, "psi": round(psi, 4),
                    "drift": ks_result["drift"] or psi > self.psi_threshold,
                }
            else:
                chi2, p_val = stats.chisquare(
                    np.histogram(prod, bins=len(np.unique(ref)))[0] + 1e-6,
                    np.histogram(ref, bins=len(np.unique(ref)))[0] + 1e-6
                )
                results[feature] = {"method": "Chi2", "chi2": round(chi2, 4),
                                    "p_value": round(p_val, 6), "drift": p_val < self.significance}
        return results

# 对抗式漂移检测：训练分类器区分参考数据和生产数据
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import cross_val_score
from typing import Dict

class AdversarialDriftDetector:
    """对抗式漂移检测器：如果能区分两个数据集，说明发生了漂移"""
    
    def __init__(self, n_estimators: int = 100):
        self.classifier = GradientBoostingClassifier(n_estimators=n_estimators, max_depth=3)
    
    def detect(self, ref_data: np.ndarray, prod_data: np.ndarray) -> Dict:
        # 构造二分类问题：0=参考数据，1=生产数据
        X = np.vstack([ref_data, prod_data])
        y = np.concatenate([np.zeros(len(ref_data)), np.ones(len(prod_data))])
        
        # 交叉验证评估分类器区分能力
        cv_scores = cross_val_score(self.classifier, X, y, cv=5, scoring="roc_auc")
        mean_auc = float(np.mean(cv_scores))
        
        # AUC = 0.5 表示完全无法区分（无漂移）
        # AUC >> 0.5 表示可以区分（存在漂移）
        drift_severity = max(0, (mean_auc - 0.5) * 2)  # 归一化到 0-1
        
        # 特征重要性分析：哪些特征贡献最大
        self.classifier.fit(X, y)
        importances = self.classifier.feature_importances_
        top_features = np.argsort(importances)[::-1][:5]
        
        return {
            "auc_roc": round(mean_auc, 4),
            "drift_severity": round(drift_severity, 4),
            "drift_detected": mean_auc > 0.65,
            "top_drift_features": top_features.tolist(),
            "feature_importances": {i: round(importances[i], 4) for i in top_features},
        }

# 预测分布监控：检测预测输出的统计异常
from typing import List, Dict
import numpy as np
from collections import deque

class PredictionDistributionMonitor:
    """监控模型预测输出的分布变化"""
    
    def __init__(self, baseline_mean: float, baseline_std: float,
                 window_size: int = 1000, z_threshold: float = 3.0):
        self.baseline_mean = baseline_mean
        self.baseline_std = baseline_std if baseline_std > 0 else 0.01
        self.window_size = window_size
        self.z_threshold = z_threshold
        self.predictions = deque(maxlen=window_size)
    
    def add_prediction(self, value: float):
        self.predictions.append(value)
    
    def check(self) -> Dict:
        if len(self.predictions) < self.window_size * 0.5:
            return {"status": "insufficient_data", "count": len(self.predictions)}
        
        preds = np.array(self.predictions)
        current_mean = np.mean(preds)
        current_std = np.std(preds)
        
        # Z-score 检测均值偏移
        z_score = abs(current_mean - self.baseline_mean) / self.baseline_std
        # 标准差变化率
        std_ratio = current_std / self.baseline_std
        # 极值比例
        outlier_ratio = np.sum(np.abs(preds - self.baseline_mean) > 3 * self.baseline_std) / len(preds)
        
        alerts = []
        if z_score > self.z_threshold:
            alerts.append(f"均值漂移: z={z_score:.2f}")
        if std_ratio > 2.0 or std_ratio < 0.5:
            alerts.append(f"方差异常: ratio={std_ratio:.2f}")
        if outlier_ratio > 0.05:
            alerts.append(f"极值过多: {outlier_ratio:.1%}")
        
        return {
            "status": "alert" if alerts else "ok",
            "current_mean": round(current_mean, 4),
            "current_std": round(current_std, 4),
            "z_score": round(z_score, 2),
            "std_ratio": round(std_ratio, 2),
            "outlier_ratio": round(outlier_ratio, 4),
            "alerts": alerts,
        }

# 业务指标关联分析：模型预测 vs 实际业务结果
import numpy as np
from typing import Dict, List
from scipy import stats

class BusinessImpactAnalyzer:
    """分析模型预测与业务结果的关联"""
    
    def __init__(self):
        self.prediction_outcome_pairs: List[tuple] = []
    
    def record(self, prediction_score: float, business_outcome: float):
        self.prediction_outcome_pairs.append((prediction_score, business_outcome))
    
    def analyze(self) -> Dict:
        if len(self.prediction_outcome_pairs) < 100:
            return {"status": "insufficient_data"}
        
        preds = np.array([p for p, _ in self.prediction_outcome_pairs])
        outcomes = np.array([o for _, o in self.prediction_outcome_pairs])
        
        # 预测与业务结果的相关性
        correlation, corr_p = stats.pearsonr(preds, outcomes)
        
        # 分桶分析：高预测分 vs 低预测分的业务表现差异
        n_bins = 5
        bin_edges = np.percentile(preds, np.linspace(0, 100, n_bins + 1))
        bin_analysis = []
        for i in range(n_bins):
            mask = (preds >= bin_edges[i]) & (preds < bin_edges[i + 1])
            if np.sum(mask) > 0:
                bin_analysis.append({
                    "bin": f"{bin_edges[i]:.2f}-{bin_edges[i+1]:.2f}",
                    "avg_outcome": round(float(np.mean(outcomes[mask])), 4),
                    "count": int(np.sum(mask)),
                })
        
        return {
            "correlation": round(correlation, 4),
            "correlation_p_value": round(corr_p, 6),
            "correlation_significant": corr_p < 0.05,
            "bin_analysis": bin_analysis,
            "total_samples": len(self.prediction_outcome_pairs),
        }

# 推理服务的监控数据采集 Sidecar 模式
import time
import json
import threading
from typing import Dict, Any
from queue import Queue

class InferenceLogger:
    """推理日志采集器：记录每次预测的元数据"""
    
    def __init__(self, batch_size: int = 100, flush_interval: float = 5.0):
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.queue: Queue = Queue(maxsize=10000)
        self.running = True
        self.flush_thread = threading.Thread(target=self._flush_loop, daemon=True)
        self.flush_thread.start()
    
    def log(self, model_id: str, input_hash: str, prediction: float,
            confidence: float, latency_ms: float, metadata: Dict = None):
        record = {
            "timestamp": time.time(),
            "model_id": model_id,
            "input_hash": input_hash,  # 隐私保护：只存哈希
            "prediction": prediction,
            "confidence": confidence,
            "latency_ms": latency_ms,
            "metadata": metadata or {},
        }
        try:
            self.queue.put_nowait(record)
        except Exception:
            pass  # 队列满时丢弃（不应阻塞推理）
    
    def _flush_loop(self):
        while self.running:
            batch = []
            start = time.time()
            while len(batch) < self.batch_size and time.time() - start < self.flush_interval:
                try:
                    batch.append(self.queue.get(timeout=0.1))
                except Exception:
                    continue
            if batch:
                # 写入 Kafka / 文件 / 远程 API
                self._write_to_sink(batch)
    
    def _write_to_sink(self, batch: list):
        # 实际实现：写入 Kafka 或远程日志服务
        pass  # 占位
    
    def stop(self):
        self.running = False

# 监控面板的数据聚合器：从原始日志到可视化指标
from collections import defaultdict
from typing import Dict, List
import numpy as np

class MetricsAggregator:
    """将原始推理日志聚合为监控指标"""
    
    def __init__(self, time_window_minutes: int = 5):
        self.window = time_window_minutes * 60
        self.records: List[Dict] = []
    
    def add_records(self, records: List[Dict]):
        self.records.extend(records)
        # 清理过期数据
        cutoff = max(r["timestamp"] for r in self.records) - self.window * 2
        self.records = [r for r in self.records if r["timestamp"] > cutoff]
    
    def aggregate_by_model(self) -> Dict[str, Dict]:
        by_model = defaultdict(list)
        for r in self.records:
            by_model[r["model_id"]].append(r)
        
        result = {}
        for model_id, recs in by_model.items():
            latencies = [r["latency_ms"] for r in recs]
            predictions = [r["prediction"] for r in recs]
            sorted_lat = sorted(latencies)
            n = len(sorted_lat)
            
            result[model_id] = {
                "request_count": len(recs),
                "latency_p50": round(sorted_lat[n // 2], 2),
                "latency_p95": round(sorted_lat[int(n * 0.95)], 2),
                "latency_p99": round(sorted_lat[int(n * 0.99)], 2),
                "pred_mean": round(np.mean(predictions), 4),
                "pred_std": round(np.std(predictions), 4),
                "pred_min": round(min(predictions), 4),
                "pred_max": round(max(predictions), 4),
                "avg_confidence": round(np.mean([r["confidence"] for r in recs]), 4),
            }
        return result

# 告警路由与去重系统
from typing import Dict, List
from datetime import datetime, timedelta
from collections import defaultdict

class AlertRouter:
    """智能告警路由：去重、分级、分发"""
    
    def __init__(self):
        self.recent_alerts: Dict[str, datetime] = {}  # 防抖动
        self.dedup_window = timedelta(hours=2)
        self.alert_history: List[Dict] = []
    
    def process_alert(self, model_id: str, alert_type: str,
                      severity: str, message: str, details: Dict = None) -> Dict:
        """处理告警：去重、分级、决定是否需要通知"""
        dedup_key = f"{model_id}:{alert_type}"
        now = datetime.now()
        
        # 防抖动：同一模型同一类型告警在窗口内只发一次
        if dedup_key in self.recent_alerts:
            last = self.recent_alerts[dedup_key]
            if now - last < self.dedup_window:
                return {"action": "suppressed", "reason": "within_dedup_window", "last_alert": str(last)}
        
        # 相关告警聚合：检查是否有相关的 P1 告警已经在处理
        related = self._find_related(model_id, alert_type)
        if related and related["severity"] == "P1":
            return {"action": "merged", "reason": f"already_handling_P1: {related['alert_type']}"}
        
        self.recent_alerts[dedup_key] = now
        alert_record = {
            "timestamp": str(now), "model_id": model_id, "type": alert_type,
            "severity": severity, "message": message, "details": details or {},
        }
        self.alert_history.append(alert_record)
        
        # 决定通知方式
        notification = self._get_notification(severity)
        return {"action": "dispatched", "alert": alert_record, "notification": notification}
    
    def _find_related(self, model_id: str, alert_type: str) -> Dict:
        for alert in reversed(self.alert_history[-50:]):
            if alert["model_id"] == model_id and alert["type"] != alert_type:
                if datetime.fromisoformat(alert["timestamp"]) > datetime.now() - self.dedup_window:
                    return alert
        return {}
    
    def _get_notification(self, severity: str) -> Dict:
        channels = {
            "P1": ["phone", "sms", "feishu", "pagerduty"],
            "P2": ["feishu", "email"],
            "P3": ["dashboard"],
        }
        return {"channels": channels.get(severity, ["dashboard"]), "priority": severity}

# 自动化响应引擎：检测到漂移后自动执行预定义动作
from enum import Enum
from typing import Callable, Dict, List
import logging

class ResponseAction(Enum):
    LOG_ONLY = "log"
    NOTIFY_TEAM = "notify"
    INCREASE_SAMPLING = "increase_sampling"
    SHADOW_TEST = "shadow_test"
    ROLLBACK = "rollback"
    RETRAIN = "retrain"

class AutomatedResponseEngine:
    """根据漂移检测结果自动执行响应动作"""
    
    def __init__(self):
        self.handlers: Dict[ResponseAction, Callable] = {}
        self.response_log: List[Dict] = []
    
    def register_handler(self, action: ResponseAction, handler: Callable):
        self.handlers[action] = handler
    
    def decide_and_execute(self, model_id: str, psi_score: float,
                           accuracy_drop: float) -> Dict:
        """根据漂移程度决定并执行响应动作"""
        actions = self._decide_actions(psi_score, accuracy_drop)
        results = []
        
        for action in actions:
            handler = self.handlers.get(action)
            if handler:
                try:
                    result = handler(model_id, psi_score, accuracy_drop)
                    results.append({"action": action.value, "status": "success", "result": result})
                except Exception as e:
                    results.append({"action": action.value, "status": "error", "error": str(e)})
        
        response_record = {
            "model_id": model_id, "psi_score": psi_score,
            "accuracy_drop": accuracy_drop, "actions": results,
        }
        self.response_log.append(response_record)
        return response_record
    
    def _decide_actions(self, psi: float, acc_drop: float) -> List[ResponseAction]:
        if psi > 0.25 or acc_drop > 0.10:
            return [ResponseAction.NOTIFY_TEAM, ResponseAction.ROLLBACK, ResponseAction.RETRAIN]
        elif psi > 0.15 or acc_drop > 0.05:
            return [ResponseAction.NOTIFY_TEAM, ResponseAction.INCREASE_SAMPLING, ResponseAction.SHADOW_TEST]
        elif psi > 0.10:
            return [ResponseAction.LOG_ONLY, ResponseAction.INCREASE_SAMPLING]
        return [ResponseAction.LOG_ONLY]

# 模型版本管理与回滚系统
import json
import hashlib
from typing import Dict, List, Optional
from datetime import datetime

class ModelRegistry:
    """模型注册表：管理所有模型版本"""
    
    def __init__(self):
        self.models: Dict[str, List[Dict]] = {}
        self.active_models: Dict[str, str] = {}  # model_id -> version_id
    
    def register(self, model_id: str, version: str, metrics: Dict,
                 training_data_hash: str, artifacts_path: str) -> Dict:
        if model_id not in self.models:
            self.models[model_id] = []
        
        model_version = {
            "version": version,
            "registered_at": datetime.now().isoformat(),
            "metrics": metrics,
            "training_data_hash": training_data_hash,
            "artifacts_path": artifacts_path,
            "status": "registered",  # registered / staging / active / rolled_back
        }
        self.models[model_id].append(model_version)
        return model_version
    
    def promote(self, model_id: str, version: str) -> Dict:
        """将模型版本提升为活跃版本"""
        model_list = self.models.get(model_id, [])
        for mv in model_list:
            if mv["version"] == version:
                old_active = self.active_models.get(model_id)
                self.active_models[model_id] = version
                mv["status"] = "active"
                # 将旧版本标记为非活跃
                for other in model_list:
                    if other["version"] != version and other["status"] == "active":
                        other["status"] = "staged_out"
                return {"status": "promoted", "previous": old_active, "current": version}
        return {"status": "error", "message": "version not found"}
    
    def rollback(self, model_id: str) -> Dict:
        """回滚到上一个活跃版本"""
        model_list = self.models.get(model_id, [])
        for mv in reversed(model_list):
            if mv["status"] in ("active", "staged_out"):
                return self.promote(model_id, mv["version"])
        return {"status": "error", "message": "no previous version found"}

# 持续重训练管道：检测到漂移后自动触发
import subprocess
import logging
from typing import Dict, Optional

class RetrainingPipeline:
    """自动重训练管道"""
    
    def __init__(self, config_path: str):
        self.config_path = config_path
        self.training_history: List[Dict] = []
    
    def trigger_retraining(self, model_id: str, reason: str,
                           drift_report: Dict) -> Dict:
        logging.info(f"触发重训练: model={model_id}, reason={reason}")
        
        # 1. 获取最新训练数据
        data_path = self._fetch_latest_data(model_id)
        
        # 2. 执行训练
        training_result = self._run_training(model_id, data_path)
        
        # 3. 验证新模型
        validation_result = self._validate_model(model_id, training_result)
        
        # 4. 记录结果
        record = {
            "model_id": model_id,
            "trigger_reason": reason,
            "drift_report": drift_report,
            "training_result": training_result,
            "validation_result": validation_result,
            "timestamp": datetime.now().isoformat(),
        }
        self.training_history.append(record)
        
        if validation_result["passed"]:
            return {"status": "success", "new_model": validation_result["model_path"]}
        else:
            return {"status": "failed", "reason": validation_result["failure_reason"]}
    
    def _fetch_latest_data(self, model_id: str) -> str:
        # 从数据仓库获取最新的标注数据
        return f"/data/latest/{model_id}"
    
    def _run_training(self, model_id: str, data_path: str) -> Dict:
        # 调用训练脚本
        result = subprocess.run(
            ["python", "train.py", "--model", model_id, "--data", data_path,
             "--config", self.config_path],
            capture_output=True, text=True, timeout=3600
        )
        return {"exit_code": result.returncode, "log": result.stdout[-500:]}
    
    def _validate_model(self, model_id: str, training_result: Dict) -> Dict:
        # 在验证集上评估新模型
        # 返回是否达到上线标准
        return {"passed": True, "model_path": f"/models/{model_id}/latest",
                "metrics": {"accuracy": 0.92, "f1": 0.89}}