diff --git a/.gitignore b/.gitignore index b2750c0..984c108 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,4 @@ backtest/__pycache__/ .env.* logs/ backtest/research/sizing_2026-09/ +backtest/research/replay_2026-09/ diff --git a/CLAUDE.md b/CLAUDE.md index d9007dd..60509e3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,7 +12,7 @@ Every numeric threshold in `config.py` and `pair_signals/xauusd.py` was backtest ```bash python test_all.py # QA suite, no network (mocked); prints its own count - # (483 tests / 32 modules as of 2026-09-08) + # (498 tests / 33 modules as of 2026-09-09) python main.py --pair XAUUSD --timeframe H1 --detail # CLI signal report (--json for full dump, # --panel for the advisory LLM agent panel, --outlook for the # hourly AI market brief; the violent-move watch line appears @@ -33,6 +33,10 @@ python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-0 # --export-features (append the 65 decision-time f_* columns; default export stays 63 cols) # --fixed-rr-tp N (research-only: hard TP at N×risk, trail off; verdict # 2026-08-05: NOT shipped — backtest/research/fixedtp_2026-08-05.md) +# --record DIR [--record-bars] (research: manifest + trades/indicators/events parquet for +# every accepted trade, and one row per analyzed bar; the book is unchanged — pair every +# recorded run with a control run and show the MD5s equal; read with +# backtest.replay_record.load_run, first look with backtest/replay_scorecard.py DIR --check) ``` **`test_all.py` is NOT pytest** — it uses a custom `@test("name")` decorator that runs each test immediately at import; don't grep for `def test_` or try `pytest test_all.py`. Run the whole file; it prints per-module results and exits non-zero on failure. There is no way to run a single test short of commenting modules out. Tests must stay green; every new gate/knob needs tests. Signal-log fixtures: use a WEEKDAY signal bar when a test exercises candle continuity — the outcome model treats weekend and 21:00 UTC gaps as scheduled, so Saturday-dated bars never produce a gap. `_candles_json` stores `df.tail(14)`, so a test snapshot longer than 14 bars silently drops its head. When patching weekend logic in tests, patch `providers.base._is_forex_weekend` — but know there are THREE independent signal-path weekend implementations (`providers/base.py`, `signal_aggregator.py`, and an inline copy in `backtest/gpu_backtest.py:_is_xau_blocked_by_calendar`) plus the advisory-only continuity copy in `advisory_outcome.py`; patching one does not reach the others. @@ -108,3 +112,4 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz - 2026-09-08 delivery-side fixes from the 3-day signal-log + telegram-log review (nothing in the signal core changed; 468 tests): (1) `_drop_stale_spread_gate_vetoes` (after every setup archetype, so the XAU_PB path is covered too) moves a core generator's stale `SPREAD GATE:` veto to `cleaned_warnings` when a later generator ships a setup — the 3-yr H1 and M15 books were regenerated before/after and are identical on every trade column (H1 253/+70.02R/PF 1.98, M15 1092/+151.90R/PF 1.38); only the exported `warnings` text differs, on 50/253 H1 and 358/1092 M15 trades, i.e. a fifth to a third of all A-grade alerts had been carrying "costs would eat the edge" beside a valid target; (2) Telegram alerts now require a trade setup (`_is_alert_candidate`, scan + breach paths; no-setup candidates log `alert skipped: … (gate reason)`); (3) every delivered alert / breach alert / violent-move heads-up logs an `… sent:` INFO line with the signal-log row id (the review had to reconstruct sends from panel rows); (4) `telegram.log` rotates daily × 30 instead of 3-hourly × 240 (old `_HH` files on the live box need one manual delete). Review facts (not evidence — three days): 8 setups, 1 right; strong readings (|score| ≥ 20) were followed by reversals in 16 of 18 episodes; hot watch level 0/25 bursts on M5 (calendar trigger never fired, 15 of 33 hot stamps were the 12–14 UTC slot on Labor Day — no holiday awareness anywhere); Oanda published no XAU bars 18:30–22:00 UTC on Labor Day and the bot scanned the stale bar 43×; the retry ladder recovered all 25 post-deploy 401 bursts (41 of 58 incidents sit on the :00 boundary); the live dashboard publisher ran only 01:00–02:20 UTC Sep 7 and is dead again. - 2026-09-06 stale-code cleanup (behaviour-neutral; suite green, M5/H1 12-month books byte-identical after every signal-side commit): `main.py --timeframe` now accepts M5/M15/H1/H4 and takes its confirmation TFs from the pair factory (the CLI's private table had paired M5 with M1+M15); `PlatformConfig.pip_value` is gone (`PAIR_INFO` is the source); removed as dead or superseded: `_get_affected_pairs`/`event_impact_on_pairs`, `SignalLogger.advance_with_bars`/`path`/`news_for`, `reset_llm_client`, the Telegram scan-summary/keyboard helpers and duplicate confirmation map, the 2026-07 Oanda-vs-TD research pager+study, `burst_profile.py`, `hour_analysis.py`, `scripts/compare_oanda_tiingo.py`, the never-implemented LLM-provider plan and the merged config.json spec. Second pass the same day (owner directive: keep only what the system needs): every research script and data file under `backtest/research/` (book_report, book_diff, the burst tools, sizing_model, commit_ladder, llm_news_eval + golden CSV, ff_calendar_scraper, calendar_impact, watch_rule_eval, the Dukascopy spread study + CSVs, the panel dry-test JSONL), the two completed study scripts in `scripts/`, the architecture diagram + generator and the data-source test plan were deleted; the campaign `.md` reports stay as the decision ledger. Git history before `953c5ad` has all of it. Third pass (owner: "remove dead functions as well"): the write-only pb/cp rejection counters, the unreachable `_TP_QUALITY_SCALE` B/C entries, the engine's unused DataFrame argument and M1 arm, the providers' generic-symbol guesses (unknown symbols now raise), the sub-$10 price format, the `=X` pair parsing and the unread `minutes_until`/`is_high_impact` news.json keys are gone. - 2026-09-08 news archive shipped (`news_archive.py`, 483 tests; nothing in the signal core changed, both news gating flags stay `False`, books untouched). Motivation: headlines only ever survived as a side effect of signal logging and `signals.db` was rotated around 2026-09-04, destroying the 3-week / 4,054-headline corpus the news study was written on — with `logs/` gitignored, git could not recover it and nobody noticed for four days. Day-one corpus after backfilling both DBs on disk: **1,016 distinct headlines, 846 of them DeepSeek-labelled**, growing ~200/day (~1,400/week). The point of collecting it: distil DeepSeek's headline scoring (78/78 on the 2026-08-06 golden set) into a numpy-only local model so the fallback stops being the keyword scorer (**24/78 = 31%**, which is what silently runs on no key / disabled / daily cap / timeout / HTTP or parse failure / batch overflow). **Do not train yet** — 800-odd labels gives a holdout too small to grade (SE ±0.032); the pre-registered plan is ~4,000 deduplicated headlines over ≥3 weeks, time-split with near-duplicates clustered first (the same story is re-titled by several outlets, so a random split scores memorisation), headline text ONLY (the teacher never sees `description` — `_llm_headline_scores` sends titles), graded against the golden set recoverable with `git show 1cebebf^:backtest/research/llm_news_golden.csv`. When it ships it slots in as DeepSeek → model → keyword and stays display-only. Training happens offline in `venv/`; a Module 32 test now AST-walks the root, `providers/` and `pair_signals/` asserting no `sklearn`/`scipy`/`lightgbm`/`torch` import and that none of them reach `requirements.txt` — that guard did not exist before. +- 2026-09-09 replay recording shipped (`backtest/replay_record.py`, `backtest/replay_scorecard.py`, Module 33, 498 tests; nothing in the signal core changed, books byte-identical — `backtest/research/replay_recording_2026-09-09.md`). `--record DIR` stores every ACCEPTED trade's decision evidence in Parquet: each indicator's vote / value / confidence / BASE weight / regime multiplier / EFFECTIVE weight / contribution on the primary AND both confirmation TFs (`indicators.parquet`, long; absence = the indicator did not fire, never a zero row), per-TF regime/composite, penalty tags, setup, outcome, and the Forex Factory calendar events from 4 h before entry to exit (`events.parquet`, all-day rows matched on the New-York date; scalar `ev_*`/`us_holiday` columns on `trades.parquet`); `--record-bars` adds one float32 row per ANALYZED bar with the same votes plus 1 h/4 h/24 h forward moves in ATRs (`bars.parquet`). The manifest carries the engine + gate fingerprints, the regime weight table in force and the book MD5. Calendar source is the gitignored `backtest/cache/calendar.parquet` (15,124 FF events 2023-09-01→2026-09-10, unix-UTC `dateline`; `timeLabel` is display-only); the scraper that built it is back at `scripts/ff_calendar_scraper.py` (resume + idempotent merge, ~1.6 s/request). The 2-year books (2024-09-05→2026-09-05, $0.65) live in `backtest/research/replay_2026-09/{m5,m15}/`: M5 781 / +146.07R / PF 1.52 (halves +28.0/+118.1), M15 793 / +117.62R / PF 1.41 (halves +38.2/+79.4); control-vs-recorded MD5 equal on both. First-look facts are descriptive only (Bollinger opposes 94% of M5 entries; divergence-opposed M5 trades did best again; every indicator's SELL votes were followed by moves against them on both TFs; the bar-level baseline drifts +1.9 ATR/24 h on M5 so BUY votes must be drift-adjusted first) — every candidate change stays a rule-1 rerun with a control book. diff --git a/backtest/gpu_backtest.py b/backtest/gpu_backtest.py index 1aac28d..3922895 100644 --- a/backtest/gpu_backtest.py +++ b/backtest/gpu_backtest.py @@ -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 " diff --git a/backtest/instructions_for_backtest.txt b/backtest/instructions_for_backtest.txt index eb63be6..04a333d 100644 --- a/backtest/instructions_for_backtest.txt +++ b/backtest/instructions_for_backtest.txt @@ -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. diff --git a/backtest/replay_record.py b/backtest/replay_record.py new file mode 100644 index 0000000..fd7b120 --- /dev/null +++ b/backtest/replay_record.py @@ -0,0 +1,1103 @@ +""" +ForexBot2 — Backtest decision recorder (RESEARCH, opt-in) +========================================================= +Turns a `--record DIR` replay into a small parquet dataset describing WHY +every accepted trade was taken: the composite maths of the primary AND +both confirmation timeframes, every indicator's value / signal / base +weight / regime multiplier / effective weight / weighted score, the +penalty tags, the setup, and the economic-calendar events around the +entry. With `--record-bars` it also stores one compact float32 row per +ANALYZED primary bar so a study can compare taken trades against the bars +that produced nothing. + +Hard contract +------------- +* **Read-only w.r.t. the signal path.** Nothing here is imported by + `signal_aggregator`, `engine` or the live surfaces, and nothing here + writes back onto a report/setup. The trade book (and therefore + `--export-csv`) is byte-identical with recording on or off. +* **Plain data only** crosses the worker boundary: `snapshot_report` + returns str/int/float/bool/None/list/dict, so pickling a `Trade` back + from a replay worker costs no enums, timestamps or dataclasses. +* Allowed imports are stdlib + numpy/pandas + `config.Timeframe` + + `engine.regime_weight_table`. Never signal_log / news_engine / + providers / gpu_backtest (circular) / sklearn. + +The composite arithmetic mirrored below is `engine.TechnicalEngine. +_compute_composite`; `composite_from_rows` exists so a study can +reproduce a composite from the stored rows and prove nothing was lost. +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +import logging +import math +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, Iterable, List, Optional, Tuple +from zoneinfo import ZoneInfo + +import numpy as np +import pandas as pd + +# Ensure project root on sys.path so this module works standalone, as +# `backtest.replay_record`, and when `backtest/` itself is on sys.path. +PROJECT_ROOT = Path(__file__).resolve().parent.parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from config import Timeframe # noqa: E402 +from engine import regime_weight_table # noqa: E402 + +logger = logging.getLogger(__name__) + + +# ═════════════════════════════════════════════════════════════════════ +# CONSTANTS +# ═════════════════════════════════════════════════════════════════════ + +INDICATORS = ( + "ema_trend", "rsi", "macd", "bollinger", "adx", + "roc", "keltner", "divergence", "candlestick", +) +ROLES = ("primary", "c1", "c2") + +# Mirrors engine.TechnicalEngine._compute_composite (engine.py ~1117): +# score = clamp((Σ wscore / Σ eff_weight) * 50, -100, 100) +# >= 40 STRONG_BUY | >= 15 BUY | <= -40 STRONG_SELL | <= -15 SELL | else NEUTRAL +COMPOSITE_THRESHOLDS = {"strong": 40.0, "weak": 15.0} + +CALENDAR_PARQUET = Path(__file__).resolve().parent / "cache" / "calendar.parquet" +EVENT_WINDOW = pd.Timedelta(hours=4) +NEAR_RELEASE_MAX_MIN = 1440.0 +_NY = ZoneInfo("America/New_York") + +REGIME_CODES = { + "unknown": 0, "trending_up": 1, "trending_down": 2, + "trending": 3, "transitioning": 4, "ranging": 5, +} +# engine._compute_atr emits exactly these four. +VOL_CODES = {"low": 0, "normal": 1, "high": 2, "extreme": 3} +QUALITY_CODES = {"": -1, "F": 0, "D": 1, "C": 2, "B": 3, "A": 4, "A+": 5} +# "XAU" is the backtester's book tag for the ATR archetype; the live +# TradeSetup.strategy for the same generator is "XAU_ATR" — both map to 2 +# so `strategy_code` never collides with 0 ("no setup"). An unrecognised +# strategy stays NaN rather than masquerading as no-setup. +STRATEGY_CODES = {"": 0, "GENERIC": 1, "XAU": 2, "XAU_ATR": 2, + "XAU_PB": 3, "XAU_CP": 4} + + +def _build_bar_cols() -> Tuple[str, ...]: + """The float32 vector layout of one recorded bar (10 + 15 + 135 = 160).""" + cols: List[str] = [ + "close", "final_score", "final_conf", "quality_code", "strategy_code", + "has_setup", "final_signal", "mtf_agreement", "penalty_total", + "exhaustion_count", + ] + for role in ROLES: + cols += [f"{role}_regime_code", f"{role}_vol_code", f"{role}_composite", + f"{role}_confidence", f"{role}_atr"] + for role in ROLES: + for ind in INDICATORS: + cols += [f"{role}_{ind}_signal", f"{role}_{ind}_value", + f"{role}_{ind}_conf", f"{role}_{ind}_eff_weight", + f"{role}_{ind}_wscore"] + return tuple(cols) + + +BAR_COLS = _build_bar_cols() +_BAR_IDX = {name: i for i, name in enumerate(BAR_COLS)} +_N_BAR_COLS = len(BAR_COLS) + +# Every column below is normalised to datetime64[ns, UTC] before writing. +_DT_COLS = ("bar_ts", "decision_time", "entry_time", "exit_time", "ts_utc") + +_EVENT_COLS = [ + "trade_id", "event_id", "ts_utc", "name", "currency", "impact", "timed", + "actual", "forecast", "previous", "revision", "better_worse", + "offset_min", "phase", +] +_IND_COLS = [ + "trade_id", "tf_role", "timeframe", "indicator", "signal", "signal_name", + "value", "confidence", "base_weight", "regime_mult", "eff_weight", + "wscore", "agrees", "regime", "description", +] +_SCALAR_COLS = [ + "ev_prev_hi_usd_min", "ev_prev_hi_usd_name", + "ev_next_hi_usd_min", "ev_next_hi_usd_name", + "ev_hi_usd_during_hold", "ev_hi_usd_during_hold_names", + "ev_any_hi_during_hold", "ev_entry_within_30m_hi_usd", "us_holiday", +] + + +# ═════════════════════════════════════════════════════════════════════ +# SMALL HELPERS +# ═════════════════════════════════════════════════════════════════════ + +def git_sha() -> str: + """Short HEAD sha of the repo, or "" when git is unavailable.""" + try: + out = subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=2, + ) + return out.stdout.strip() if out.returncode == 0 else "" + except Exception: # pragma: no cover + return "" + + +def _finite(x) -> Optional[float]: + """float(x) when it is a real finite number, else None.""" + if x is None: + return None + try: + v = float(x) + except (TypeError, ValueError): + return None + if math.isnan(v) or math.isinf(v): + return None + return v + + +def _num(x) -> float: + """_finite with NaN instead of None (for float32 buffers).""" + v = _finite(x) + return np.nan if v is None else v + + +def _agrees(signal_int: int, direction: str) -> str: + """Same rule as gpu_backtest._build_indicator_breakdown.""" + is_buy = direction == "BUY" + if signal_int > 0: # BUY / STRONG_BUY + return "agree" if is_buy else "oppose" + if signal_int < 0: # SELL / STRONG_SELL + return "oppose" if is_buy else "agree" + return "neutral" + + +def _sig_int(signal) -> int: + return int(getattr(signal, "value", 0) or 0) + + +def _sanitize(name: str) -> str: + return "".join(ch if (ch.isalnum() or ch == "_") else "_" for ch in str(name)) + + +# ═════════════════════════════════════════════════════════════════════ +# DECISION SNAPSHOT (runs inside the replay worker) +# ═════════════════════════════════════════════════════════════════════ + +def _tf_block(analysis, role: str, direction: str) -> Dict: + """One timeframe's plain-data block (scored indicators only). + + Indicators are skipped under exactly the composite's rule — private + `_`-prefixed series and weight-0 context indicators (that is what + excludes `atr`) — so Σ eff_weight / Σ wscore over the stored rows + reproduces the composite. + """ + inds: Dict[str, Dict] = {} + for name, ind in (getattr(analysis, "indicators", None) or {}).items(): + if name.startswith("_") or ind.weight == 0: + continue + sig_int = _sig_int(ind.signal) + base = _finite(ind.weight) + eff = _finite(getattr(ind, "effective_weight", None)) + regime_mult = (eff / base) if (eff is not None and base) and base > 0 else None + inds[name] = { + "signal": sig_int, + "signal_name": str(getattr(ind.signal, "name", "")), + "value": _finite(ind.value), + "confidence": _finite(ind.confidence), + "base_weight": base, + "regime_mult": regime_mult, + "eff_weight": eff, + "wscore": _finite(getattr(ind, "weighted_score", None)), + "agrees": _agrees(sig_int, direction), + "description": str(ind.description or ""), + } + levels = getattr(analysis, "key_levels", None) or {} + return { + "tf": str(getattr(analysis, "timeframe", "") or ""), + "role": role, + "regime": str(getattr(analysis, "market_regime", "") or ""), + "vol": str(getattr(analysis, "volatility_regime", "") or ""), + "composite_score": _finite(getattr(analysis, "composite_score", None)), + "composite_signal": _sig_int(getattr(analysis, "composite_signal", None)), + "confidence": _finite(getattr(analysis, "confidence", None)), + "atr": _finite(levels.get("atr")), + "atr_pct": _finite(levels.get("atr_pct")), + "pattern_signals": [str(p) for p in (getattr(analysis, "pattern_signals", None) or [])], + "indicators": inds, + } + + +def _role_analyses(report, confirmation_tfs) -> List[Tuple[str, object]]: + """[(role, analysis)] — primary first, then c1/c2 by confirmation order.""" + out: List[Tuple[str, object]] = [] + primary = getattr(report, "primary_analysis", None) + if primary is not None: + out.append(("primary", primary)) + order = [tf.value for tf in (confirmation_tfs or [])] + for analysis in (getattr(report, "confirmation_analyses", None) or []): + tf_name = str(getattr(analysis, "timeframe", "") or "") + try: + i = order.index(tf_name) + except ValueError: + logger.debug("record: confirmation analysis %s is not in %s — skipped", + tf_name, order) + continue + if i + 1 >= len(ROLES): + logger.debug("record: more confirmation TFs than roles (%s)", tf_name) + continue + out.append((ROLES[i + 1], analysis)) + return out + + +def snapshot_report(report, setup, decision_time, + primary_tf, confirmation_tfs) -> Dict: + """Plain-data decision snapshot; `json.dumps` must succeed on it. + + Called inside the replay worker for every accepted candidate, so it + must stay allocation-light and must not touch the report. + """ + direction = str(getattr(setup, "direction", "") or "") + primary = getattr(report, "primary_analysis", None) + # report.timestamp is wall-clock in analyze_from_dataframes; the + # primary analysis carries the bar LABEL (same string signal_log + # stores as bar_ts / indicators_json). + bar_ts = str(getattr(primary, "timestamp", "") or "") or str(report.timestamp) + + try: + decision_iso = pd.Timestamp(decision_time).isoformat() + except Exception: # pragma: no cover + decision_iso = str(decision_time) + + return { + "bar_ts": bar_ts, + "decision_time": decision_iso, + "final_signal": _sig_int(report.final_signal), + "final_score": _finite(report.final_score), + "final_confidence": _finite(report.final_confidence), + "quality": str(report.signal_quality or ""), + "mtf_agreement": bool(report.mtf_agreement), + "mtf_signals": {str(k): _sig_int(v) + for k, v in (report.mtf_signals or {}).items()}, + "mtf_scores": {str(k): _finite(v) + for k, v in (report.mtf_scores or {}).items()}, + "penalty_tags": {str(k): _finite(v) + for k, v in (report.penalty_tags or {}).items()}, + "penalty_total": _finite(report.penalty_total), + "momentum_fading": bool(report.momentum_fading), + "exhaustion_count": int(report.exhaustion_count or 0), + "spread_estimate": _finite(report.spread_estimate), + "current_price": _finite(report.current_price), + "warnings": [str(w) for w in (report.warnings or [])], + "cleaned_warnings": [str(w) for w in (report.cleaned_warnings or [])], + "bullish_reasons": [str(r) for r in (report.bullish_reasons or [])], + "bearish_reasons": [str(r) for r in (report.bearish_reasons or [])], + "setup": { + "strategy": str(getattr(setup, "strategy", "") or ""), + "setup_pattern": (str(setup.setup_pattern) + if getattr(setup, "setup_pattern", None) else None), + "round_trip_cost": _finite(getattr(setup, "round_trip_cost", None)), + "trail_arm_r": _finite(getattr(setup, "trail_arm_r", None)), + "trail_dist_r": _finite(getattr(setup, "trail_dist_r", None)), + "entry_price": _finite(getattr(setup, "entry_price", None)), + "stop_loss": _finite(getattr(setup, "stop_loss", None)), + "take_profit_1": _finite(getattr(setup, "take_profit_1", None)), + "take_profit_2": _finite(getattr(setup, "take_profit_2", None)), + "risk_reward": _finite(getattr(setup, "risk_reward", None)), + }, + "tfs": [_tf_block(analysis, role, direction) + for role, analysis in _role_analyses(report, confirmation_tfs)], + } + + +def composite_from_rows(eff_weights: Iterable, wscores: Iterable) -> Tuple[float, int]: + """Reproduce (composite score, signal int) from stored indicator rows.""" + ew = np.asarray(list(eff_weights), dtype=float) + ws = np.asarray(list(wscores), dtype=float) + total_w = float(np.nansum(ew)) if ew.size else 0.0 + total_s = float(np.nansum(ws)) if ws.size else 0.0 + if total_w == 0: + return 0.0, 0 + score = max(-100.0, min(100.0, (total_s / total_w) * 50.0)) + strong = COMPOSITE_THRESHOLDS["strong"] + weak = COMPOSITE_THRESHOLDS["weak"] + if score >= strong: + sig = 2 + elif score >= weak: + sig = 1 + elif score <= -strong: + sig = -2 + elif score <= -weak: + sig = -1 + else: + sig = 0 + return score, sig + + +# ═════════════════════════════════════════════════════════════════════ +# TRADE + INDICATOR TABLES +# ═════════════════════════════════════════════════════════════════════ + +def _blocks_by_role(snap: Dict) -> Dict[str, Dict]: + return {str(b.get("role", "")): b for b in (snap.get("tfs") or [])} + + +def build_trades_table(trades, book_df: pd.DataFrame, + primary_tf: Timeframe) -> pd.DataFrame: + """One row per accepted trade: the 63 book columns + decision context. + + `book_df` is `gpu_backtest.trades_to_dataframe(trades)` (passed in so + this module never imports the backtester) and is copied verbatim, in + order, so a recorded run is a strict superset of the pinned book. + """ + trades = list(trades or []) + book = (book_df.copy().reset_index(drop=True) + if book_df is not None else pd.DataFrame()) + book_cols = list(book.columns) + + warned_empty = False + pen_tags: List[str] = [] + seen_tags = set() + for t in trades: + for tag in (getattr(t, "snapshot", None) or {}).get("penalty_tags", {}) or {}: + if tag not in seen_tags: + seen_tags.add(tag) + pen_tags.append(str(tag)) + + rows: List[Dict] = [] + bar_ts_raw: List[Optional[str]] = [] + dec_ts_raw: List[Optional[str]] = [] + + for i, t in enumerate(trades): + snap = getattr(t, "snapshot", None) or {} + if not snap and not warned_empty: + warned_empty = True + logger.warning( + "--record: trade snapshots are empty — decision context " + "columns will be NaN/'' (was the run recorded?)") + blocks = _blocks_by_role(snap) + setup = snap.get("setup") or {} + bar_ts_raw.append(snap.get("bar_ts") or None) + dec_ts_raw.append(snap.get("decision_time") or None) + + row: Dict = { + "tp2": _finite(getattr(t, "tp2", None)), + "risk_pips": _finite(getattr(t, "risk_pips", None)), + "reward_pips": _finite(getattr(t, "reward_pips", None)), + "pip_value": _finite(getattr(t, "pip_value", None)), + "stake_multiplier": _finite(getattr(t, "stake_multiplier", None)), + } + for role in ROLES: + b = blocks.get(role) + if b is None: + row.update({ + f"{role}_tf": "", f"{role}_regime": "", f"{role}_vol": "", + f"{role}_composite": np.nan, f"{role}_composite_signal": pd.NA, + f"{role}_confidence": np.nan, f"{role}_atr": np.nan, + f"{role}_n_voting": pd.NA, + f"{role}_sum_eff_weight": np.nan, + f"{role}_sum_wscore": np.nan, + }) + continue + inds = b.get("indicators") or {} + effs = [v.get("eff_weight") for v in inds.values()] + wss = [v.get("wscore") for v in inds.values()] + row.update({ + f"{role}_tf": str(b.get("tf") or ""), + f"{role}_regime": str(b.get("regime") or ""), + f"{role}_vol": str(b.get("vol") or ""), + f"{role}_composite": _num(b.get("composite_score")), + f"{role}_composite_signal": int(b.get("composite_signal") or 0), + f"{role}_confidence": _num(b.get("confidence")), + f"{role}_atr": _num(b.get("atr")), + f"{role}_n_voting": len(inds), + f"{role}_sum_eff_weight": float( + np.nansum([_num(x) for x in effs])) if effs else np.nan, + f"{role}_sum_wscore": float( + np.nansum([_num(x) for x in wss])) if wss else np.nan, + }) + + row["mtf_agreement"] = bool(snap.get("mtf_agreement", False)) + row["final_signal"] = int(snap.get("final_signal") or 0) + row["penalty_total"] = _num(snap.get("penalty_total")) + tags = snap.get("penalty_tags") or {} + for tag in pen_tags: + row[f"pen_{_sanitize(tag)}"] = _num(tags.get(tag)) if tag in tags else np.nan + row["exhaustion_count"] = int(snap.get("exhaustion_count") or 0) + row["momentum_fading"] = bool(snap.get("momentum_fading", False)) + + row["setup_strategy"] = str(setup.get("strategy") or "") + row["setup_pattern"] = str(setup.get("setup_pattern") or "") + row["round_trip_cost"] = _num(setup.get("round_trip_cost")) + row["setup_trail_arm_r"] = _num(setup.get("trail_arm_r")) + row["setup_trail_dist_r"] = _num(setup.get("trail_dist_r")) + + warns = snap.get("warnings") or [] + row["warnings_all"] = " || ".join(str(w) for w in warns) + row["cleaned_warnings"] = " || ".join( + str(w) for w in (snap.get("cleaned_warnings") or [])) + row["n_warnings"] = len(warns) + row["bullish_reasons_all"] = " || ".join( + str(r) for r in (snap.get("bullish_reasons") or [])) + row["bearish_reasons_all"] = " || ".join( + str(r) for r in (snap.get("bearish_reasons") or [])) + row["spread_estimate"] = _num(snap.get("spread_estimate")) + row["current_price"] = _num(snap.get("current_price")) + rows.append(row) + + extra_cols = _trade_extra_columns(pen_tags) + extras = pd.DataFrame(rows, columns=extra_cols) if rows else \ + pd.DataFrame({c: pd.Series(dtype="object") for c in extra_cols}) + + head = pd.DataFrame({ + "trade_id": pd.Series(np.arange(len(trades)), dtype="int32"), + "bar_ts": pd.to_datetime(pd.Series(bar_ts_raw, dtype="object"), + utc=True, errors="coerce"), + "decision_time": pd.to_datetime(pd.Series(dec_ts_raw, dtype="object"), + utc=True, errors="coerce"), + }) + if not rows: + head = pd.DataFrame({ + "trade_id": pd.Series(dtype="int32"), + "bar_ts": pd.Series(dtype="datetime64[ns, UTC]"), + "decision_time": pd.Series(dtype="datetime64[ns, UTC]"), + }) + + df = pd.concat( + [head.reset_index(drop=True), book.reset_index(drop=True), + extras.reset_index(drop=True)], axis=1) + df = df[["trade_id", "bar_ts", "decision_time"] + book_cols + extra_cols] + + # Compact dtypes (nullable where a value can legitimately be absent). + if len(df): + for role in ROLES: + df[f"{role}_composite_signal"] = df[f"{role}_composite_signal"].astype("Int8") + df[f"{role}_n_voting"] = df[f"{role}_n_voting"].astype("Int16") + df["final_signal"] = df["final_signal"].astype("int8") + df["exhaustion_count"] = df["exhaustion_count"].astype("int16") + df["n_warnings"] = df["n_warnings"].astype("int8") + df["mtf_agreement"] = df["mtf_agreement"].astype(bool) + df["momentum_fading"] = df["momentum_fading"].astype(bool) + return df + + +def _trade_extra_columns(pen_tags: List[str]) -> List[str]: + cols = ["tp2", "risk_pips", "reward_pips", "pip_value", "stake_multiplier"] + for role in ROLES: + cols += [f"{role}_tf", f"{role}_regime", f"{role}_vol", + f"{role}_composite", f"{role}_composite_signal", + f"{role}_confidence", f"{role}_atr", f"{role}_n_voting", + f"{role}_sum_eff_weight", f"{role}_sum_wscore"] + cols += ["mtf_agreement", "final_signal", "penalty_total"] + cols += [f"pen_{_sanitize(t)}" for t in pen_tags] + cols += ["exhaustion_count", "momentum_fading", "setup_strategy", + "setup_pattern", "round_trip_cost", "setup_trail_arm_r", + "setup_trail_dist_r", "warnings_all", "cleaned_warnings", + "n_warnings", "bullish_reasons_all", "bearish_reasons_all", + "spread_estimate", "current_price"] + return cols + + +def build_indicators_table(trades) -> pd.DataFrame: + """LONG table: one row per (trade, timeframe role, indicator that VOTED). + + Divergence and candlestick only exist when they fired — an absent row + means "no vote", never a zero vote. + """ + rows: List[Dict] = [] + for i, t in enumerate(list(trades or [])): + snap = getattr(t, "snapshot", None) or {} + for block in (snap.get("tfs") or []): + regime = str(block.get("regime") or "") + tf_name = str(block.get("tf") or "") + role = str(block.get("role") or "") + for name, ind in (block.get("indicators") or {}).items(): + rows.append({ + "trade_id": i, + "tf_role": role, + "timeframe": tf_name, + "indicator": str(name), + "signal": int(ind.get("signal") or 0), + "signal_name": str(ind.get("signal_name") or ""), + "value": _num(ind.get("value")), + "confidence": _num(ind.get("confidence")), + "base_weight": _num(ind.get("base_weight")), + "regime_mult": _num(ind.get("regime_mult")), + "eff_weight": _num(ind.get("eff_weight")), + "wscore": _num(ind.get("wscore")), + "agrees": str(ind.get("agrees") or ""), + "regime": regime, + "description": str(ind.get("description") or ""), + }) + if not rows: + return pd.DataFrame({c: pd.Series(dtype="object") for c in _IND_COLS}) + df = pd.DataFrame(rows, columns=_IND_COLS) + df["trade_id"] = df["trade_id"].astype("int32") + df["signal"] = df["signal"].astype("int8") + for col in ("value", "confidence", "base_weight", "regime_mult", + "eff_weight", "wscore"): + df[col] = df[col].astype("float64") + return df + + +# ═════════════════════════════════════════════════════════════════════ +# ECONOMIC CALENDAR +# ═════════════════════════════════════════════════════════════════════ + +def load_calendar(path=CALENDAR_PARQUET) -> pd.DataFrame: + """Forex-Factory calendar parquet, normalised. + + `timeLabel` is a display string and is NEVER used for timing — + `dateline` (unix seconds UTC) is the only clock. `timeMasked` marks + all-day rows, whose `dateline` is only a date anchor. + """ + cal = pd.read_parquet(path) + cal = cal.copy() + cal["ts_utc"] = pd.to_datetime( + cal["dateline"], unit="s", utc=True).astype("datetime64[ns, UTC]") + cal["timed"] = ~cal["timeMasked"].astype(bool) + cal["et_date"] = cal["ts_utc"].dt.tz_convert(_NY).dt.date + cal = cal.rename(columns={"impactName": "impact", + "actualBetterWorse": "better_worse"}) + for col in ("actual", "forecast", "previous", "revision"): + if col in cal.columns: + cal[col] = cal[col].fillna("").astype(str) + return cal.sort_values("ts_utc").reset_index(drop=True) + + +def calendar_fingerprint(path=CALENDAR_PARQUET) -> Optional[Dict]: + p = Path(path) + if not p.exists(): + return None + try: + cal = load_calendar(p) + return { + "path": str(p), + "md5": hashlib.md5(p.read_bytes()).hexdigest(), + "n_rows": int(len(cal)), + "min_ts": cal["ts_utc"].min().isoformat() if len(cal) else None, + "max_ts": cal["ts_utc"].max().isoformat() if len(cal) else None, + } + except Exception as exc: # pragma: no cover + logger.warning("--record: calendar fingerprint failed: %s", exc) + return None + + +def _ns(series: pd.Series) -> np.ndarray: + """tz-aware datetimes -> int64 nanoseconds (NaT -> iinfo.min).""" + s = pd.to_datetime(series, utc=True, errors="coerce").astype("datetime64[ns, UTC]") + return s.astype("int64").to_numpy() + + +_NAT_NS = np.iinfo(np.int64).min + + +def _empty(cols: List[str]) -> pd.DataFrame: + return pd.DataFrame({c: pd.Series(dtype="object") for c in cols}) + + +def build_events_table(trades_df: pd.DataFrame, cal: pd.DataFrame) -> pd.DataFrame: + """LONG table: calendar rows around each trade. + + Timed rows land in the window ``[entry - 4h, max(exit, entry + 4h)]`` + (both ends inclusive) and are phased `before` / `during_hold` / + `after_exit`. All-day (masked) rows are attached when their New-York + calendar date equals the entry's New-York date. + """ + if trades_df is None or trades_df.empty or cal is None or cal.empty: + return _empty(_EVENT_COLS) + + timed = cal[cal["timed"]].reset_index(drop=True) + masked = cal[~cal["timed"]].reset_index(drop=True) + t_ns = _ns(timed["ts_utc"]) if len(timed) else np.zeros(0, dtype=np.int64) + + by_date: Dict[object, List[int]] = {} + for i, d in enumerate(masked["et_date"].tolist()): + by_date.setdefault(d, []).append(i) + + entry_ns = _ns(trades_df["entry_time"]) + exit_ns = (_ns(trades_df["exit_time"]) if "exit_time" in trades_df.columns + else np.full(len(trades_df), _NAT_NS, dtype=np.int64)) + entry_et = pd.to_datetime(trades_df["entry_time"], utc=True) \ + .dt.tz_convert(_NY).dt.date.tolist() + trade_ids = (trades_df["trade_id"].tolist() if "trade_id" in trades_df.columns + else list(range(len(trades_df)))) + win_ns = int(EVENT_WINDOW.value) + + rows: List[Dict] = [] + for i in range(len(trades_df)): + e = int(entry_ns[i]) + if e == _NAT_NS: + continue + x = int(exit_ns[i]) + has_exit = x != _NAT_NS + # No exit yet -> treat the hold as zero-length; the window still + # runs to entry + 4h so the release context is recorded. + x_eff = x if has_exit else e + lo = e - win_ns + hi = max(x_eff, e + win_ns) + if len(t_ns): + a = int(np.searchsorted(t_ns, lo, side="left")) + b = int(np.searchsorted(t_ns, hi, side="right")) + else: + a = b = 0 + for j in range(a, b): + ts = int(t_ns[j]) + if ts < e: + phase = "before" + elif ts <= x_eff: + phase = "during_hold" + else: + phase = "after_exit" + r = timed.iloc[j] + rows.append({ + "trade_id": trade_ids[i], + "event_id": int(r["id"]), + "ts_utc": r["ts_utc"], + "name": str(r["name"]), + "currency": str(r["currency"]), + "impact": str(r["impact"]), + "timed": True, + "actual": str(r["actual"]), + "forecast": str(r["forecast"]), + "previous": str(r["previous"]), + "revision": str(r["revision"]), + "better_worse": int(r["better_worse"]), + "offset_min": (ts - e) / 6e10, + "phase": phase, + }) + for j in by_date.get(entry_et[i], ()): + r = masked.iloc[j] + rows.append({ + "trade_id": trade_ids[i], + "event_id": int(r["id"]), + "ts_utc": r["ts_utc"], + "name": str(r["name"]), + "currency": str(r["currency"]), + "impact": str(r["impact"]), + "timed": False, + "actual": str(r["actual"]), + "forecast": str(r["forecast"]), + "previous": str(r["previous"]), + "revision": str(r["revision"]), + "better_worse": int(r["better_worse"]), + "offset_min": np.nan, + "phase": "all_day", + }) + + if not rows: + return _empty(_EVENT_COLS) + df = pd.DataFrame(rows, columns=_EVENT_COLS) + df["event_id"] = df["event_id"].astype("int64") + df["better_worse"] = df["better_worse"].astype("int8") + df["offset_min"] = df["offset_min"].astype("float64") + df["timed"] = df["timed"].astype(bool) + return df + + +def calendar_scalars(trades_df: pd.DataFrame, cal: pd.DataFrame) -> pd.DataFrame: + """Per-trade calendar summary, index-aligned to `trades_df`.""" + n = 0 if trades_df is None else len(trades_df) + out = pd.DataFrame({ + "ev_prev_hi_usd_min": np.full(n, np.nan), + "ev_prev_hi_usd_name": [""] * n, + "ev_next_hi_usd_min": np.full(n, np.nan), + "ev_next_hi_usd_name": [""] * n, + "ev_hi_usd_during_hold": np.zeros(n, dtype="int16"), + "ev_hi_usd_during_hold_names": [""] * n, + "ev_any_hi_during_hold": np.zeros(n, dtype="int16"), + "ev_entry_within_30m_hi_usd": np.zeros(n, dtype=bool), + "us_holiday": np.zeros(n, dtype=bool), + }, index=(trades_df.index if n else None)) + if not n or cal is None or cal.empty: + return out + + hi_all = cal[(cal["impact"] == "high") & cal["timed"]].reset_index(drop=True) + hi_usd = hi_all[hi_all["currency"] == "USD"].reset_index(drop=True) + hi_all_ns = _ns(hi_all["ts_utc"]) if len(hi_all) else np.zeros(0, np.int64) + hi_usd_ns = _ns(hi_usd["ts_utc"]) if len(hi_usd) else np.zeros(0, np.int64) + usd_names = hi_usd["name"].astype(str).tolist() + + hol = cal[(cal["currency"] == "USD") & (cal["impact"] == "holiday") + & (~cal["timed"])] + hol_dates = set(hol["et_date"].tolist()) + + entry_ns = _ns(trades_df["entry_time"]) + exit_ns = (_ns(trades_df["exit_time"]) if "exit_time" in trades_df.columns + else np.full(n, _NAT_NS, dtype=np.int64)) + entry_et = pd.to_datetime(trades_df["entry_time"], utc=True) \ + .dt.tz_convert(_NY).dt.date.tolist() + half_hour = int(30 * 6e10) + + for i in range(n): + e = int(entry_ns[i]) + if e == _NAT_NS: + continue + pos = int(np.searchsorted(hi_usd_ns, e, side="right")) + if pos > 0: + mins = (e - int(hi_usd_ns[pos - 1])) / 6e10 + if mins <= NEAR_RELEASE_MAX_MIN: + out.iat[i, 0] = mins + out.iat[i, 1] = usd_names[pos - 1] + if pos < len(hi_usd_ns): + mins = (int(hi_usd_ns[pos]) - e) / 6e10 + if mins <= NEAR_RELEASE_MAX_MIN: + out.iat[i, 2] = mins + out.iat[i, 3] = usd_names[pos] + + x = int(exit_ns[i]) + if x != _NAT_NS and x > e: + lo = int(np.searchsorted(hi_usd_ns, e, side="right")) + hi = int(np.searchsorted(hi_usd_ns, x, side="right")) + out.iat[i, 4] = int(max(0, hi - lo)) + out.iat[i, 5] = "|".join(usd_names[lo:hi]) + lo_a = int(np.searchsorted(hi_all_ns, e, side="right")) + hi_a = int(np.searchsorted(hi_all_ns, x, side="right")) + out.iat[i, 6] = int(max(0, hi_a - lo_a)) + + if len(hi_usd_ns): + a = int(np.searchsorted(hi_usd_ns, e - half_hour, side="left")) + b = int(np.searchsorted(hi_usd_ns, e + half_hour, side="right")) + out.iat[i, 7] = bool(b > a) + out.iat[i, 8] = entry_et[i] in hol_dates + + out["ev_hi_usd_during_hold"] = out["ev_hi_usd_during_hold"].astype("int16") + out["ev_any_hi_during_hold"] = out["ev_any_hi_during_hold"].astype("int16") + out["ev_entry_within_30m_hi_usd"] = out["ev_entry_within_30m_hi_usd"].astype(bool) + out["us_holiday"] = out["us_holiday"].astype(bool) + return out + + +# ═════════════════════════════════════════════════════════════════════ +# PER-BAR RECORDER (--record-bars) +# ═════════════════════════════════════════════════════════════════════ + +def new_bar_buffer() -> Dict: + return {"rows": [], "pos": [], "ts_ns": [], "warning0": []} + + +def add_bar_row(buf: Dict, pos: int, ts, report, + primary_tf, confirmation_tfs) -> None: + """Append one compact float32 row for an ANALYZED primary bar. + + Runs ~140k times on a 2-year M5 replay, so it allocates exactly one + NaN row and fills it through a precomputed column-index map. An + absent indicator or a missing confirmation block stays NaN — never 0 + (0 is a real NEUTRAL vote). + """ + row = np.full(_N_BAR_COLS, np.nan, dtype=np.float32) + idx = _BAR_IDX + setup = getattr(report, "trade_setup", None) + + row[idx["close"]] = _num(report.current_price) + row[idx["final_score"]] = _num(report.final_score) + row[idx["final_conf"]] = _num(report.final_confidence) + row[idx["quality_code"]] = QUALITY_CODES.get( + str(report.signal_quality or ""), np.nan) + row[idx["strategy_code"]] = ( + STRATEGY_CODES.get(str(getattr(setup, "strategy", "") or ""), np.nan) + if setup is not None else 0) + row[idx["has_setup"]] = 1.0 if setup is not None else 0.0 + row[idx["final_signal"]] = _sig_int(report.final_signal) + row[idx["mtf_agreement"]] = 1.0 if report.mtf_agreement else 0.0 + row[idx["penalty_total"]] = _num(report.penalty_total) + row[idx["exhaustion_count"]] = int(report.exhaustion_count or 0) + + for role, analysis in _role_analyses(report, confirmation_tfs): + levels = getattr(analysis, "key_levels", None) or {} + row[idx[f"{role}_regime_code"]] = REGIME_CODES.get( + str(getattr(analysis, "market_regime", "") or ""), np.nan) + row[idx[f"{role}_vol_code"]] = VOL_CODES.get( + str(getattr(analysis, "volatility_regime", "") or ""), np.nan) + row[idx[f"{role}_composite"]] = _num(getattr(analysis, "composite_score", None)) + row[idx[f"{role}_confidence"]] = _num(getattr(analysis, "confidence", None)) + row[idx[f"{role}_atr"]] = _num(levels.get("atr")) + for name, ind in (getattr(analysis, "indicators", None) or {}).items(): + if name.startswith("_") or ind.weight == 0: + continue + key = f"{role}_{name}_" + base = idx.get(key + "signal") + if base is None: + continue # indicator outside INDICATORS + row[base] = _sig_int(ind.signal) + row[base + 1] = _num(ind.value) + row[base + 2] = _num(ind.confidence) + row[base + 3] = _num(getattr(ind, "effective_weight", None)) + row[base + 4] = _num(getattr(ind, "weighted_score", None)) + + buf["rows"].append(row) + buf["pos"].append(int(pos)) + buf["ts_ns"].append(int(pd.Timestamp(ts).value)) + warns = report.warnings or [] + buf["warning0"].append(str(warns[0]) if warns else "") + + +def finish_bar_rows(buf: Optional[Dict]): + """(matrix float32 [n,K], ts_ns int64, pos int64, warning0 list) or None.""" + if not buf or not buf.get("rows"): + return None + mat = np.vstack(buf["rows"]).astype(np.float32, copy=False) + return (mat, + np.asarray(buf["ts_ns"], dtype=np.int64), + np.asarray(buf["pos"], dtype=np.int64), + list(buf["warning0"])) + + +def forward_moves(close: np.ndarray, high: np.ndarray, low: np.ndarray, + atr: np.ndarray, k: int) -> Dict[str, np.ndarray]: + """ATR-normalised forward move / MFE / MAE over the next `k` bars. + + NaN where the horizon runs off the end of the series or the anchor + bar has no usable ATR. + """ + close = np.asarray(close, dtype=np.float64) + high = np.asarray(high, dtype=np.float64) + low = np.asarray(low, dtype=np.float64) + atr = np.asarray(atr, dtype=np.float64) + n = close.size + fwd = np.full(n, np.nan) + mfe = np.full(n, np.nan) + mae = np.full(n, np.nan) + if k <= 0 or n <= k: + return {"fwd": fwd, "mfe": mfe, "mae": mae} + + m = n - k # anchors 0 .. n-k-1 + with np.errstate(invalid="ignore", divide="ignore"): + denom = atr[:m] + fwd[:m] = (close[k:] - close[:m]) / denom + win_hi = np.lib.stride_tricks.sliding_window_view(high[1:], k)[:m] + win_lo = np.lib.stride_tricks.sliding_window_view(low[1:], k)[:m] + mfe[:m] = (win_hi.max(axis=1) - close[:m]) / denom + mae[:m] = (close[:m] - win_lo.min(axis=1)) / denom + + bad = ~np.isfinite(atr) | (atr <= 0) + fwd[bad] = np.nan + mfe[bad] = np.nan + mae[bad] = np.nan + return {"fwd": fwd, "mfe": mfe, "mae": mae} + + +def build_bars_table(chunks, sim_df: pd.DataFrame, primary_tf: Timeframe, + primary_atr: Optional[np.ndarray] = None) -> Optional[pd.DataFrame]: + """Concat worker bar chunks into one frame + ATR-normalised forward moves.""" + mats, tss, poss = [], [], [] + warns: List[str] = [] + for chunk in (chunks or []): + if chunk is None: + continue + mat, ts_ns, pos, warn0 = chunk + mats.append(mat) + tss.append(ts_ns) + poss.append(pos) + warns.extend(warn0) + if not mats: + return None + + mat = np.vstack(mats) + ts_ns = np.concatenate(tss) + pos = np.concatenate(poss) + order = np.argsort(pos, kind="stable") + mat = mat[order] + ts_ns = ts_ns[order] + pos = pos[order] + warns = [warns[i] for i in order] + + bar_ts = pd.to_datetime(ts_ns, utc=True) + df = pd.DataFrame(mat, columns=list(BAR_COLS)) + df.insert(0, "warning0", warns) + df.insert(0, "raw_pos", np.zeros(len(df), dtype=np.int32)) + df.insert(0, "pos", pos.astype(np.int32)) + df.insert(0, "bar_ts", bar_ts) + + if sim_df is None or sim_df.empty: + return df + + raw_pos = np.asarray(sim_df.index.get_indexer(bar_ts), dtype=np.int64) + df["raw_pos"] = raw_pos.astype(np.int32) + valid = raw_pos >= 0 + + close = sim_df["Close"].to_numpy(dtype=np.float64) + high = sim_df["High"].to_numpy(dtype=np.float64) + low = sim_df["Low"].to_numpy(dtype=np.float64) + n_raw = close.size + + atr_vals = (np.asarray(primary_atr, dtype=np.float64) + if primary_atr is not None + else df["primary_atr"].to_numpy(dtype=np.float64)) + atr_full = np.full(n_raw, np.nan) + if valid.any(): + atr_full[raw_pos[valid]] = atr_vals[valid] + + bar_seconds = int(pd.Timedelta(primary_tf.value).total_seconds()) + horizons = {"1h": 3600, "4h": 4 * 3600, "24h": 24 * 3600} + ks = {label: max(1, secs // max(bar_seconds, 1)) + for label, secs in horizons.items()} + + def _pick(vec: np.ndarray) -> np.ndarray: + out = np.full(len(df), np.nan) + if valid.any(): + out[valid] = vec[raw_pos[valid]] + return out + + moves = {label: forward_moves(close, high, low, atr_full, k) + for label, k in ks.items()} + df["fwd_1h_atr"] = _pick(moves["1h"]["fwd"]) + df["fwd_4h_atr"] = _pick(moves["4h"]["fwd"]) + df["fwd_24h_atr"] = _pick(moves["24h"]["fwd"]) + df["mfe_4h_atr"] = _pick(moves["4h"]["mfe"]) + df["mae_4h_atr"] = _pick(moves["4h"]["mae"]) + df["mfe_24h_atr"] = _pick(moves["24h"]["mfe"]) + df["mae_24h_atr"] = _pick(moves["24h"]["mae"]) + + k24 = ks["24h"] + avail = np.full(len(df), -1, dtype=np.int64) + if valid.any(): + avail[valid] = np.minimum(k24, n_raw - 1 - raw_pos[valid]) + df["fwd_bars_24h"] = avail.astype("int32") + return df + + +# ═════════════════════════════════════════════════════════════════════ +# RUN WRITER / READER +# ═════════════════════════════════════════════════════════════════════ + +def _normalise_datetimes(df: pd.DataFrame) -> pd.DataFrame: + for col in df.columns: + if col in _DT_COLS or pd.api.types.is_datetime64_any_dtype(df[col]): + df[col] = pd.to_datetime( + df[col], utc=True, errors="coerce").astype("datetime64[ns, UTC]") + return df + + +def _write(df: Optional[pd.DataFrame], path: Path) -> Optional[int]: + if df is None: + return None + df = _normalise_datetimes(df.copy()) + df.to_parquet(path, index=False) + return int(len(df)) + + +def write_run(record_dir, *, trades, book_df, bt_cfg, platform_cfg, aggregator, + run_info: Dict, sim_df=None, bar_chunks=None) -> Path: + """Write manifest.json + the parquet tables for one replay.""" + out = Path(record_dir) + out.mkdir(parents=True, exist_ok=True) + run_info = dict(run_info or {}) + + trades_df = build_trades_table(trades, book_df, bt_cfg.primary_tf) + ind_df = build_indicators_table(trades) + + cal_fp = calendar_fingerprint(CALENDAR_PARQUET) + events_df: Optional[pd.DataFrame] = None + if cal_fp is not None: + try: + cal = load_calendar(CALENDAR_PARQUET) + events_df = build_events_table(trades_df, cal) + scalars = calendar_scalars(trades_df, cal) + for col in _SCALAR_COLS: + trades_df[col] = scalars[col].to_numpy() if len(scalars) else scalars[col] + except Exception as exc: + logger.warning("--record: calendar join failed (%s) — " + "events.parquet skipped", exc) + events_df = None + cal_fp = None + else: + logger.warning("--record: %s not found — no events table", + CALENDAR_PARQUET) + + engine_fp = aggregator.engine_config_fingerprint() + gate_fp = aggregator.gate_config_fingerprint() + trades_df["engine_fingerprint"] = engine_fp + trades_df["gate_fingerprint"] = gate_fp + + bars_df = (build_bars_table(bar_chunks, sim_df, bt_cfg.primary_tf) + if bar_chunks else None) + + n_trades = _write(trades_df, out / "trades.parquet") + n_ind = _write(ind_df, out / "indicators.parquet") + n_events = _write(events_df, out / "events.parquet") + n_bars = _write(bars_df, out / "bars.parquet") + + base_weights: Dict[str, List[float]] = {} + if len(ind_df): + for name, grp in ind_df.groupby("indicator")["base_weight"]: + base_weights[str(name)] = sorted( + {float(v) for v in grp.dropna().unique()}) + + skipped = dict(run_info.get("skipped") or {}) + n_in_window = run_info.get("n_bars_in_window") + n_analyzed = (int(n_in_window) - int(sum(skipped.values())) + if n_in_window is not None else None) + + manifest = { + "created_at": datetime.now(timezone.utc).isoformat(), + "cli": sys.argv[1:], + "git_sha": git_sha(), + "backend": run_info.get("backend"), + "pair": bt_cfg.pair, + "display_name": platform_cfg.display_name, + "primary_tf": bt_cfg.primary_tf.value, + "confirmation_tfs": [tf.value for tf in bt_cfg.confirmation_tfs], + "window": {"start": bt_cfg.start.isoformat(), "end": bt_cfg.end.isoformat()}, + "spread": bt_cfg.spread, + "quality_filter": list(bt_cfg.quality_filter), + "strategies_only": list(bt_cfg.strategies_only), + "warmup_bars": bt_cfg.warmup_bars, + "cooldown_bars": bt_cfg.cooldown_bars, + "max_hold_bars": bt_cfg.max_hold_bars, + "fixed_rr_tp": bt_cfg.fixed_rr_tp, + "workers": run_info.get("workers"), + "engine_fingerprint": engine_fp, + "gate_fingerprint": gate_fp, + "technical_config": dataclasses.asdict(platform_cfg.technical), + "signals_config": dataclasses.asdict(platform_cfg.signals), + "regime_weight_table": regime_weight_table(platform_cfg.technical), + "base_weights_observed": base_weights, + "composite_thresholds": dict(COMPOSITE_THRESHOLDS), + "bars": { + "first_pos": run_info.get("first_pos"), + "last_pos": run_info.get("last_pos"), + "n_primary_bars": run_info.get("n_primary_bars"), + "n_in_window": n_in_window, + "n_analyzed": n_analyzed, + "skipped": skipped, + }, + "n_candidates": run_info.get("n_candidates"), + "n_trades": int(len(trades_df)), + "cache_files": run_info.get("cache_files"), + "calendar": cal_fp, + "tables": {"trades": n_trades, "indicators": n_ind, + "events": n_events, "bars": n_bars}, + "book_csv_md5": hashlib.md5( + book_df.to_csv(index=False).encode()).hexdigest() + if book_df is not None else None, + "replay_seconds": run_info.get("replay_seconds"), + } + with open(out / "manifest.json", "w", encoding="utf-8") as fh: + json.dump(manifest, fh, indent=2, default=str) + return out + + +def load_run(record_dir) -> Dict: + """Read a recorded run back: manifest + tables (None when absent).""" + path = Path(record_dir) + with open(path / "manifest.json", encoding="utf-8") as fh: + manifest = json.load(fh) + def _read(name): + p = path / name + return pd.read_parquet(p) if p.exists() else None + return { + "manifest": manifest, + "trades": _read("trades.parquet"), + "indicators": _read("indicators.parquet"), + "events": _read("events.parquet"), + "bars": _read("bars.parquet"), + } diff --git a/backtest/replay_scorecard.py b/backtest/replay_scorecard.py new file mode 100644 index 0000000..06d752f --- /dev/null +++ b/backtest/replay_scorecard.py @@ -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()) diff --git a/backtest/research/replay_recording_2026-09-09.md b/backtest/research/replay_recording_2026-09-09.md new file mode 100644 index 0000000..7c91380 --- /dev/null +++ b/backtest/research/replay_recording_2026-09-09.md @@ -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. diff --git a/scripts/ff_calendar_scraper.py b/scripts/ff_calendar_scraper.py new file mode 100644 index 0000000..8e27723 --- /dev/null +++ b/scripts/ff_calendar_scraper.py @@ -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() diff --git a/test_all.py b/test_all.py index db380c0..2fda1d3 100644 --- a/test_all.py +++ b/test_all.py @@ -11281,6 +11281,586 @@ 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" + + # ══════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════