""" entry_audit_v1.py ================== Reconciles the discrepancy between: - Full backtest: 4,019 trades over 22 years (182/year, 3.5/week) - Entry window test: 1,607 trades (73/year, 1.4/week) Investigates: 1. How did the full backtest find entries? What was the actual entry logic? 2. What % of the 7,595 signals get filled at 0.786? 3. What happens to the 78.8% of signals that never reach 0.786? - Do they go on to be profitable (we are missing money)? - Or unprofitable (filter is working correctly)? 4. What fill rate is needed to achieve 3.5 trades/week? 5. What entry method achieves that fill rate with best win rate? This is a CRITICAL audit before deciding whether to change entry logic. Place in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\entry_audit_v1.py Run from project root: python src/research/entry_audit_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" 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 HOLD_HOURS = 52 ZSCORE_BANDS = [ (2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0), ] def get_multiplier(z): for lo, hi, mult in ZSCORE_BANDS: if lo <= z < hi: return mult return ZSCORE_BANDS[-1][2] def simulate_single_entry(m15, signal_df, entry_time, entry_price, signal_dir, zscore_abs): """Simulate a single trade from entry. Returns ret, exit_reason.""" entry_idx = get_first_m15_idx_at_or_after(m15, entry_time) if entry_idx is None: return None, None hold_bars = HOLD_HOURS * 4 exit_idx = min(entry_idx + hold_bars, len(m15) - 1) path = m15.iloc[entry_idx:exit_idx + 1] if path.empty: return None, None 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(): bt = 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"] <= bt] 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 return raw_ret - SPREAD_COST, exit_reason def main(): print("=" * 70) print("ENTRY METHOD AUDIT — Critical Investigation") print("=" * 70) print("\nLoading data...") m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) signals = build_frozen_signals(threshold=THRESHOLD, allowed_hours=ALLOWED_HOURS) 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 loading signal_df: {e}") signal_df = pd.DataFrame(columns=["datetime","lag_zscore_24h_v3"]) # Load the validated trade log to check what entries were used trade_log_path = TRADES_DIR / "trades_real_costs.csv" if not trade_log_path.exists(): trade_log_path = TRADES_DIR / "trades_eurusd_final.csv" validated_trades = pd.read_csv(trade_log_path, parse_dates=["entry_time","exit_time"]) print(f" Validated backtest trades: {len(validated_trades):,}") print(f" Total signals available : {len(signals):,}") print(f" Signals per year : {len(signals)/YEARS:.1f}") # ── PART 1: Check entry timing in validated backtest ───────────────────── print(f"\n{'─'*70}") print("PART 1: HOW DID THE FULL BACKTEST FIND ITS ENTRIES?") print(f"{'─'*70}") # Load the base (non-real-costs) trade log for entry analysis base_log = None for fname in ["trades_growth_52h.csv", "trades_eurusd_final.csv"]: p = TRADES_DIR / fname if p.exists(): base_log = pd.read_csv(p, parse_dates=["entry_time","exit_time"]) print(f"\n Using: {fname} ({len(base_log):,} trades)") break if base_log is not None: # Check if entry_time == signal bar time (immediate entry) # vs entry_time > signal bar time (waited for pullback) # Merge signals onto trades by nearest time signals_sorted = signals.sort_values("datetime") trades_sorted = base_log.sort_values("entry_time") # For each trade, find the most recent signal before entry lags = [] for _, trade in trades_sorted.head(500).iterrows(): prior = signals_sorted[signals_sorted["datetime"] <= trade["entry_time"]] if not prior.empty: last_sig = prior.iloc[-1] lag_h = (trade["entry_time"] - last_sig["datetime"]).total_seconds() / 3600 lags.append(lag_h) lags = [l for l in lags if 0 <= l <= 48] # filter noise if lags: print(f"\n Entry timing analysis (sample of 500 trades):") print(f" Avg lag signal→entry : {np.mean(lags):.2f} hours") print(f" Median lag : {np.median(lags):.2f} hours") print(f" < 1h lag : {sum(1 for l in lags if l < 1)/len(lags)*100:.1f}%") print(f" 1-6h lag : {sum(1 for l in lags if 1<=l<6)/len(lags)*100:.1f}%") print(f" > 6h lag : {sum(1 for l in lags if l >= 6)/len(lags)*100:.1f}%") print(f"\n If avg lag ≈ 0h: backtest entered at signal bar (not Fib pullback)") print(f" If avg lag > 0h: backtest genuinely waited for 0.786 pullback") # ── PART 2: What happens to unfilled signals? ───────────────────────────── print(f"\n{'─'*70}") print("PART 2: WHAT HAPPENS TO UNFILLED SIGNALS?") print(f" (signals where price never reaches 0.786 within 6h)") print(f"{'─'*70}") fib_filled = [] # signals that got a Fib entry fib_unfilled = [] # signals that never reached 0.786 last_exit_fib = None last_exit_mkt = None # Track market entry results for unfilled signals unfilled_mkt_results = [] print("\n Scanning all signals...", end="", flush=True) for _, sig in signals.iterrows(): sig_time = sig["datetime"] signal_dir = int(sig["signal"]) zscore_abs = abs(float(sig["lag_zscore_24h_v3"])) # Check Fib entry got_fib = False if last_exit_fib is None or sig_time >= last_exit_fib: entry = find_entry_pullback(m15=m15, signal_row=sig, fib=FIB, wait_hours=6) if entry: got_fib = True fib_filled.append({ "sig_time" : sig_time, "signal_dir": signal_dir, "zscore_abs": zscore_abs, }) last_exit_fib = entry["entry_time"] + pd.Timedelta(hours=HOLD_HOURS) if not got_fib: # For unfilled signals — simulate what WOULD have happened # if we entered at market at the signal bar's next 15M close if last_exit_mkt is None or sig_time >= last_exit_mkt: entry_idx = get_first_m15_idx_at_or_after(m15, sig_time) if entry_idx is not None and entry_idx + 1 < len(m15): next_bar = m15.iloc[entry_idx + 1] entry_price = float(next_bar["open"]) entry_time = next_bar["datetime"] ret, reason = simulate_single_entry( m15, signal_df, entry_time, entry_price, signal_dir, zscore_abs ) if ret is not None: unfilled_mkt_results.append({ "sig_time" : sig_time, "ret" : ret, "exit_reason": reason, "win" : int(ret > 0), "zscore_abs": zscore_abs, }) last_exit_mkt = entry_time + pd.Timedelta(hours=HOLD_HOURS) fib_unfilled.append(sig_time) else: fib_unfilled.append(sig_time) print(f" done.\n") n_filled = len(fib_filled) n_unfilled = len(fib_unfilled) n_total = n_filled + n_unfilled print(f" Total signals scanned : {n_total:,}") print(f" Got Fib entry (filled) : {n_filled:,} ({n_filled/n_total*100:.1f}%)") print(f" No Fib entry (unfilled) : {n_unfilled:,} ({n_unfilled/n_total*100:.1f}%)") if unfilled_mkt_results: df_u = pd.DataFrame(unfilled_mkt_results) wr = df_u["win"].mean() * 100 avg = df_u["ret"].mean() * 100 n_u = len(df_u) gross_p = df_u[df_u["ret"]>0]["ret"].sum() * BASE_NOTIONAL gross_l = abs(df_u[df_u["ret"]<0]["ret"].sum()) * BASE_NOTIONAL pf = gross_p / gross_l if gross_l > 0 else 0 print(f"\n UNFILLED SIGNALS — if traded at market:") print(f" Sample size : {n_u:,} trades simulated") print(f" Win rate : {wr:.1f}%") print(f" Avg return : {avg:.4f}%") print(f" Profit factor : {pf:.3f}") exits = df_u["exit_reason"].value_counts(normalize=True)*100 print(f" TP exits : {exits.get('tp',0):.1f}%") print(f" Stop exits : {exits.get('stop',0):.1f}%") print(f" Z-exits : {exits.get('zexit',0):.1f}%") print(f"\n COMPARE — Fib-filled signals (from full backtest):") print(f" Win rate : ~72.8%") print(f" Avg return : ~+0.068%") print(f" Profit factor : ~2.28") print(f"\n VERDICT:") if wr > 55 and avg > 0 and pf > 1.3: print(f" *** SIGNIFICANT FINDING ***") print(f" Unfilled signals have positive expectancy (WR={wr:.1f}%, PF={pf:.2f})") print(f" The 0.786 filter is DISCARDING profitable trades.") print(f" Consider: enter all signals at market OR widen entry to 0.618") print(f" Additional trades available: ~{n_u/YEARS:.0f}/year = ~{n_u/YEARS/52:.1f}/week") elif wr < 50 or avg < 0 or pf < 1.0: print(f" Unfilled signals are LOSING trades (WR={wr:.1f}%, PF={pf:.2f})") print(f" The 0.786 filter is correctly EXCLUDING poor quality entries.") print(f" 6h window and 0.786 level are well-calibrated.") else: print(f" Unfilled signals are borderline (WR={wr:.1f}%, PF={pf:.2f})") print(f" May be worth testing a 0.618 entry as a compromise level.") # ── PART 3: Realistic trade frequency ──────────────────────────────────── print(f"\n{'─'*70}") print("PART 3: REALISTIC LIVE TRADE FREQUENCY") print(f"{'─'*70}") fib_per_year = n_filled / YEARS fib_per_week = fib_per_year / 52 print(f""" Fib-filled trades per year : {fib_per_year:.0f} Fib-filled trades per week : {fib_per_week:.1f} Backtest reported 4,019 trades / 22 years = {4019/YEARS:.0f}/year = {4019/YEARS/52:.1f}/week Entry window test found {n_filled:,} trades / 22 years = {fib_per_year:.0f}/year = {fib_per_week:.1f}/week GAP: {4019-n_filled:,} trades ({(4019-n_filled)/4019*100:.0f}% of backtest total) Possible explanations for the gap: 1. Full backtest entered at hourly bar close (not strict 0.786 on 15M) 2. find_entry_pullback has stricter logic than backtest used 3. Some signals in backtest didn't require Fib pullback (e.g. immediate entry) IMPLICATION: If live system gets {fib_per_week:.1f} trades/week, FTMO Phase 1 timeline extends: At {fib_per_week:.1f} trades/week with {72.8:.0f}% WR and {0.068:.4f}% avg ret: Monthly PnL estimate = {fib_per_week*52/12 * 0.00068 * BASE_NOTIONAL:,.0f} Months to Phase 1 (10k) = {10000 / (fib_per_week*52/12 * 0.00068 * BASE_NOTIONAL):.1f} months """) # Save results if unfilled_mkt_results: pd.DataFrame(unfilled_mkt_results).to_csv( TRADES_DIR / "unfilled_signal_results.csv", index=False ) print(f" Saved: unfilled_signal_results.csv") print(f"\n{'='*70}") print("BOTTOM LINE") print(f"{'='*70}") print(f""" The 21% fill rate means {fib_per_week:.1f} live trades/week not 3.5. The critical question is whether the unfilled 79% are: A) Profitable trades being filtered out (fix: widen or remove Fib filter) B) Losing trades being correctly excluded (keep current filter) The win rate and profit factor of unfilled signals above answers this. If A: reconsider entry method entirely. If B: accept lower trade frequency — quality over quantity. """) if __name__ == "__main__": main()