diff --git a/.github/workflows/monthly_publish.yml b/.github/workflows/monthly_publish.yml index b325132..add7ffd 100644 --- a/.github/workflows/monthly_publish.yml +++ b/.github/workflows/monthly_publish.yml @@ -85,6 +85,13 @@ jobs: - name: Write Monthly Telegram Preview run: python scripts/run_monthly_build_telegram.py --print-only --output-path data/output/monthly_telegram.txt + - name: Generate Strategy Metrics + run: | + set -euo pipefail + python scripts/generate_strategy_metrics.py + env: + STRATEGY_METRICS_REPO: ${{ github.repository }} + - name: Assemble Monthly Report Bundle id: monthly-report run: | @@ -97,6 +104,16 @@ jobs: print(f"artifact_name={payload['artifact_name']}") PY + - name: Include Strategy Metrics in Report Bundle + run: | + set -euo pipefail + if [ -f data/output/strategy_metrics.json ]; then + cp data/output/strategy_metrics.json data/output/monthly_report_bundle/strategy_metrics.json + echo "Included strategy_metrics.json in report bundle" + else + echo "Warning: strategy_metrics.json not found — watcher will skip" + fi + - name: Append Monthly Report Job Summary run: cat data/output/monthly_report_bundle/job_summary.md >> "$GITHUB_STEP_SUMMARY" diff --git a/scripts/generate_strategy_metrics.py b/scripts/generate_strategy_metrics.py new file mode 100644 index 0000000..1f7521d --- /dev/null +++ b/scripts/generate_strategy_metrics.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""Generate strategy_metrics.json consumed by AIAuditBridge strategy optimization watcher. + +Reads the monthly shadow build summary and per-track release indexes, then writes a +JSON payload that the watcher can evaluate against degradation thresholds. + +The watcher (AIAuditBridge service/strategy_optimization_policy.py) checks: + sharpe, cagr, calmar, win_rate (higher-is-better, relative-drop threshold) + max_dd (lower-is-better, absolute-worsening threshold) + +Metrics that are not yet available (e.g. backtest returns) are omitted — the watcher +gracefully skips missing metrics without raising false degradation signals. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +DEFAULT_REPO = "QuantStrategyLab/CryptoLivePoolPipelines" + + +def _safe_float(value: Any) -> float | None: + if value is None or isinstance(value, bool): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _track_metrics(index_table: pd.DataFrame) -> dict[str, Any]: + """Extract current (latest period) and baseline (all-time average) metrics.""" + if index_table.empty: + return {"current_metrics": {}, "baseline_metrics": {}} + latest = index_table.iloc[-1] + numeric_columns = index_table.select_dtypes(include=["number"]).columns.tolist() + + current: dict[str, float] = {} + baseline: dict[str, float] = {} + for col in numeric_columns: + cur = _safe_float(latest.get(col)) + base = _safe_float(index_table[col].mean()) + if cur is not None: + current[col] = cur + if base is not None: + baseline[col] = base + + return {"current_metrics": current, "baseline_metrics": baseline} + + +def generate_strategy_metrics( + summary_path: Path, + *, + repo: str = DEFAULT_REPO, + output_path: Path | None = None, +) -> dict[str, Any]: + """Generate the strategy_metrics.json payload.""" + if not summary_path.exists(): + raise FileNotFoundError(f"shadow build summary not found: {summary_path}") + + summary = json.loads(summary_path.read_text(encoding="utf-8")) + generated_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + snapshots: list[dict[str, Any]] = [] + + # ── official baseline track ────────────────────────────────────── + baseline_cfg = summary.get("official_baseline", {}) + baseline_profile = str(baseline_cfg.get("profile") or "baseline_blended_rank") + + baseline_live_pool = baseline_cfg.get("live_pool_path") + if baseline_live_pool: + baseline_dir = Path(baseline_live_pool).parent.parent # .../version/live_pool.json → parent dir + baseline_index = baseline_dir / "release_index.csv" + if baseline_index.exists(): + index = pd.read_csv(baseline_index) + metrics = _track_metrics(index) + snapshots.append( + { + "repo": repo, + "strategy_profile": baseline_profile, + "plugin": "", + "current_metrics": metrics["current_metrics"], + "baseline_metrics": metrics["baseline_metrics"], + "source": str(summary_path), + "generated_at": generated_at, + } + ) + + # ── shadow candidate tracks ───────────────────────────────────── + shadow_cfg = summary.get("shadow_candidate_tracks", {}) + tracks = shadow_cfg.get("tracks", []) + if not isinstance(tracks, list): + tracks = [] + + _root_dir = Path(shadow_cfg.get("root_dir", "")) # noqa: F841 + for track in tracks: + if not isinstance(track, dict): + continue + track_id = str(track.get("track_id") or "") + profile_name = str(track.get("profile_name") or "") + release_index_path = track.get("release_index_path") + + index_path: Path | None = None + if release_index_path: + index_path = Path(release_index_path) + if not index_path.is_absolute(): + index_path = PROJECT_ROOT / index_path + + if index_path is None or not index_path.exists(): + snapshots.append( + { + "repo": repo, + "strategy_profile": profile_name, + "plugin": track_id, + "current_metrics": {}, + "baseline_metrics": {}, + "source": str(index_path) if index_path else "", + "generated_at": generated_at, + } + ) + continue + + index = pd.read_csv(index_path) + metrics = _track_metrics(index) + snapshots.append( + { + "repo": repo, + "strategy_profile": profile_name, + "plugin": track_id, + "current_metrics": metrics["current_metrics"], + "baseline_metrics": metrics["baseline_metrics"], + "source": str(index_path), + "generated_at": generated_at, + } + ) + + payload: dict[str, Any] = { + "repo": repo, + "generated_at": generated_at, + "source": "monthly_shadow_build", + "snapshots": snapshots, + } + + if output_path is not None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text( + json.dumps(payload, ensure_ascii=False, indent=2, default=str), + encoding="utf-8", + ) + print(f"Wrote strategy_metrics.json → {output_path}") + print(f" snapshots: {len(snapshots)}") + for s in snapshots: + profile = s.get("strategy_profile", "") + n_metrics = len(s.get("current_metrics", {})) + print(f" - {profile}: {n_metrics} metrics") + + return payload + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Generate strategy_metrics.json for the AIAuditBridge strategy optimization watcher." + ) + parser.add_argument( + "--summary", + default="data/output/monthly_shadow_build_summary.json", + help="Path to the shadow build summary JSON (default: data/output/monthly_shadow_build_summary.json)", + ) + parser.add_argument( + "--repo", + default=DEFAULT_REPO, + help="Repository identifier for the metrics payload", + ) + parser.add_argument( + "--output", + default="data/output/strategy_metrics.json", + help="Output path for strategy_metrics.json", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + summary_path = Path(args.summary) + if not summary_path.is_absolute(): + summary_path = PROJECT_ROOT / summary_path + + output_path = Path(args.output) + if not output_path.is_absolute(): + output_path = PROJECT_ROOT / output_path + + try: + generate_strategy_metrics( + summary_path, + repo=args.repo, + output_path=output_path, + ) + except FileNotFoundError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + except (json.JSONDecodeError, ValueError) as exc: + print(f"Error: invalid input format — {exc}", file=sys.stderr) + return 1 + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())