From da1fbb1ff7e6fa5054524ffe3755b28d8bea4bb2 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:49:09 +0800 Subject: [PATCH 1/4] fix: fail closed on quant monitor delivery Co-Authored-By: Codex --- .../scripts/daily_briefing_builder.py | 27 +++++- ops/quant-monitor/scripts/health_cycle.py | 91 +++++++++++++++++-- .../tests/test_monitor_fail_closed.py | 85 +++++++++++++++++ service/briefing_consumer.py | 7 +- tests/test_briefing_model_router.py | 12 +++ 5 files changed, 212 insertions(+), 10 deletions(-) create mode 100644 ops/quant-monitor/tests/test_monitor_fail_closed.py diff --git a/ops/quant-monitor/scripts/daily_briefing_builder.py b/ops/quant-monitor/scripts/daily_briefing_builder.py index 97509d36..7b52d71c 100755 --- a/ops/quant-monitor/scripts/daily_briefing_builder.py +++ b/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}") diff --git a/ops/quant-monitor/scripts/health_cycle.py b/ops/quant-monitor/scripts/health_cycle.py index dccebe88..f19e2650 100755 --- a/ops/quant-monitor/scripts/health_cycle.py +++ b/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"), } diff --git a/ops/quant-monitor/tests/test_monitor_fail_closed.py b/ops/quant-monitor/tests/test_monitor_fail_closed.py new file mode 100644 index 00000000..7dda0673 --- /dev/null +++ b/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() diff --git a/service/briefing_consumer.py b/service/briefing_consumer.py index 1f66c0c1..0b27daf4 100644 --- a/service/briefing_consumer.py +++ b/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 "")) ) diff --git a/tests/test_briefing_model_router.py b/tests/test_briefing_model_router.py index da4c9c56..272cb309 100644 --- a/tests/test_briefing_model_router.py +++ b/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 From cffd40353a25a001ba3917a942dbfb8adcf9df5e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:49:18 +0800 Subject: [PATCH 2/4] fix: keep quant monitor fail closed Co-Authored-By: Codex --- .../scripts/daily_briefing_builder.py | 8 +++ ops/quant-monitor/scripts/health_cycle.py | 21 ++++++- .../tests/test_monitor_fail_closed.py | 56 +++++++++++++++++++ 3 files changed, 83 insertions(+), 2 deletions(-) diff --git a/ops/quant-monitor/scripts/daily_briefing_builder.py b/ops/quant-monitor/scripts/daily_briefing_builder.py index 7b52d71c..e1ee48dc 100755 --- a/ops/quant-monitor/scripts/daily_briefing_builder.py +++ b/ops/quant-monitor/scripts/daily_briefing_builder.py @@ -57,10 +57,16 @@ def main() -> int: build_dashboard(output_dir=tmp, output_format="json") dash_path = Path(tmp) / "strategy_health_dashboard.json" strategies_raw: list[dict[str, Any]] = [] + dashboard_error: dict[str, str] | None = None if dash_path.is_file(): payload = json.loads(dash_path.read_text(encoding="utf-8")) if isinstance(payload.get("strategies"), list): strategies_raw = [row for row in payload["strategies"] if isinstance(row, dict)] + else: + dashboard_error = { + "code": "dashboard_data_unavailable", + "error_type": "FileNotFoundError", + } by_domain: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in strategies_raw: @@ -74,6 +80,8 @@ def main() -> int: strategies = by_domain.get(domain, []) summary = _status_counts(strategies) domain_errors = [drift_errors[domain]] if domain in drift_errors else [] + if dashboard_error: + domain_errors.append(dashboard_error) report = { "domain": domain, "ok": not domain_errors, diff --git a/ops/quant-monitor/scripts/health_cycle.py b/ops/quant-monitor/scripts/health_cycle.py index f19e2650..33bfba60 100755 --- a/ops/quant-monitor/scripts/health_cycle.py +++ b/ops/quant-monitor/scripts/health_cycle.py @@ -51,6 +51,8 @@ def _is_duplicate_alert(root: Path, fingerprint: str) -> bool: payload = json.loads(_alert_state_path(root).read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False + if not isinstance(payload, dict): + return False return str(payload.get("fingerprint") or "") == fingerprint @@ -79,6 +81,19 @@ def _clear_alert(root: Path) -> None: pass +def _create_issues_for_available_domains( + drift_results: dict[str, list[Any]], + create_issues_for_domain, + *, + domains=DOMAINS, +) -> list[dict[str, Any]]: + issue_results: list[dict[str, Any]] = [] + for domain in domains: + if domain in drift_results: + issue_results.extend(create_issues_for_domain(domain)) + return issue_results + + def _send_telegram(text: str) -> bool: token = (os.environ.get("TELEGRAM_TOKEN") or os.environ.get("TG_TOKEN") or "").strip() chat = (os.environ.get("GLOBAL_TELEGRAM_CHAT_ID") or "").strip() @@ -198,7 +213,6 @@ def main() -> int: domain = str(row.get("domain") or "?") telegram_lines.append(f"[{domain}] {profile}: health_score={score:.1f}") - issue_results: list[dict[str, Any]] = [] for domain in DOMAINS: drifts = drift_results.get(domain, []) for drift in drifts: @@ -209,7 +223,10 @@ def main() -> int: elif score >= DRIFT_REVIEW: pass # tracked via create_issues_for_domain below - issue_results.extend(create_issues_for_domain(domain)) + issue_results = _create_issues_for_available_domains( + drift_results, + create_issues_for_domain, + ) for line in critical_lines: _create_owner_issue( diff --git a/ops/quant-monitor/tests/test_monitor_fail_closed.py b/ops/quant-monitor/tests/test_monitor_fail_closed.py index 7dda0673..f067e27c 100644 --- a/ops/quant-monitor/tests/test_monitor_fail_closed.py +++ b/ops/quant-monitor/tests/test_monitor_fail_closed.py @@ -1,7 +1,10 @@ import importlib.util +import json +import os import tempfile import unittest from pathlib import Path +from unittest import mock ROOT = Path(__file__).resolve().parents[1] @@ -80,6 +83,59 @@ def test_health_cycle_alert_fingerprint_is_deduplicated_until_recovery(self) -> HEALTH_CYCLE._clear_alert(root) self.assertFalse(HEALTH_CYCLE._is_duplicate_alert(root, fingerprint)) + def test_health_cycle_non_object_alert_state_is_a_cache_miss(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + state_path = HEALTH_CYCLE._alert_state_path(root) + state_path.parent.mkdir(parents=True) + for payload in (None, [], "invalid"): + state_path.write_text(json.dumps(payload), encoding="utf-8") + self.assertFalse(HEALTH_CYCLE._is_duplicate_alert(root, "fingerprint")) + + def test_health_cycle_skips_issue_creation_when_drift_is_unavailable(self) -> None: + created_for: list[str] = [] + + results = HEALTH_CYCLE._create_issues_for_available_domains( + {"us_equity": []}, + lambda domain: created_for.append(domain) or [], + domains=("us_equity", "crypto"), + ) + + self.assertEqual(results, []) + self.assertEqual(created_for, ["us_equity"]) + + def test_daily_briefing_marks_missing_dashboard_unavailable(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + with ( + mock.patch.dict( + os.environ, + {"QUANT_MONITOR_ROOT": str(root), "DAY": "2026-07-30"}, + ), + mock.patch( + "quant_platform_kit.strategy_lifecycle.drift_detector.run_drift_detection", + return_value=[], + ), + mock.patch( + "quant_platform_kit.strategy_lifecycle.health_dashboard.build_dashboard", + return_value=None, + ), + ): + self.assertEqual(DAILY_BRIEFING.main(), 0) + + for domain in DAILY_BRIEFING.DOMAINS: + report = json.loads( + (root / "data" / "daily-reports" / "2026-07-30" / f"{domain}.json").read_text( + encoding="utf-8" + ) + ) + self.assertFalse(report["ok"]) + self.assertEqual(report["data_status"], "unavailable") + self.assertIn( + {"code": "dashboard_data_unavailable", "error_type": "FileNotFoundError"}, + report["errors"], + ) + if __name__ == "__main__": unittest.main() From acf38f5aeb1a89d64656d75fa43961265625a059 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:51:02 +0800 Subject: [PATCH 3/4] test: isolate quant monitor dependency doubles Co-Authored-By: Codex --- .../tests/test_monitor_fail_closed.py | 23 +++++++++++++------ 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/ops/quant-monitor/tests/test_monitor_fail_closed.py b/ops/quant-monitor/tests/test_monitor_fail_closed.py index f067e27c..bad55bf0 100644 --- a/ops/quant-monitor/tests/test_monitor_fail_closed.py +++ b/ops/quant-monitor/tests/test_monitor_fail_closed.py @@ -1,7 +1,9 @@ import importlib.util import json import os +import sys import tempfile +import types import unittest from pathlib import Path from unittest import mock @@ -107,18 +109,25 @@ def test_health_cycle_skips_issue_creation_when_drift_is_unavailable(self) -> No def test_daily_briefing_marks_missing_dashboard_unavailable(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) + qpk = types.ModuleType("quant_platform_kit") + lifecycle = types.ModuleType("quant_platform_kit.strategy_lifecycle") + drift_detector = types.ModuleType("quant_platform_kit.strategy_lifecycle.drift_detector") + health_dashboard = types.ModuleType("quant_platform_kit.strategy_lifecycle.health_dashboard") + drift_detector.run_drift_detection = lambda _domain: [] + health_dashboard.build_dashboard = lambda **_kwargs: None with ( mock.patch.dict( os.environ, {"QUANT_MONITOR_ROOT": str(root), "DAY": "2026-07-30"}, ), - mock.patch( - "quant_platform_kit.strategy_lifecycle.drift_detector.run_drift_detection", - return_value=[], - ), - mock.patch( - "quant_platform_kit.strategy_lifecycle.health_dashboard.build_dashboard", - return_value=None, + mock.patch.dict( + sys.modules, + { + "quant_platform_kit": qpk, + "quant_platform_kit.strategy_lifecycle": lifecycle, + "quant_platform_kit.strategy_lifecycle.drift_detector": drift_detector, + "quant_platform_kit.strategy_lifecycle.health_dashboard": health_dashboard, + }, ), ): self.assertEqual(DAILY_BRIEFING.main(), 0) From c25a2d910779e2c8f4d677a7c3b3aff3c341f868 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:00:58 +0800 Subject: [PATCH 4/4] fix: deduplicate stable monitor incidents Co-Authored-By: Codex --- ops/quant-monitor/scripts/health_cycle.py | 48 +++++++++++++------ .../tests/test_monitor_fail_closed.py | 22 +++++++++ 2 files changed, 56 insertions(+), 14 deletions(-) diff --git a/ops/quant-monitor/scripts/health_cycle.py b/ops/quant-monitor/scripts/health_cycle.py index 33bfba60..d575bd05 100755 --- a/ops/quant-monitor/scripts/health_cycle.py +++ b/ops/quant-monitor/scripts/health_cycle.py @@ -42,6 +42,21 @@ def _alert_fingerprint(lines: list[str]) -> str: return hashlib.sha256(payload.encode("utf-8")).hexdigest() +def _strategy_health_alert(row: dict[str, Any]) -> tuple[str, str] | None: + try: + score = float(row.get("overall_score")) + except (TypeError, ValueError): + return None + if score >= SCORE_ALERT: + return None + profile = str(row.get("strategy_profile") or "?") + domain = str(row.get("domain") or "?") + return ( + f"[{domain}] {profile}: health_score={score:.1f}", + f"strategy_health_below_{SCORE_ALERT:g}:{domain}:{profile}", + ) + + def _alert_state_path(root: Path) -> Path: return root / _ALERT_STATE_RELATIVE_PATH @@ -201,17 +216,14 @@ def main() -> int: telegram_lines: list[str] = [] critical_lines: list[str] = [] + alert_identities: list[str] = [] for row in strategies: - try: - score = float(row.get("overall_score")) - except (TypeError, ValueError): - continue - if score >= SCORE_ALERT: - continue - profile = str(row.get("strategy_profile") or "?") - domain = str(row.get("domain") or "?") - telegram_lines.append(f"[{domain}] {profile}: health_score={score:.1f}") + alert = _strategy_health_alert(row) + if alert: + line, identity = alert + telegram_lines.append(line) + alert_identities.append(identity) for domain in DOMAINS: drifts = drift_results.get(domain, []) @@ -220,6 +232,9 @@ def main() -> int: label = f"[{domain}] {drift.strategy_profile}: drift_score={score:.2f}" if score >= DRIFT_CRITICAL: critical_lines.append(label) + alert_identities.append( + f"critical_drift:{domain}:{drift.strategy_profile}" + ) elif score >= DRIFT_REVIEW: pass # tracked via create_issues_for_domain below @@ -234,18 +249,23 @@ def main() -> int: body=f"Quant-monitor detected critical drift.\n\n- {line}", ) - data_error_lines = [ - f"[{error['domain']}] {error['code']} ({error['error_type']})" - for error in drift_errors - ] + data_error_lines: list[str] = [] + for error in drift_errors: + data_error_lines.append( + f"[{error['domain']}] {error['code']} ({error['error_type']})" + ) + alert_identities.append( + f"data_error:{error['domain']}:{error['code']}:{error['error_type']}" + ) if collector_payload_invalid: data_error_lines.append("[collector] dashboard_data_unavailable") + alert_identities.append("data_error: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) - fingerprint = _alert_fingerprint(notify_lines) + fingerprint = _alert_fingerprint(alert_identities) duplicate_alert_suppressed = _is_duplicate_alert(root, fingerprint) if not duplicate_alert_suppressed: telegram_sent = _send_telegram(body) diff --git a/ops/quant-monitor/tests/test_monitor_fail_closed.py b/ops/quant-monitor/tests/test_monitor_fail_closed.py index bad55bf0..e2d3328b 100644 --- a/ops/quant-monitor/tests/test_monitor_fail_closed.py +++ b/ops/quant-monitor/tests/test_monitor_fail_closed.py @@ -85,6 +85,28 @@ def test_health_cycle_alert_fingerprint_is_deduplicated_until_recovery(self) -> HEALTH_CYCLE._clear_alert(root) self.assertFalse(HEALTH_CYCLE._is_duplicate_alert(root, fingerprint)) + def test_health_cycle_score_changes_keep_the_same_incident_fingerprint(self) -> None: + first_line, first_identity = HEALTH_CYCLE._strategy_health_alert( + { + "domain": "us_equity", + "strategy_profile": "example", + "overall_score": 59.9, + } + ) + second_line, second_identity = HEALTH_CYCLE._strategy_health_alert( + { + "domain": "us_equity", + "strategy_profile": "example", + "overall_score": 59.8, + } + ) + + self.assertNotEqual(first_line, second_line) + self.assertEqual( + HEALTH_CYCLE._alert_fingerprint([first_identity]), + HEALTH_CYCLE._alert_fingerprint([second_identity]), + ) + def test_health_cycle_non_object_alert_state_is_a_cache_miss(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp)