""" 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"), # >0 means some bar-analyses fell back to the live engine's pandas # math instead of the pre-computed cache (a parity hazard). "cache_fallbacks": run_info.get("cache_fallbacks"), "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"), }