harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

backtest: walk-forward mode + extended-range data safety

Browse Source 
--walk-forward slices ONE continuous replay into rolling train/test
folds by entry time (default 12mo/3mo/3mo; no per-fold re-optimization)
and reports per-fold PF / expectancy / max DD plus aggregated OOS stats,
flagging test folds with PF < 1. Off by default — default runs are
byte-identical (scripts/compare_baseline.py 32/32 PASS with this diff).

Data safety for deep-history fetches: fetch_history refuses to recreate
any baseline-pinned parquet filename and records the sha256 of every
freshly written parquet in backtest/cache/data_manifest.json.

Tests: Module 23 (fold builder, slicing, flagging, OOS dedup, CLI
defaults, pin guard, manifest recording) — 367 total.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017FBjfomAipxvPWwRJEDDyJ
This commit is contained in:
Harwinder Sekhon
2026-08-07 18:21:39 -04:00
parent 1c264d5315
commit b357ab1e87
3 changed files with 500 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtest/gpu_backtest.py
+253
View File
@@ -25,11 +25,14 @@ Usage:
python backtest/gpu_backtest.py --pair XAUUSD --days 365
python backtest/gpu_backtest.py --pair XAUUSD --start 2025-01-01 --end 2025-06-01
python backtest/gpu_backtest.py --pair XAUUSD --days 180 --no-cache
python backtest/gpu_backtest.py --pair XAUUSD --start 2021-07-30 --end 2026-07-30 \
--walk-forward # rolling 12mo-train/3mo-test folds, step 3mo
"""
from __future__ import annotations
import argparse
import hashlib
import json
import logging
import multiprocessing as mp
@@ -972,6 +975,65 @@ class GPUTechnicalEngine(TechnicalEngine):
CACHE_DIR = Path(__file__).resolve().parent / "cache"
#: Root of the byte-for-byte reference set (scripts/capture_baseline.py).
BASELINE_ROOT = PROJECT_ROOT / "baseline"
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 _baseline_pinned_names() -> set:
"""Parquet filenames claimed by any baseline/<commit>/manifest.json.
These are the pinned inputs of the byte-for-byte reference runs
(scripts/compare_baseline.py). fetch_history must never write one:
the write path only runs on a cache miss, so reaching it for a pinned
name means the original file was deleted — and recreating it from a
live fetch would put NEW bars under the baseline's filename, silently
poisoning the reference set. Extended-range windows always resolve to
different filenames (the cache key embeds start/end), so they land in
separate parquets and never collide with the pinned set.
"""
names: set = set()
if not BASELINE_ROOT.is_dir():
return names
for manifest in BASELINE_ROOT.glob("*/manifest.json"):
try:
names.update(json.loads(manifest.read_text()).get("cache_files", {}))
except (OSError, json.JSONDecodeError):
continue
return names
def _record_in_data_manifest(path: Path) -> None:
"""Record a freshly written parquet's sha256 in cache/data_manifest.json.
Uses the baseline manifest's ``cache_files`` mapping format
({filename: sha256}) so extended-range data files carry the same
provenance as the pinned baseline set. Best-effort: a manifest write
failure must not fail the fetch that produced the data.
"""
manifest_path = CACHE_DIR / "data_manifest.json"
try:
manifest = json.loads(manifest_path.read_text())
except (OSError, json.JSONDecodeError):
manifest = {}
files = manifest.get("cache_files", {})
files[path.name] = _sha256_file(path)
manifest["cache_files"] = dict(sorted(files.items()))
manifest["updated_utc"] = datetime.now(timezone.utc).strftime(
"%Y-%m-%dT%H:%M:%SZ")
try:
manifest_path.write_text(
json.dumps(manifest, indent=2, sort_keys=True) + "\n")
except OSError as exc:
logger.warning(f"data_manifest.json not updated: {exc}")
def _cache_path(
symbol: str, tf: Timeframe, start: datetime, end: datetime,
@@ -1031,7 +1093,15 @@ def fetch_history(
df = provider.fetch_range(symbol, timeframe, start, end)
if use_cache:
if path.name in _baseline_pinned_names():
raise RuntimeError(
f"refusing to write {path.name}: this filename is pinned by a "
f"baseline manifest under baseline/ and freshly fetched bars "
f"are not the baseline's bars. Restore the original parquet "
f"(git/backup) instead of refetching, or use a different "
f"--start/--end window.")
df.to_parquet(path)
_record_in_data_manifest(path)
logger.info(f"Cached {len(df)} raw bars to {path.name}")
return provider.validate_and_clean(df, symbol, timeframe) if clean else df
@@ -2157,6 +2227,154 @@ def print_trade_reasoning(trades: List[Trade], limit: int = 5) -> None:
print(f" MFE / MAE: {t.mfe_r:.2f}R / {t.mae_r:.2f}R")
# ═════════════════════════════════════════════════════════════════════
# WALK-FORWARD ANALYSIS (--walk-forward)
# ═════════════════════════════════════════════════════════════════════
#
# Rolling-fold out-of-sample evaluation of ONE continuous replay: the
# window is replayed exactly once with the run's fixed config (there is
# no per-fold re-optimization — nothing here retunes anything), then the
# trade book is sliced into rolling train/test folds by ENTRY time.
# Slicing a single replay keeps the book identical to a normal run of the
# same window: warmup happens once and the entry cooldown crosses fold
# boundaries exactly as it would live. Train columns are in-sample
# context; the aggregated stats over all test folds are the OOS read.
def _add_months(dt: datetime, months: int) -> datetime:
"""Calendar-month arithmetic (tz preserved; clamps at month end)."""
return (pd.Timestamp(dt) + pd.DateOffset(months=months)).to_pydatetime()
def build_walkforward_folds(
start: datetime, end: datetime,
train_months: int, test_months: int, step_months: int,
) -> List[Tuple[datetime, datetime, datetime, datetime]]:
"""Rolling (train_start, train_end, test_start, test_end) folds.
test_start == train_end; a fold is emitted only if it fits entirely
inside [start, end]. Returns [] when the window is too short.
"""
if min(train_months, test_months, step_months) <= 0:
raise ValueError("train/test/step months must all be > 0")
folds = []
i = 0
while True:
tr_s = _add_months(start, i * step_months)
tr_e = _add_months(tr_s, train_months)
te_e = _add_months(tr_e, test_months)
if te_e > end:
break
folds.append((tr_s, tr_e, tr_e, te_e))
i += 1
return folds
def walkforward_stats(
trades: List[Trade],
folds: List[Tuple[datetime, datetime, datetime, datetime]],
) -> Dict:
"""Per-fold train/test Metrics + deduped aggregated OOS stats.
Fold membership is by entry time (half-open [lo, hi)); a trade's
resolution may extend past its fold end, matching how a live account
would hold through the boundary. With step < test length the test
windows overlap — the OOS aggregate counts each trade exactly once,
in chronological order.
"""
fold_stats: List[Dict] = []
oos_trades: List[Trade] = []
seen_ids: set = set()
for k, (tr_s, tr_e, te_s, te_e) in enumerate(folds, 1):
lo_tr, hi_tr = pd.Timestamp(tr_s), pd.Timestamp(tr_e)
lo_te, hi_te = pd.Timestamp(te_s), pd.Timestamp(te_e)
test_slice = [t for t in trades if lo_te <= t.entry_time < hi_te]
train_m = compute_metrics(
[t for t in trades if lo_tr <= t.entry_time < hi_tr])
test_m = compute_metrics(test_slice)
fold_stats.append({
"fold": k,
"train_start": tr_s, "train_end": tr_e,
"test_start": te_s, "test_end": te_e,
"train": train_m, "test": test_m,
# PF < 1.0 on the fold's OOS slice; empty folds are not flagged
# (no evidence either way) but are visible as n=0 in the table.
"flagged": test_m.n_trades > 0 and test_m.profit_factor < 1.0,
})
for t in test_slice:
if id(t) not in seen_ids:
seen_ids.add(id(t))
oos_trades.append(t)
oos_trades.sort(key=lambda t: t.entry_time)
return {
"folds": fold_stats,
"oos": compute_metrics(oos_trades),
"n_flagged": sum(1 for f in fold_stats if f["flagged"]),
}
def _fmt_pf(m: Metrics) -> str:
if m.n_trades == 0:
return "—"
return "inf" if np.isinf(m.profit_factor) else f"{m.profit_factor:.2f}"
def print_walkforward_report(
bt_cfg: BacktestConfig, stats: Dict,
train_months: int, test_months: int, step_months: int,
) -> None:
bar = "═" * 78
folds = stats["folds"]
print(f"\n{bar}")
print(f" WALK-FORWARD ANALYSIS "
f"({bt_cfg.start.date()} → {bt_cfg.end.date()})")
print(f" Train {train_months}mo / Test {test_months}mo / "
f"Step {step_months}mo → {len(folds)} folds "
f"(fixed config — no per-fold re-optimization)")
print(f"{bar}")
rows = []
for f in folds:
tr, te = f["train"], f["test"]
rows.append([
str(f["fold"]),
f"{f['train_start']:%Y-%m-%d}→{f['train_end']:%Y-%m-%d}",
str(tr.n_trades), _fmt_pf(tr),
f"{tr.avg_r:+.3f}" if tr.n_trades else "—",
f"{tr.max_drawdown_r:.2f}" if tr.n_trades else "—",
f"{f['test_start']:%Y-%m-%d}→{f['test_end']:%Y-%m-%d}",
str(te.n_trades), _fmt_pf(te),
f"{te.avg_r:+.3f}" if te.n_trades else "—",
f"{te.max_drawdown_r:.2f}" if te.n_trades else "—",
"⚠ PF<1" if f["flagged"] else "",
])
print("\n ── Rolling folds (expR = expectancy per trade, "
"DD = max drawdown in R) ──")
print(_fmt_table(
["Fold", "Train window", "n", "PF", "expR", "DD",
"Test window", "n", "PF", "expR", "DD", "Flag"], rows))
o = stats["oos"]
print("\n ── Aggregated out-of-sample (union of test folds, "
"each trade once) ──")
if o.n_trades == 0:
print(" No trades in any test fold.")
else:
print(_fmt_table(["Metric", "Value"], [
["OOS trades", f"{o.n_trades}"],
["Win rate", f"{o.win_rate:.1f}%"],
["Total R", f"{o.total_r:+.2f}"],
["Expectancy / trade", f"{o.avg_r:+.3f}"],
["Profit factor", _fmt_pf(o)],
["Max DD (R)", f"{o.max_drawdown_r:.2f}"],
["Folds flagged PF<1", f"{stats['n_flagged']} / {len(folds)}"],
]))
if stats["n_flagged"]:
flagged = [str(f["fold"]) for f in folds if f["flagged"]]
print(f"\n ⚠ {stats['n_flagged']} fold(s) with out-of-sample "
f"PF < 1.0: fold {', '.join(flagged)}")
print(f"{bar}\n")
def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame:
"""Export trades for downstream analysis (e.g. parquet/CSV).
@@ -2217,6 +2435,19 @@ def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
p.add_argument("--days", type=int, default=180, help="Days of history (ignored if --start/--end set)")
p.add_argument("--start", help="ISO date YYYY-MM-DD (overrides --days)")
p.add_argument("--end", help="ISO date YYYY-MM-DD (defaults to today)")
p.add_argument("--walk-forward", action="store_true",
help="Walk-forward analysis: replay the window once with "
"the run's fixed config, then slice the book into "
"rolling train/test folds and report per-fold PF / "
"expectancy / max DD / trade count plus aggregated "
"out-of-sample stats (folds with test PF < 1.0 are "
"flagged). No per-fold re-optimization.")
p.add_argument("--wf-train-months", type=int, default=12,
help="Walk-forward: train window length (calendar months)")
p.add_argument("--wf-test-months", type=int, default=3,
help="Walk-forward: test window length (calendar months)")
p.add_argument("--wf-step-months", type=int, default=3,
help="Walk-forward: step between fold starts (calendar months)")
p.add_argument("--primary-tf", default="15min", help="Primary TF (1min/5min/15min/1h/4h)")
p.add_argument("--confirm-tfs", default=None,
help="Comma-separated confirmation TFs. Default: the pair "
@@ -2398,6 +2629,24 @@ def main(argv: Optional[List[str]] = None) -> int:
else:
start = end - timedelta(days=args.days)
# Fail fast on an unusable walk-forward request — before any data
# loading. With walk-forward off nothing below changes.
wf_folds = None
if args.walk_forward:
try:
wf_folds = build_walkforward_folds(
start, end, args.wf_train_months,
args.wf_test_months, args.wf_step_months)
except ValueError as exc:
raise SystemExit(f"--walk-forward: {exc}")
if not wf_folds:
raise SystemExit(
f"--walk-forward: window {start.date()} → {end.date()} "
f"({(end - start).days} days) is shorter than one fold "
f"(train {args.wf_train_months}mo + test "
f"{args.wf_test_months}mo); pass a longer --start/--end "
f"range.")
spread = (args.spread if args.spread is not None
else PAIR_INFO[pair].get("default_spread", 0.30))
@@ -2427,6 +2676,10 @@ def main(argv: Optional[List[str]] = None) -> int:
trades, _ = run_backtest(bt_cfg, platform_cfg=platform_cfg)
metrics = compute_metrics(trades)
print_report(pair, bt_cfg, metrics, trades=trades, detail=args.detail)
if wf_folds:
print_walkforward_report(
bt_cfg, walkforward_stats(trades, wf_folds),
args.wf_train_months, args.wf_test_months, args.wf_step_months)
print_trade_reasoning(trades, limit=args.reasoning_samples)
if args.export_csv: