#!/usr/bin/env python3 """ backtest/replay_scorecard.py — read a replay record, score the indicators ========================================================================= READ-ONLY analysis of one (or several) replay-record folders written by ``backtest/replay_record.py``: which indicator votes actually earned their weight in the book, and what would have changed if one of them had been removed or re-weighted. Six lenses: 1. INDICATOR SCORECARD — for every timeframe role (primary / c1 / c2) and every indicator: mean R when that indicator AGREED with the trade, when it OPPOSED it, when it was NEUTRAL, and when it did not vote at all ("absent" — divergence and candlestick only fire on a pattern, so a missing row is "no opinion", never a zero vote). 2. CONTRIBUTION CORRELATION — rank correlation between how much an indicator pushed the composite in the trade's own direction and the R the trade made. Positive = pushing harder went with better trades. 3. SCORECARD BY REGIME — the same agree/oppose/neutral split, primary timeframe only, cut by the market regime at decision time. Weights are already regime-dependent; this says whether that is paying. 4. CALENDAR BUCKETS — trades entered just before / just after a high-impact USD release, trades that were held through one, quiet trades, and US holidays. 5. WHAT-IF (--what-if) — recompute each trade's primary composite with one indicator dropped or re-weighted, and count the trades that would have lost their direction or changed strength band. IN-BOOK ONLY: it can never show the trades a change would ADD, and it ignores every gate. Any real change needs a full backtest rerun (CLAUDE.md rule 1). 6. BAR-LEVEL (--bars, needs bars.parquet) — the same what-if over EVERY analyzed bar, not just the accepted trades, scored against the plain forward move in ATRs; plus a per-indicator table asking whether that indicator's vote, on its own, pointed the right way. ``--check`` re-derives the record's own arithmetic (weights, weighted scores, composites, regime multipliers, bar/trade counts) and exits non-zero if any invariant is broken. Usage: python backtest/replay_scorecard.py backtest/records/m5_12mo python backtest/replay_scorecard.py RUN_A RUN_B --min-n 30 python backtest/replay_scorecard.py RUN --what-if python backtest/replay_scorecard.py RUN --bars python backtest/replay_scorecard.py RUN --check """ from __future__ import annotations import argparse import os import sys from pathlib import Path from typing import Dict, List, Optional, Sequence, Tuple import numpy as np import pandas as pd # Ensure project root on sys.path so this script can be invoked from # anywhere (same idiom as backtest/gpu_backtest.py). PROJECT_ROOT = Path(__file__).resolve().parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) # Contract defaults. backtest.replay_record owns the real values; they are # copied in by _load_record_module() at CLI time so that the compute # functions below stay unit-testable without that module present. INDICATORS: Tuple[str, ...] = ( "ema_trend", "rsi", "macd", "bollinger", "adx", "roc", "keltner", "divergence", "candlestick", ) ROLES: Tuple[str, ...] = ("primary", "c1", "c2") COMPOSITE_THRESHOLDS: Dict[str, float] = {"strong": 40.0, "weak": 15.0} # Monkeypatch hook: set this to a callable # ``(eff_weights, wscores) -> (score, signal_int)`` to run the compute # functions without backtest.replay_record on the path. _composite_from_rows = None BUCKETS = ("agree", "oppose", "neutral", "absent") # ── record module (lazy — compute functions must not need it) ───────── def _load_record_module(): """Import backtest.replay_record and adopt its constants.""" try: from backtest import replay_record # noqa: E402 (lazy by design) except Exception as exc: # pragma: no cover - CLI path raise ImportError( "backtest/replay_record.py is not importable " f"({exc}). It is the loader for replay-record folders; " "this script cannot read a run without it." ) from exc global INDICATORS, ROLES, COMPOSITE_THRESHOLDS, _composite_from_rows INDICATORS = tuple(getattr(replay_record, "INDICATORS", INDICATORS)) ROLES = tuple(getattr(replay_record, "ROLES", ROLES)) COMPOSITE_THRESHOLDS = dict( getattr(replay_record, "COMPOSITE_THRESHOLDS", COMPOSITE_THRESHOLDS)) if _composite_from_rows is None: _composite_from_rows = replay_record.composite_from_rows return replay_record def _composite(eff_weights, wscores) -> Tuple[float, int]: """composite_from_rows, via the hook or a lazy import.""" global _composite_from_rows if _composite_from_rows is None: _composite_from_rows = _load_record_module().composite_from_rows return _composite_from_rows(eff_weights, wscores) def _weak() -> float: return float(COMPOSITE_THRESHOLDS.get("weak", 15.0)) def _strong() -> float: return float(COMPOSITE_THRESHOLDS.get("strong", 40.0)) # ── small formatting helpers ───────────────────────────────────────── def _fmt_r(value) -> str: return " -" if value is None or pd.isna(value) else f"{value:+6.3f}" def _fmt_total(value) -> str: return " -" if value is None or pd.isna(value) else f"{value:+7.1f}" def _fmt_pct(value) -> str: return " -" if value is None or pd.isna(value) else f"{100 * value:3.0f}%" def _fmt_num(value, nd: int = 3) -> str: return " -" if value is None or pd.isna(value) else f"{value:+.{nd}f}" def _dir_sign(direction) -> float: return -1.0 if str(direction).upper().startswith("S") else 1.0 def _stats(r: pd.Series) -> Dict[str, float]: """n / mean R / win rate / total R for one bucket of trades.""" r = pd.to_numeric(pd.Series(r), errors="coerce").dropna() if r.empty: return {"n": 0, "mean_r": float("nan"), "win_rate": float("nan"), "total_r": 0.0} return {"n": int(len(r)), "mean_r": float(r.mean()), "win_rate": float((r > 0).mean()), "total_r": float(r.sum())} def _spearman(a: pd.Series, b: pd.Series) -> float: """Rank correlation without scipy (average-tie ranks + corrcoef).""" frame = pd.DataFrame({"a": pd.to_numeric(a, errors="coerce"), "b": pd.to_numeric(b, errors="coerce")}).dropna() if len(frame) < 3: return float("nan") ra, rb = frame["a"].rank(), frame["b"].rank() if ra.nunique() < 2 or rb.nunique() < 2: return float("nan") return float(np.corrcoef(ra.to_numpy(), rb.to_numpy())[0, 1]) def _indicator_names(indicators: pd.DataFrame) -> List[str]: """Contract order first, then anything extra the record carried.""" seen = set(indicators["indicator"].dropna().unique()) \ if "indicator" in indicators.columns else set() names = [i for i in INDICATORS] names += sorted(n for n in seen if n not in set(names)) return names def _roles_present(indicators: pd.DataFrame) -> List[str]: if "tf_role" not in indicators.columns: return [] seen = set(indicators["tf_role"].dropna().unique()) roles = [r for r in ROLES if r in seen] roles += sorted(r for r in seen if r not in set(ROLES)) return roles # ── 1. indicator scorecard ─────────────────────────────────────────── def indicator_scorecard(trades: pd.DataFrame, indicators: pd.DataFrame, min_n: int = 20) -> pd.DataFrame: """Mean R per (timeframe role, indicator, agree/oppose/neutral/absent). The trade universe for a role is every trade that has at least one indicator row for that role, so "absent" means "this indicator did not vote on that trade", not "the role was missing". """ out: List[dict] = [] hidden = 0 if trades.empty or indicators.empty: frame = pd.DataFrame(out, columns=["tf_role", "indicator", "bucket", "n", "mean_r", "win_rate", "total_r"]) frame.attrs["n_hidden"] = 0 frame.attrs["min_n"] = min_n return frame r_by_trade = trades.set_index("trade_id")["r_multiple"] for role in _roles_present(indicators): role_rows = indicators[indicators["tf_role"] == role] universe = pd.Index(role_rows["trade_id"].unique()) universe = universe.intersection(r_by_trade.index) if len(universe) == 0: continue for name in _indicator_names(indicators): sub = role_rows[role_rows["indicator"] == name] agrees = dict(zip(sub["trade_id"], sub["agrees"])) groups: Dict[str, List] = {b: [] for b in BUCKETS} for tid in universe: bucket = agrees.get(tid) key = "absent" if bucket is None or pd.isna(bucket) \ else str(bucket) groups.setdefault(key, []).append(tid) for bucket, tids in groups.items(): stat = _stats(r_by_trade.loc[tids]) if tids else _stats( pd.Series(dtype=float)) if stat["n"] < min_n: hidden += 1 if stat["n"] else 0 continue out.append({"tf_role": role, "indicator": name, "bucket": bucket, **stat}) frame = pd.DataFrame(out, columns=["tf_role", "indicator", "bucket", "n", "mean_r", "win_rate", "total_r"]) frame.attrs["n_hidden"] = hidden frame.attrs["min_n"] = min_n return frame def contribution_corr(trades: pd.DataFrame, indicators: pd.DataFrame) -> pd.DataFrame: """Rank correlation of (push in the trade's direction) vs R made.""" out: List[dict] = [] if trades.empty or indicators.empty: return pd.DataFrame(out, columns=["tf_role", "indicator", "n", "spearman"]) keep = trades[["trade_id", "direction", "r_multiple"]].copy() keep["dir_sign"] = keep["direction"].map(_dir_sign) merged = indicators.merge(keep, on="trade_id", how="inner") merged["contribution"] = pd.to_numeric( merged["wscore"], errors="coerce") * merged["dir_sign"] for role in _roles_present(indicators): for name in _indicator_names(indicators): sub = merged[(merged["tf_role"] == role) & (merged["indicator"] == name)] if sub.empty: continue out.append({"tf_role": role, "indicator": name, "n": int(len(sub)), "spearman": _spearman(sub["contribution"], sub["r_multiple"])}) return pd.DataFrame(out, columns=["tf_role", "indicator", "n", "spearman"]) def print_indicator_scorecard(frame: pd.DataFrame, corr: Optional[pd.DataFrame] = None) -> None: print(f"\n{'─' * 74}\n INDICATOR SCORECARD — mean R by what the " f"indicator said about the trade\n{'─' * 74}") print(" Read: 'agree' = the indicator pointed the same way as the trade " "and\n 'absent' = it never voted (divergence/candlestick only fire " "on a pattern).") if frame.empty: print(" (nothing to show)") return min_n = frame.attrs.get("min_n", 0) for role in frame["tf_role"].drop_duplicates(): print(f"\n {role.upper()} timeframe") print(f" {'indicator':12} {'bucket':8} {'n':>5} {'mean R':>6} " f"{'win':>4} {'total R':>7}") for _, row in frame[frame["tf_role"] == role].iterrows(): print(f" {row['indicator']:12} {row['bucket']:8} " f"{int(row['n']):5d} {_fmt_r(row['mean_r'])} " f"{_fmt_pct(row['win_rate'])} " f"{_fmt_total(row['total_r'])}") hidden = frame.attrs.get("n_hidden", 0) if hidden: print(f"\n ({hidden} buckets hidden — fewer than {min_n} trades)") if corr is not None and not corr.empty: print("\n Rank correlation, how hard it pushed vs what the trade " "made\n (positive = pushing harder went with better trades; " "0 = no relation):") for role in corr["tf_role"].drop_duplicates(): part = corr[corr["tf_role"] == role] cells = " ".join( f"{r['indicator']}:{_fmt_num(r['spearman'], 2)}" for _, r in part.iterrows()) print(f" {role:8} {cells}") # ── 2. scorecard by regime ─────────────────────────────────────────── def scorecard_by_regime(trades: pd.DataFrame, indicators: pd.DataFrame, min_n: int = 20) -> pd.DataFrame: """Primary-timeframe agree/oppose/neutral split, cut by market regime.""" out: List[dict] = [] cols = ["regime", "indicator", "bucket", "n", "mean_r", "win_rate", "total_r"] if trades.empty or indicators.empty or "regime" not in trades.columns: return pd.DataFrame(out, columns=cols) prim = indicators[indicators["tf_role"] == "primary"] if prim.empty: return pd.DataFrame(out, columns=cols) regime_of = trades.set_index("trade_id")["regime"] r_by_trade = trades.set_index("trade_id")["r_multiple"] for regime in sorted(str(x) for x in regime_of.dropna().unique()): tids = pd.Index(regime_of[regime_of == regime].index) sub = prim[prim["trade_id"].isin(tids)] if sub.empty: continue for name in _indicator_names(indicators): rows = sub[sub["indicator"] == name] for bucket in ("agree", "oppose", "neutral"): sel = rows[rows["agrees"] == bucket]["trade_id"] stat = _stats(r_by_trade.reindex(sel)) if stat["n"] < min_n: continue out.append({"regime": regime, "indicator": name, "bucket": bucket, **stat}) return pd.DataFrame(out, columns=cols) def print_scorecard_by_regime(frame: pd.DataFrame, min_n: int = 20) -> None: print(f"\n{'─' * 74}\n BY MARKET REGIME — primary timeframe only" f"\n{'─' * 74}") print(" Read: the same agree/oppose/neutral split, but only inside one " "kind of\n market — weights already change by regime, this says " "whether that pays.") if frame.empty: print(f" (no bucket reached {min_n} trades)") return for regime in frame["regime"].drop_duplicates(): print(f"\n regime: {regime}") for _, row in frame[frame["regime"] == regime].iterrows(): print(f" {row['indicator']:12} {row['bucket']:8} " f"{int(row['n']):5d} {_fmt_r(row['mean_r'])} " f"{_fmt_pct(row['win_rate'])} " f"{_fmt_total(row['total_r'])}") # ── 3. calendar buckets ────────────────────────────────────────────── def calendar_buckets(trades: pd.DataFrame) -> pd.DataFrame: """Trades grouped by their position around high-impact USD releases.""" cols = ["bucket", "n", "mean_r", "win_rate", "total_r"] if trades.empty: return pd.DataFrame([], columns=cols) def col(name, default=np.nan) -> pd.Series: if name in trades.columns: return trades[name] return pd.Series([default] * len(trades), index=trades.index) nxt = pd.to_numeric(col("ev_next_hi_usd_min"), errors="coerce") prev = pd.to_numeric(col("ev_prev_hi_usd_min"), errors="coerce") during = pd.to_numeric(col("ev_hi_usd_during_hold"), errors="coerce") holiday = col("us_holiday", False).fillna(False).astype(bool) r = trades["r_multiple"] masks = [ ("all trades", pd.Series(True, index=trades.index)), ("entered <=30 min BEFORE a USD high-impact release", nxt <= 30), ("entered <=30 min AFTER one", prev <= 30), ("held through >=1 release", during >= 1), ("quiet (no release within 24 h either side, none during hold)", nxt.isna() & prev.isna() & (during.fillna(0) == 0)), ("US holiday", holiday), ] out = [{"bucket": label, **_stats(r[mask.fillna(False)])} for label, mask in masks] return pd.DataFrame(out, columns=cols) def print_calendar_buckets(frame: pd.DataFrame) -> None: print(f"\n{'─' * 74}\n CALENDAR — trades around high-impact USD news" f"\n{'─' * 74}") print(" Read: one trade can land in more than one row (a trade entered " "before a\n release is usually also holding through it), so the " "rows do not add up.") if frame.empty: print(" (nothing to show)") return for _, row in frame.iterrows(): print(f" {row['bucket']:58} n={int(row['n']):5d} " f"{_fmt_r(row['mean_r'])} {_fmt_pct(row['win_rate'])} " f"{_fmt_total(row['total_r'])}") # ── 4. what-if (in-book only) ──────────────────────────────────────── WHATIF_HEADER = ( "IN-BOOK ONLY — this cannot add trades the change would create and " "ignores the gates;\n any real change needs a full backtest rerun " "(rule 1)." ) def _primary_rows_by_trade(indicators: pd.DataFrame) -> Dict[object, Tuple]: """trade_id -> (indicator names, eff_weights, wscores) for the primary TF.""" cache: Dict[object, Tuple] = {} if indicators.empty: return cache prim = indicators[indicators["tf_role"] == "primary"] for tid, group in prim.groupby("trade_id"): cache[tid] = (group["indicator"].to_numpy(), pd.to_numeric(group["eff_weight"], errors="coerce").to_numpy(float), pd.to_numeric(group["wscore"], errors="coerce").to_numpy(float)) return cache def what_if(trades: pd.DataFrame, indicators: pd.DataFrame, drop: Optional[str] = None, scale: Optional[Tuple[str, float]] = None, _rows: Optional[Dict[object, Tuple]] = None) -> pd.DataFrame: """Recompute each trade's primary composite with one indicator removed (``drop="bollinger"``) or re-weighted (``scale=("bollinger", 0.5)``).""" cols = ["trade_id", "direction", "r_multiple", "composite_recorded", "composite_new", "loses_direction", "band_change"] if trades.empty: return pd.DataFrame([], columns=cols) rows = _primary_rows_by_trade(indicators) if _rows is None else _rows weak, strong = _weak(), _strong() out: List[dict] = [] for _, trade in trades.iterrows(): tid = trade["trade_id"] names, eff, ws = rows.get( tid, (np.array([]), np.array([]), np.array([]))) eff, ws = eff.copy(), ws.copy() if drop is not None and len(names): keep = names != drop eff, ws = eff[keep], ws[keep] elif scale is not None and len(names): target, factor = scale hit = names == target eff = np.where(hit, eff * float(factor), eff) ws = np.where(hit, ws * float(factor), ws) new_score, _ = _composite(eff, ws) recorded = trade.get("primary_composite", np.nan) recorded = float(recorded) if pd.notna(recorded) else float("nan") sign = _dir_sign(trade.get("direction")) loses = bool(abs(new_score) < weak or np.sign(new_score) != np.sign(sign)) band = bool((abs(recorded) >= strong) != (abs(new_score) >= strong)) \ if not np.isnan(recorded) else False out.append({"trade_id": tid, "direction": trade.get("direction"), "r_multiple": trade.get("r_multiple"), "composite_recorded": recorded, "composite_new": float(new_score), "loses_direction": loses, "band_change": band}) return pd.DataFrame(out, columns=cols) def what_if_summary(frame: pd.DataFrame) -> Dict[str, float]: """How many trades the change would unsettle, and the R they carried.""" if frame.empty: return {"n": 0, "n_loses_direction": 0, "r_of_lost": 0.0, "n_band_change": 0, "r_of_band_change": 0.0} lost = frame[frame["loses_direction"].astype(bool)] band = frame[frame["band_change"].astype(bool)] return {"n": int(len(frame)), "n_loses_direction": int(len(lost)), "r_of_lost": float(pd.to_numeric( lost["r_multiple"], errors="coerce").sum()), "n_band_change": int(len(band)), "r_of_band_change": float(pd.to_numeric( band["r_multiple"], errors="coerce").sum())} def what_if_grid(trades: pd.DataFrame, indicators: pd.DataFrame) -> pd.DataFrame: """drop / x0.5 / x1.5 for every indicator, one row each.""" rows = _primary_rows_by_trade(indicators) out: List[dict] = [] for name in _indicator_names(indicators): for label, kwargs in (("drop", {"drop": name}), ("x0.5", {"scale": (name, 0.5)}), ("x1.5", {"scale": (name, 1.5)})): summary = what_if_summary( what_if(trades, indicators, _rows=rows, **kwargs)) out.append({"indicator": name, "change": label, **summary}) return pd.DataFrame(out, columns=["indicator", "change", "n", "n_loses_direction", "r_of_lost", "n_band_change", "r_of_band_change"]) def print_what_if(frame: pd.DataFrame) -> None: print(f"\n{'─' * 74}\n WHAT-IF — drop or re-weight one indicator" f"\n{'─' * 74}") print(f" {WHATIF_HEADER}") print(" Read: 'loses direction' = the trade's own score would no longer " "clear the\n entry band (or flips sign); 'band change' = it would " "cross the strong/weak line.") if frame.empty: print(" (nothing to show)") return print(f"\n {'indicator':12} {'change':6} {'lost':>5} {'R lost':>8} " f"{'band':>5} {'R band':>8}") for _, row in frame.iterrows(): print(f" {row['indicator']:12} {row['change']:6} " f"{int(row['n_loses_direction']):5d} " f"{_fmt_total(row['r_of_lost'])} " f"{int(row['n_band_change']):5d} " f"{_fmt_total(row['r_of_band_change'])}") # ── 5. bar-level lenses ────────────────────────────────────────────── def _bar_composites(bars: pd.DataFrame, drop: Optional[str] = None, scale: Optional[Tuple[str, float]] = None) -> np.ndarray: """Primary composite for every bar, recomputed from the stored parts.""" names = [n for n in INDICATORS if f"primary_{n}_eff_weight" in bars.columns and f"primary_{n}_wscore" in bars.columns] if not names: return np.zeros(len(bars)) # copy=True: a to_numpy() view can be read-only under copy-on-write. eff = bars[[f"primary_{n}_eff_weight" for n in names]].to_numpy( dtype=float, copy=True) ws = bars[[f"primary_{n}_wscore" for n in names]].to_numpy( dtype=float, copy=True) if drop is not None and drop in names: idx = names.index(drop) eff[:, idx] = np.nan ws[:, idx] = np.nan elif scale is not None and scale[0] in names: idx = names.index(scale[0]) eff[:, idx] = eff[:, idx] * float(scale[1]) ws[:, idx] = ws[:, idx] * float(scale[1]) total = np.nansum(eff, axis=1) numer = np.nansum(ws, axis=1) score = np.divide(numer, total, out=np.zeros_like(numer), where=total > 0) * 50.0 return np.clip(score, -100.0, 100.0) def _signed_mean(bars: pd.DataFrame, mask: np.ndarray, sign: np.ndarray, column: str) -> float: if column not in bars.columns or not mask.any(): return float("nan") values = pd.to_numeric(bars[column], errors="coerce").to_numpy(float) picked = values[mask] * sign[mask] picked = picked[np.isfinite(picked)] return float(picked.mean()) if picked.size else float("nan") def bars_what_if(bars: pd.DataFrame, drop: Optional[str] = None, scale: Optional[Tuple[str, float]] = None) -> Dict[str, dict]: """Every analyzed bar, not just the trades: which bars would newly become directional, and which would fall back to neutral.""" label = f"drop {drop}" if drop else ( f"{scale[0]} x{scale[1]}" if scale else "unchanged") if bars is None or bars.empty: return {"label": label, "n_bars": 0, "new_cross": {"n": 0}, "drop_out": {"n": 0}} recorded = pd.to_numeric(bars.get("primary_composite"), errors="coerce").to_numpy(float) new = _bar_composites(bars, drop=drop, scale=scale) weak = _weak() rec_dir = np.abs(np.nan_to_num(recorded)) >= weak new_dir = np.abs(new) >= weak cross = (~rec_dir) & new_dir fell = rec_dir & (~new_dir) sign_new = np.sign(new) sign_rec = np.sign(np.nan_to_num(recorded)) out = {"label": label, "n_bars": int(len(bars))} for key, mask, sign in (("new_cross", cross, sign_new), ("drop_out", fell, sign_rec)): out[key] = { "n": int(mask.sum()), "fwd_4h_atr": _signed_mean(bars, mask, sign, "fwd_4h_atr"), "fwd_24h_atr": _signed_mean(bars, mask, sign, "fwd_24h_atr"), } return out def bars_indicator_edge(bars: pd.DataFrame) -> pd.DataFrame: """Does this indicator's vote, on its own, point the right way?""" cols = ["indicator", "bucket", "n", "fwd_4h_signed", "fwd_24h_signed"] if bars is None or bars.empty: return pd.DataFrame([], columns=cols) out: List[dict] = [] for name in INDICATORS: column = f"primary_{name}_signal" if column not in bars.columns: continue signal = pd.to_numeric(bars[column], errors="coerce").to_numpy(float) buckets = (("buy-side", signal > 0, np.ones(len(bars))), ("sell-side", signal < 0, -np.ones(len(bars))), ("neutral", signal == 0, np.ones(len(bars))), ("absent", np.isnan(signal), np.ones(len(bars)))) for bucket, mask, sign in buckets: out.append({ "indicator": name, "bucket": bucket, "n": int(mask.sum()), "fwd_4h_signed": _signed_mean(bars, mask, sign, "fwd_4h_atr"), "fwd_24h_signed": _signed_mean(bars, mask, sign, "fwd_24h_atr")}) return pd.DataFrame(out, columns=cols) def print_bars_indicator_edge(frame: pd.DataFrame) -> None: print(f"\n{'─' * 74}\n EVERY BAR — where did price go after each vote?" f"\n{'─' * 74}") print(" Read: the number is the average move over the next 4 h / 24 h in " "ATRs,\n flipped so that positive always means the vote was right. " "'neutral' and\n 'absent' rows are the plain move (no direction " "claimed).") if frame.empty: print(" (no bars table in this run)") return print(f"\n {'indicator':12} {'vote':10} {'n':>7} {'4 h':>7} " f"{'24 h':>7}") for _, row in frame.iterrows(): print(f" {row['indicator']:12} {row['bucket']:10} " f"{int(row['n']):7d} {_fmt_num(row['fwd_4h_signed'])} " f"{_fmt_num(row['fwd_24h_signed'])}") def print_bars_what_if(results: Sequence[dict]) -> None: print(f"\n{'─' * 74}\n EVERY BAR — what-if: bars gained and bars lost" f"\n{'─' * 74}") print(" Read: 'new' bars would start calling a direction they do not " "call today,\n 'lost' bars would stop. The R-free forward move " "says whether that is good.") print(f"\n {'change':16} {'new':>6} {'4 h':>7} {'24 h':>7} " f"{'lost':>6} {'4 h':>7} {'24 h':>7}") for res in results: new, fell = res["new_cross"], res["drop_out"] print(f" {res['label']:16} {new['n']:6d} " f"{_fmt_num(new.get('fwd_4h_atr'))} " f"{_fmt_num(new.get('fwd_24h_atr'))} " f"{fell['n']:6d} {_fmt_num(fell.get('fwd_4h_atr'))} " f"{_fmt_num(fell.get('fwd_24h_atr'))}") # ── 6. invariant checks ────────────────────────────────────────────── CHECK_NAMES = ("n_trades", "primary_tf", "confirm_roles", "eff_weight", "wscore", "composite", "regime_mult", "bars") def _ts_key(value) -> str: try: stamp = pd.Timestamp(value) if stamp.tzinfo is None: stamp = stamp.tz_localize("UTC") return stamp.tz_convert("UTC").isoformat() except Exception: return str(value) def check_run(run: dict) -> List[str]: """Re-derive the record's own arithmetic. Empty list = everything held.""" fails: List[str] = [] manifest = run.get("manifest") or {} trades = run.get("trades") indicators = run.get("indicators") bars = run.get("bars") trades = pd.DataFrame() if trades is None else trades indicators = pd.DataFrame() if indicators is None else indicators want = manifest.get("n_trades") if want is not None and int(want) != len(trades): fails.append(f"n_trades: manifest says {want}, trades.parquet has " f"{len(trades)}") if "primary_tf" in trades.columns: blank = trades["primary_tf"].isna() | ( trades["primary_tf"].astype(str).str.strip() == "") if blank.any(): fails.append(f"primary_tf: {int(blank.sum())} trades have no " "primary timeframe") elif len(trades): fails.append("primary_tf: column missing from trades") for role in ("c1", "c2"): column = f"{role}_tf" if column not in trades.columns or not len(trades): continue present = trades[column].notna() & ( trades[column].astype(str).str.strip() != "") share = present.mean() if share <= 0.99: fails.append(f"confirm_roles: {role} present on " f"{int(present.sum())}/{len(trades)} trades " f"({100 * share:.1f}%)") if not indicators.empty: base = pd.to_numeric(indicators["base_weight"], errors="coerce") mult = pd.to_numeric(indicators["regime_mult"], errors="coerce") eff = pd.to_numeric(indicators["eff_weight"], errors="coerce") bad = (base * mult - eff).abs() > 1e-9 if bad.any(): fails.append(f"eff_weight: {int(bad.sum())} rows where " "base_weight x regime_mult != eff_weight") signal = pd.to_numeric(indicators["signal"], errors="coerce") conf = pd.to_numeric(indicators["confidence"], errors="coerce") ws = pd.to_numeric(indicators["wscore"], errors="coerce") bad = (signal * eff * conf / 100.0 - ws).abs() > 1e-9 if bad.any(): fails.append(f"wscore: {int(bad.sum())} rows where " "signal x eff_weight x confidence/100 != wscore") offenders: List[str] = [] for (tid, role), group in indicators.groupby(["trade_id", "tf_role"]): column = f"{role}_composite" if column not in trades.columns: continue match = trades[trades["trade_id"] == tid] if match.empty: continue want_score = match.iloc[0][column] if pd.isna(want_score): continue got, _ = _composite(group["eff_weight"].tolist(), group["wscore"].tolist()) if abs(float(got) - float(want_score)) > 1e-6: offenders.append(f"trade {tid} {role}: rows give {got:.6f}, " f"record says {float(want_score):.6f}") if offenders: fails.append(f"composite: {len(offenders)} trade/role composites " f"disagree with their rows — " + "; ".join(offenders[:5])) table = manifest.get("regime_weight_table") or {} if table and "regime" in indicators.columns: bad_pairs: List[str] = [] pairs = indicators[["regime", "indicator", "regime_mult"]].drop_duplicates() for _, row in pairs.iterrows(): want_mult = float(table.get(str(row["regime"]), {}).get( str(row["indicator"]), 1.0)) if abs(float(row["regime_mult"]) - want_mult) > 1e-9: bad_pairs.append(f"{row['regime']}/{row['indicator']}: " f"{row['regime_mult']} != {want_mult}") if bad_pairs: fails.append("regime_mult: " + "; ".join(bad_pairs[:5])) if bars is not None and not bars.empty: n_candidates = manifest.get("n_candidates") n_trades = manifest.get("n_trades", len(trades)) if "has_setup" in bars.columns and n_candidates is not None: setups = int(pd.to_numeric(bars["has_setup"], errors="coerce").fillna(0).sum()) if not (setups >= int(n_candidates) >= int(n_trades)): fails.append(f"bars: has_setup={setups} >= n_candidates=" f"{n_candidates} >= n_trades={n_trades} does not " "hold") n_analyzed = (manifest.get("bars") or {}).get("n_analyzed") if n_analyzed is not None and int(n_analyzed) != len(bars): fails.append(f"bars: manifest n_analyzed={n_analyzed}, " f"bars.parquet has {len(bars)} rows") if "bar_ts" in bars.columns and "bar_ts" in trades.columns: known = {_ts_key(v) for v in bars["bar_ts"]} missing = [t for t in trades["bar_ts"] if _ts_key(t) not in known] if missing: fails.append(f"bars: {len(missing)} trade bar_ts values are " f"not in bars.parquet (e.g. {missing[0]})") return fails def print_check(fails: Sequence[str]) -> None: print(f"\n{'─' * 74}\n CHECK — does the record agree with itself?" f"\n{'─' * 74}") for name in CHECK_NAMES: hits = [f for f in fails if f.startswith(name + ":")] if hits: for message in hits: print(f" FAIL {message}") else: print(f" PASS {name}") extra = [f for f in fails if not any(f.startswith(n + ":") for n in CHECK_NAMES)] for message in extra: print(f" FAIL {message}") print("\n " + ("all invariants held" if not fails else f"{len(fails)} invariant(s) broken")) # ── run header ─────────────────────────────────────────────────────── def print_run_header(manifest: dict, path: str) -> None: window = manifest.get("window") if isinstance(window, dict): window = f"{window.get('start')} -> {window.get('end')}" confirm = manifest.get("confirmation_tfs") if isinstance(confirm, (list, tuple)): confirm = ",".join(str(c) for c in confirm) print(f"\n{'═' * 74}") print(f" RUN {path}") print(f"{'═' * 74}") print(f" primary {manifest.get('primary_tf', '?')} " f"confirm {confirm or '?'} window {window or '?'}") print(f" trades {manifest.get('n_trades', '?')} " f"candidates {manifest.get('n_candidates', '?')} " f"analyzed bars " f"{(manifest.get('bars') or {}).get('n_analyzed', '?')} " f"spread {manifest.get('spread', '?')}") if manifest.get("book_csv_md5"): print(f" book md5 {manifest['book_csv_md5']}") # ── main ───────────────────────────────────────────────────────────── def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[1]) parser.add_argument("dirs", nargs="+", metavar="DIR", help="replay-record folder(s) written by " "backtest/replay_record.py") parser.add_argument("--what-if", action="store_true", help="drop / re-weight each indicator (in-book only)") parser.add_argument("--bars", action="store_true", help="bar-level lenses (needs bars.parquet)") parser.add_argument("--check", action="store_true", help="verify the record's own arithmetic") parser.add_argument("--min-n", type=int, default=20, help="hide buckets with fewer trades than this") args = parser.parse_args(argv) try: record = _load_record_module() except ImportError as exc: print(str(exc)) return 2 failed = False for path in args.dirs: if not os.path.isdir(path): print(f"not a directory: {path}") failed = True continue try: run = record.load_run(path) except Exception as exc: print(f"could not read {path}: {exc}") failed = True continue manifest = run.get("manifest") or {} trades = run.get("trades") indicators = run.get("indicators") bars = run.get("bars") print_run_header(manifest, path) if trades is None or trades.empty: print(" (no trades in this run)") continue print_indicator_scorecard( indicator_scorecard(trades, indicators, min_n=args.min_n), contribution_corr(trades, indicators)) print_scorecard_by_regime( scorecard_by_regime(trades, indicators, min_n=args.min_n), min_n=args.min_n) print_calendar_buckets(calendar_buckets(trades)) if args.what_if: print_what_if(what_if_grid(trades, indicators)) if args.bars: if bars is None or bars.empty: print("\n --bars: this run has no bars.parquet") else: print_bars_indicator_edge(bars_indicator_edge(bars)) print_bars_what_if( [bars_what_if(bars, drop=name) for name in _indicator_names(indicators)]) print(f"\n {WHATIF_HEADER}") if args.check: fails = check_run(run) print_check(fails) failed = failed or bool(fails) return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())