mymt5opp/scripts/walk_forward.py

"""Walk-forward OOS evaluation of the Optuna finalists (doc 06 §4).

The 3 diverse finalists were selected on the IS window (2025-01-01 → 2025-12-15,
2-week purge after). This script runs each finalist's FIXED params on the OOS
window (2026-01-01 → 2026-07-01) and compares:

  OOS metric / IS metric

A robust finalist keeps most of its edge out-of-sample. A fragile one keeps
its edge only in IS — typically trade-count collapses or PF falls below 1.

Also runs the IS numbers with the same fixed params so the ratio is computed
on identical configurations (the study's stored metrics are valid but we
recompute here for the same OOS script path / instrument).

OOS uses the same M1 tick-level exit simulation (mandatory for trailing/BE).
"""
from __future__ import annotations
import sys
from pathlib import Path
PROJECT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT))

import optuna
import pandas as pd

from shared.core.engine import SizingInputs
from shared.core.metrics import compute_metrics
from shared.data.loaders import load_bars
from shared.optimizer.selector import select_diverse_topn
from strategies.gold_scalper_pro.instruments import XAUUSD_REAL
from strategies.gold_scalper_pro.scalper_engine import (
    ScalperEngine,
    engine_kwargs_from_params,
)
from strategies.gold_scalper_pro.search_space import (
    FROZEN_BASELINE,
    SEARCH_SPACE,
)
from strategies.gold_scalper_pro.signals import build_signals


IS_START  = pd.Timestamp("2025-01-01 00:00:00")
IS_END    = pd.Timestamp("2025-12-15 00:00:00")
OOS_START = pd.Timestamp("2026-01-01 00:00:00")
OOS_END   = pd.Timestamp("2026-07-01 00:00:00")
INITIAL_DEPOSIT = 1000.0


def slice_window(df: pd.DataFrame, start: pd.Timestamp, end: pd.Timestamp) -> pd.DataFrame:
    return df[(df["timestamp"] >= start) & (df["timestamp"] < end)].reset_index(drop=True)


def run_with_params(params: dict, bars: pd.DataFrame, m1: pd.DataFrame) -> dict:
    """Run the engine with FIXED params on a window, return metrics dict."""
    pack = build_signals(params, bars, XAUUSD_REAL)
    engine = ScalperEngine()
    result = engine.run(
        bars, pack.signals_long, pack.signals_short,
        pack.sl_prices, pack.tp_prices,
        XAUUSD_REAL, SizingInputs(), INITIAL_DEPOSIT,
        m1_bars=m1,
        **engine_kwargs_from_params(params),
    )
    m = compute_metrics(result, periods_per_year=252 * 24 * 12)
    return {
        "net": m.net_profit,
        "PF": m.profit_factor,
        "trades": m.total_trades,
        "DD%": m.max_equity_dd_pct,
        "sharpe": m.sharpe,
        "win_rate": m.win_rate,
    }


def main() -> int:
    db = PROJECT / "studies" / "optuna" / "gold_scalper_pro_is2025.db"
    study = optuna.load_study(
        study_name="gold_scalper_pro_is2025",
        storage=f"sqlite:///{db}",
    )
    finalists = select_diverse_topn(study, n=3, ranges=SEARCH_SPACE)
    if not finalists:
        print("no constraint-passing finalists to walk-forward.")
        return 1

    print("loading bars ...")
    m5 = load_bars(PROJECT / "data" / "XAUUSD_M5_2024-06-26_2026-06-26.parquet")
    m1 = load_bars(PROJECT / "data" / "XAUUSD_M1_2024-06-26_2026-06-26.parquet")
    is_bars = slice_window(m5, IS_START, IS_END)
    is_m1   = slice_window(m1, IS_START, IS_END)
    oos_bars = slice_window(m5, OOS_START, OOS_END)
    oos_m1   = slice_window(m1, OOS_START, OOS_END)
    print(f"  IS bars : {len(is_bars):,}   IS M1 : {len(is_m1):,}")
    print(f"  OOS bars: {len(oos_bars):,}   OOS M1: {len(oos_m1):,}")
    print(f"  IS window : {IS_START.date()} → {IS_END.date()}  (11.5 months)")
    print(f"  OOS window: {OOS_START.date()} → {OOS_END.date()}  (6 months)")

    print(f"\n{'='*100}")
    print(f"{'metric':12s} {'IS':>14s} {'OOS':>14s} {'OOS/IS':>10s}   notes")
    print(f"{'-'*100}")

    for i, t in enumerate(finalists, 1):
        merged = {**FROZEN_BASELINE, **t.params}
        print(f"\n--- finalist #{i}  trial #{t.number}  score={t.value:.2f} ---")
        is_m = run_with_params(merged, is_bars, is_m1)
        oos_m = run_with_params(merged, oos_bars, oos_m1)
        _print_row("net",       is_m["net"],  oos_m["net"],  ratio=oos_m["net"]/is_m["net"] if is_m["net"] != 0 else None)
        _print_row("PF",        is_m["PF"],   oos_m["PF"],   ratio=oos_m["PF"]/is_m["PF"] if is_m["PF"] != 0 else None)
        _print_row("trades",    is_m["trades"], oos_m["trades"], ratio=oos_m["trades"]/is_m["trades"] if is_m["trades"] else None)
        _print_row("DD%",       is_m["DD%"],  oos_m["DD%"],  ratio=oos_m["DD%"]/is_m["DD%"] if is_m["DD%"] else None)
        _print_row("sharpe",    is_m["sharpe"], oos_m["sharpe"], ratio=oos_m["sharpe"]/is_m["sharpe"] if is_m["sharpe"] else None)
        _print_row("win_rate",  is_m["win_rate"], oos_m["win_rate"], ratio=oos_m["win_rate"]/is_m["win_rate"] if is_m["win_rate"] else None)

    print(f"\n{'='*100}")
    print("interpretation (doc 06 §4 walk-forward):")
    print("  OOS/IS ≥ ~0.6 on net and PF  → edge holds out-of-sample (robust)")
    print("  OOS/IS < ~0.5 on PF, or OOS PF < 1.0  → fragile; IS-only edge")
    print("  trade-count ratio drops sharply  → signal degraded in the new regime")
    return 0


def _print_row(label: str, is_v: float, oos_v: float, *, ratio: float | None) -> None:
    if ratio is None:
        r = "  n/a"
    else:
        r = f"  {ratio:6.2f}"
    print(f"{label:12s} {is_v:14.2f} {oos_v:14.2f} {r:>10s}")


if __name__ == "__main__":
    raise SystemExit(main())