fix: fail closed on quant monitor delivery

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

ops/quant-monitor/scripts/daily_briefing_builder.py

@@ -15,6 +15,20 @@
 DOMAINS = ("cn_equity", "hk_equity", "us_equity", "crypto")
 
 
+def _collect_drift_results(run_drift_detection, *, domains=DOMAINS):
+    results: dict[str, list[Any]] = {}
+    errors: dict[str, dict[str, str]] = {}
+    for domain in domains:
+        try:
+            results[domain] = list(run_drift_detection(domain))
+        except Exception as exc:
+            errors[domain] = {
+                "code": "drift_data_unavailable",
+                "error_type": type(exc).__name__,
+            }
+    return results, errors
+
+
 def _status_counts(strategies: list[dict[str, Any]]) -> dict[str, int]:
     counts = {"healthy": 0, "watch": 0, "review": 0, "critical": 0}
     for row in strategies:
@@ -33,9 +47,10 @@ def main() -> int:
     from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection
     from quant_platform_kit.strategy_lifecycle.health_dashboard import build_dashboard
 
+    drift_results, drift_errors = _collect_drift_results(run_drift_detection)
     drift_by_key: dict[tuple[str, str], float] = {}
-    for domain in DOMAINS:
-        for drift in run_drift_detection(domain):
+    for domain, domain_results in drift_results.items():
+        for drift in domain_results:
             drift_by_key[(domain, drift.strategy_profile)] = float(drift.drift_score or 0.0)
 
     with tempfile.TemporaryDirectory() as tmp:
@@ -58,13 +73,19 @@ def main() -> int:
     for domain in DOMAINS:
         strategies = by_domain.get(domain, [])
         summary = _status_counts(strategies)
+        domain_errors = [drift_errors[domain]] if domain in drift_errors else []
         report = {
             "domain": domain,
-            "ok": True,
+            "ok": not domain_errors,
+            "data_status": "unavailable" if domain_errors else "ready",
             "as_of": datetime.now(timezone.utc).isoformat(),
             "strategies": strategies,
             "summary": summary,
+            "errors": domain_errors,
         }
+        if domain_errors:
+            error = domain_errors[0]
+            report["error"] = f"{error['code']}:{error['error_type']}"
         path = out_dir / f"{domain}.json"
         path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
         print(f"[briefing] wrote {path}")

ops/quant-monitor/scripts/health_cycle.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import subprocess
@@ -16,6 +17,66 @@
 SCORE_ALERT = 60.0
 DRIFT_REVIEW = 0.50
 DRIFT_CRITICAL = 0.75
