"""Phase 6 — Optuna optimization for GoldScalperPro (doc 06). Two modes via CLI flags: --smoke 30 trials, IS = 2025 H1 only (~6 months). Validates the objective + storage + scoring end-to-end in <2 min before committing to the full study. Run this first, always. (default) 500 trials, IS = 2025 full year (with a 2-week purge gap before year-end so OOS walk-forward is leak-free). TPE sampler, SQLite-persisted so the study resumes/inspects mid-run. Runs in the foreground with progress bar; for a long run launch with `start /b python scripts\\optimize.py` (Windows) and poll the .db file separately. Why M1 bars are mandatory here (doc 03 §7 / §8, CLAUDE.md standing rule): GoldScalperPro uses break-even + trailing stops, so the bar-level engine produces a −40% to −50% hidden gap vs MT5. The objective passes ``m1_bars`` to ``engine.run`` so the engine switches to tick-level exit simulation. Window design (doc 06 §4 walk-forward): IS = 2025-01-01 00:00 → 2025-12-15 00:00 (exclusive end, ~11.5 months) purge = 2025-12-15 .. 2025-12-31 (2-week gap, no trades counted either side) OOS = 2026-01-01 00:00 → 2026-07-01 00:00 (~6 months, fixed finalist params) Constraints (user-confirmed "strict" preset): min_trades=40, min_profit_factor=1.5, max_equity_dd_pct=0.25 """ from __future__ import annotations import argparse 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.optimizer.objective import ( Constraints, ObjectiveConfig, build_objective, ) 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 from strategies.gold_scalper_pro.search_space import ( FROZEN_BASELINE, INT_PARAMS, SEARCH_SPACE, ) from strategies.gold_scalper_pro.signals import build_signals from strategies.gold_scalper_pro.scalper_engine import engine_kwargs_from_params # ── Windows ──────────────────────────────────────────────────────────────── IS_START = pd.Timestamp("2025-01-01 00:00:00") IS_END = pd.Timestamp("2025-12-15 00:00:00") # exclusive end (purge after) OOS_START = pd.Timestamp("2026-01-01 00:00:00") OOS_END = pd.Timestamp("2026-07-01 00:00:00") # exclusive end INITIAL_DEPOSIT = 1000.0 SEED = 42 def slice_window(df: pd.DataFrame, start: pd.Timestamp, end: pd.Timestamp) -> pd.DataFrame: """Slice bars to [start, end) — exclusive end matches MT5 tester semantics.""" return df[(df["timestamp"] >= start) & (df["timestamp"] < end)].reset_index(drop=True) def make_objective_config( bars_is: pd.DataFrame, m1_is: pd.DataFrame, full_m5: pd.DataFrame, ) -> ObjectiveConfig: """Assemble the ObjectiveConfig for the IS window. ``bars_is`` / ``m1_is`` are trimmed to the IS evaluation window — the engine runs on these (initial_deposit reset, no open position at IS_START, matching MT5 Strategy Tester). ``full_m5`` is the FULL M5 history (data starts 2024-06-26 → ~6 months of pre-IS warmup, well beyond EMA(160)'s ~14h requirement). It is passed as ``signals_full_bars`` so the objective builds signals on it, then slices the signal arrays to ``bars_is``' time range — mirroring MT5 tester's pre-test chart-history indicator warmup. Without this, EMA/RSI/ATR would only start warming up at IS_START and the first Python trade would land ~14h late vs MT5 (the original warmup bug — see reeval_finalist_forward.py docstring). """ constraints = Constraints( min_trades=40, min_profit_factor=1.5, max_equity_dd_pct=0.25, ) return ObjectiveConfig( engine=ScalperEngine(), bars=bars_is, instrument=XAUUSD_REAL, sizing=SizingInputs(), initial_deposit=INITIAL_DEPOSIT, search_space=SEARCH_SPACE, int_params=INT_PARAMS, frozen_baseline=FROZEN_BASELINE, constraints=constraints, dd_weight=1.0, build_signals=build_signals, build_engine_kwargs=engine_kwargs_from_params, m1_bars=m1_is, # mandatory for trailing/BE EA signals_full_bars=full_m5, # indicator warmup (doc 03 §8) ) def run_smoke() -> int: """30 trials, IS H1 2025 only. Validates the script end-to-end.""" print("=== Phase 6 SMOKE TEST ===") m5 = load_m5() m1 = load_m1() # Shorter IS window for the smoke run. bars_is = slice_window(m5, pd.Timestamp("2025-01-01"), pd.Timestamp("2025-07-01")) m1_is = slice_window(m1, pd.Timestamp("2025-01-01"), pd.Timestamp("2025-07-01")) print(f" IS bars : {len(bars_is):,} M1 bars: {len(m1_is):,}") print(f" warmup : {len(m5):,} full M5 bars (signals_full_bars)") cfg = make_objective_config(bars_is, m1_is, full_m5=m5) objective = build_objective(cfg) study = optuna.create_study( direction="maximize", sampler=optuna.samplers.TPESampler(seed=SEED), ) print(" running 30 trials ...") study.optimize(objective, n_trials=30, show_progress_bar=False) print(f" done. best value = {study.best_value:.2f}") print(f" best params: {study.best_params}") print(f" best attrs : net={study.best_trial.user_attrs['net_profit']:.2f}, " f"PF={study.best_trial.user_attrs['profit_factor']:.2f}, " f"trades={study.best_trial.user_attrs['total_trades']}, " f"DD%={study.best_trial.user_attrs['max_equity_dd_pct']:.2%}") return 0 def run_full(study_db: Path, n_trials: int) -> int: """Full study, IS = 2025 (with 2-week purge). SQLite-persisted.""" print(f"=== Phase 6 FULL STUDY ({n_trials} trials) ===") m5 = load_m5() m1 = load_m1() bars_is = slice_window(m5, IS_START, IS_END) m1_is = slice_window(m1, IS_START, IS_END) print(f" IS window: {IS_START.date()} → {IS_END.date()} (exclusive)") print(f" IS bars : {len(bars_is):,} M1 bars: {len(m1_is):,}") print(f" warmup : {len(m5):,} full M5 bars (signals_full_bars)") cfg = make_objective_config(bars_is, m1_is, full_m5=m5) objective = build_objective(cfg) study_db.parent.mkdir(parents=True, exist_ok=True) storage = f"sqlite:///{study_db}" study = optuna.create_study( direction="maximize", sampler=optuna.samplers.TPESampler(seed=SEED), storage=storage, study_name="gold_scalper_pro_is2025", load_if_exists=True, ) n_existing = len([t for t in study.trials if t.state.name == "COMPLETE"]) if n_existing > 0: print(f" resumed existing study: {n_existing} complete trials so far") print(f" running {n_trials} trials (foreground; Ctrl+C to stop — study is saved) ...") study.optimize(objective, n_trials=n_trials, show_progress_bar=True) print(f"\n === best trial ===") print(f" value = {study.best_value:.2f}") print(f" params:") for k, v in study.best_params.items(): print(f" {k:24s} = {v}") a = study.best_trial.user_attrs print(f" metrics: net={a['net_profit']:.2f}, PF={a['profit_factor']:.2f}, " f"trades={a['total_trades']}, DD%={a['max_equity_dd_pct']:.2%}") if a.get("violations"): print(f" violations: {a['violations']}") # Diverse top-3 finalists (doc 06 §3). print(f"\n === diverse top-3 finalists ===") finalists = select_diverse_topn(study, n=3, ranges=SEARCH_SPACE) if not finalists: print(" no constraint-passing trials found.") return 1 for i, t in enumerate(finalists, 1): d = t.user_attrs print(f" finalist #{i}: trial #{t.number} value={t.value:.2f}") print(f" net={d['net_profit']:.2f}, PF={d['profit_factor']:.2f}, " f"trades={d['total_trades']}, DD%={d['max_equity_dd_pct']:.2%}") print(f" params: {t.params}") print(f"\n study DB: {study_db}") return 0 def load_m5() -> pd.DataFrame: from shared.data.loaders import load_bars p = PROJECT / "data" / "XAUUSD_M5_2024-06-26_2026-06-26.parquet" if not p.exists(): sys.exit(f"missing M5 data: {p}") return load_bars(p) def load_m1() -> pd.DataFrame: from shared.data.loaders import load_bars p = PROJECT / "data" / "XAUUSD_M1_2024-06-26_2026-06-26.parquet" if not p.exists(): sys.exit(f"missing M1 data: {p} — run scripts/download_xauusd_m1.py first") return load_bars(p) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--smoke", action="store_true", help="30 trials on IS H1 2025; validate the script end-to-end") ap.add_argument("--trials", type=int, default=500, help="trial budget for the full study (default 500)") ap.add_argument("--db", type=Path, default=PROJECT / "studies" / "optuna" / "gold_scalper_pro_is2025.db", help="SQLite path for the full study (resumable)") args = ap.parse_args() if args.smoke: return run_smoke() return run_full(args.db, args.trials) if __name__ == "__main__": raise SystemExit(main())