"""Shared machinery for the backtest baseline capture/compare pair. `capture_baseline.py` and `compare_baseline.py` both import everything here so the two sides produce their outputs from literally the same code: the run-spec table, the subprocess runner, the equity/metrics derivations, and the cache/stderr sanity checks. Reproducibility invariants encoded here (see baseline//README.md): * --start/--end always pinned — `--days` bakes wall-clock time-of-day into the first replay bar (gpu_backtest.py resolves `end` with datetime.now when --end is absent). * --warmup-bars pinned at 250 — it feeds pad_days and therefore the parquet cache FILENAME. * Subprocess env drops CUDA_PATH (pins the NumPy CPU backend; setting it would silently flip _simulate_trailing onto the CuPy path) and sets PYTHONHASHSEED=0 (removes str-hash set-ordering as a variable). * Data must come from backtest/cache/ — a cache miss would hit the Oanda API (and the stale .env token 401s), and freshly fetched bars are not the baseline's bars. Any "Fetching" line in stderr fails the run. * stdout (summary.txt) is byte-comparable; stderr (run.log) carries %(asctime)s timestamps and elapsed times and is provenance-only. """ from __future__ import annotations import hashlib import json import math import os import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Tuple import pandas as pd ROOT = Path(__file__).resolve().parents[1] VENV_PYTHON = ROOT / "venv" / "bin" / "python" BACKTESTER_REL = "backtest/gpu_backtest.py" CACHE_DIR = ROOT / "backtest" / "cache" BASELINE_ROOT = ROOT / "baseline" WINDOW_START = "2025-08-04" WINDOW_END = "2026-08-04" #: Files that must match byte-for-byte on compare. run.log is deliberately #: absent (timestamped stderr, never comparable). COMPARED_FILES = ("trades.csv", "summary.txt", "equity.csv", "metrics.json") #: Canonical factory confirm map (pair_signals/xauusd.py) — pinned on the #: command line so each invocation is self-contained. CONFIRMS = { "5min": "15min,1h", "15min": "5min,1h", "1h": "15min,4h", "4h": "1h,1d", } #: Mirrors BaseDataProvider.BARS_PER_DAY for the TFs the baseline touches. #: Kept local so the scripts never import repo modules (importing app_config #: pulls the .env into os.environ). Drift is self-detecting: if the pad #: math ever changes upstream, the expected cache file goes missing and the #: pre-flight aborts before any network fetch. _BARS_PER_DAY = {"5min": 288, "15min": 96, "1h": 24, "4h": 6, "1d": 1} #: fetch_history logs an ERROR for parquets older than the raw-cache format #: change (2026-07-08) — those contain CLEANED bars and are invalid input. RAW_CACHE_EPOCH = 1783468800.0 # ───────────────────────────────────────────────────────────────────── # Run specs # ───────────────────────────────────────────────────────────────────── def run_specs() -> Dict[str, dict]: """The 8-run baseline matrix: 4 primary TFs × spreads {0.30, 0.65}.""" specs: Dict[str, dict] = {} for key, tf in (("m5", "5min"), ("m15", "15min"), ("h1", "1h"), ("h4", "4h")): for spread in ("0.30", "0.65"): run_key = f"{key}_s{spread.replace('.', '')}" # 0.30 → s030 specs[run_key] = { "primary_tf": tf, "confirm_tfs": CONFIRMS[tf], "spread": spread, "argv": build_argv(tf, spread), } return specs def build_argv(primary_tf: str, spread: str) -> List[str]: """Backtester argv with every reproducibility-relevant knob pinned. --export-csv is appended at run time (path depends on the output dir). """ return [ BACKTESTER_REL, "--pair", "XAUUSD", "--primary-tf", primary_tf, "--confirm-tfs", CONFIRMS[primary_tf], "--start", WINDOW_START, "--end", WINDOW_END, "--spread", spread, "--quality", "A+,A,B,C", "--warmup-bars", "250", "--cooldown-bars", "5", "--max-hold-bars", "300", "--workers", "4", "--reasoning-samples", "3", "--detail", ] def expected_cache_files(primary_tf: str) -> List[str]: """Parquet filenames a run will read (mirrors gpu_backtest pad math). pad_days = max(60, warmup_bars // 4) = 62 at the pinned 250; per-TF floor min_pad_days = ceil(210 / bars_per_day * 7/5) + 5. """ start = datetime.strptime(WINDOW_START, "%Y-%m-%d") end = datetime.strptime(WINDOW_END, "%Y-%m-%d") pad_days = max(60, 250 // 4) names = [] for tf in [primary_tf] + CONFIRMS[primary_tf].split(","): min_pad = math.ceil(210 / _BARS_PER_DAY[tf] * 7 / 5) + 5 tf_start = start - timedelta(days=max(pad_days, min_pad)) names.append( f"oanda_XAU_USD_{tf}_{tf_start:%Y%m%d}_{end:%Y%m%d}.parquet") return names def all_expected_cache_files(specs: Dict[str, dict]) -> List[str]: seen: List[str] = [] for spec in specs.values(): for name in expected_cache_files(spec["primary_tf"]): if name not in seen: seen.append(name) return seen # ───────────────────────────────────────────────────────────────────── # Subprocess runner + stderr sanity # ───────────────────────────────────────────────────────────────────── def run_backtest(argv: List[str], run_dir: Path) -> None: """Run one backtest: stdout → summary.txt, stderr → run.log. Raises RuntimeError (with the log path) on a nonzero exit. """ run_dir.mkdir(parents=True, exist_ok=True) env = {k: v for k, v in os.environ.items() if k != "CUDA_PATH"} env["PYTHONHASHSEED"] = "0" cmd = ([str(VENV_PYTHON)] + argv + ["--export-csv", str(run_dir / "trades.csv")]) with open(run_dir / "summary.txt", "wb") as out, \ open(run_dir / "run.log", "wb") as err: rc = subprocess.run(cmd, cwd=ROOT, env=env, stdout=out, stderr=err).returncode if rc != 0: raise RuntimeError( f"backtest exited {rc} — see {run_dir / 'run.log'}") def check_run_log(run_dir: Path, primary_tf: str) -> Tuple[List[str], str]: """Sanity-check a run's stderr. Returns (problems, backend_string). Hard requirements: every TF loaded from the parquet cache (zero "Fetching" lines — a live fetch means the input data is not the baseline's data) and a parseable Backend line. """ text = (run_dir / "run.log").read_text(errors="replace") problems: List[str] = [] if "Fetching " in text: problems.append("live fetch attempted (cache miss) — data is not " "the baseline's data") for tf in [primary_tf] + CONFIRMS[primary_tf].split(","): if f"Loading cached XAU/USD {tf}:" not in text: problems.append(f"no cache-hit line for TF {tf}") if "predates the raw-cache format" in text: problems.append("pre-2026-07-08 cache file in use (cleaned bars)") backend = "" for line in text.splitlines(): if "Backend: " in line: backend = line.split("Backend: ", 1)[1].split(" | ", 1)[0].strip() break if not backend: problems.append("no Backend line found in run.log") return problems, backend # ───────────────────────────────────────────────────────────────────── # Derived outputs (the backtester exports no equity curve — derive it # from the trade book instead of modifying gpu_backtest.py) # ───────────────────────────────────────────────────────────────────── def _resolved_trades(trades_csv: Path) -> pd.DataFrame: """Resolved trades in book order (matches compute_metrics' filter: unresolved '' / NO_DATA rows are exported to CSV but never scored).""" df = pd.read_csv(trades_csv, dtype=str, keep_default_na=False) df["r_multiple"] = pd.to_numeric(df["r_multiple"], errors="coerce") mask = (df["exit_reason"] != "") & (df["exit_reason"] != "NO_DATA") \ & df["r_multiple"].notna() return df[mask].reset_index(drop=True) def derive_equity(trades_csv: Path, out_path: Path) -> None: """Cumulative-R equity curve over resolved trades in book order.""" df = _resolved_trades(trades_csv) eq = pd.DataFrame({ "trade_n": range(1, len(df) + 1), "entry_time": df["entry_time"], "exit_time": df["exit_time"], "quality": df["quality"], "strategy": df["strategy"], "r_multiple": df["r_multiple"], }) eq["cum_r"] = eq["r_multiple"].cumsum() eq["high_water"] = eq["cum_r"].cummax().clip(lower=0.0) eq["drawdown_r"] = eq["cum_r"] - eq["high_water"] eq.to_csv(out_path, index=False, float_format="%.6f") def derive_metrics(trades_csv: Path, out_path: Path) -> dict: """Headline + grouped stats as sorted, fixed-precision JSON.""" df = _resolved_trades(trades_csv) metrics = {"overall": _stats(df["r_multiple"]), "by_quality": {}, "by_strategy": {}} for col, bucket in (("quality", "by_quality"), ("strategy", "by_strategy")): for value in sorted(df[col].unique()): metrics[bucket][value] = _stats(df.loc[df[col] == value, "r_multiple"]) out_path.write_text(json.dumps(metrics, indent=2, sort_keys=True) + "\n") return metrics def _stats(r: pd.Series) -> dict: n = len(r) wins = int((r > 0).sum()) gross_win = float(r[r > 0].sum()) gross_loss = float(-r[r < 0].sum()) cum = r.cumsum() dd = cum - cum.cummax().clip(lower=0.0) return { "trades": n, "total_r": round(float(r.sum()), 6), "win_rate_pct": round(wins / n * 100, 6) if n else 0.0, "profit_factor": (round(gross_win / gross_loss, 6) if gross_loss > 0 else None), "max_dd_r": round(float(dd.min()), 6) if n else 0.0, } # ───────────────────────────────────────────────────────────────────── # Hashing / manifest helpers # ───────────────────────────────────────────────────────────────────── def sha256_file(path: Path) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def config_inputs() -> dict: """Non-secret config.json fields that alter backtest results (the pair factory overlays `pairs`, and `data_providers.active` keys the cache).""" cfg_path = ROOT / "config.json" if not cfg_path.exists(): return {"data_providers.active": None, "pairs": None} cfg = json.loads(cfg_path.read_text()) return { "data_providers.active": cfg.get("data_providers", {}).get("active"), "pairs": cfg.get("pairs"), } def git_info() -> dict: def _git(*args: str) -> str: return subprocess.run(["git", *args], cwd=ROOT, capture_output=True, text=True).stdout.strip() return { "commit": _git("rev-parse", "HEAD"), "commit_short": _git("rev-parse", "--short", "HEAD"), # -uno: untracked files (e.g. these scripts pre-commit) can't # change the backtester; only tracked modifications count. "dirty": bool(_git("status", "--porcelain", "-uno")), } def tool_versions() -> dict: import numpy versions = { "python": ".".join(map(str, sys.version_info[:3])), "numpy": numpy.__version__, "pandas": pd.__version__, } try: from importlib.metadata import version versions["cupy"] = version("cupy") except Exception: versions["cupy"] = None return versions def utc_now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")