ForexBot2/scripts/baseline_common.py

"""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/<commit>/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")