xauusd M15: halve the RSI regime multiplier in every regime (M15 override 5, rule-1 validated)

Owner-approved 2026-09-09 after the pre-registered weight test (backtest/research/weight_test_2026-09-09.md). M15-only technical.regime_weights_override = the shipped table with rsi x 0.5 (0.5 trending family / 0.75 ranging / 0.5 transitioning), derived in the factory and pinned literally by the Module 9 equivalence test; a Module 33 test locks that M5/H1/H4 carry no override. Evidence, 2-year Oanda window 2024-09-05..2026-09-05 at $0.65: 793/+117.62R PF 1.41 -> 931/+146.14R PF 1.43 (fixed halves +27.9/+0.6R vs control); OOS 2024-06-01..2025-03-30 300/+40.59R PF 1.38 -> 355/+60.67R PF 1.49; 3-year +151.90R -> +186.61R (max DD 15.5 -> 19.3R, 2026 -4R). Parity: the factory run with no --set flag reproduces the validated candidate books byte-for- byte (2-yr MD5 d185f099..., OOS 6f8f24b4...). M15 canonical book re-pinned to backtest/oanda_m15_2yr_065_20260909.csv. Docs carry a one-line note. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Nc9Ninzyi18QkWezrxeBvB
research: weight-change test on the recorded replay files (pre-registered; M15 RSI x0.5 passes, not shipped)
2026-09-09 17:36:41 -04:00 · 2026-09-09 17:26:21 -04:00 · 2026-09-09 16:41:49 -04:00
13 changed files with 3252 additions and 9 deletions
@@ -30,3 +30,4 @@ backtest/__pycache__/
 .env.*
 logs/
 backtest/research/sizing_2026-09/
 backtest/research/replay_2026-09/
@@ -63,6 +63,7 @@ from providers import BaseDataProvider, make_data_provider  # noqa: E402
 from engine import IndicatorSignal, TechnicalAnalysis, TechnicalEngine  # noqa: E402
 from signal_aggregator import SignalAggregator, SignalReport, TradeSetup  # noqa: E402
 import position_sizing  # noqa: E402
 from backtest import replay_record  # noqa: E402
 # ─────────────────────────────────────────────────────────────────────
 #  GPU shim — CuPy if available AND CUDA runtime works, else NumPy.
@@ -1079,6 +1080,11 @@ class Trade:
    # untouched.
    features: Dict[str, float] = field(default_factory=dict)
    stake_multiplier: float = 1.0
    # RESEARCH (--record only): plain-data decision snapshot built by
    # replay_record.snapshot_report — indicator weights/values on every
    # timeframe, penalties, setup. Empty on normal runs, so pickling a
    # Trade back from a worker costs nothing and the export is untouched.
    snapshot: Dict = field(default_factory=dict)
    # Filled on exit:
    exit_time: Optional[pd.Timestamp] = None
@@ -1465,6 +1471,13 @@ class BacktestConfig:
    # columns would break comparison against the pinned 63-column books,
    # and the feature build costs time per candidate.
    export_features: bool = False
    # RESEARCH (--record DIR): write manifest.json + trades/indicators/
    # events parquet describing every accepted trade to this directory.
    # Read-only w.r.t. the signal path — the book is unchanged.
    record_dir: Optional[str] = None
    # With record_dir set: also store one compact row per ANALYZED
    # primary bar (bars.parquet), not just the accepted trades.
    record_bars: bool = False
 def _is_xau_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optional[str]:
@@ -1518,6 +1531,9 @@ def _replay_span(span: Tuple[int, int]):
    skipped: Dict[str, int] = {}
    candidates: List[Tuple[int, Trade]] = []
    is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
    record = bool(bt_cfg.record_dir)
    record_bars = record and bt_cfg.record_bars
    bar_buf = replay_record.new_bar_buffer() if record_bars else None
    for pos in range(span[0], span[1]):
        ts = primary_df.index[pos]
@@ -1575,6 +1591,12 @@ def _replay_span(span: Tuple[int, int]):
        if platform_cfg.backtest.regime_filter:
            aggregator._apply_regime_filter_live(report, slice_dfs)
        # RESEARCH (--record-bars): every ANALYZED bar, setup or not.
        if record_bars:
            replay_record.add_bar_row(
                bar_buf, pos, ts, report,
                bt_cfg.primary_tf, bt_cfg.confirmation_tfs)
        if report.trade_setup is None:
            continue
        if report.signal_quality not in bt_cfg.quality_filter:
@@ -1632,6 +1654,13 @@ def _replay_span(span: Tuple[int, int]):
                list(bt_cfg.confirmation_tfs),
            )
        # RESEARCH (--record): plain-data decision snapshot. Read-only —
        # it never touches the report or the setup.
        snap = (replay_record.snapshot_report(
            report, ts_setup, decision_time,
            bt_cfg.primary_tf, list(bt_cfg.confirmation_tfs))
            if record else {})
        candidates.append((pos, Trade(
            pair=bt_cfg.pair,
            entry_time=entry_time,
@@ -1658,6 +1687,7 @@ def _replay_span(span: Tuple[int, int]):
            indicators_neutral=neutral,
            indicators_total=total,
            features=feats,
            snapshot=snap,
            stake_multiplier=float(ts_setup.stake_multiplier
                                   if ts_setup.stake_multiplier is not None
                                   else 1.0),
@@ -1667,7 +1697,8 @@ def _replay_span(span: Tuple[int, int]):
                          if tag == "XAU_PB" else 0.0),
        )))
