Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions docs/ai_autonomy_architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -278,6 +278,18 @@ GitHub Codex App 是唯一 AI PR reviewer。AIAuditBridge 只保留月度审计

这样可以把策略优化先收敛为可审计、可回放的建议流,再逐步扩展到受控执行面。

Quant Monitor 的具体路由遵循同一边界:

- lifecycle score、status 或 drift 越过阈值,只生成 monitoring evidence;
- evidence 写入对应策略仓的去重 optimization issue,供 AI 诊断和有界、no-order
实验规划;
- 监控不因分数低或漂移高直接发送 Telegram;
- 数据/可信工件不可用、circuit breaker 或 optimization issue 记录失败,才升级为
人工运维告警。

“监测到劣化”只代表产生优化触发证据,不代表已授权启动优化周期,更不代表可以影响
broker、订单、资金或 live deployment。

---

## 4. 可落地的阶段性改造计划
Expand Down
14 changes: 8 additions & 6 deletions ops/quant-monitor/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,23 +13,24 @@ VPS Codex 定时监控(`codex-quant.timer` 每 30 分钟)+ 收盘简报(`c
| `QUANT_PLATFORM_KIT_ROOT` | `~/Projects/QuantPlatformKit` |
| `GLOBAL_TELEGRAM_CHAT_ID` | systemd 注入,勿提交 git |
| `GH_TOKEN` | `gh` 拉仓 + 开 Issue |
| `QSL_MONITOR_ISSUE_OWNER` | 3σ 漂移 Issue @ 的用户(默认 `Pigbibi`) |
| `QSL_GITHUB_REPO` | 非策略类 briefing Issue 的默认仓库;策略证据按 domain 写入对应策略仓 |

凭证:`scripts/load_telegram_env.sh` 从 GCP `quant-sentinel-telegram-bot-token` 加载。

## 每 30 分钟(health_check.sh)

1. `sync_strategy_repos.sh` — `git pull` 四策略仓 + QPK
2. `health_cycle.py` — `build_dashboard` + `run_drift_detection`
3. lifecycle `overall_score < 60` → Telegram 量化哨兵(研究/回测健康,不代表平台执行故障)
4. drift ≥ 0.50(~2σ)→ `create_issues_for_domain` 开 Issue(不 @)
5. drift ≥ 0.75(~3σ)→ Telegram + Issue **@owner**
3. lifecycle `overall_score < 60` 或 drift ≥ 0.50 → 生成 monitoring evidence
4. evidence → 对应策略仓的去重、issue-only AI optimization proposal
5. 分数或漂移本身不发 Telegram;数据/工件不可用或 Issue 记录失败才通知人工

## 每日收盘后(daily_briefing_pipeline.sh)

1. `daily_briefing_builder.py` → `data/daily-reports/YYYY-MM-DD/<domain>.json`
2. `AIAuditBridge/scripts/consume_daily_briefing.py --dispatch`
3. 正常 → quiet;review → Issue;critical → Telegram
3. 正常 → quiet;review/critical → issue-only AI optimization proposal
4. data unavailable、circuit breaker 或 proposal 记录失败 → Telegram

## 部署

Expand All @@ -41,4 +42,5 @@ bash ops/quant-monitor/scripts/deploy_to_vps.sh

- 不要手填 token/chat id 到仓库
- 报警只走量化哨兵 bot
- 日报默认不通知人,除非 `briefing_consumer` 判定 critical
- 策略健康证据只进入可审计的 issue-only 优化队列,不自动改策略、参数、仓位或部署
- 策略劣化记录成功后不通知人;只对数据/运行风险和记录失败 fail-closed 通知
12 changes: 12 additions & 0 deletions ops/quant-monitor/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,18 @@ contract 和大小限制校验后原子切换;代码仓库与 lifecycle 数据

Token 从 GCP Secret `quant-sentinel-telegram-bot-token` 加载;**不要**把 token 或 chat id 写进 git。

告警路由:

| 事件 | 处置 |
|------|------|
| lifecycle score / drift 劣化 | 写入对应策略仓的去重、issue-only AI optimization proposal |
| 数据或可信工件不可用 | Telegram |
| circuit breaker / runtime risk | Telegram |
| optimization proposal 记录失败 | Telegram |

监控证据只触发研究审查,不自动修改策略代码、live 参数、仓位、风险预算,不自动
merge 或 deploy。成功记录策略劣化后,monitor 正常结束,不再重复通知人工。

| 变量 | 说明 |
|------|------|
| `QUANT_SENTINEL_TELEGRAM_SECRET_NAME` | 默认 `quant-sentinel-telegram-bot-token` |
Expand Down
194 changes: 100 additions & 94 deletions ops/quant-monitor/scripts/health_cycle.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,12 @@
#!/usr/bin/env python3
"""VPS health cycle — roadmap task 7 (scores, drift, issues, Telegram)."""
"""VPS health cycle — route strategy evidence to issue-only optimization monitoring."""

from __future__ import annotations

import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
from datetime import datetime, timedelta, timezone
Expand Down Expand Up @@ -151,25 +150,10 @@ 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}: lifecycle_health_score={score:.1f}",
f"strategy_health_below_{SCORE_ALERT:g}:{domain}:{profile}",
)


def _build_alert_body(lines: list[str]) -> str:
return (
"🚨 quant-monitor strategy_lifecycle\n"
"• scope: research/backtest lifecycle health; not platform runtime execution\n"
"🚨 quant-monitor operational\n"
"• scope: data/evidence or optimization-record delivery failure\n"
+ "\n".join(f"• {line}" for line in lines)
)

Expand Down Expand Up @@ -213,17 +197,83 @@ def _clear_alert(root: Path) -> None:
pass


def _create_issues_for_available_domains(
def _build_monitoring_findings(
strategies: list[dict[str, Any]],
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
) -> list[Any]:
from service.strategy_watch import build_strategy_monitoring_finding

records: dict[tuple[str, str], dict[str, Any]] = {}
metric_keys = (
"overall_score",
"performance_score",
"risk_score",
"decay_score",
"stability_score",
"operational_score",
"status",
"as_of",
)
for row in strategies:
try:
score = float(row.get("overall_score"))
except (TypeError, ValueError):
continue
Comment on lines +218 to +221

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve status-only lifecycle findings

When a lifecycle row is review or critical but overall_score is missing or non-numeric, this early continue drops the row before its status is considered; the dashboard normalizer permits such rows by retaining the status and representing an invalid score as null. A status-only degradation therefore produces neither an optimization issue nor an operational error, so classify status independently of score parsing.

AGENTS.md reference: ops/quant-monitor/AGENTS.md:L45-L45

Useful? React with 👍 / 👎.

if score >= SCORE_ALERT:
continue
domain = str(row.get("domain") or "").strip()
profile = str(row.get("strategy_profile") or "").strip()
if not domain or not profile:
continue
record = records.setdefault(
(domain, profile),
{"metrics": {}, "signals": [], "severity": "medium", "generated_at": ""},
)
record["metrics"].update({key: row[key] for key in metric_keys if key in row})
record["signals"].append(
{
"metric": "overall_score",
"reason": f"overall_score={score:.1f} is below monitoring threshold {SCORE_ALERT:.1f}",
}
)
record["generated_at"] = str(row.get("as_of") or "")
if str(row.get("status") or "").lower() == "critical" or score <= 40.0:
record["severity"] = "high"

for domain, drifts in drift_results.items():
for drift in drifts:
score = float(drift.drift_score or 0.0)
if score < DRIFT_REVIEW:
continue
profile = str(drift.strategy_profile or "").strip()
if not profile:
continue
record = records.setdefault(
(domain, profile),
{"metrics": {}, "signals": [], "severity": "medium", "generated_at": ""},
)
record["metrics"]["drift_score"] = score
record["signals"].append(
{
"metric": "drift_score",
"reason": f"drift_score={score:.2f} exceeds monitoring threshold {DRIFT_REVIEW:.2f}",
}
)
if score >= DRIFT_CRITICAL:
record["severity"] = "high"

return [
build_strategy_monitoring_finding(
domain=domain,
profile=profile,
severity=record["severity"],
metrics=record["metrics"],
signals=record["signals"],
source="quant-monitor/health-cycle",
generated_at=record["generated_at"],
)
for (domain, profile), record in sorted(records.items())
]


def _send_telegram(text: str) -> bool:
Expand All @@ -239,46 +289,16 @@ def _send_telegram(text: str) -> bool:
return False


def _create_owner_issue(*, title: str, body: str) -> str | None:
repo = (os.environ.get("QSL_GITHUB_REPO") or "QuantStrategyLab/CnEquityStrategies").strip()
owner = (os.environ.get("QSL_MONITOR_ISSUE_OWNER") or "Pigbibi").strip()
full_body = f"{body}\n\ncc @{owner}"
try:
out = subprocess.check_output(
[
"gh",
"issue",
"create",
"--repo",
repo,
"--title",
title,
"--body",
full_body,
"--label",
"monitoring",
"--label",
"drift-critical",
],
text=True,
stderr=subprocess.STDOUT,
env={**os.environ, "GH_PROMPT": "disabled"},
)
return out.strip()
except Exception:
return None


def main() -> int:
root = Path(os.environ.get("QUANT_MONITOR_ROOT") or Path(__file__).resolve().parents[1])
out_dir = root / "data" / "health"
dash_dir = out_dir / "dashboard"
out_dir.mkdir(parents=True, exist_ok=True)

from quant_platform_kit.strategy_lifecycle.codex_integration import create_issues_for_domain
from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection
from quant_platform_kit.strategy_lifecycle.health_dashboard import build_dashboard
from quant_platform_kit.strategy_lifecycle.performance_monitor import run_monitor
from scripts.run_strategy_optimization_watcher import dispatch_strategy_watch_findings

ready_domains, artifact_errors = _load_lifecycle_artifact_status(root)
snapshot_results, drift_results, lifecycle_errors = _refresh_and_collect_drift(
Expand Down Expand Up @@ -338,41 +358,14 @@ def main() -> int:
elif not json_path.is_file():
collector_payload_invalid = True

telegram_lines: list[str] = []
critical_lines: list[str] = []
alert_identities: list[str] = []

for row in strategies:
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, [])
for drift in drifts:
score = float(drift.drift_score or 0.0)
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

issue_results = _create_issues_for_available_domains(
drift_results,
create_issues_for_domain,
monitoring_findings = _build_monitoring_findings(strategies, drift_results)
optimization_watch = dispatch_strategy_watch_findings(

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Route watcher exceptions through the operational fallback

When gh is missing or cannot be executed, list_open_issue_urls raises an OSError/FileNotFoundError, which the watcher helper does not convert into its errors result; this unguarded call therefore terminates health_cycle.py before optimization_error_lines and _send_telegram run. Catch watcher-level exceptions here and represent them as an issue-record failure so the promised fail-closed Telegram notification is still delivered.

AGENTS.md reference: ops/quant-monitor/AGENTS.md:L45-L46

Useful? React with 👍 / 👎.

monitoring_findings,
dry_run=False,
comment_existing=False,
)

for line in critical_lines:
_create_owner_issue(
title=f"[monitor] critical drift — {line}",
body=f"Quant-monitor detected critical drift.\n\n- {line}",
)

data_error_lines: list[str] = []
for error in data_errors:
data_error_lines.append(
Expand All @@ -384,7 +377,16 @@ def main() -> int:
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
optimization_error_lines: list[str] = []
optimization_errors = int(optimization_watch.get("errors") or 0)
if optimization_errors:
optimization_error_lines.append(
f"[optimization] issue_record_failed ({optimization_errors})"
)
alert_identities.append(
f"optimization_error:issue_record_failed:{optimization_errors}"
)
Comment on lines +386 to +388

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Fingerprint the failed optimization records, not only their count

When consecutive cycles each have the same number of issue-record failures but for different repositories or strategies, this identity is unchanged, so _is_duplicate_alert suppresses the later Telegram alert as if it were the same incident. Include the failed repository and watcher issue key in the fingerprint so a new delivery failure cannot be hidden merely because its count matches the preceding cycle.

AGENTS.md reference: ops/quant-monitor/AGENTS.md:L45-L46

Useful? React with 👍 / 👎.

notify_lines = data_error_lines + optimization_error_lines
telegram_sent = False
duplicate_alert_suppressed = False
if notify_lines:
Expand All @@ -407,7 +409,11 @@ def main() -> int:
"duplicate_alert_suppressed": duplicate_alert_suppressed,
"data_errors": data_errors,
"snapshot_count": sum(len(rows) for rows in snapshot_results.values()),
"issues_created": len([r for r in issue_results if r.get("issue_url")]),
"optimization_findings": len(monitoring_findings),
"optimization_issues_created": len(
[result for result in optimization_watch.get("issues", []) if result.get("created")]
),
"optimization_issue_errors": optimization_errors,
"ok": not notify_lines and not collector_payload_invalid,
"collector_payload_valid": not collector_payload_invalid,
"snapshot_data_status": normalized_payload.get("data_status"),
Expand Down
1 change: 0 additions & 1 deletion ops/quant-monitor/systemd/codex-quant.service.example
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ Environment=QUANT_SENTINEL_GCP_PROJECT=firstradequant
# Set on VPS only — do not commit real chat id to git:
Environment=GLOBAL_TELEGRAM_CHAT_ID=
Environment=QSL_GITHUB_REPO=QuantStrategyLab/CnEquityStrategies
Environment=QSL_MONITOR_ISSUE_OWNER=Pigbibi
RuntimeDirectory=quant-monitor
RuntimeDirectoryMode=0750
ExecStartPre=/bin/bash /home/ubuntu/Projects/AIAuditBridge/ops/quant-monitor/scripts/load_telegram_env.sh /run/quant-monitor/telegram.env
Expand Down
Loading