模型监控告警与测试触发联动实战

# -*- coding: utf-8 -*-"""模型监控告警与测试触发联动 - 简化模拟"""import random  class MetricsMonitor:    """监控指标, 计算Z-score并判定异常"""     def __init__(self, history):        self.history = history        self.alert_count = 0     def mean_std(self, data):        n = len(data)        mean = sum(data) / n        var = sum((x - mean) ** 2 for x in data) / n        return mean, var ** 0.5     def check(self, new_value):        mean, std = self.mean_std(self.history)        if std == 0:            return False        z_score = (new_value - mean) / std        is_abnormal = abs(z_score) > 3        self.history.append(new_value)        if is_abnormal:            self.alert_count += 1        return is_abnormal  class AlertRouter:    """告警路由, 按级别分发事件"""     def __init__(self):        self.rules = {            "P0": ["on_call", "test_pipeline"],            "P1": ["test_pipeline"],            "P2": ["dashboard"]        }     def route(self, alert_level, event):        targets = self.rules.get(alert_level, [])        print("[路由] %s级告警 -> %s" % (alert_level, ", ".join(targets)))        return targets  class TestTrigger:    """测试触发, 执行回归测试"""     def __init__(self, suites):        self.suites = suites        self.trigger_count = 0     def run(self, suite_name="regression"):        self.trigger_count += 1        print("[测试] 执行%s套件..." % suite_name)        passed = random.random() > 0.3        return {            "suite": suite_name,            "passed": passed,            "duration": round(random.uniform(1, 10), 2)        }  class DeployGate:    """部署联动, 测试通过后触发部署"""     def __init__(self):        self.deploy_count = 0     def verify_and_deploy(self, test_result):        if not test_result["passed"]:            print("[部署] 测试未通过, 阻断部署")            return False        self.deploy_count += 1        print("[部署] 前置检查通过, 开始滚动发布...")        return True  def main():    random.seed(42)    history = [round(random.gauss(0.5, 0.05), 3) for _ in range(100)]    monitor = MetricsMonitor(history)    router = AlertRouter()    test_trigger = TestTrigger(["smoke", "regression"])    deploy_gate = DeployGate()     events = [        {"level": "P1", "value": round(random.gauss(0.8, 0.1), 3)},        {"level": "P2", "value": round(random.gauss(0.55, 0.05), 3)},        {"level": "P0", "value": round(random.gauss(0.05, 0.02), 3)},    ]     for i, evt in enumerate(events):        print("\n=== 事件%d ===" % (i + 1))        abnormal = monitor.check(evt["value"])        if not abnormal:            print("[监控] 指标 %.3f 正常, 无需处理" % evt["value"])            continue        print("[监控] 指标 %.3f 异常, 触发 %s 级告警" % (evt["value"], evt["level"]))        targets = router.route(evt["level"], "accuracy_drop")        if "test_pipeline" in targets:            result = test_trigger.run()            deploy_gate.verify_and_deploy(result)     print("\n=== 汇总 ===")    print("告警次数: %d, 测试触发: %d, 部署次数: %d" % (        monitor.alert_count, test_trigger.trigger_count, deploy_gate.deploy_count))  if __name__ == "__main__":    main()