-    return skipped, candidates
+    return skipped, candidates, (replay_record.finish_bar_rows(bar_buf)
                                 if record_bars else None)
 def _reanchor_entries_to_raw(
@@ -1791,6 +1822,9 @@ def run_backtest(
    fetch_start = bt_cfg.start - timedelta(days=pad_days)
    dfs_full: Dict[Timeframe, pd.DataFrame] = {}
    # Which parquet each TF came from (recorded in --record's manifest so
    # a study can prove two runs read the same bars).
    cache_names: Dict[str, str] = {}
    for tf in timeframes:
        # Slow TFs need a longer pad: 60 days of D1 is only ~44 bars — no
        # EMA200, truncated swing/divergence windows — while live always has
@@ -1800,6 +1834,8 @@ def run_backtest(
        bars_per_day = BaseDataProvider.BARS_PER_DAY[tf]
        min_pad_days = int(np.ceil(210 / bars_per_day * 7 / 5)) + 5
        tf_start = bt_cfg.start - timedelta(days=max(pad_days, min_pad_days))
        cache_names[tf.value] = _cache_path(
            bt_cfg.pair, tf, tf_start, bt_cfg.end, provider.name).name
        df = fetch_history(
            provider, bt_cfg.pair, tf,
            start=tf_start, end=bt_cfg.end,
@@ -1874,10 +1910,13 @@ def run_backtest(
    skipped_reasons: Dict[str, int] = {}
    candidates: List[Tuple[int, Trade]] = []
-    for skips, cands in results:
+    bar_chunks: List = []
    for skips, cands, bars in results:
        for reason, count in skips.items():
            skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
        candidates.extend(cands)
        if bars is not None:
            bar_chunks.append(bars)
    candidates.sort(key=lambda pc: pc[0])
    # ── Phase 2: entry cooldown — the only sequential rule.
@@ -1885,6 +1924,7 @@ def run_backtest(
    trades: List[Trade] = [t for _, t in accepted]
    signal_ts_list: List[pd.Timestamp] = [primary_df.index[p] for p, _ in accepted]
    replay_seconds = time.perf_counter() - t_loop
    logger.info(
        f"Replay complete: {len(trades)} entries in "
        f"{time.perf_counter() - t_loop:.1f}s "
@@ -1920,6 +1960,19 @@ def run_backtest(
        f"(GPU={'yes' if _GPU_AVAILABLE else 'no'})"
    )
    if bt_cfg.record_dir:
        replay_record.write_run(
            bt_cfg.record_dir, trades=trades, book_df=trades_to_dataframe(trades),
            bt_cfg=bt_cfg, platform_cfg=platform_cfg, aggregator=aggregator,
            run_info=dict(first_pos=first_eligible_pos, last_pos=last_pos,
                          n_primary_bars=len(primary_df), n_bars_in_window=n_bars,
                          skipped=skipped_reasons, n_candidates=len(candidates),
                          workers=workers,
                          backend="CuPy GPU" if _GPU_AVAILABLE else "NumPy CPU fallback",
                          cache_files=cache_names, replay_seconds=replay_seconds),
            sim_df=sim_df, bar_chunks=bar_chunks or None)
        logger.info(f"--record: wrote run to {bt_cfg.record_dir}")
    return trades, dfs_full
@@ -2313,6 +2366,13 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
                        "(+ stake_multiplier, r_sized) to --export-csv; "
                        "research flag, default off so pinned 63-column "
                        "books stay comparable")
    p.add_argument("--record", metavar="DIR", default=None,
                   help="RESEARCH: write manifest.json + trades/indicators/"
                        "events parquet for every accepted trade to DIR; "
                        "the book and CSV are unchanged")
    p.add_argument("--record-bars", action="store_true",
                   help="with --record: also write one row per analyzed "
                        "primary bar (bars.parquet, ~140k rows on 2-yr M5)")
    p.add_argument("--detail", action="store_true",
                   help="Print full per-trade reasoning (indicators, signals, agreement, why entered)")
    p.add_argument("--workers", type=int, default=0,
@@ -2320,7 +2380,10 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
    p.add_argument("--reasoning-samples", type=int, default=3,
                   help="If --detail not set, print this many sample trades' full reasoning")
    p.add_argument("--log-level", default="INFO")
-    return p.parse_args(argv)
+    args = p.parse_args(argv)
    if args.record_bars and not args.record:
        p.error("--record-bars requires --record")
    return args
 def _resolve_pair(s: str) -> str:
@@ -2466,7 +2529,16 @@ def main(argv: Optional[List[str]] = None) -> int:
        workers=args.workers,
        fixed_rr_tp=args.fixed_rr_tp or 0.0,
        export_features=args.export_features,
        record_dir=args.record,
        record_bars=args.record_bars,
    )
    if args.record:
        logger.info(
            f"--record ON: decision snapshots → {args.record} "
            f"(trades/indicators/events"
            f"{'/bars' if args.record_bars else ''} parquet + manifest.json; "
            f"the trade book is unchanged)"
        )
    if args.export_features:
        logger.info(
            f"--export-features ON: appending stake_multiplier, r_sized and "
@@ -24,3 +24,17 @@ re-implementation.
 - Any sizing report must show the FLAT book and the SIZED book side by side and the
  label-shuffle null; the 2026-09-06 campaign verdict (nothing ships,
  backtest/research/sizing_campaign_2026-09.md) is the reference.
 --- 2026-09-09: replay recording (--record DIR [--record-bars]) ---
 - `--record DIR` writes manifest.json + trades/indicators/events parquet for every ACCEPTED trade:
  every indicator's signal / value / confidence / BASE weight / regime multiplier / EFFECTIVE
  weight / contribution on the primary AND both confirmation timeframes, penalties, the setup, the
  outcome, and the Forex Factory calendar events around the trade (backtest/cache/calendar.parquet,
  refreshed with scripts/ff_calendar_scraper.py). `--record-bars` adds one row per ANALYZED primary
  bar with the same votes plus forward moves (1h/4h/24h in ATR units).
 - Recording is read-only w.r.t. the signal path: any recorded run must be paired with a control
  run (same CLI without --record) and the report must show the two books' MD5s equal
  (manifest.book_csv_md5 is the md5 of the 63-column book).
 - Scorecards and "what-if" tables from these files (backtest/replay_scorecard.py) are IN-BOOK
  descriptive statistics: they cannot add trades a change would create and ignore the gates.
  Any weight / indicator change they suggest is a hypothesis for a full rule-1 rerun, never a decision.
@@ -0,0 +1,894 @@
 #!/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())
@@ -0,0 +1,148 @@
 # Replay recording campaign — 2026-09-09 (2-year Oanda window, M5 + M15, $0.65)
 ## Purpose and owner decisions
 The owner asked for a 2-year replay of the M5 and M15 timeframes that writes down, for
 every trade setup the system generates, the indicator weights and values the decision was
 made with and the economic-calendar events around it, in a form that a later study can use
 to decide whether indicator weights should change, or indicators be dropped or replaced.
 Owner decisions (2026-09-09): also record one compact row per ANALYZED bar (not only the
 bars that became trades), and restore the Forex Factory calendar scraper to `scripts/`.
 Nothing in the signal core changed. The recorder is opt-in (`--record DIR`, `--record-bars`)
 and read-only with respect to the signal path; the trade book with recording on is
 byte-identical to the book with it off (MD5 below).
 ## Harness
 `backtest/gpu_backtest.py --pair XAUUSD --start 2024-09-05 --end 2026-09-05 --spread 0.65
 --workers 12 --primary-tf {5min|15min}`, pair-factory config (no --set/--quality/--strategies/
 --confirm-tfs; confirmation TFs from the factory map: M5→15min,1h and M15→5min,1h),
 `Backend: NumPy CPU fallback` (CuPy cannot find CUDA on this box). Code at git `28b591d`
 plus the recorder (this campaign). Bars fetched once from Oanda into
 `backtest/cache/oanda_XAU_USD_{5min,15min,1h}_20240705_20260905.parquet` (62-day warm-up pad)
 and read from the cache by every later run. Each timeframe was run twice: a CONTROL run
 (`--export-csv book_control.csv`) and a RECORDED run (`--export-csv book.csv --record DIR
 --record-bars`). Outputs in `backtest/research/replay_2026-09/{m5,m15}/` (gitignored).
 ## Books (identical for control and recorded — that is the point)
 | book | trades | total R | WR | PF | max DD |
 |---|---|---|---|---|---|
 | M5 whole window | 781 | +146.07 | 52.9% | 1.52 | 10.1 |
 |   first half | 189 | +28.01 | 49.7% | 1.41 | 8.4 |
 |   second half | 592 | +118.06 | 53.9% | 1.56 | 10.1 |
 | M15 whole window | 793 | +117.62 | 60.4% | 1.41 | 14.8 |
 |   first half | 383 | +38.22 | 58.5% | 1.26 | 14.8 |
 |   second half | 410 | +79.40 | 62.2% | 1.55 | 9.9 |
 Halves split the entry-time range in two; M5 trades cluster in the second year.
 ### Behaviour-neutrality check
 | TF | control MD5 | recorded MD5 | manifest `book_csv_md5` |
 |---|---|---|---|
 | M5 | 29120a1ee32979431d062f210d41a9d1 | 29120a1ee32979431d062f210d41a9d1 | same |
 | M15 | e62d3ded56f35572f52a96f604a99cf4 | e62d3ded56f35572f52a96f604a99cf4 | same |
 Replay time M5 114 s / M15 35 s on 12 workers (control runs 110 s / 35 s — recording costs
 seconds). `backtest/replay_scorecard.py DIR --check` passes all eight self-consistency
 invariants on both runs: trade counts match the manifest; every trade has its primary block
 and both confirmation blocks; base weight × regime multiplier = effective weight and
 contribution = vote × effective weight × confidence/100 on every one of the 17,869 (M5) /
 18,070 (M15) indicator rows; the composite score is reproduced from the stored rows on every
 trade and timeframe; every regime multiplier equals the table in the manifest; the bar table
 has exactly one row per analyzed bar and contains every trade's bar.
 ### Cross-check against the owner's 3-year baseline books (same window, pre-2026-09-07 code)
 | TF | 2-yr run | baseline `sizing_2026-09/base_*_3yr_065.csv` cut to the window | shared entries |
 |---|---|---|---|
 | M5 | 781 / +146.07R | 769 / +153.58R | 757 (24 only new, 12 only baseline) |
 | M15 | 793 / +117.62R | 788 / +115.78R | 774 (19 only new, 14 only baseline) |
 The baseline files are dated 2026-09-05 and predate the 2026-09-07 outlier-filter change
 (which moved the 3-yr M5 book 857→870 trades, +161.3→+153.6R), so the differences are the
 expected ones; byte identity was never expected here. The behaviour-neutrality proof is the
 MD5 pair above, on identical code.
 ## What was recorded (storage decision: Parquet, one folder per run)
 Parquet keeps exact typed numbers and UTC timestamps, compresses well (M5: 20 MB for 86k
 bars × 172 columns), and pandas reads a file in one line; the repo already uses it for the
 bar cache. SQLite would need JSON blobs or many tables with string timestamps; CSV loses
 types and is several times larger. DuckDB can query the parquet files directly if SQL is
 ever wanted.
 | file | grain | M5 rows | M15 rows |
 |---|---|---|---|
 | `manifest.json` | CLI, window, spread, TFs, engine + gate fingerprints, git SHA, backend, regime weight table in force, observed base weights, composite thresholds, bar counts + skip reasons, cache file names, calendar fingerprint, table row counts, book MD5 | 1 | 1 |
 | `trades.parquet` | one row per ACCEPTED trade: the 63 book columns + per-timeframe regime / composite / confidence / ATR / vote sums, penalty tags, setup fields, all warnings and reasons, fingerprints, calendar scalars | 781 | 793 |
 | `indicators.parquet` | one row per trade × timeframe × indicator that voted: vote, value, confidence, base weight, regime multiplier, effective weight, contribution, agree/oppose/neutral, regime | 17,869 | 18,070 |
 | `events.parquet` | one row per trade × calendar event from 4 h before entry to exit (or 4 h after entry), plus all-day rows on the entry's New-York date, with phase before / during_hold / after_exit / all_day | 7,803 | 12,932 |
 | `bars.parquet` | one row per ANALYZED primary bar (after weekend/session filters, before the setup gates): final score/signal/quality, has_setup, per-timeframe regime/composite and per-indicator vote/value/confidence/effective weight/contribution as float32, plus forward moves 1 h / 4 h / 24 h and MFE/MAE in ATRs on the raw bars | 86,192 | 29,917 |
 Candidates before the 5-bar cooldown: M5 1,279 → 781 accepted; M15 1,535 → 793. An
 indicator that did not fire (divergence, candlestick) has NO row / NaN — never a zero vote.
 Reader: `backtest.replay_record.load_run(dir)`. First analysis pass:
 `backtest/replay_scorecard.py DIR --what-if --bars` (full output saved as `scorecard.txt`
 in each run folder).
 ## First look (descriptive statistics only — NOT evidence for a change)
 Everything below is in-book: it cannot see trades a change would have created, and it
 ignores the MTF/penalty/gate chain that sits between the composite and the entry. The 2-year
 window is a strong gold up-trend, so the "no vote" baseline itself drifts up about +0.37 ATR
 per 4 h and +1.9 ATR per 24 h on M5 (+0.15 / +0.9 on M15); any BUY-vote number must be read
 against that baseline, not against zero.
 Trade-level (primary timeframe):
 - Bollinger opposes the entry on 731 of 781 M5 trades (94%): a mean-reversion reading
  inside a trend-following entry — the "constant bias" seen in the 2026-08-04 audit — and
  those trades carry +152R. RSI is neutral on 73% of M5 entries.
 - M5 trades where divergence OPPOSED the entry did best (158 trades, +0.37R mean, 63% wins)
  and trades where it was absent did worst (546, +0.13R); the same "inverted" pattern as the
  2026-08-04 in-book audit, which replayed WORSE when "fixed" — the reminder that this table
  is a hypothesis generator.
 - What-if (M5): removing ema_trend would drop 206 accepted trades carrying +36.8R below the
  entry band; removing MACD 211 / +21.2R; ADX 127 / +22.6R. On M15 hardly any trade loses
  direction from a single-indicator removal (the entries sit further above the band); the
  changes show up as strong/weak band moves instead.
 - Calendar: M5 trades held through at least one high-impact USD release did better
  (69 trades, +0.42R mean vs +0.19 overall; M15 224 trades, +0.23 vs +0.15) — the same
  direction as the 2026-09-06 addendum. Entries within 30 min after a release are slightly
  negative on both TFs (40 and 35 trades — too few to act on).
 Bar-level (all analyzed bars, drift NOT yet removed):
 - Every indicator's SELL-side votes were followed, on average, by a move AGAINST the vote
  on both timeframes (e.g. M5 ema_trend −0.27 ATR / 4 h, divergence −0.63); this is the
  up-drift plus the known "SELL negative everywhere" finding. The next analysis step is to
  subtract the per-bar baseline (or split by regime) before scoring votes.
 - Above-baseline BUY callers on M5 at 4 h: ADX +0.50, keltner +0.50, RSI +0.45,
  candlestick +0.46, ROC +0.46, ema_trend (+0.67 vs baseline +0.32 among its own neutral
  bars); below baseline: divergence +0.30, Bollinger +0.36.
 - Bars what-if (M5): dropping divergence would newly qualify 2,268 bars whose later move
  averaged +0.26 ATR / 4 h and drop 3,611 bars averaging −0.26 — the one single-indicator
  removal where both sides point the same way. Dropping ema_trend or ADX points the other
  way (the bars they add did worse, the bars they remove did better).
 ## Pre-registered rule for anything that comes out of these files
 A candidate change (weight, regime multiplier, dropped or replaced indicator) is a
 hypothesis until it passes CLAUDE.md rule 1 on a full rerun through
 `analyze_from_dataframes`: positive total R in BOTH halves of a ≥12-month window, PF ≥ 1.3
 at $0.65, trailing exits, next-open fills, plus the 2024-06→2025-03 out-of-sample window for
 marginal edges, compared with a CONTROL produced on the same harness. Nothing in this
 campaign changes any number.
 ## Suggested next analysis (not done here)
 1. Drift-adjust the bar table (subtract the neutral/absent baseline per regime and hour) and
   re-score each indicator's standalone hit rate by timeframe role.
 2. For the 2–3 most suspect indicators (divergence, Bollinger, RSI on M5), run the real
   rule-1 rerun with `--set technical.regime_weights_override=...` or `disabled_indicators`
   and compare against the control books in this folder.
 3. Re-run this recording after any config change so the folders stay comparable (the
   manifest carries the engine fingerprint and the regime table for exactly that purpose).
 Tests: Module 33 "backtest replay recording" (12 tests); suite green.
@@ -0,0 +1,187 @@
 # Weight-change test campaign — 2026-09-09 (does the recorded evidence improve predictions?)
 ## Purpose
 First use of the replay-recording files (`replay_recording_2026-09-09.md`): take candidate
 changes to the regime weight table suggested by the per-bar evidence, rerun the FULL backtest
 with each, and compare against the same-harness control books. This is the "before versus
 after" that measures improvement. Everything below the rule was written before any candidate
 was run.
 ## Harness (identical to the recording campaign)
 `backtest/gpu_backtest.py --pair XAUUSD --start 2024-09-05 --end 2026-09-05 --spread 0.65
 --workers N --primary-tf {5min|15min}` + `--set technical.regime_weights_override=<full table>`
 (the field replaces the whole table, so every candidate passes the complete table with one
 cell family changed). Controls: `backtest/research/replay_2026-09/{m5,m15}/book_control.csv`
 (same code, no override). OOS window: `--start 2024-06-01 --end 2025-03-30` (cached
 `oanda_XAU_USD_*_20240331_20250330.parquet`), control produced on the same harness.
 Backend NumPy CPU. Code: git `af06685`.
 ## Pre-registered candidate-selection rule (written 2026-09-09 before the screen ran)
 1. A candidate changes ONE indicator's regime multipliers only (no code, no gate, no
   threshold change), on ONE timeframe. Allowed forms: multiplier → 0 in all regimes,
   × 0.5 in all regimes, × 2 in all regimes, or the same applied to a single regime family
   (trending / ranging / transitioning) when the screen shows the effect is regime-specific.
 2. The screen (scratch script over `bars.parquet`, both TFs): for each candidate, recompute
   every analyzed bar's PRIMARY composite from the recorded effective weights and
   contributions, find the bars that would NEWLY clear ±15 and the bars that would DROP OUT,
   and score the candidate by the net drift-adjusted signed 4-hour forward move
   (gained bars minus lost bars; drift = the mean unsigned 4-hour move of all analyzed bars
   in the same regime). Top 3 per timeframe by that score are run; nothing else.
 3. Earlier verdicts on the same knob are recorded next to the candidate but do not veto it
   (this harness — 2-year Oanda window, post-2026-09-07 outlier rule — is new).
 ## Pre-registered acceptance rule (vs the same-harness control)
 A candidate PASSES only if ALL of the following hold on the 2-year window at $0.65:
 - a. ΔR ≥ +5R overall versus the control (smaller differences are noise on ~800 trades);
 - b. ΔR > 0 in BOTH fixed halves: 2024-09-05→2025-09-05 and 2025-09-05→2026-09-05;
 - c. PF ≥ 1.3 and not more than 0.05 below the control's PF;
 - d. max drawdown not more than 2R worse than the control's;
 - e. trade count within ±25% of the control's (guards against "fewer trades" artefacts);
 - f. OOS window 2024-06-01→2025-03-30: ΔR ≥ 0 versus the OOS control.
 Failing candidates are recorded and closed. No rescue: no multiplier fine-tuning after
 seeing a result, no second window shopping. A passing candidate is a SHIPPING PROPOSAL for
 the owner (rule 2: it would land as a per-timeframe `regime_weights_override` in
 `pair_signals/xauusd.py` with tests), not an automatic change.
 ## Screen results (scratch script over `bars.parquet`, both TFs; composite reproduced from the stored columns within 3.4e-6 on every bar)
 Drift per regime (mean 4 h move of all analyzed bars, ATR units, M5): trending_up +0.55,
 trending_down +0.16, trending +0.35, transitioning +0.64, ranging +0.38 — a strong up-drift
 that every BUY vote inherits, hence the per-regime adjustment.
 Standalone votes after drift removal (mean signed 4 h move; positive = the vote pointed the
 right way), M5 highlights:
 | indicator | regime family | buy votes | sell votes | reading |
 |---|---|---|---|---|
 | macd | trending | −0.079 (n 8,717) | −0.154 (n 8,891) | wrong way on BOTH sides; also at 24 h (−0.22 / −0.56) |
 | macd | transitioning | +0.273 (1,738) | +0.288 (1,885) | best caller in that regime |
 | rsi | transitioning | −1.089 (357) | −0.146 (570) | wrong way, small n |
 | bollinger | transitioning | −0.261 (2,668) | −0.337 (2,675) | wrong way |
 | bollinger | trending | +0.084 | +0.091 | mildly right |
 | ema_trend | ranging / transitioning | +0.33 / +0.42 | ≈0 | buy side right, sell side nothing |
 | adx | trending | +0.048 | +0.041 | never fires outside trending |
 Candidate screen (net drift-adjusted 4 h move of gained minus lost bars; the 24 h column must
 agree in sign for a candidate to be run). No candidate flips a bar's direction — one
 indicator's weight only moves bars in and out of the neutral band.
 M5 top of the board: `macd:zero_trending` +1235 (24 h +4181, agrees; 3,406 gained / 7,339 lost,
 touches 23% of directional bars), `macd:double_ranging` +1186 (24 h −21, rejected),
 `divergence:zero_all` +1026 (24 h −633, rejected), `bollinger:zero_ranging` +940 (24 h −728,
 rejected), `bollinger:zero_all` +800 (24 h −1654, rejected), `macd:zero_all` +705 (24 h +1644,
 agrees; touches 50% of directional bars), `macd:half_trending` +689 (24 h +334, agrees).
 Bottom: `ema_trend:zero_all` −389, `macd:double_trending` −540, `macd:zero_transitioning` −607.
 M15 top of the board: `ema_trend:double_all` +329 (24 h +1209, agrees), `macd:zero_all` +293
 (24 h −79, rejected), `divergence:zero_all` +264 (24 h −480, rejected),
 `divergence:zero_trending` +246 (24 h +390, agrees), `rsi:half_all` +221 (24 h +99, agrees).
 Bottom: `macd:double_trending` −337, `macd:double_all` −323.
 Candidates run (rule 2 above, 24 h-sign filter applied):
 - M5-1 `macd:zero_trending` — MACD multiplier 2.0 → 0 in trending_up/trending_down/trending.
 - M5-2 `macd:zero_all` — MACD multiplier → 0 in every regime.
 - M5-3 `macd:half_trending` — MACD multiplier 2.0 → 1.0 in the trending family.
 - M15-1 `ema_trend:double_all` — ema_trend multiplier × 2 in every regime.
 - M15-2 `divergence:zero_trending` — divergence multiplier 1.0 → 0 in the trending family.
 - M15-3 `rsi:half_all` — RSI multiplier × 0.5 in every regime.
 Prior verdicts on the same knobs: none for MACD or ema_trend; "divergence-down" on M5 was
 rejected 2026-08-04 and "divergence-off" on M15 was declined 2026-09 (R up, 5-yr DD 35.8 →
 47.4R); neither is the same change as M15-2 (trending family only).
 Half of each board's top entries fail the 24 h check, some violently (`bollinger:zero_all` on
 M5: +800 at 4 h, −1654 at 24 h) — a 4 h-only screen is a coin flip. Scores are sums, so
 volume can carry them: `macd:zero_all` scores by removing 16,170 directional bars whose mean
 was −0.07, while its gained bars average −0.09.
 ## Rerun results (2-year window 2024-09-05→2026-09-05, $0.65; halves fixed at 2025-09-05; OOS 2024-06-01→2025-03-30)
 | candidate | trades | total R | ΔR | H1 ΔR | H2 ΔR | PF | max DD | OOS ΔR | verdict |
 |---|---|---|---|---|---|---|---|---|---|
 | M5 control | 781 | +146.07 | | | | 1.52 | 10.1 | (86 / +9.47, PF 1.33) | |
 | M5-1 macd → 0 in trending | 808 | +105.35 | −40.7 | +1.9 | −42.6 | 1.35 | 17.1 | +0.2 | FAIL a,b,c,d |
 | M5-2 macd → 0 everywhere | 699 | +99.79 | −46.3 | +0.5 | −46.8 | 1.38 | 17.2 | −1.8 | FAIL a,b,c,d,f |
 | M5-3 macd × 0.5 in trending | 804 | +136.12 | −10.0 | +7.4 | −17.4 | 1.47 | 15.2 | +6.8 | FAIL a,b,c,d |
 | M15 control | 793 | +117.62 | | | | 1.41 | 14.8 | (300 / +40.59, PF 1.38) | |
 | M15-1 ema_trend × 2 everywhere | 1229 | +148.20 | +30.6 | +10.0 | +20.5 | 1.32 | 26.4 | +0.7 | FAIL c,d,e (+55% trades) |
 | M15-2 divergence → 0 in trending | 861 | +127.69 | +10.1 | −3.0 | +13.1 | 1.41 | 18.4 | +3.6 | FAIL b,d |
 | **M15-3 rsi × 0.5 everywhere** | **931** | **+146.14** | **+28.5** | **+27.9** | **+0.6** | **1.43** | **16.1** | **+20.1 (355 / +60.67, PF 1.49)** | **PASS a–f** |
 The M5 screen was wrong in every case: the three MACD candidates that scored best on the
 per-bar screen (and were the only ones whose 24 h score agreed) lose 10–46R in the full
 replay, all of it in the second year. The per-bar composite screen sees threshold crossings;
 it does not see that MACD's "wrong-way" votes in trending regimes are doing useful work
 downstream (the exhaustion penalties, the M5 min-score gate and the setup generators all sit
 between the composite and a trade). Lesson recorded: the screen ranks hypotheses, the replay
 judges them, and on M5 the two disagreed 3/3.
 ### M15-3 trade diff (control → candidate, 2-year)
 - 721 shared entries: control +124.7R → candidate +119.4R (56 shared trades changed R because
  the score moved their grade and therefore the setup distances).
 - 210 added trades: +26.8R, mean +0.127R, WR 59.5% — 157 BUY (+32.5R) and 53 SELL (−5.8R);
  156 are B-grade (+19.1R), 201 sit in the 25–40 score band; by regime ranging 63 (+13.1R),
  trending_up 88 (+12.1R), transitioning 32 (+2.9R), trending_down 27 (−1.3R).
 - 72 removed trades: −7.1R, WR 50%.
 - By year: 2024 (Sep–Dec) +19.4 → +26.7R, 2025 +51.9 → +77.4R, 2026 (Jan–Sep) +46.3 → +42.1R.
  The 2-year gain is concentrated in 2025; the fixed second half passes rule b by only +0.6R.
 - OOS (2024-06→2025-03): 300 → 355 trades, +40.6 → +60.7R, PF 1.38 → 1.49, WR 59.3 → 60.8%.
 Mechanism in plain terms: RSI is neutral on most M15 entries and, when it votes, it votes
 against the trend (a mean-reversion reading). Halving its weight lets more trend-following
 entries clear the score band, mostly BUYs, in a data set that is a gold bull market
 throughout. The OOS window improves in quality (PF, win rate), not only in count, which is
 the strongest single fact in the candidate's favour; the year-by-year unevenness is the
 strongest fact against.
 ### Robustness check on the 3-year window (owner's maximum; 2023-09-05→2026-09-05, cached bars)
 | book | trades | total R | WR | PF | max DD |
 |---|---|---|---|---|---|
 | M15 control 3-yr | 1092 | +151.90 | 59.4% | 1.38 | 15.5 |
 | M15-3 rsi × 0.5, 3-yr | 1273 | +186.61 | 59.4% | 1.39 | 19.3 |
 | extra year only (2023-09-05→2024-09-05, never seen by the screen): control | 299 | +33.23 | 56.9% | 1.29 | 14.9 |
 | extra year only: candidate | 342 | +39.43 | 56.4% | 1.29 | 19.3 |
 Calendar years, control → candidate: 2023 (Sep–Dec) +10.1 → +13.2R; 2024 +42.6 → +52.9R;
 2025 +51.9 → +77.4R; 2026 (Jan–Sep) +47.3 → +43.2R. Three of four years improve; the most
 recent one does not, and the 3-year max drawdown is 3.8R deeper (all of it in the 2023-24 year).
 ## Verdict
 - **The process works end to end**: recorded evidence → screen → pre-registered rule → full
  replay against same-harness controls → one pass, five fails, each with the reason recorded.
 - **M5**: all three MACD candidates FAIL (−10 to −46R). Closed. The per-bar screen is not to
  be trusted on M5 without a replay; it ranked them first.
 - **M15-1 (ema_trend × 2)** FAILS on PF, drawdown and trade count; **M15-2 (divergence → 0 in
  trending)** FAILS the first half and the drawdown. Both closed.
 - **M15-3 (RSI multiplier × 0.5 in every regime) PASSES the pre-registered rule** (a–f) and the
  3-year robustness check is positive (+34.7R, PF 1.38 → 1.39, extra year +6.2R). It is a
  SHIPPING PROPOSAL for the owner, with these facts on the other side of the scale: the 2-year
  second half passes by only +0.6R, 2026 is −4R, the added trades are mostly B-grade BUYs in a
  bull-market data set (added SELLs lose), and the 3-year max drawdown deepens 15.5 → 19.3R.
  If shipped it lands as an M15-only `technical.regime_weights_override` in
  `pair_signals/xauusd.py` (rule 2) with the full table (RSI 1.0 → 0.5 in the trending family,
  1.5 → 0.75 in ranging, 1.0 → 0.5 in transitioning), a factory-equivalence test update and a
  docs note; the M15 canonical book would then be re-pinned from `candidates/m15_rsi:half_all_2yr.csv`
  (931 / +146.14R / PF 1.43 at $0.65). Not shipped in this campaign.
 Books and logs: `backtest/research/replay_2026-09/candidates/` and `.../oos/` (gitignored).
 ## Shipped (owner decision 2026-09-09, same day)
 M15 override 5 in `pair_signals/xauusd.py`: `technical.regime_weights_override` = the shipped
 table with the RSI multiplier halved in every regime (0.5 trending family / 0.75 ranging /
 0.5 transitioning), derived from the default table and pinned literally by the Module 9
 factory-equivalence test; one more Module 33 test locks that M5/H1/H4 carry no override.
 Parity proof — the factory run with NO `--set` flag reproduces the validated candidate books
 byte-for-byte: 2-year MD5 `d185f099482a07f29502cb5a371a9bbf` (931 / +146.14R / PF 1.43) and
 OOS MD5 `6f8f24b49597ced28f50a19e08c3472f` (355 / +60.67R / PF 1.49). Canonical M15 book
 re-pinned to `backtest/oanda_m15_2yr_065_20260909.csv` (copy of the 2-year factory book).
 Docs: `docs/user_guide.html` and `docs/beginner_guide.html` carry a one-line note (PDF builds
 remain behind, as before). Suite 499 tests. Live M15 scans get a new `engine_fingerprint`
 from the next deploy; signal-log comparisons across it must split at the deploy time.
@@ -138,6 +138,7 @@ The message says: <i>"once price hits 4,159, trail SL $33 behind best price."</i
 <li><b>H1 trades are held longer</b> — often overnight, sometimes over days. That is normal. Your stop loss is working for you while you sleep.</li>
 <li><b>H1 is BUY-only.</b> All H1 SELL signals are blocked (since 2026-09-03) because they were a losing side on this timeframe — expect to see BUY signals here, never SELL.</li>
 <li><b>M5 also holds back some SELL signals</b> — specifically while the market is in a choppy "transitioning" phase, since those shorts tested poorly.</li>
 <li><b>M15 got a small tune-up on 2026-09-09</b> — one of its nine indicators (RSI) now counts for half as much, which means a few more M15 signals per week. Nothing changes in how you act on a signal.</li>
 <li><b>Never take the same signal on two timeframes.</b> They are usually the same gold move seen at different zoom levels. Taking both doubles your risk, not your profit.</li>
 </ul>
@@ -98,7 +98,7 @@
 </table>
 <ul>
 <li><b>M5 — intraday, execution-critical.</b> Highest total edge, but stops are tight: a couple of dollars of slippage or a late entry eats a meaningful share of every trade. Suitable only if you can act within a minute or two of the alert (or automate). Trades open and close within a session; you sleep flat. SELL signals are held back while the market regime reads "transitioning" — a validated filter, not a bug if you notice fewer shorts there.</li>
-<li><b>M15 — the middle path.</b> Same frequency as M5, wider stops, up to 15 minutes to act. About half your trades will still be open overnight.</li>
+<li><b>M15 — the middle path.</b> Same frequency as M5, wider stops, up to 15 minutes to act. About half your trades will still be open overnight. Since 2026-09-09 the M15 scoring gives its RSI reading half the weight it used to (RSI mostly voted against the trend on this timeframe), so expect roughly 9 signals a week instead of 7–8; in the two-year test (Sep 2024 – Sep 2026) that change lifted M15 from +117.6R to +146.1R with the same win rate and a slightly better profit factor, and it also improved the separate out-of-sample period. The M15 column above predates it.</li>
 <li><b>H1 — swing trading around a normal life.</b> Best quality per trade and very forgiving of slow manual execution (you have up to an hour). The costs: positions held for days, through news and overnight sessions, and a <b>lumpy</b> equity curve — one quarter of the test year was negative and most of the profit came in a single quarter. Judge it on quarters, not weeks. H1 no longer issues SELL signals at all (blocked since 2026-09-03) — it is BUY-only on this timeframe.</li>
 </ul>
 <div class="callout"><b>If you trade manually from the Telegram alerts, H1 (or M15) is the honest choice.</b> M5's backtested edge assumes near-instant fills that a human copying signals rarely achieves.</div>
@@ -90,6 +90,23 @@ metric. One clause waived BY OWNER DECISION 2026-07-21: 5yr max DD is
 accepted in exchange for +20% total R. M5 keeps the global 0.6 arm
 (same per-TF divergence pattern as tp_floor_rr and the hour-9 block).
 M15 override 5 — RSI regime multiplier x 0.5 in every regime
 (validated 2026-09-09, 2-year Oanda window 2024-09-05 -> 2026-09-05 at
 $0.65, pre-registered rule in backtest/research/
 weight_test_2026-09-09.md): 793/+117.62R PF 1.41 DD 14.8R ->
 931/+146.14R PF 1.43 DD 16.1R, fixed halves +27.9/+0.6R vs control;
 OOS 2024-06-01 -> 2025-03-30 300/+40.59R PF 1.38 -> 355/+60.67R
 PF 1.49; 3-year 1092/+151.90R PF 1.38 -> 1273/+186.61R PF 1.39 (max DD
 15.5 -> 19.3R, all in the 2023-24 year; 2026 -4R is the one calendar
 year that does not improve). Mechanism: RSI is neutral on ~73% of M15
 entries and votes against the trend when it fires; halving it lets more
 trend-following entries clear the score band (210 added trades +26.8R,
 mostly B-grade BUYs; 72 removed -7.1R). Candidate came from the
 recorded replay screen (replay_recording_2026-09-09.md); the same
 screen's M5 picks (MACD-down) all FAILED the replay, and ema_trend x 2
 / divergence -> 0 on M15 failed too. Canonical M15 book re-pinned to
 backtest/oanda_m15_2yr_065_20260909.csv.
 M5 override (validated 2026-07-11, full-history replay 2020-03-17 ->
 2026-07-11 + 12mo book, $0.30 spread): block hour 9 UTC (post-London-
 open whipsaw). Hour-9 entries lost -28.9R/143 over 6 years, negative
@@ -210,8 +227,8 @@ bollinger all look mis-weighted in-book and all replay worse when
 oppose vote is a load-bearing quality bar, like ROC).
 """
-from typing import Optional, List
+from typing import Dict, Optional, List
-from config import PlatformConfig, Timeframe
+from config import PlatformConfig, TechnicalConfig, Timeframe
 from app_config import apply_pair_overrides
 _CONFIRMATIONS = {
@@ -222,6 +239,19 @@ _CONFIRMATIONS = {
 }
 def _m15_regime_weights() -> Dict[str, Dict[str, float]]:
    """Shipped regime weight table with the RSI multiplier halved in
    every regime — the M15-only override validated 2026-09-09 (see the
    module docstring, M15 override 5). Derived from the default table so
    the intent ("RSI x 0.5, everything else as shipped") stays explicit;
    the Module 9 equivalence test pins the resulting numbers."""
    from engine import regime_weight_table  # local import: keep the factory import-light
    table = regime_weight_table(TechnicalConfig())   # a fresh TechnicalConfig has no override -> the shipped table
    return {regime: {name: (mult * 0.5 if name == "rsi" else mult)
                     for name, mult in weights.items()}
            for regime, weights in table.items()}
 def make_config(
    primary_tf: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
@@ -272,6 +302,9 @@ def make_config(
        # Later trailing-stop arm on M15 only — see module docstring.
        config.signals.trail_arm_r = 0.70
        config.signals.trail_dist_r = 0.75
        # RSI regime multiplier × 0.5 in every regime on M15 only — see
        # module docstring (M15 override 5, validated 2026-09-09).
        config.technical.regime_weights_override = _m15_regime_weights()
    if primary_tf == Timeframe.H1:
        # Block every H1 SELL — the SELL quality gate (6e in
        # signal_aggregator.py) downgrades any actionable-grade short
@@ -0,0 +1,144 @@
 """Forex Factory calendar scraper -> backtest/cache/calendar.parquet.
 Fetches weekly calendar pages and parses the event JSON embedded in
 `window.calendarComponentStates` (the `days:` array). Event `dateline`
 is a unix epoch, so timestamps carry no timezone ambiguity — verified
 2026-07-11 against known release times (NFP 12:30/13:30 UTC across DST,
 ECB 12:15/13:15) and against EUR/USD M15 volatility: the bar containing
 a high-impact event averages 3.5x normal range, and shifting event
 times by ±1h destroys that alignment.
 Usage:
    python scripts/ff_calendar_scraper.py --start 2024-01-01 --end 2026-07-13
    python scripts/ff_calendar_scraper.py            # refresh: last stored week -> today
 Merges into the existing parquet on event (id, dateline), so refreshing
 is idempotent and also picks up revised `actual` values for recent weeks.
 Be polite: ~1.6s between requests (a full year is ~52 requests).
 """
 import argparse
 import json
 import time
 from datetime import date, timedelta
 from pathlib import Path
 from typing import List, Optional
 import pandas as pd
 import requests
 CACHE_DIR = Path(__file__).resolve().parents[1] / "backtest" / "cache"
 PARQUET = CACHE_DIR / "calendar.parquet"
 UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/126.0 Safari/537.36")
 FIELDS = ["id", "dateline", "name", "country", "currency", "impactName",
          "timeLabel", "timeMasked", "actual", "forecast", "previous",
          "revision", "actualBetterWorse"]
 def week_url(d: date) -> str:
    return (f"https://www.forexfactory.com/calendar?week="
            f"{d.strftime('%b').lower()}{d.day}.{d.year}")
 def extract_days(html: str) -> Optional[list]:
    """Balanced-bracket extraction of the `days:` JSON array."""
    i = html.find("days:")
    if i < 0:
        return None
    j = html.find("[", i)
    depth, in_str, esc = 0, False, False
    for k in range(j, len(html)):
        ch = html[k]
        if in_str:
            if esc:
                esc = False
            elif ch == "\\":
                esc = True
            elif ch == '"':
                in_str = False
        else:
            if ch == '"':
                in_str = True
            elif ch == "[":
                depth += 1
            elif ch == "]":
                depth -= 1
                if depth == 0:
                    return json.loads(html[j:k + 1])
    return None
 def fetch_week(sess: requests.Session, d: date, retries: int = 4) -> Optional[List[dict]]:
    for attempt in range(retries):
        try:
            r = sess.get(week_url(d), timeout=30)
            if r.status_code == 200:
                days = extract_days(r.text)
                if days is not None:
                    return [{k: e.get(k) for k in FIELDS}
                            for day in days for e in day.get("events", [])]
        except requests.RequestException:
            pass
        time.sleep(5 * (attempt + 1))
    return None
 def scrape(start: date, end: date) -> pd.DataFrame:
    sess = requests.Session()
    sess.headers["User-Agent"] = UA
    rows: List[dict] = []
    failed: List[date] = []
    d = start
    while d <= end:
        week_rows = fetch_week(sess, d)
        if week_rows is None:
            failed.append(d)
            print(f"FAILED week {d}")
        else:
            rows.extend(week_rows)
        d += timedelta(days=7)
        time.sleep(1.6)
    if failed:
        print(f"WARNING: {len(failed)} weeks failed: {failed}")
    return pd.DataFrame(rows)
 def merge_into_store(new: pd.DataFrame, path: Path = PARQUET) -> pd.DataFrame:
    if path.exists():
        old = pd.read_parquet(path)
        combined = pd.concat([old, new])
    else:
        combined = new
    # keep="last": a refresh overwrites stale rows (actuals fill in
    # after release, revisions land weeks later).
    combined = (combined.drop_duplicates(subset=["id", "dateline"], keep="last")
                .sort_values("dateline").reset_index(drop=True))
    combined.to_parquet(path, index=False)
    return combined
 def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--start", type=date.fromisoformat, default=None)
    ap.add_argument("--end", type=date.fromisoformat, default=date.today())
    args = ap.parse_args()
    start = args.start
    if start is None:
        if PARQUET.exists():
            last = pd.read_parquet(PARQUET).dateline.max()
            start = pd.Timestamp(last, unit="s").date() - timedelta(days=7)
        else:
            start = date(2024, 1, 1)
    print(f"Scraping {start} -> {args.end}")
    new = scrape(start, args.end)
    combined = merge_into_store(new)
    print(f"Store: {len(combined)} events "
          f"({pd.Timestamp(combined.dateline.min(), unit='s')} -> "
          f"{pd.Timestamp(combined.dateline.max(), unit='s')}) -> {PARQUET}")
 if __name__ == "__main__":
    main()
@@ -2078,6 +2078,35 @@ def _():
            inline_cfg.xau_strategy.sell_blocked_hours_utc = [15, 17]
            inline_cfg.signals.trail_arm_r = 0.70
            inline_cfg.signals.trail_dist_r = 0.75
            # M15 override 5 (2026-09-09): RSI x 0.5 in every regime —
            # LITERAL so a silent default-table change fails loudly here.
            inline_cfg.technical.regime_weights_override = {
                "trending_up": {
                    "ema_trend": 2.0, "rsi": 0.5, "macd": 2.0,
                    "bollinger": 0.5, "keltner": 1.5, "roc": 1.5,
                    "candlestick": 0.8, "divergence": 1.0, "adx": 0.8,
                },
                "trending_down": {
                    "ema_trend": 2.0, "rsi": 0.5, "macd": 2.0,
                    "bollinger": 0.5, "keltner": 1.5, "roc": 1.5,
                    "candlestick": 0.8, "divergence": 1.0, "adx": 0.8,
                },
                "trending": {
                    "ema_trend": 2.0, "rsi": 0.5, "macd": 2.0,
                    "bollinger": 0.5, "keltner": 1.5, "roc": 1.5,
                    "candlestick": 0.8, "divergence": 1.0, "adx": 0.8,
                },
                "ranging": {
                    "ema_trend": 0.3, "rsi": 0.75, "macd": 1.5,
                    "bollinger": 1.5, "keltner": 2.0, "roc": 0.5,
                    "candlestick": 1.0, "divergence": 2.5, "adx": 0.5,
                },
                "transitioning": {
                    "ema_trend": 1.5, "rsi": 0.5, "macd": 2.5,
                    "bollinger": 1.0, "keltner": 2.5, "roc": 2.0,
                    "candlestick": 1.0, "divergence": 2.0, "adx": 1.0,
                },
            }
        if tf == Timeframe.M5:
            # Documented per-TF overrides (see xauusd.py docstring):
            # hour 9 UTC blocked, hour 21 UNBLOCKED (2026-07-30
@@ -11281,6 +11310,617 @@ def _():
        assert name not in req, f"{name} must not enter requirements.txt"
 # ══════════════════════════════════════════════════════════════════════
 #  MODULE 33: backtest replay recording (--record, research-only)
 # ══════════════════════════════════════════════════════════════════════
 print("\n═══ Module 33: backtest replay recording ═══")
 import contextlib as _ctx33
 import dataclasses as _dc33
 import hashlib as _hashlib33
 import pickle as _pickle33
 import types as _types33
 from backtest import gpu_backtest as _gb33
 from backtest import replay_record as _rr
 _M33_CACHE = {}
 def _m33_report():
    """One REAL M5 analyze_from_dataframes run + a hand-built setup.
    Cached — the engine replay is the slow part and two tests need the
    same report. Returns (cfg, aggregator, report, setup)."""
    if "report" not in _M33_CACHE:
        from pair_signals.xauusd import make_config as _mk33
        cfg = _mk33(Timeframe.M5, [Timeframe.M15, Timeframe.H1])
        dfs = {
            Timeframe.M5: make_xau_bars(n=400, freq="5min",
                                        end="2026-07-08 16:55"),
            Timeframe.M15: make_xau_bars(n=300, freq="15min",
                                         end="2026-07-08 16:45", seed=8),
            Timeframe.H1: make_xau_bars(n=300, freq="60min",
                                        end="2026-07-08 16:00", seed=9),
        }
        agg = SignalAggregator(cfg)
        report = agg.analyze_from_dataframes(XAUUSD, dfs, use_news=False)
        px = report.current_price
        # Direction only drives the `agrees` label; this fixture's score
        # is bearish, so a SELL keeps the labels meaningful.
        setup = TradeSetup(
            pair=XAUUSD, display_name="XAU/USD", direction="SELL",
            entry_price=px, stop_loss=px + 6.0, take_profit_1=px - 12.0,
            take_profit_2=px - 20.0, risk_reward=2.0, risk_pips=60.0,
            reward_pips=120.0, pip_value=0.10, strategy="XAU_ATR")
        _M33_CACHE["report"] = (cfg, agg, report, setup)
    return _M33_CACHE["report"]
 def _m33_snapshot():
    """snapshot_report() of the cached real report."""
    cfg, _agg, report, setup = _m33_report()
    return _rr.snapshot_report(
        report, setup, pd.Timestamp("2026-07-08 17:00", tz="UTC"),
        Timeframe.M5, [Timeframe.M15, Timeframe.H1])
 def _m33_hand_analysis():
    """A TechnicalAnalysis with exactly 4 VOTING indicators + a weight-0
    ATR, pushed through the real _compute_composite so the effective
    weight / weighted score stamps are the engine's own."""
    cfg, _agg, _rep, _setup = _m33_report()
    result = TechnicalAnalysis(
        symbol=XAUUSD, timeframe=Timeframe.M5.value,
        timestamp="2026-07-08 16:55:00+00:00",
        indicators={
            "ema_trend": IndicatorSignal("EMA", 1.0, SignalStrength.BUY,
                                         60.0, "ema up", 1.0),
            "rsi": IndicatorSignal("RSI", 55.0, SignalStrength.NEUTRAL,
                                   50.0, "rsi mid", 1.0),
            "macd": IndicatorSignal("MACD", 0.4, SignalStrength.BUY,
                                    70.0, "macd up", 1.5),
            "keltner": IndicatorSignal("KC", 2.0, SignalStrength.SELL,
                                       55.0, "kc high", 1.5),
            # Context only (weight 0) — the composite skips it, so the
            # recorder must too.
            "atr": IndicatorSignal("ATR", 5.0, SignalStrength.NEUTRAL,
                                   50.0, "atr", 0.0),
        },
        market_regime="ranging", volatility_regime="normal",
        key_levels={"atr": 5.0, "atr_pct": 0.15})
    TechnicalEngine(cfg.technical)._compute_composite(result)
    return result
 def _m33_duck_report(analysis):
    """Minimal duck-typed report carrying every attribute snapshot_report
    and add_bar_row read (they never touch anything else)."""
    return _types33.SimpleNamespace(
        timestamp="2026-07-08 16:55:00+00:00",
        primary_analysis=analysis, confirmation_analyses=[],
        final_signal=SignalStrength.BUY, final_score=22.0,
        final_confidence=58.0, signal_quality="B",
        mtf_agreement=False, mtf_signals={}, mtf_scores={},
        penalty_tags={}, penalty_total=0.0, momentum_fading=False,
        exhaustion_count=0, spread_estimate=0.30, current_price=3300.0,
        warnings=["a warning"], cleaned_warnings=[],
        bullish_reasons=["b"], bearish_reasons=[], trade_setup=None)
 def _m33_cal_row(row_id, ts, name, currency="USD", impact="high",
                 masked=False, label="1:30pm"):
    """One Forex-Factory calendar row in the parquet's own schema."""
    return {
        "id": row_id, "dateline": int(pd.Timestamp(ts).timestamp()),
        "name": name, "country": currency[:2], "currency": currency,
        "impactName": impact, "timeLabel": label, "timeMasked": masked,
        "actual": "1.0", "forecast": "0.9", "previous": "0.8",
        "revision": "", "actualBetterWorse": 1,
    }
 def _m33_span_frames(drift=0.8):
    """600 M5 bars + long-enough M15/H1 frames that produce candidates."""
    return {
        Timeframe.M5: make_xau_bars(n=600, freq="5min",
                                    end="2026-07-08 16:55", drift=drift, seed=7),
        Timeframe.M15: make_xau_bars(n=300, freq="15min",
                                     end="2026-07-08 16:45", drift=drift, seed=8),
        Timeframe.H1: make_xau_bars(n=300, freq="60min",
                                    end="2026-07-08 16:00", drift=drift, seed=9),
    }
@test("replay record: snapshot reproduces the engine's weights and composite on a real report")
 def _():
    cfg, _agg, report, setup = _m33_report()
    snap = _m33_snapshot()
    assert [b["role"] for b in snap["tfs"]] == ["primary", "c1", "c2"], \
        [b["role"] for b in snap["tfs"]]
    assert [b["tf"] for b in snap["tfs"]] == ["5min", "15min", "1h"], \
        [b["tf"] for b in snap["tfs"]]
    table = regime_weight_table(cfg.technical)
    analyses = [report.primary_analysis] + list(report.confirmation_analyses)
    for block, analysis in zip(snap["tfs"], analyses):
        # The stored set is exactly the composite's voting set.
        voting = {name for name, ind in analysis.indicators.items()
                  if not name.startswith("_") and ind.weight != 0}
        assert set(block["indicators"]) == voting, \
            (block["role"], sorted(block["indicators"]), sorted(voting))
        mults = table.get(analysis.market_regime, {})
        for name, stored in block["indicators"].items():
            ind = analysis.indicators[name]
            expect_eff = stored["base_weight"] * mults.get(name, 1.0)
            assert abs(stored["eff_weight"] - expect_eff) < 1e-9, \
                (name, stored["eff_weight"], expect_eff)
            expect_ws = (stored["signal"] * stored["eff_weight"]
                         * stored["confidence"] / 100.0)
            assert abs(stored["wscore"] - expect_ws) < 1e-9, \
                (name, stored["wscore"], expect_ws)
            assert abs(stored["regime_mult"] * stored["base_weight"]
                       - stored["eff_weight"]) < 1e-9, name
            assert stored["signal"] == ind.signal.value
        # Σ wscore / Σ eff_weight reproduces the composite exactly.
        effs = [v["eff_weight"] for v in block["indicators"].values()]
        wscores = [v["wscore"] for v in block["indicators"].values()]
        score, sig = _rr.composite_from_rows(effs, wscores)
        assert abs(score - analysis.composite_score) < 1e-9, \
            (block["role"], score, analysis.composite_score)
        assert sig == analysis.composite_signal.value, \
            (block["role"], sig, analysis.composite_signal.value)
    # Plain data only: it must survive JSON and the worker pickle boundary.
    json.dumps(snap)
    assert _pickle33.loads(_pickle33.dumps(snap)) == snap
    # bar_ts is the bar LABEL (what signal_log stores), not wall clock.
    assert snap["bar_ts"] == str(report.primary_analysis.timestamp), \
        (snap["bar_ts"], report.primary_analysis.timestamp)
    assert snap["setup"]["strategy"] == "XAU_ATR"
@test("replay record: an indicator that did not fire is no vote, not a zero")
 def _():
    # Divergence and candlestick only exist when they fire. A zero row
    # would read as a NEUTRAL vote and drag every study's average.
    analysis = _m33_hand_analysis()
    report = _m33_duck_report(analysis)
    setup = _types33.SimpleNamespace(
        direction="BUY", strategy="XAU_ATR", setup_pattern=None,
        round_trip_cost=0.30, trail_arm_r=0.6, trail_dist_r=0.8,
        entry_price=3300.0, stop_loss=3294.0, take_profit_1=3312.0,
        take_profit_2=3320.0, risk_reward=2.0)
    snap = _rr.snapshot_report(
        report, setup, pd.Timestamp("2026-07-08 17:00", tz="UTC"),
        Timeframe.M5, [Timeframe.M15, Timeframe.H1])
    assert set(snap["tfs"][0]["indicators"]) == {
        "ema_trend", "rsi", "macd", "keltner"}, sorted(snap["tfs"][0]["indicators"])
    trade = _m29_trade()
    trade.snapshot = snap
    ind_df = _rr.build_indicators_table([trade])
    got = set(ind_df["indicator"])
    assert got == {"ema_trend", "rsi", "macd", "keltner"}, sorted(got)
    for absent in ("candlestick", "divergence", "atr"):
        assert absent not in got, f"{absent} must have NO row at all"
    # The per-bar float32 row: absent indicator / absent TF stays NaN.
    buf = _rr.new_bar_buffer()
    _rr.add_bar_row(buf, 5, pd.Timestamp("2026-07-08 16:55", tz="UTC"),
                    report, Timeframe.M5, [Timeframe.M15, Timeframe.H1])
    mat, ts_ns, pos, warn0 = _rr.finish_bar_rows(buf)
    bars = pd.DataFrame(mat, columns=list(_rr.BAR_COLS))
    cs_cols = [c for c in _rr.BAR_COLS if c.startswith("primary_candlestick_")]
    assert len(cs_cols) == 5, cs_cols
    assert bars[cs_cols].isna().all(axis=None), bars[cs_cols].iloc[0].tolist()
    conf_cols = [c for c in _rr.BAR_COLS
                 if c.startswith("c1_") or c.startswith("c2_")]
    assert bars[conf_cols].isna().all(axis=None), \
        "a missing confirmation block must be NaN, never 0"
    assert warn0 == ["a warning"] and list(pos) == [5]
    effs = [v["eff_weight"] for v in snap["tfs"][0]["indicators"].values()]
    wscores = [v["wscore"] for v in snap["tfs"][0]["indicators"].values()]
    score, sig = _rr.composite_from_rows(effs, wscores)
    assert abs(score - analysis.composite_score) < 1e-9, \
        (score, analysis.composite_score)
    assert sig == analysis.composite_signal.value
@test("replay record: Trade.snapshot defaults empty, is not shared, and pickles")
 def _():
    t1, t2 = _m29_trade(), _m29_trade()
    assert t1.snapshot == {}, t1.snapshot
    assert t2.snapshot == {}, t2.snapshot
    assert t1.snapshot is not t2.snapshot, "mutable default trap"
    t1.snapshot = {"tfs": [{"role": "primary", "indicators": {"rsi": {"wscore": 1.0}}}]}
    back = _pickle33.loads(_pickle33.dumps(t1))
    assert back.snapshot == t1.snapshot, back.snapshot
    assert t2.snapshot == {}, "a snapshot on one trade must not reach another"
@test("replay record: trades_to_dataframe ignores the snapshot field (63 columns, identical frame)")
 def _():
    # The pinned books must stay byte-identical with recording on.
    plain = _m29_trade(r_multiple=1.5)
    recorded = _m29_trade(r_multiple=1.5)
    recorded.snapshot = _m33_snapshot()
    off = _gb33.trades_to_dataframe([plain])
    on = _gb33.trades_to_dataframe([recorded])
    assert len(off.columns) == 63, len(off.columns)
    assert "snapshot" not in off.columns and "snapshot" not in on.columns
    pd.testing.assert_frame_equal(off, on)
@test("replay record: composite thresholds match the engine")
 def _():
    # One indicator, base weight 1.0, regime "unknown" (no multiplier
    # table entry -> x1.0), so composite = signal x confidence / 2.
    tech = TechnicalConfig()
    for target in (39.99, 40.0, 14.99, 15.0, -14.99, -15.0, -39.99, -40.0, 0.0):
        signal = SignalStrength.BUY if target >= 0 else SignalStrength.SELL
        result = TechnicalAnalysis(
            symbol=XAUUSD, timeframe=Timeframe.M5.value, timestamp="t",
            indicators={"rsi": IndicatorSignal(
                "RSI", 50.0, signal, abs(target) * 2.0, "", 1.0)},
            market_regime="unknown")
        TechnicalEngine(tech)._compute_composite(result)
        assert abs(result.composite_score - target) < 1e-9, \
            (target, result.composite_score)
        ind = result.indicators["rsi"]
        assert abs(ind.effective_weight - 1.0) < 1e-9, ind.effective_weight
        score, sig = _rr.composite_from_rows(
            [ind.effective_weight], [ind.weighted_score])
        assert abs(score - result.composite_score) < 1e-9, (score, target)
        assert sig == result.composite_signal.value, \
            (target, sig, result.composite_signal.value)
    # Both sides of both thresholds really were exercised.
    assert _rr.COMPOSITE_THRESHOLDS == {"strong": 40.0, "weak": 15.0}
@test("replay record: calendar join — window edges, phases, all-day, holiday, offsets")
 def _():
    entry = pd.Timestamp("2026-09-07 13:00", tz="UTC")     # Monday
    exit_ts = entry + pd.Timedelta(hours=1)
    # Window = [entry-4h, max(exit, entry+4h)] = [09:00, 17:00], inclusive.
    rows = {
        "A": _m33_cal_row(1, entry - pd.Timedelta(hours=4), "A prev release"),
        "B": _m33_cal_row(2, entry - pd.Timedelta(hours=4, seconds=60),
                          "B outside window"),
        "C": _m33_cal_row(3, entry + pd.Timedelta(minutes=30), "C in hold"),
        # medium impact: in `events` as after_exit, but never counted high.
        "D": _m33_cal_row(4, exit_ts + pd.Timedelta(hours=2), "D after exit",
                          impact="medium"),
        "E": _m33_cal_row(5, entry + pd.Timedelta(hours=25), "E far ahead"),
        "F": _m33_cal_row(6, entry + pd.Timedelta(minutes=45), "F eur low",
                          currency="EUR", impact="low"),
        # All-day US holiday: its dateline is 2026-09-08 00:30 UTC =
        # 2026-09-07 20:30 ET, so it joins on the entry's ET date.
        "G": _m33_cal_row(7, pd.Timestamp("2026-09-08 00:30", tz="UTC"),
                          "Bank Holiday", impact="holiday", masked=True,
                          label="All Day"),
        # A TIMED holiday row is not an all-day closure and must not
        # flag us_holiday for a trade entered that day.
        "H": _m33_cal_row(8, pd.Timestamp("2026-11-02 14:00", tz="UTC"),
                          "Daylight Saving Time Shift", impact="holiday"),
        "I": _m33_cal_row(9, entry + pd.Timedelta(minutes=20), "I imminent"),
    }
    trades_df = pd.DataFrame({
        "trade_id": [0, 1],
        "entry_time": pd.to_datetime(
            [entry, pd.Timestamp("2026-11-02 12:00", tz="UTC")], utc=True),
        # Trade 1 is still open: exit NaT must not raise.
        "exit_time": pd.to_datetime([exit_ts, pd.NaT], utc=True),
    })
    with tempfile.TemporaryDirectory() as d:
        full_path = os.path.join(d, "cal_full.parquet")
        pd.DataFrame([rows[k] for k in "ABCDEFGHI"]).to_parquet(
            full_path, index=False)
        # Same calendar minus every USD-high release inside 24h AFTER the
        # entry, so the "no next release" branch is reachable.
        quiet_path = os.path.join(d, "cal_quiet.parquet")
        pd.DataFrame([rows[k] for k in "ABDEFGH"]).to_parquet(
            quiet_path, index=False)
        cal = _rr.load_calendar(full_path)
        events = _rr.build_events_table(trades_df, cal)
        assert str(events["ts_utc"].dtype) == "datetime64[ns, UTC]", \
            events["ts_utc"].dtype
        t0 = events[events["trade_id"] == 0]
        phases = dict(zip(t0["name"], t0["phase"]))
        assert phases == {
            "A prev release": "before", "I imminent": "during_hold",
            "C in hold": "during_hold", "F eur low": "during_hold",
            "D after exit": "after_exit", "Bank Holiday": "all_day",
        }, phases
        assert "B outside window" not in phases, "entry-4h-60s is outside"
        assert "E far ahead" not in phases, "entry+25h is outside"
        offsets = dict(zip(t0["name"], t0["offset_min"]))
        assert offsets["A prev release"] == -240.0, offsets["A prev release"]
        assert offsets["I imminent"] == 20.0
        assert offsets["D after exit"] == 180.0
        assert np.isnan(offsets["Bank Holiday"]), "all-day rows have no offset"
        timed = dict(zip(t0["name"], t0["timed"]))
        assert timed["Bank Holiday"] is np.False_ or timed["Bank Holiday"] == False
        assert timed["A prev release"] == True
        scal = _rr.calendar_scalars(trades_df, cal)
        assert scal["ev_prev_hi_usd_min"].iloc[0] == 240.0
        assert scal["ev_prev_hi_usd_name"].iloc[0] == "A prev release"
        assert scal["ev_next_hi_usd_min"].iloc[0] == 20.0
        assert scal["ev_next_hi_usd_name"].iloc[0] == "I imminent"
        assert bool(scal["ev_entry_within_30m_hi_usd"].iloc[0]) is True
        # I and C are USD-high inside the hold; F (EUR low) and D
        # (USD medium) are not.
        assert scal["ev_hi_usd_during_hold"].iloc[0] == 2, \
            scal["ev_hi_usd_during_hold"].iloc[0]
        assert scal["ev_hi_usd_during_hold_names"].iloc[0] == \
            "I imminent|C in hold"
        assert scal["ev_any_hi_during_hold"].iloc[0] == 2
        assert bool(scal["us_holiday"].iloc[0]) is True, \
            "the all-day USD holiday shares the entry's ET date"
        # Open trade: no exit -> no hold window, and no crash.
        assert scal["ev_hi_usd_during_hold"].iloc[1] == 0
        assert bool(scal["us_holiday"].iloc[1]) is False, \
            "a TIMED holiday row must not flag us_holiday"
        # Quiet calendar: the nearest USD-high after entry is 25h out,
        # beyond NEAR_RELEASE_MAX_MIN, so both branches read empty.
        quiet = _rr.load_calendar(quiet_path)
        scal_q = _rr.calendar_scalars(trades_df, quiet)
        assert scal_q["ev_prev_hi_usd_min"].iloc[0] == 240.0
        assert np.isnan(scal_q["ev_next_hi_usd_min"].iloc[0]), \
            scal_q["ev_next_hi_usd_min"].iloc[0]
        assert scal_q["ev_next_hi_usd_name"].iloc[0] == ""
        assert bool(scal_q["ev_entry_within_30m_hi_usd"].iloc[0]) is False
        assert scal_q["ev_hi_usd_during_hold"].iloc[0] == 0
@test("replay record: write_run + load_run round-trip with a full manifest")
 def _():
    cfg, agg, _report, _setup = _m33_report()
    snap = _m33_snapshot()
    trades = [_m29_trade(r_multiple=1.5), _m29_trade(r_multiple=-1.0)]
    for t in trades:
        t.snapshot = copy.deepcopy(snap)
    book_df = _gb33.trades_to_dataframe(trades)
    assert len(book_df.columns) == 63, len(book_df.columns)
    run_info = dict(first_pos=250, last_pos=400, n_primary_bars=400,
                    n_bars_in_window=150, skipped={"SESSION_WINDOW": 10},
                    n_candidates=3, workers=1, backend="NumPy CPU fallback",
                    cache_files={"5min": "x.parquet"}, replay_seconds=0.1)
    with tempfile.TemporaryDirectory() as d:
        # NEVER read backtest/cache/calendar.parquet from a test.
        cal_path = os.path.join(d, "cal.parquet")
        pd.DataFrame([_m33_cal_row(
            1, pd.Timestamp("2026-01-05 13:30", tz="UTC"),
            "Non-Farm Employment Change")]).to_parquet(cal_path, index=False)
        rec_dir = os.path.join(d, "run")
        bt_cfg = _gb33.BacktestConfig(
            pair=XAUUSD, start=datetime(2026, 7, 1), end=datetime(2026, 7, 8),
            primary_tf=Timeframe.M5,
            confirmation_tfs=[Timeframe.M15, Timeframe.H1],
            spread=0.65, record_dir=rec_dir)
        saved_cal = _rr.CALENDAR_PARQUET
        try:
            _rr.CALENDAR_PARQUET = cal_path
            _rr.write_run(rec_dir, trades=trades, book_df=book_df,
                          bt_cfg=bt_cfg, platform_cfg=cfg, aggregator=agg,
                          run_info=run_info)
        finally:
            _rr.CALENDAR_PARQUET = saved_cal
        run = _rr.load_run(rec_dir)
        man = run["manifest"]
        assert man["engine_fingerprint"] == agg.engine_config_fingerprint()
        assert man["gate_fingerprint"] == agg.gate_config_fingerprint()
        assert man["regime_weight_table"] == regime_weight_table(cfg.technical)
        assert man["base_weights_observed"], "base weights must be recorded"
        for name, seen in man["base_weights_observed"].items():
            assert set(seen) <= {0.5, 1.0, 1.5}, (name, seen)
        assert man["calendar"]["md5"] == _hashlib33.md5(
            open(cal_path, "rb").read()).hexdigest()
        expect_md5 = _hashlib33.md5(
            book_df.to_csv(index=False).encode()).hexdigest()
        assert man["book_csv_md5"] == expect_md5, man["book_csv_md5"]
        csv_path = os.path.join(d, "book.csv")
        book_df.to_csv(csv_path, index=False)
        assert expect_md5 == _hashlib33.md5(
            open(csv_path, "rb").read()).hexdigest(), \
            "the manifest md5 must match the exported CSV byte-for-byte"
        assert man["n_trades"] == 2, man["n_trades"]
        assert man["spread"] == 0.65
        assert man["tables"]["bars"] is None, "no --record-bars in this run"
        trades_tbl = run["trades"]
        assert trades_tbl.columns[0] == "trade_id", trades_tbl.columns[0]
        cols = list(trades_tbl.columns)
        book_cols = list(book_df.columns)
        start = cols.index(book_cols[0])
        assert cols[start:start + len(book_cols)] == book_cols, \
            "the 63 book columns must stay a contiguous, ordered slice"
        assert str(run["indicators"]["signal"].dtype) == "int8", \
            run["indicators"]["signal"].dtype
        assert run["indicators"]["signal"].dtype == np.int8
        assert str(trades_tbl["entry_time"].dtype) == "datetime64[ns, UTC]", \
            trades_tbl["entry_time"].dtype
        assert isinstance(run["events"], pd.DataFrame), type(run["events"])
        assert run["bars"] is None, "bars.parquet must not exist"
@test("replay record: _replay_span with --record returns the same candidates plus snapshots")
 def _():
    from pair_signals.xauusd import make_config as _mk33
    cfg = _mk33(Timeframe.M5, [Timeframe.M15, Timeframe.H1])
    dfs_full = _m33_span_frames()
    caches = {tf.value: _gb33.precompute_indicators(df, cfg.technical)
              for tf, df in dfs_full.items()}
    agg = SignalAggregator(cfg)
    agg.tech_engine = _gb33.GPUTechnicalEngine(cfg.technical, caches)
    bt_off = _gb33.BacktestConfig(
        pair=XAUUSD, start=datetime(2026, 7, 6), end=datetime(2026, 7, 9),
        primary_tf=Timeframe.M5,
        confirmation_tfs=[Timeframe.M15, Timeframe.H1])
    bt_on = _dc33.replace(bt_off, record_dir="unused", record_bars=True)
    span = (300, 600)
    try:
        _gb33._REPLAY_CTX.update(
            primary_df=dfs_full[Timeframe.M5], dfs_full=dfs_full,
            timeframes=[Timeframe.M5, Timeframe.M15, Timeframe.H1],
            aggregator=agg, platform_cfg=cfg, bt_cfg=bt_off,
            primary_dur=pd.Timedelta(minutes=5))
        skip_off, cands_off, bars_off = _gb33._replay_span(span)
        _gb33._REPLAY_CTX["bt_cfg"] = bt_on
        skip_on, cands_on, bars_on = _gb33._replay_span(span)
    finally:
        _gb33._REPLAY_CTX.clear()
    assert len(cands_off) > 0, "the fixture must produce candidates"
    assert skip_off == skip_on, (skip_off, skip_on)
    assert [p for p, _ in cands_off] == [p for p, _ in cands_on], \
        "recording must not move a single entry"
    for (_p_off, t_off), (_p_on, t_on) in zip(cands_off, cands_on):
        d_off = _dc33.asdict(t_off)
        d_on = _dc33.asdict(t_on)
        d_off.pop("snapshot")
        snap = d_on.pop("snapshot")
        differing = [k for k in d_off if d_off[k] != d_on[k]]
        assert not differing, differing
        assert t_off.snapshot == {}, "recording off must leave the field empty"
        assert snap and len(snap["tfs"]) == 3, \
            "a recorded trade carries primary + both confirmation blocks"
    assert bars_off is None, "no bar chunk without --record-bars"
    mat, ts_ns, pos, warn0 = bars_on
    assert mat.dtype == np.float32, mat.dtype
    assert mat.shape[1] == len(_rr.BAR_COLS), mat.shape
    n_analyzed = (span[1] - span[0]) - sum(skip_off.values())
    assert mat.shape[0] == n_analyzed, (mat.shape[0], n_analyzed)
    assert len(cands_on) <= mat.shape[0] <= (span[1] - span[0])
    assert len(ts_ns) == len(pos) == len(warn0) == mat.shape[0]
@test("replay record: forward moves on a hand-built series")
 def _():
    close = np.array([10.0, 11.0, 13.0, 12.0, 15.0, 14.0])
    high, low = close + 1.0, close - 1.0
    atr = np.array([1.0, 1.0, 1.0, 1.0, 1.0, 0.0])
    mv = _rr.forward_moves(close, high, low, atr, 2)
    assert mv["fwd"][0] == 3.0, mv["fwd"][0]
    assert mv["mfe"][0] == 4.0, mv["mfe"][0]     # max(11,13)+1 - 10
    assert mv["mae"][0] == 0.0, mv["mae"][0]     # 10 - (min(11,13)-1)
    assert mv["fwd"][3] == 2.0, mv["fwd"][3]
    for i in (4, 5):
        for key in ("fwd", "mfe", "mae"):
            assert np.isnan(mv[key][i]), (key, i, mv[key][i])
    # A zero ATR blanks its own row even with a full horizon ahead.
    mv2 = _rr.forward_moves(close, high, low,
                            np.array([1.0, 0.0, 1.0, 1.0, 1.0, 1.0]), 2)
    assert np.isnan(mv2["fwd"][1]) and np.isnan(mv2["mfe"][1]) \
        and np.isnan(mv2["mae"][1]), mv2["fwd"][1]
    assert mv2["fwd"][0] == 3.0 and mv2["fwd"][2] == 2.0
@test("replay record: --record/--record-bars parse, default None/False, bars need record")
 def _():
    args = _gb33._parse_args(["--pair", "XAUUSD"])
    assert args.record is None, args.record
    assert args.record_bars is False, args.record_bars
    assert _gb33._parse_args(
        ["--pair", "XAUUSD", "--record", "/tmp/x"]).record == "/tmp/x"
    try:
        with open(os.devnull, "w") as _null, _ctx33.redirect_stderr(_null):
            _gb33._parse_args(["--pair", "XAUUSD", "--record-bars"])
        raise AssertionError("--record-bars without --record must exit")
    except SystemExit:
        pass
    bare = _gb33.BacktestConfig(pair=XAUUSD, start=datetime(2026, 7, 1),
                                end=datetime(2026, 7, 8))
    assert bare.record_dir is None and bare.record_bars is False
@test("replay record: replay_record imports none of the live writers or the backtester")
 def _():
    # The recorder runs inside replay workers and must stay a leaf: no
    # live writer, no news, no provider, and no import cycle back into
    # the backtester.
    banned = {"signal_log", "news_engine", "providers", "telegram_bot",
              "news_archive", "agent_panel", "violent_watch",
              "market_outlook", "gpu_backtest", "sklearn", "scipy",
              "lightgbm", "torch"}
    path = Path("backtest/replay_record.py")
    tree = ast.parse(path.read_text(encoding="utf-8"), str(path))
    seen, offenders = set(), []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            names = [a.name for a in node.names]
        elif isinstance(node, ast.ImportFrom):
            names = [node.module or ""]
        else:
            continue
        for name in names:
            root = name.split(".")[0]
            seen.add(root)
            if root in banned:
                offenders.append(f"{path}:{node.lineno} {name}")
    assert not offenders, f"replay_record must stay a leaf module: {offenders}"
    # ...and the guard is not vacuous — it really does import these two.
    assert "engine" in seen and "config" in seen, sorted(seen)
@test("replay record: the backtester still satisfies its source guards with the recorder wired in")
 def _():
    src = open(_gb33.__file__).read()
    for banned in ("import signal_log", "scan_outcomes", "label_scan_outcomes",
                   "update_watch", "log_watch_update", "NewsEngine(",
                   "news_archive", "import violent_watch", "from violent_watch"):
        assert banned not in src, f"backtests must stay write-free: {banned}"
    for required in ("use_news=False",
                     "_apply_regime_filter_live(report, slice_dfs)",
                     "from backtest import replay_record", "snapshot=snap"):
        assert required in src, f"missing: {required}"
    assert "_compute_composite" not in _gb33.GPUTechnicalEngine.__dict__, \
        "the GPU engine must inherit the live composite, never shadow it"
@test("replay record follow-up: M15 factory halves RSI in every regime, other TFs carry no override")
 def _():
    # M15 override 5 (validated 2026-09-09): the M15 factory ships a
    # regime_weights_override that is the shipped table with the RSI
    # multiplier halved everywhere and every other cell untouched.
    import engine as _eng
    from pair_signals import PAIR_PRODUCERS
    default = _eng.regime_weight_table(TechnicalConfig())
    m15 = PAIR_PRODUCERS[XAUUSD](Timeframe.M15, [Timeframe.M5, Timeframe.H1])
    override = m15.technical.regime_weights_override
    assert set(override) == set(default), \
        f"M15 override must cover exactly the shipped regimes: {sorted(override)}"
    for regime, weights in default.items():
        got = override[regime]
        assert set(got) == set(weights), \
            f"{regime}: indicator set changed: {sorted(got)}"
        for name, mult in weights.items():
            want = mult * 0.5 if name == "rsi" else mult
            assert got[name] == want, \
                f"{regime}.{name}: {got[name]} != {want} (default {mult})"
    # The engine really runs with it (regime_weight_table prefers the override).
    assert _eng.regime_weight_table(m15.technical) == override
    # ...and it is M15-only — same per-TF pattern as every other override.
    for tf, ctfs in [(Timeframe.M5, [Timeframe.M15, Timeframe.H1]),
                     (Timeframe.H1, [Timeframe.M15, Timeframe.H4]),
                     (Timeframe.H4, [Timeframe.H1, Timeframe.D1])]:
        cfg = PAIR_PRODUCERS[XAUUSD](tf, ctfs)
        assert cfg.technical.regime_weights_override == {}, \
            f"{tf} must carry no regime weight override"
 # ══════════════════════════════════════════════════════════════════════
 #  SUMMARY
 # ══════════════════════════════════════════════════════════════════════