+_ALERT_STATE_RELATIVE_PATH = Path("data/alert-state/health_cycle.json")
+
+
+def _collect_drift_results(run_drift_detection, *, domains=DOMAINS):
+    results: dict[str, list[Any]] = {}
+    errors: list[dict[str, str]] = []
+    for domain in domains:
+        try:
+            results[domain] = list(run_drift_detection(domain))
+        except Exception as exc:
+            errors.append(
+                {
+                    "domain": domain,
+                    "code": "drift_data_unavailable",
+                    "error_type": type(exc).__name__,
+                }
+            )
+    return results, errors
+
+
+def _alert_fingerprint(lines: list[str]) -> str:
+    payload = "\n".join(sorted(str(line) for line in lines))
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _alert_state_path(root: Path) -> Path:
+    return root / _ALERT_STATE_RELATIVE_PATH
+
+
+def _is_duplicate_alert(root: Path, fingerprint: str) -> bool:
+    try:
+        payload = json.loads(_alert_state_path(root).read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return False
+    return str(payload.get("fingerprint") or "") == fingerprint
+
+
+def _record_alert(root: Path, fingerprint: str) -> None:
+    path = _alert_state_path(root)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temp_path = path.with_suffix(".tmp")
+    temp_path.write_text(
+        json.dumps(
+            {
+                "schema_version": "quant_monitor_alert_state.v1",
+                "fingerprint": fingerprint,
+            },
+            sort_keys=True,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    temp_path.replace(path)
+
+
+def _clear_alert(root: Path) -> None:
+    try:
+        _alert_state_path(root).unlink()
+    except FileNotFoundError:
+        pass
 
 
 def _send_telegram(text: str) -> bool:
@@ -26,8 +87,7 @@ def _send_telegram(text: str) -> bool:
     try:
         from quant_platform_kit.notifications.telegram import send_telegram_message
 
-        send_telegram_message(bot_token=token, chat_ids=chat, text=text)
-        return True
+        return bool(send_telegram_message(bot_token=token, chat_ids=chat, text=text))
     except Exception:
         return False
 
@@ -72,6 +132,7 @@ def main() -> int:
     from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection
     from quant_platform_kit.strategy_lifecycle.health_dashboard import build_dashboard
 
+    drift_results, drift_errors = _collect_drift_results(run_drift_detection)
     build_dashboard(output_dir=str(dash_dir), output_format="json")
 
     strategies: list[dict[str, Any]] = []
@@ -139,7 +200,7 @@ def main() -> int:
 
     issue_results: list[dict[str, Any]] = []
     for domain in DOMAINS:
-        drifts = run_drift_detection(domain)
+        drifts = drift_results.get(domain, [])
         for drift in drifts:
             score = float(drift.drift_score or 0.0)
             label = f"[{domain}] {drift.strategy_profile}: drift_score={score:.2f}"
@@ -156,18 +217,36 @@ def main() -> int:
             body=f"Quant-monitor detected critical drift.\n\n- {line}",
         )
 
-    notify_lines = telegram_lines + critical_lines
+    data_error_lines = [
+        f"[{error['domain']}] {error['code']} ({error['error_type']})"
+        for error in drift_errors
+    ]
+    if collector_payload_invalid:
+        data_error_lines.append("[collector] dashboard_data_unavailable")
+    notify_lines = telegram_lines + critical_lines + data_error_lines
+    telegram_sent = False
+    duplicate_alert_suppressed = False
     if notify_lines:
         body = "🚨 quant-monitor health_cycle\n" + "\n".join(f"• {line}" for line in notify_lines)
-        _send_telegram(body)
+        fingerprint = _alert_fingerprint(notify_lines)
+        duplicate_alert_suppressed = _is_duplicate_alert(root, fingerprint)
+        if not duplicate_alert_suppressed:
+            telegram_sent = _send_telegram(body)
+            if telegram_sent:
+                _record_alert(root, fingerprint)
+    else:
+        _clear_alert(root)
 
     summary = {
         "as_of": datetime.now(timezone.utc).isoformat(),
         "domains": list(DOMAINS),
         "strategy_count": len(strategies),
         "telegram_alerts": notify_lines,
+        "telegram_sent": telegram_sent,
+        "duplicate_alert_suppressed": duplicate_alert_suppressed,
+        "data_errors": drift_errors,
         "issues_created": len([r for r in issue_results if r.get("issue_url")]),
-        "ok": not notify_lines,
+        "ok": not notify_lines and not collector_payload_invalid,
         "collector_payload_valid": not collector_payload_invalid,
         "snapshot_data_status": normalized_payload.get("data_status"),
     }

ops/quant-monitor/tests/test_monitor_fail_closed.py

@@ -0,0 +1,85 @@
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_script(name: str):
+    spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+HEALTH_CYCLE = _load_script("health_cycle")
+DAILY_BRIEFING = _load_script("daily_briefing_builder")
+
+
+class MonitorFailClosedTests(unittest.TestCase):
+    def test_health_cycle_collects_drift_errors_without_aborting(self) -> None:
+        def unavailable(_domain):
+            raise RuntimeError("sensitive path must not escape")
+
+        results, errors = HEALTH_CYCLE._collect_drift_results(
+            unavailable,
+            domains=("cn_equity", "us_equity"),
+        )
+
+        self.assertEqual(results, {})
+        self.assertEqual(
+            errors,
+            [
+                {
+                    "domain": "cn_equity",
+                    "code": "drift_data_unavailable",
+                    "error_type": "RuntimeError",
+                },
+                {
+                    "domain": "us_equity",
+                    "code": "drift_data_unavailable",
+                    "error_type": "RuntimeError",
+                },
+            ],
+        )
+        self.assertNotIn("sensitive path", str(errors))
+
+    def test_daily_briefing_collects_drift_errors_without_aborting(self) -> None:
+        def unavailable(_domain):
+            raise RuntimeError("sensitive path must not escape")
+
+        results, errors = DAILY_BRIEFING._collect_drift_results(
+            unavailable,
+            domains=("crypto",),
+        )
+
+        self.assertEqual(results, {})
+        self.assertEqual(
+            errors,
+            {
+                "crypto": {
+                    "code": "drift_data_unavailable",
+                    "error_type": "RuntimeError",
+                }
+            },
+        )
+        self.assertNotIn("sensitive path", str(errors))
+
+    def test_health_cycle_alert_fingerprint_is_deduplicated_until_recovery(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            fingerprint = HEALTH_CYCLE._alert_fingerprint(["same failure"])
+
+            self.assertFalse(HEALTH_CYCLE._is_duplicate_alert(root, fingerprint))
+            HEALTH_CYCLE._record_alert(root, fingerprint)
+            self.assertTrue(HEALTH_CYCLE._is_duplicate_alert(root, fingerprint))
+
+            HEALTH_CYCLE._clear_alert(root)
+            self.assertFalse(HEALTH_CYCLE._is_duplicate_alert(root, fingerprint))
+
+
+if __name__ == "__main__":
+    unittest.main()

service/briefing_consumer.py

@@ -167,7 +167,12 @@ def _classify_report_payload(
 
     if payload.get("ok") is False:
         error = str(payload.get("error") or "ok=false")
-        level = BriefingAction.TELEGRAM if "circuit" in error.lower() else BriefingAction.GITHUB_ISSUE
+        data_unavailable = str(payload.get("data_status") or "").strip().lower() == "unavailable"
+        level = (
+            BriefingAction.TELEGRAM
+            if data_unavailable or "circuit" in error.lower()
+            else BriefingAction.GITHUB_ISSUE
+        )
         findings.append(
             BriefingFinding(source=source, level=level, reason=error, domain=str(payload.get("domain") or ""))
         )

tests/test_briefing_model_router.py

@@ -97,6 +97,18 @@ def test_telegram_for_critical_drift(self) -> None:
         )
         self.assertEqual(findings[0].level, BriefingAction.TELEGRAM)
 
+    def test_telegram_when_briefing_data_is_unavailable(self) -> None:
+        findings = consume_briefing_report(
+            {
+                "ok": False,
+                "data_status": "unavailable",
+                "domain": "us_equity",
+                "error": "drift_data_unavailable:RuntimeError",
+            }
+        )
+        self.assertEqual(len(findings), 1)
+        self.assertEqual(findings[0].level, BriefingAction.TELEGRAM)
+
     def test_consume_briefing_dir_reads_files(self) -> None:
         import json
         import tempfile