126 lines
4.7 KiB
Python
126 lines
4.7 KiB
Python
"""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()
|