Add S/R v2 research and validation harness

scripts/compare_sr_variants.py

@@ -0,0 +1,125 @@
+"""Compare two audited local S/R backtest reports by setup identity.
+
+Reports must be generated with ``--sr-audit``. The comparison is read-only
+apart from its explicit CSV/JSON outputs under the caller-selected paths.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import math
+from pathlib import Path
+
+
+def _args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("control")
+    parser.add_argument("variant")
+    parser.add_argument("--out-csv", required=True)
+    parser.add_argument("--out-json", required=True)
+    return parser.parse_args()
+
+
+def _load(path: str) -> dict:
+    with Path(path).open(encoding="utf-8") as handle:
+        report = json.load(handle)
+    if report.get("sr_candidate_audit") is None:
+        raise SystemExit(f"Report lacks sr_candidate_audit; rerun with --sr-audit: {path}")
+    return report
+
+
+def _key(row: dict) -> tuple[str, str, str]:
+    return row["symbol"], row["date"], row["direction"]
+
+
+def _cohort_stats(rows: list[dict]) -> dict:
+    net = [float(row.get("net_r", 0.0)) for row in rows]
+    hold = [float(row.get("hold30_r", 0.0)) for row in rows]
+    trimmed = sorted(net, reverse=True)[math.ceil(len(net) * 0.05):]
+    return {
+        "count": len(rows),
+        "net_avg_r": round(sum(net) / len(net), 4) if net else None,
+        "net_avg_r_ex_top5": round(sum(trimmed) / len(trimmed), 4) if trimmed else None,
+        "hold30_avg_r": round(sum(hold) / len(hold), 4) if hold else None,
+    }
+
+
+def _production_book(report: dict) -> dict | None:
+    runs = ((report.get("portfolio_monitor") or {}).get("runs") or [])
+    row = next(
+        (
+            run for run in runs
+            if run.get("is_production") and run.get("lookback") == "all"
+        ),
+        None,
+    )
+    if row is None:
+        return None
+    return {
+        key: row.get(key)
+        for key in ("sharpe", "cagr_pct", "max_drawdown_pct", "trades", "skipped_book_full")
+    }
+
+
+def main() -> None:
+    args = _args()
+    control = _load(args.control)
+    variant = _load(args.variant)
+    control_rows = {_key(row): row for row in control["sr_candidate_audit"]}
+    variant_rows = {_key(row): row for row in variant["sr_candidate_audit"]}
+    control_q = {key for key, row in control_rows.items() if row.get("qualified")}
+    variant_q = {key for key, row in variant_rows.items() if row.get("qualified")}
+
+    retained = control_q & variant_q
+    added = variant_q - control_q
+    removed = control_q - variant_q
+    union = sorted(control_q | variant_q, key=lambda key: (key[1], key[0], key[2]))
+
+    csv_path = Path(args.out_csv)
+    csv_path.parent.mkdir(parents=True, exist_ok=True)
+    fields = [
+        "symbol", "date", "direction", "cohort",
+        "control_rr", "variant_rr", "control_prob", "variant_prob",
+        "control_sources", "variant_sources", "control_net_r", "variant_net_r",
+        "control_hold30_r", "variant_hold30_r",
+    ]
+    with csv_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fields)
+        writer.writeheader()
+        for key in union:
+            c = control_rows.get(key) or {}
+            v = variant_rows.get(key) or {}
+            cohort = "retained" if key in retained else "added" if key in added else "removed"
+            writer.writerow({
+                "symbol": key[0], "date": key[1], "direction": key[2], "cohort": cohort,
+                "control_rr": c.get("rr"), "variant_rr": v.get("rr"),
+                "control_prob": c.get("primary_prob"), "variant_prob": v.get("primary_prob"),
+                "control_sources": "+".join(c.get("primary_sources") or []),
+                "variant_sources": "+".join(v.get("primary_sources") or []),
+                "control_net_r": c.get("net_r"), "variant_net_r": v.get("net_r"),
+                "control_hold30_r": c.get("hold30_r"), "variant_hold30_r": v.get("hold30_r"),
+            })
+
+    summary = {
+        "control_report": str(Path(args.control)),
+        "variant_report": str(Path(args.variant)),
+        "control_variant": (control.get("params") or {}).get("sr_variant"),
+        "variant": (variant.get("params") or {}).get("sr_variant"),
+        "retained": _cohort_stats([variant_rows[key] for key in retained]),
+        "added": _cohort_stats([variant_rows[key] for key in added]),
+        "removed": _cohort_stats([control_rows[key] for key in removed]),
+        "control_book": _production_book(control),
+        "variant_book": _production_book(variant),
+    }
+    json_path = Path(args.out_json)
+    json_path.parent.mkdir(parents=True, exist_ok=True)
+    with json_path.open("w", encoding="utf-8") as handle:
+        json.dump(summary, handle, indent=2)
+        handle.write("\n")
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == "__main__":
+    main()