""" walk_forward_validation_v1.py ============================== Walk-forward validation of the best configuration from take_profit_exit_v1.py. Configuration to validate -------------------------- Entry : Fibonacci 0.786 pullback on 15M within 6h of signal Signal : lag_zscore_24h_v3 >= 2.75 (US-DE 2Y spread) Stop : 0.25% hard tail risk TP : 0.30% take profit Z-exit : |z-score| crosses ±1.5 opposite direction Hold : 52h maximum Session: London + NY (hours 7-16) Why Walk-Forward Validation Matters ------------------------------------- Method B (TP 0.30%) was discovered by testing 8 configurations on the full 22-year dataset. The risk is that 0.30% was the best TP on THIS data by chance rather than genuine edge. Walk-forward tests whether the finding holds on data the optimisation never saw. Method ------- Split the full dataset into: IN-SAMPLE (train): 2003 - 2018 (~15 years, 68%) OUT-OF-SAMPLE (test): 2019 - 2026 (~7 years, 32%) For the test period, run all TP levels (0.20% - 0.80%) independently. If 0.30% is the best on the in-sample AND performs well on out-of-sample, the finding is validated. If a different TP is best on OOS, the 0.30% finding was likely overfitted to the historical data. Additional rolling window test -------------------------------- Also runs a rolling 3-year window validation (12 windows) to check consistency across all market regimes including: 2003-2008: Pre-crisis, low vol 2008-2012: Crisis and recovery 2012-2016: QE era, low vol 2016-2020: Normalisation, Brexit, COVID 2020-2026: Post-COVID, hiking cycle Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\walk_forward_validation_v1.py Run from project root: python src/research/walk_forward_validation_v1.py """ import pandas as pd import numpy as np from pathlib import Path import sys BASE_PATH = Path(__file__).resolve().parents[2] SRC_PATH = BASE_PATH / "src" if str(SRC_PATH) not in sys.path: sys.path.append(str(SRC_PATH)) from research.combined_candidate_matrix_v1 import ( build_frozen_signals, find_entry_pullback, get_first_m15_idx_at_or_after, ) from ingestion.price_loader_15m import load_eurusd_15m from features.spot_lag_v3 import get_model_ready_spot_lag_v3 TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" TRADES_DIR.mkdir(parents=True, exist_ok=True) # ── Parameters ──────────────────────────────────────────────────────────────── THRESHOLD = 2.75 FIB = 0.786 HOLD_HOURS = 52 STOP = 0.0025 SPREAD_COST = 0.0001 ALLOWED_HOURS = set(range(7, 17)) ZSCORE_EXIT = 1.5 # Walk-forward split IS_END = "2018-12-31" # end of in-sample OOS_START = "2019-01-01" # start of out-of-sample # TP levels to test at each window TP_LEVELS = [None, 0.0020, 0.0025, 0.0030, 0.0035, 0.0040, 0.0050, 0.0060, 0.0075, 0.0100] # Rolling window size (years) ROLLING_WINDOW_YEARS = 3 YEARS = 22.0 # ── Trade simulator ─────────────────────────────────────────────────────────── def simulate_trade( m15 : pd.DataFrame, signal_df : pd.DataFrame, entry_time : pd.Timestamp, entry_price: float, signal : int, tp : float | None, z_thresh : float = ZSCORE_EXIT, hold_hours : int = HOLD_HOURS, stop : float = STOP, spread_cost: float = SPREAD_COST, ) -> dict | None: entry_idx = get_first_m15_idx_at_or_after(m15, entry_time) if entry_idx is None: return None hold_bars = hold_hours * 4 exit_idx = min(entry_idx + hold_bars, len(m15) - 1) path = m15.iloc[entry_idx : exit_idx + 1].copy() if path.empty: return None signal_window = signal_df[ (signal_df["datetime"] >= entry_time) & (signal_df["datetime"] <= path.iloc[-1]["datetime"]) ].copy() if signal == 1: full_mae = (path["low"].min() - entry_price) / entry_price full_mfe = (path["high"].max() - entry_price) / entry_price else: full_mae = -((path["high"].max() - entry_price) / entry_price) full_mfe = (entry_price - path["low"].min()) / entry_price exit_price = float(path.iloc[-1]["close"]) exit_reason = "time" stop_hit = False final_bar_i = len(path) - 1 for bar_i in range(len(path)): bar = path.iloc[bar_i] bar_time = bar["datetime"] # TP check if tp is not None: if signal == 1: tp_hit = (float(bar["high"]) - entry_price) / entry_price >= tp else: tp_hit = (entry_price - float(bar["low"])) / entry_price >= tp if tp_hit: exit_price = entry_price*(1+tp) if signal==1 else entry_price*(1-tp) exit_reason = "tp" final_bar_i = bar_i break # Stop check if signal == 1: adverse = (float(bar["low"]) - entry_price) / entry_price else: adverse = -((float(bar["high"]) - entry_price) / entry_price) if adverse <= -stop: exit_price = entry_price*(1-stop) if signal==1 else entry_price*(1+stop) exit_reason = "stop" stop_hit = True final_bar_i = bar_i break # Z-score exit z_bars = signal_window[signal_window["datetime"] <= bar_time] if not z_bars.empty: cz = float(z_bars.iloc[-1]["lag_zscore_24h_v3"]) if (signal == 1 and cz <= -z_thresh) or (signal == -1 and cz >= z_thresh): exit_price = float(bar["close"]) exit_reason = "zscore_reversal" final_bar_i = bar_i break raw_ret = (-stop if stop_hit else (exit_price - entry_price)/entry_price if signal == 1 else -(exit_price - entry_price)/entry_price) return { "entry_time" : entry_time, "exit_time" : path.iloc[final_bar_i]["datetime"], "signal" : signal, "entry_price": entry_price, "exit_price" : exit_price, "stop_hit" : stop_hit, "exit_reason": exit_reason, "mae" : full_mae, "mfe" : full_mfe, "return" : raw_ret - spread_cost, } # ── Run signals through a date-filtered period ──────────────────────────────── def run_period( signals : pd.DataFrame, m15 : pd.DataFrame, signal_df : pd.DataFrame, tp : float | None, date_start : str, date_end : str, ) -> pd.DataFrame: """Runs the trade simulation filtered to a specific date range.""" period_signals = signals[ (signals["datetime"] >= pd.Timestamp(date_start)) & (signals["datetime"] <= pd.Timestamp(date_end)) ].copy().reset_index(drop=True) if period_signals.empty: return pd.DataFrame() trades = [] last_exit_time = None for _, sig in period_signals.iterrows(): if last_exit_time is not None and sig["datetime"] < last_exit_time: continue entry = find_entry_pullback( m15=m15, signal_row=sig, fib=FIB, wait_hours=6 ) if entry is None: continue trade = simulate_trade( m15 =m15, signal_df =signal_df, entry_time =entry["entry_time"], entry_price=entry["entry_price"], signal =int(sig["signal"]), tp =tp, ) if trade is None: continue trade["zscore_abs"] = abs(float(sig["lag_zscore_24h_v3"])) trades.append(trade) last_exit_time = trade["exit_time"] if not trades: return pd.DataFrame() df = pd.DataFrame(trades) df["equity"] = (1 + df["return"]).cumprod() df["peak"] = df["equity"].cummax() df["drawdown"] = df["equity"] / df["peak"] - 1 df["win"] = (df["return"] > 0).astype(int) return df # ── Summarise a period result ───────────────────────────────────────────────── def summarise_period(df: pd.DataFrame, period_years: float) -> dict: if df.empty: return {k: np.nan for k in ["trades", "per_yr", "win_rate", "avg_return", "final_eq", "max_dd", "positive"]} n = len(df) return { "trades" : n, "per_yr" : n / period_years, "win_rate" : df["win"].mean() * 100, "avg_return": df["return"].mean(), "final_eq" : df["equity"].iloc[-1], "max_dd" : df["drawdown"].min() * 100, "positive" : int(df["return"].mean() > 0), } # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 70) print("WALK-FORWARD VALIDATION V1") print(f" In-sample : 2003 - {IS_END} (~15 years)") print(f" Out-of-sample: {OOS_START} - 2026 (~7 years)") print(f" TP levels tested: {[f'{t*100:.2f}%' if t else 'None' for t in TP_LEVELS]}") print("=" * 70) print("\nLoading data...") m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) signal_df = get_model_ready_spot_lag_v3().copy() signal_df["datetime"] = pd.to_datetime(signal_df["datetime"]) signal_df = signal_df.sort_values("datetime").reset_index(drop=True) signals = build_frozen_signals( threshold=THRESHOLD, allowed_hours=ALLOWED_HOURS ) print(f" 15M bars: {len(m15):,} | Signals: {len(signals)}") # ── Part 1: In-sample vs Out-of-sample comparison ───────────────────────── print(f"\n{'─'*70}") print("PART 1 — IN-SAMPLE vs OUT-OF-SAMPLE") print(f"{'─'*70}") print(f"\n Testing each TP level on both periods...") print(f" KEY TEST: Does TP 0.30% remain best on OOS data?\n") is_years = 15.0 oos_years = 7.0 is_rows = [] oos_rows = [] for tp in TP_LEVELS: tp_label = f"{tp*100:.2f}%" if tp else "No TP" df_is = run_period(signals, m15, signal_df, tp, "2003-01-01", IS_END) df_oos = run_period(signals, m15, signal_df, tp, OOS_START, "2026-12-31") s_is = summarise_period(df_is, is_years) s_oos = summarise_period(df_oos, oos_years) s_is["tp"] = tp_label s_oos["tp"] = tp_label is_rows.append(s_is) oos_rows.append(s_oos) print(f" TP {tp_label:<8}" f" IS: n={s_is['trades']:>4} WR={s_is['win_rate']:>5.1f}%" f" AvgRet={s_is['avg_return']:>9.6f}" f" FinalEq={s_is['final_eq']:>7.4f}" f" MaxDD={s_is['max_dd']:>7.2f}%") print(f" {'':8}" f" OOS: n={s_oos['trades']:>4} WR={s_oos['win_rate']:>5.1f}%" f" AvgRet={s_oos['avg_return']:>9.6f}" f" FinalEq={s_oos['final_eq']:>7.4f}" f" MaxDD={s_oos['max_dd']:>7.2f}%\n") # ── Find best IS and OOS ────────────────────────────────────────────────── is_df_results = pd.DataFrame(is_rows) oos_df_results = pd.DataFrame(oos_rows) best_is = is_df_results.loc[is_df_results["avg_return"].idxmax(), "tp"] best_oos = oos_df_results.loc[oos_df_results["avg_return"].idxmax(), "tp"] print(f"\n Best TP by avg return:") print(f" In-sample : {best_is}") print(f" Out-of-sample : {best_oos}") if best_is == best_oos: print(f"\n VALIDATED — Same TP is best on both periods.") print(f" The {best_is} finding is NOT an in-sample artefact.") elif best_oos in ["0.30%", "0.25%", "0.35%"]: print(f"\n PARTIALLY VALIDATED — OOS best is close to IS best.") print(f" The general principle (small TP) holds out-of-sample.") print(f" Minor calibration difference is acceptable.") else: print(f"\n WARNING — IS and OOS disagree significantly.") print(f" IS best: {best_is} | OOS best: {best_oos}") print(f" The 0.30% TP may be over-fitted. Use caution.") # ── OOS performance of validated config ─────────────────────────────────── print(f"\n{'─'*70}") print("PART 2 — OOS DETAILED RESULTS FOR KEY CONFIGS") print(f"{'─'*70}") key_tps = [None, 0.0025, 0.0030, 0.0035, 0.0040, 0.0050] print(f"\n {'TP':<10}{'N':>5}{'WR%':>7}{'AvgRet':>10}" f"{'FinalEq':>10}{'MaxDD%':>8}{'Pass?':>8}") print(f" {'─'*58}") for tp in key_tps: tp_label = f"{tp*100:.2f}%" if tp else "No TP" row = next((r for r in oos_rows if r["tp"] == tp_label), None) if row: is_valid = row["avg_return"] > 0 marker = " ✓ VALIDATED" if (tp == 0.0030 and is_valid) else \ " ✓" if is_valid else " ✗" print(f" {tp_label:<10}{row['trades']:>5}" f"{row['win_rate']:>7.1f}{row['avg_return']:>10.6f}" f"{row['final_eq']:>10.4f}{row['max_dd']:>8.2f}{marker}") # ── Part 3: Rolling 3-year windows ─────────────────────────────────────── print(f"\n{'─'*70}") print("PART 3 — ROLLING 3-YEAR WINDOW CONSISTENCY") print(f" TP=0.30% tested across all 3-year windows") print(f" A robust edge should be positive in most windows") print(f"{'─'*70}\n") windows = [] start_year = 2003 end_year = 2026 window_years = ROLLING_WINDOW_YEARS for yr in range(start_year, end_year - window_years + 1, 1): w_start = f"{yr}-01-01" w_end = f"{yr + window_years}-01-01" windows.append((w_start, w_end)) positive_windows = 0 total_windows = 0 print(f" {'Window':<24}{'Trades':>7}{'WR%':>7}{'AvgRet':>10}" f"{'FinalEq':>10}{'MaxDD%':>8}{'Edge?':>7}") print(f" {'─'*73}") for w_start, w_end in windows: df_w = run_period(signals, m15, signal_df, 0.0030, w_start, w_end) if df_w.empty or len(df_w) < 10: continue s = summarise_period(df_w, window_years) positive = s["avg_return"] > 0 positive_windows += int(positive) total_windows += 1 marker = "+" if positive else "-" print(f" {w_start[:4]}-{w_end[:4]:<20}" f"{s['trades']:>7}{s['win_rate']:>7.1f}" f"{s['avg_return']:>10.6f}{s['final_eq']:>10.4f}" f"{s['max_dd']:>8.2f} {marker}") print(f"\n Positive windows : {positive_windows}/{total_windows} " f"({positive_windows/total_windows*100:.1f}%)") print(f" Target : >= 70% of windows positive") if positive_windows / total_windows >= 0.70: print(f"\n ROLLING VALIDATION PASSED") print(f" The 0.30% TP configuration produces positive returns") print(f" in {positive_windows/total_windows*100:.0f}% of all 3-year windows.") print(f" This includes crisis periods, low-vol regimes, and") print(f" the 2022 hiking cycle. The edge is regime-robust.") else: print(f"\n ROLLING VALIDATION FAILED") print(f" Too many negative windows — the edge is regime-dependent.") # ── Final verdict ───────────────────────────────────────────────────────── print(f"\n{'='*70}") print("OVERALL VALIDATION VERDICT") print(f"{'='*70}") oos_030 = next((r for r in oos_rows if r["tp"] == "0.30%"), None) rolling_pass = positive_windows / total_windows >= 0.70 if total_windows > 0 else False oos_pass = oos_030 is not None and oos_030["avg_return"] > 0 checks = [ ("OOS avg return positive", oos_pass), ("OOS final equity > 1.0", oos_030 is not None and oos_030.get("final_eq", 0) > 1.0), ("Rolling windows >= 70% positive", rolling_pass), ("IS best TP matches OOS vicinity", best_oos in ["0.25%", "0.30%", "0.35%", "0.20%"]), ] passed = sum(1 for _, v in checks if v) print(f"\n Validation checks ({passed}/{len(checks)} passed):") for label, result in checks: print(f" {'✓' if result else '✗'} {label}") if passed >= 3: print(f""" VERDICT: VALIDATED The TP 0.30% + z-exit 1.5 configuration survives walk-forward testing. The edge is genuine and not solely an in-sample fitting artefact. Validated parameters to freeze: Stop : 0.25% TP : 0.30% Z-exit : 1.5 reversal threshold Hold : 52h maximum Next steps: 1. Run AUDUSD with identical parameters 2. Build 2-pair portfolio FTMO simulation 3. If portfolio validated, proceed to execution framework """) elif passed == 2: print(f""" VERDICT: PARTIALLY VALIDATED — USE WITH CAUTION The edge exists but shows some regime-dependence. Recommended: proceed but size conservatively (0.25% risk vs 0.30%). """) else: print(f""" VERDICT: NOT VALIDATED The 0.30% TP result does not hold out-of-sample. Do not proceed with this configuration. Review the OOS results above to identify a more robust TP level. """) # Save results pd.DataFrame(is_rows).to_csv( TRADES_DIR / "wf_is_results.csv", index=False) pd.DataFrame(oos_rows).to_csv( TRADES_DIR / "wf_oos_results.csv", index=False) print(f" Results saved to: {TRADES_DIR}") if __name__ == "__main__": main()