feat(strategy): require real crypto performance evidence #125
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,197 @@ | ||
| #!/usr/bin/env python3 | ||
| """Produce a fail-closed S2 baseline review for crypto live-pool rotation.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import argparse | ||
| import csv | ||
| import json | ||
| import math | ||
| from datetime import datetime, timezone | ||
| from pathlib import Path | ||
| from typing import Any | ||
|
|
||
| PROFILE = "crypto_live_pool_rotation" | ||
| GATE_NAMES = { | ||
| "H1": "falsifiable_hypothesis", | ||
| "H2": "data_provenance", | ||
| "H3": "sample_adequacy", | ||
| "H4": "benchmark_comparison", | ||
| "H5": "cost_model", | ||
| "H6": "cost_stress", | ||
| "H7": "oos_folds", | ||
| "H8": "purge_embargo", | ||
| "H9": "leakage_control", | ||
| "H10": "parameter_stability", | ||
| "H11": "risk_metrics", | ||
| "H12": "reproducibility", | ||
| } | ||
|
|
||
|
|
||
| def _gate(gate_id: str, reason: str, *, status: str = "insufficient_evidence", refs: list[str] | None = None) -> dict[str, Any]: | ||
| return { | ||
| "id": gate_id, | ||
| "name": GATE_NAMES[gate_id], | ||
| "status": status, | ||
| "reason_codes": [reason], | ||
| "evidence_refs": refs or [], | ||
| } | ||
|
|
||
|
|
||
| def build_review(*, performance_summary: Path, walkforward_summary: Path) -> dict[str, Any]: | ||
| missing = [] | ||
| if not _usable_performance_csv(performance_summary): | ||
| missing.append(str(performance_summary)) | ||
| if not _usable_walkforward_csv(walkforward_summary): | ||
| missing.append(str(walkforward_summary)) | ||
| reason = "MISSING_OR_INVALID_REAL_PERFORMANCE_ARTIFACT" if missing else "BASELINE_REVIEW_NOT_YET_FROZEN" | ||
| gates = [_gate(gate_id, reason) for gate_id in GATE_NAMES] | ||
| return { | ||
| "schema_version": "strategy_review.v1", | ||
| "profile": PROFILE, | ||
| "decision": "insufficient_evidence", | ||
| "promotion_allowed": False, | ||
| "score": 0, | ||
| "hard_gates": gates, | ||
| "scorecard": {"total": 0, "max": 100, "scored_gates": 0}, | ||
| "blocking_reason_codes": [reason], | ||
| "evidence": { | ||
| "metrics_kind": "performance", | ||
| "data_source": "unavailable" if missing else "unfrozen", | ||
| "sample_count": 0, | ||
| "oos_folds": 0, | ||
| "placeholder_metrics": False, | ||
| "provenance": { | ||
| "snapshot": {"source_revision": "unavailable", "cost_model": "unavailable", "data_timestamp": "unavailable", "status": "unavailable"}, | ||
| "backtest": {"source_revision": "unavailable", "cost_model": "unavailable", "data_timestamp": "unavailable", "status": "unavailable"}, | ||
| }, | ||
| "missing_artifacts": missing, | ||
| "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), | ||
| }, | ||
| "decision_packet": { | ||
| "strategy_what": "按评分选择加密资产并进行 live-pool rotation；当前仅评审流程，不执行交易。", | ||
| "return_source": "真实 performance artifacts 未提供，收益来源无法确认。", | ||
| "loss_scenarios": "未完成真实回测，主要亏损场景无法确认；不得用 placeholder 指标替代。", | ||
| "max_risk": "证据不足，最大回撤、容量和流动性风险均不可确认。", | ||
| "evidence_sufficiency": "insufficient_evidence", | ||
| "version_change": "新增 fail-closed 基线评审与人工决策 packet；未改变 live 参数。", | ||
| "system_recommendation": "insufficient_evidence", | ||
| "technical_evidence_refs": missing, | ||
| "automation_boundary": { | ||
| "research_auto_after_hard_gates": True, | ||
| "shadow_auto_after_hard_gates": True, | ||
| "canary_mode": "bounded_preapproved_only", | ||
| "canary_limits": {"max_capital": 1000.0, "capital_currency": "USD", "max_duration_days": 14, "max_drawdown_fraction": 0.05, "max_leverage": 1.0, "max_concurrency": 1}, | ||
| "auto_scale_allowed": False, | ||
| "normal_live_requires_human": True, | ||
| "funding_leverage_risk_override_requires_human": True, | ||
| "hard_risk_auto_pause_rollback": True, | ||
| }, | ||
| "allowed_human_decisions": ["approve_research", "reject_rollback"], | ||
| }, | ||
| } | ||
|
|
||
|
|
||
| def _read_csv(path: Path) -> tuple[list[str], list[list[str]]] | None: | ||
| if not path.exists() or not path.is_file(): | ||
| return None | ||
| try: | ||
| with path.open(newline="", encoding="utf-8") as handle: | ||
| reader = csv.reader(handle) | ||
| header = next(reader, None) | ||
| data_rows = list(reader) | ||
| if not header or not data_rows: | ||
| return None | ||
| return ["".join(ch for ch in cell.lower() if ch.isalnum() or ch == "_") for cell in header], data_rows | ||
| except (OSError, UnicodeError, csv.Error): | ||
| return None | ||
|
|
||
|
|
||
| def _has_numeric_value(header: list[str], rows: list[list[str]], candidates: set[str]) -> bool: | ||
| for row in rows: | ||
| for index, name in enumerate(header): | ||
| if name not in candidates or index >= len(row): | ||
| continue | ||
| try: | ||
| value = float(row[index]) | ||
| if math.isfinite(value): | ||
| return True | ||
| except (TypeError, ValueError): | ||
| continue | ||
| return False | ||
|
|
||
|
|
||
| def _usable_performance_csv(path: Path) -> bool: | ||
| parsed = _read_csv(path) | ||
| if parsed is None: | ||
| return False | ||
| header, rows = parsed | ||
| metrics = {"cagr", "sharpe", "calmar", "maxdrawdown", "max_dd", "annualizedvolatility"} | ||
| if "strategy" in header: | ||
| strategy_index = header.index("strategy") | ||
| if not any(len(row) > strategy_index and row[strategy_index].strip().lower() == "final_score" for row in rows): | ||
| return False | ||
| return bool(set(header).intersection(metrics)) and _has_numeric_value(header, rows, metrics) | ||
|
There was a problem hiding this comment.
When Useful? React with 👍 / 👎. |
||
|
|
||
|
|
||
| def _usable_walkforward_csv(path: Path) -> bool: | ||
| parsed = _read_csv(path) | ||
| if parsed is None: | ||
| return False | ||
| header, rows = parsed | ||
| required = {"window_id", "test_start", "test_end"} | ||
| metrics = {"window_cagr", "window_sharpe", "window_max_drawdown", "h30_precision", "h60_precision", "h90_precision"} | ||
| if not required.issubset(set(header)) or not set(header).intersection(metrics): | ||
| return False | ||
| positions = {name: header.index(name) for name in required} | ||
| for row in rows: | ||
| try: | ||
| if any(not row[index].strip() for index in positions.values()): | ||
| continue | ||
| int(row[positions["window_id"]]) | ||
| datetime.fromisoformat(row[positions["test_start"]].replace("Z", "+00:00")) | ||
| datetime.fromisoformat(row[positions["test_end"]].replace("Z", "+00:00")) | ||
| except (IndexError, TypeError, ValueError): | ||
|
Comment on lines
+153
to
+154
There was a problem hiding this comment.
When a hand-built or corrupted Useful? React with 👍 / 👎. |
||
| continue | ||
| if _has_numeric_value(header, [row], metrics): | ||
| return True | ||
| return False | ||
|
|
||
|
|
||
| def render_markdown(review: dict[str, Any]) -> str: | ||
| lines = [ | ||
| f"# Baseline review: `{review['profile']}`", | ||
| "", | ||
| f"- Decision: **{review['decision']}**", | ||
| f"- Promotion allowed: **{review['promotion_allowed']}**", | ||
| f"- Score: **{review['score']}/100**", | ||
| "", | ||
| "This report is fail-closed. No real performance artifact is promoted by this command.", | ||
| "", | ||
| "| Gate | Status | Reason |", | ||
| "|---|---|---|", | ||
| ] | ||
| for gate in review["hard_gates"]: | ||
| lines.append(f"| {gate['id']} {gate['name']} | {gate['status']} | {', '.join(gate['reason_codes'])} |") | ||
| lines.extend(["", "## Blocking reasons", "", *[f"- `{reason}`" for reason in review["blocking_reason_codes"]]]) | ||
| return "\n".join(lines) + "\n" | ||
|
|
||
|
|
||
| def main(argv: list[str] | None = None) -> int: | ||
| parser = argparse.ArgumentParser() | ||
| parser.add_argument("--performance-summary", type=Path, default=Path("data/reports/performance_summary.csv")) | ||
| parser.add_argument("--walkforward-summary", type=Path, default=Path("data/reports/walkforward_validation_summary.csv")) | ||
| parser.add_argument("--output-json", type=Path, required=True) | ||
| parser.add_argument("--output-md", type=Path, required=True) | ||
| args = parser.parse_args(argv) | ||
| review = build_review(performance_summary=args.performance_summary, walkforward_summary=args.walkforward_summary) | ||
| args.output_json.parent.mkdir(parents=True, exist_ok=True) | ||
| args.output_md.parent.mkdir(parents=True, exist_ok=True) | ||
| args.output_json.write_text(json.dumps(review, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") | ||
| args.output_md.write_text(render_markdown(review), encoding="utf-8") | ||
| print(json.dumps({"decision": review["decision"], "promotion_allowed": False, "output": str(args.output_json)})) | ||
| return 0 | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| raise SystemExit(main()) | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When the publish job reaches this verification step, it calls the lifecycle runner with an empty param set, so
CryptoLivePoolBacktestRunnerfalls back to its hard-codedDEFAULT_BACKTEST_CONFIG(top_n=2,weighting='equal'). I checked the production research path:src/pipeline.pybacktests usingconfig/default.yaml, whose strategy istop_n: 3andweighting: inverse_vol. This means the uploaded preflight can pass while validating a different portfolio than the crypto live-pool rotation operators actually run; pass the loaded strategy config or make the runner defaults match the live config.Useful? React with 👍 / 👎.