""" entry_window_test_v1.py ======================== Tests different Fibonacci entry window durations to find the optimal wait time for the 0.786 pullback after a signal fires. Tests: 1h, 2h, 3h, 4h, 6h (current), 8h, 12h, 24h For each window measures: - Fill rate (% of signals that get an entry) - Win rate on filled trades - Avg return per filled trade - Sharpe ratio - Profit factor - Avg entry quality (how close to ideal price) Key question: is 6h the optimal balance between: - Too short (missing valid pullbacks = low fill rate) - Too long (entering on stale signals = lower win rate) Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\entry_window_test_v1.py Run from project root: python src/research/entry_window_test_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 TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" CHARTS_DIR = BASE_PATH / "data" / "processed" THRESHOLD = 2.75 FIB = 0.786 STOP = 0.0025 TP = 0.0020 ZSCORE_EXIT = 1.5 SPREAD_COST = 0.0001 ALLOWED_HOURS = set(range(7, 17)) ACCOUNT_START = 100_000.0 BASE_NOTIONAL = 300_000.0 YEARS = 22.0 RISK_FREE = 0.04 ZSCORE_BANDS = [ (2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0), ] WINDOWS_TO_TEST = [1, 2, 3, 4, 6, 8, 12, 24] def get_multiplier(z): for lo, hi, mult in ZSCORE_BANDS: if lo <= z < hi: return mult return ZSCORE_BANDS[-1][2] def simulate_trades_for_window(signals, m15, signal_df, wait_hours): """ Runs the full backtest with a given entry window duration. Returns a list of trade results. """ trades = [] last_exit = None hold_bars = 52 * 4 # 52 hours in 15M bars for _, sig in signals.iterrows(): sig_time = sig["datetime"] # Skip if still in a trade if last_exit is not None and sig_time < last_exit: continue # Find Fibonacci entry within wait_hours window entry = find_entry_pullback( m15=m15, signal_row=sig, fib=FIB, wait_hours=wait_hours, ) if entry is None: continue entry_time = entry["entry_time"] entry_price = entry["entry_price"] signal_dir = int(sig["signal"]) zscore_abs = abs(float(sig["lag_zscore_24h_v3"])) multiplier = get_multiplier(zscore_abs) notional = min(BASE_NOTIONAL * multiplier, BASE_NOTIONAL * 3) # Find entry bar in 15M entry_idx = get_first_m15_idx_at_or_after(m15, entry_time) if entry_idx is None: continue exit_idx = min(entry_idx + hold_bars, len(m15) - 1) path = m15.iloc[entry_idx:exit_idx + 1] if path.empty: continue # Get z-score series for z-exit check sig_window = signal_df[ (signal_df["datetime"] >= entry_time) & (signal_df["datetime"] <= path.iloc[-1]["datetime"]) ] exit_price = float(path.iloc[-1]["close"]) exit_reason = "time" for _, bar in path.iterrows(): bar_time = bar["datetime"] if signal_dir == 1: tp_hit = (float(bar["high"]) - entry_price) / entry_price >= TP stop_hit = (float(bar["low"]) - entry_price) / entry_price <= -STOP else: tp_hit = (entry_price - float(bar["low"])) / entry_price >= TP stop_hit = (entry_price - float(bar["high"])) / entry_price <= -STOP if tp_hit: exit_price = entry_price * (1 + TP) if signal_dir == 1 else entry_price * (1 - TP) exit_reason = "tp" break if stop_hit: exit_price = entry_price * (1 - STOP) if signal_dir == 1 else entry_price * (1 + STOP) exit_reason = "stop" break z_bars = sig_window[sig_window["datetime"] <= bar_time] if not z_bars.empty: cz = float(z_bars.iloc[-1]["lag_zscore_24h_v3"]) if (signal_dir == 1 and cz <= -ZSCORE_EXIT) or \ (signal_dir == -1 and cz >= ZSCORE_EXIT): exit_price = float(bar["close"]) exit_reason = "zexit" break raw_ret = (exit_price - entry_price) / entry_price * signal_dir net_ret = raw_ret - SPREAD_COST trades.append({ "entry_time" : entry_time, "exit_reason" : exit_reason, "ret" : net_ret, "notional" : notional, "dollar_pnl" : net_ret * notional, "win" : int(net_ret > 0), "zscore_abs" : zscore_abs, "wait_hours" : wait_hours, "signal_time" : sig_time, "entry_lag_h" : (entry_time - sig_time).total_seconds() / 3600, }) last_exit = entry_time + pd.Timedelta(hours=52) return trades def calc_metrics(trades, wait_hours, total_signals): if not trades: return None df = pd.DataFrame(trades) returns = df["ret"].values dollar_pnl = df["dollar_pnl"].values n = len(df) fill_rate = n / total_signals * 100 equity = ACCOUNT_START + dollar_pnl.cumsum() peak = np.maximum.accumulate(equity) dd = (equity - peak) / peak * 100 per_yr = n / YEARS rf_pt = (1 + RISK_FREE) ** (1/per_yr) - 1 pct_ret = dollar_pnl / ACCOUNT_START excess = pct_ret - rf_pt sharpe = (excess.mean() / excess.std() * np.sqrt(per_yr) if excess.std() > 0 else 0) gross_p = dollar_pnl[dollar_pnl > 0].sum() gross_l = abs(dollar_pnl[dollar_pnl < 0].sum()) pf = gross_p / gross_l if gross_l > 0 else 999 exit_counts = df["exit_reason"].value_counts(normalize=True) * 100 avg_lag = df["entry_lag_h"].mean() return { "wait_hours" : wait_hours, "n_trades" : n, "fill_rate" : fill_rate, "win_rate" : df["win"].mean() * 100, "avg_ret" : returns.mean() * 100, "sharpe" : sharpe, "pf" : pf, "max_dd" : dd.min(), "net_pnl" : dollar_pnl.sum(), "tp_pct" : exit_counts.get("tp", 0), "stop_pct" : exit_counts.get("stop", 0), "zexit_pct" : exit_counts.get("zexit", 0), "time_pct" : exit_counts.get("time", 0), "avg_lag_h" : avg_lag, "cagr" : (equity[-1] / ACCOUNT_START) ** (1/YEARS) * 100 - 100, } def main(): print("=" * 75) print("ENTRY WINDOW SENSITIVITY TEST") print(f" Testing Fibonacci pullback wait times: {WINDOWS_TO_TEST} hours") print(f" Current validated value: 6h") print("=" * 75) print("\nLoading data...") m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) signals = build_frozen_signals( threshold=THRESHOLD, allowed_hours=ALLOWED_HOURS ) # Load signal df for z-exit checks try: from features.spot_lag_v3 import get_model_ready_spot_lag_v3 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) except Exception as e: print(f" Warning: could not load signal_df for z-exit: {e}") signal_df = pd.DataFrame(columns=["datetime", "lag_zscore_24h_v3"]) total_signals = len(signals) print(f" {total_signals:,} signals to test across {len(WINDOWS_TO_TEST)} windows") print(f" This will take a few minutes...\n") results = [] for wait_h in WINDOWS_TO_TEST: print(f" Testing {wait_h}h window...", end="", flush=True) trades = simulate_trades_for_window(signals, m15, signal_df, wait_h) metrics = calc_metrics(trades, wait_h, total_signals) if metrics: results.append(metrics) print(f" {len(trades):>4} trades WR={metrics['win_rate']:.1f}% " f"Sharpe={metrics['sharpe']:.2f} Fill={metrics['fill_rate']:.1f}%") else: print(" No trades") if not results: print("ERROR: No results generated") return # ── Print summary table ─────────────────────────────────────────────────── print(f"\n{'='*90}") print("RESULTS — ENTRY WINDOW SENSITIVITY") print(f"{'='*90}") print(f"\n {'Window':<10}{'Trades':>7}{'Fill%':>7}{'WR%':>7}{'Sharpe':>8}" f"{'PF':>7}{'CAGR%':>8}{'MaxDD%':>8}{'NetPnL':>10}" f"{'TP%':>6}{'Stop%':>7}{'AvgLag':>8}") print(f" {'─'*90}") best_sharpe = max(r["sharpe"] for r in results) for r in results: flag = " <- CURRENT" if r["wait_hours"] == 6 else "" flag = " <- BEST" if r["sharpe"] == best_sharpe and r["wait_hours"] != 6 else flag flag = " <- CURRENT + BEST" if r["sharpe"] == best_sharpe and r["wait_hours"] == 6 else flag print(f" {str(r['wait_hours'])+'h':<10}{r['n_trades']:>7}" f"{r['fill_rate']:>7.1f}{r['win_rate']:>7.1f}" f"{r['sharpe']:>8.2f}{r['pf']:>7.2f}" f"{r['cagr']:>8.2f}{r['max_dd']:>8.2f}" f" {r['net_pnl']:>8,.0f}" f"{r['tp_pct']:>6.1f}{r['stop_pct']:>7.1f}" f"{r['avg_lag_h']:>8.2f}{flag}") print(f"\n Key: Fill% = % of signals that get a trade entry") print(f" AvgLag = avg hours between signal and actual entry") # ── Insight ─────────────────────────────────────────────────────────────── best = max(results, key=lambda r: r["sharpe"]) current = next(r for r in results if r["wait_hours"] == 6) short_1h = next((r for r in results if r["wait_hours"] == 1), None) print(f"\n{'='*75}") print("ANALYSIS") print(f"{'='*75}") print(f""" Current window (6h): Fill rate : {current['fill_rate']:.1f}% Win rate : {current['win_rate']:.1f}% Sharpe : {current['sharpe']:.2f} Avg lag : {current['avg_lag_h']:.2f}h from signal to entry Best window ({best['wait_hours']}h): Fill rate : {best['fill_rate']:.1f}% Win rate : {best['win_rate']:.1f}% Sharpe : {best['sharpe']:.2f} Avg lag : {best['avg_lag_h']:.2f}h from signal to entry Sharpe vs 6h: {(best['sharpe']-current['sharpe'])/current['sharpe']*100:+.1f}% """) if short_1h: print(f" 1h window (tightest): " f"Fill={short_1h['fill_rate']:.1f}% " f"WR={short_1h['win_rate']:.1f}% " f"Sharpe={short_1h['sharpe']:.2f}") print(f" Fill rate gain 1h->6h: " f"{current['fill_rate']-short_1h['fill_rate']:+.1f}% " f"(signals caught by waiting longer)") print(f" Win rate cost 1h->6h: " f"{current['win_rate']-short_1h['win_rate']:+.1f}% " f"(quality dilution from longer wait)") print(f""" INTERPRETATION: - Shorter windows = fewer entries but higher quality (tighter to signal) - Longer windows = more entries but risk of stale signal - The optimal window balances fill rate vs signal freshness - If best window is close to 6h: current setting is well-calibrated - If best window is much shorter: consider tightening the entry window - If best window is much longer: consider widening it """) # Save results csv_path = TRADES_DIR / "entry_window_results.csv" pd.DataFrame(results).to_csv(csv_path, index=False) print(f" Results saved: {csv_path}") if __name__ == "__main__": main()