r""" trade_by_trade_diagnostic.py ============================= For each EU trade the backtest would have taken in Sep 2025 - Mar 2026, diagnose what the LIVE monitor would have done at that exact moment. For each backtest trade, we classify it into one of these buckets: 1. WOULD_MATCH: Live would have taken this trade too 2. SESSION_FILTER: Signal time outside live monitor's session window 3. POSITION_OPEN: Live monitor already had a position open 4. ARMED_ACTIVE: Live monitor was already "armed" for a previous signal 5. ZSCORE_MISMATCH: Live's z-score below threshold while backtest's was above 6. FIB_NOT_HIT_LIVE: Live's Fib target (from latest 15M bar) wouldn't have filled 7. UNCLEAR: Doesn't cleanly fit any bucket This gives concrete, per-trade attribution of where the gap comes from. Run after live_vs_backtest_30d.py has been validated. cd C:\Users\paul_\OneDrive\fx_macro_intraday python src\research\trade_by_trade_diagnostic.py """ import sys import warnings from pathlib import Path from datetime import datetime, timedelta import pandas as pd import numpy as np warnings.filterwarnings("ignore") BASE = Path(__file__).resolve().parents[2] SRC = BASE / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) 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 # ── Parameters (match live monitor and backtest) ────────────────────────────── EU_THRESHOLD = 2.75 EU_FIB = 0.786 EU_HOLD_HOURS = 52 EU_STOP = 0.0025 EU_TP = 0.0020 EU_ALLOWED_HOURS = set(range(7, 17)) # backtest session UTC 7-16 # Live monitor session: (UTC + 2) % 24 between 7 and 17 inclusive # At UTC h: live considers it in-session if (h+2)%24 in [7,17] # → UTC 5..15 inclusive def live_in_session(utc_hour: int) -> bool: eet_hour = (utc_hour + 2) % 24 return 7 <= eet_hour <= 17 # ── Period to analyse ───────────────────────────────────────────────────────── DEFAULT_START = "2025-09-01" DEFAULT_END = "2026-03-13" # ────────────────────────────────────────────────────────────────────────────── # Replicate the BACKTEST entry behavior (same code as combined_candidate_matrix) # ────────────────────────────────────────────────────────────────────────────── def backtest_simulate(start_date: pd.Timestamp, end_date: pd.Timestamp): """Returns list of trades the backtest WOULD take in the window.""" signals = build_frozen_signals( threshold=EU_THRESHOLD, allowed_hours=EU_ALLOWED_HOURS, ) signals["datetime"] = pd.to_datetime(signals["datetime"]) recent = signals[ (signals["datetime"] >= start_date) & (signals["datetime"] <= end_date) ].copy().reset_index(drop=True) m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) trades = [] last_exit_time = None skipped_position_open = [] for _, sig in recent.iterrows(): if last_exit_time is not None and sig["datetime"] < last_exit_time: # backtest skipped this signal because previous position still open skipped_position_open.append(sig["datetime"]) continue entry = find_entry_pullback( m15=m15, signal_row=sig, fib=EU_FIB, wait_hours=6 ) if entry is None: continue entry_idx = get_first_m15_idx_at_or_after(m15, entry["entry_time"]) if entry_idx is None: continue hold_bars = EU_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: continue entry_price = entry["entry_price"] signal = int(sig["signal"]) exit_price = float(path.iloc[-1]["close"]) exit_time = path.iloc[-1]["datetime"] exit_reason = "time" for _, bar in path.iterrows(): if signal == 1: tp_hit = (float(bar["high"]) - entry_price) / entry_price >= EU_TP stop_hit = (float(bar["low"]) - entry_price) / entry_price <= -EU_STOP else: tp_hit = (entry_price - float(bar["low"])) / entry_price >= EU_TP stop_hit = -((float(bar["high"]) - entry_price) / entry_price) <= -EU_STOP if tp_hit: exit_price = entry_price*(1+EU_TP) if signal==1 else entry_price*(1-EU_TP) exit_time = bar["datetime"] exit_reason = "tp" break if stop_hit: exit_price = entry_price*(1-EU_STOP) if signal==1 else entry_price*(1+EU_STOP) exit_time = bar["datetime"] exit_reason = "stop" break last_exit_time = exit_time trades.append({ "signal_time": sig["datetime"], "signal_close": float(sig["close"]), "signal_high": float(sig["high"]), "signal_low": float(sig["low"]), "entry_time": entry["entry_time"], "entry_price": entry_price, "exit_time": exit_time, "exit_price": exit_price, "exit_reason": exit_reason, "signal": signal, "zscore_bt": float(sig["lag_zscore_24h_v3"]), }) return trades, m15, recent # ────────────────────────────────────────────────────────────────────────────── # Replicate the LIVE monitor's Fib target computation # Live uses the LATEST 15M bar at signal-check time (NOT the signal H1 bar) # ────────────────────────────────────────────────────────────────────────────── def live_fib_target_at(signal_time: pd.Timestamp, signal: int, m15: pd.DataFrame): """Compute the Fib target that the LIVE monitor would have set at signal_time. Live monitor logic at signal-check time (typically signal_time + 1 minute): bars_15m = last 10 bars last_bar = bars_15m.iloc[-1] pull = last_bar.close - last_bar.low (LONG) or last_bar.high - last_bar.close (SHORT) target = last_bar.close - fib*pull (LONG) or last_bar.close + fib*pull (SHORT) The "latest 15M bar at signal-check time" is the LAST CLOSED 15M bar when the live monitor wakes up at signal_time+1min. That's the bar ending just before signal_time (or at signal_time if signal_time is on a 15M boundary). Since signal_time is on the hour, the latest closed 15M bar would be the one ending at signal_time itself (which spans signal_time - 15min to signal_time, with closing time = signal_time). Note: signal_time is the H1 bar close time. The 15M bar ending at signal_time is the natural "latest closed 15M bar". """ # Find the 15M bar ending at signal_time (or the closest one before) idx = m15["datetime"].searchsorted(signal_time, side="left") # If exact match, that's the bar ending at signal_time (the one we want) # If not, take the bar just before if idx >= len(m15): return None if m15.iloc[idx]["datetime"] != signal_time: idx -= 1 if idx < 0: return None last_bar = m15.iloc[idx] bh = float(last_bar["high"]) bl = float(last_bar["low"]) bc = float(last_bar["close"]) if signal == 1: pull = bc - bl if pull <= 0.0001: return None return bc - EU_FIB * pull else: pull = bh - bc if pull <= 0.0001: return None return bc + EU_FIB * pull def live_target_hit_within_window( target: float, signal: int, signal_time: pd.Timestamp, m15: pd.DataFrame, wait_hours: int = 6, ) -> tuple[bool, pd.Timestamp]: """Did price hit the live monitor's target within wait_hours?""" idx = m15["datetime"].searchsorted(signal_time, side="left") end_time = signal_time + pd.Timedelta(hours=wait_hours) end_idx = m15["datetime"].searchsorted(end_time, side="left") window = m15.iloc[idx:end_idx] if window.empty: return False, None if signal == 1: hit = window[window["low"] <= target] else: hit = window[window["high"] >= target] if hit.empty: return False, None return True, hit.iloc[0]["datetime"] # ────────────────────────────────────────────────────────────────────────────── # Diagnose each backtest trade # ────────────────────────────────────────────────────────────────────────────── def diagnose(trades: list[dict], m15: pd.DataFrame, all_signals: pd.DataFrame): """For each backtest trade, classify what live would have done.""" # Simulate the live monitor's position-open state through the window. # Live takes a trade if: # - signal_time in live session # - position not currently open from a previous trade # - not already armed for a previous signal episode # - fib target gets hit in next 6h # Walk through trades in order. Track live's position open/closed state. live_position_until = None # exit time of live's currently-open position results = [] for t in trades: sig_time = pd.Timestamp(t["signal_time"]) signal = t["signal"] utc_hour = sig_time.hour # CHECK 1: Session filter in_live_session = live_in_session(utc_hour) # CHECK 2: Position open in live? position_blocked = (live_position_until is not None and sig_time < live_position_until) # CHECK 3: Live's Fib target — would it have been hit? live_target = live_fib_target_at(sig_time, signal, m15) if live_target is not None: target_hit, hit_time = live_target_hit_within_window( live_target, signal, sig_time, m15, wait_hours=6 ) else: target_hit, hit_time = False, None # Classify if not in_live_session: verdict = "SESSION_FILTER" elif position_blocked: verdict = "POSITION_OPEN" elif not target_hit: verdict = "FIB_NOT_HIT_LIVE" else: verdict = "WOULD_MATCH" # Simulate live taking this trade — apply same outcome as backtest # (approximation: we use backtest exit time as live exit time since # the actual hold logic is identical if entry happens) # Note: live's entry price differs from backtest's, but we're # primarily concerned with WHETHER trade fires, not exact P&L. live_position_until = pd.Timestamp(t["exit_time"]) # Compare backtest fib target vs live fib target if signal == 1: bt_pull = t["signal_close"] - t["signal_low"] bt_target = t["signal_close"] - EU_FIB * bt_pull else: bt_pull = t["signal_high"] - t["signal_close"] bt_target = t["signal_close"] + EU_FIB * bt_pull results.append({ "signal_time": sig_time, "signal": "L" if signal == 1 else "S", "z_bt": t["zscore_bt"], "bt_target": bt_target, "live_target": live_target, "target_hit_live": target_hit, "in_live_session": in_live_session, "position_blocked": position_blocked, "verdict": verdict, }) return results # ────────────────────────────────────────────────────────────────────────────── # Main # ────────────────────────────────────────────────────────────────────────────── def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument("--start", default=DEFAULT_START) parser.add_argument("--end", default=DEFAULT_END) args = parser.parse_args() start_date = pd.Timestamp(args.start) end_date = pd.Timestamp(args.end) print("=" * 80) print(" TRADE-BY-TRADE DIAGNOSTIC") print("=" * 80) print(f" Period: {start_date.date()} → {end_date.date()}") print(f" For each EU trade the backtest would have taken, diagnose what live would do.") print() print(" Running backtest simulation...") trades, m15, all_signals = backtest_simulate(start_date, end_date) print(f" Backtest found {len(trades)} EU trades") print() print(" Diagnosing live behavior for each trade...") results = diagnose(trades, m15, all_signals) # Display per-trade table print() print("=" * 80) print(" PER-TRADE DIAGNOSIS") print("=" * 80) print(f" {'Signal time':<17} {'Sig':>3} {'Z':>7} {'BT tgt':>9} {'Live tgt':>9} " f"{'Hit?':>5} {'Sess':>5} {'PosOp':>5} {'Verdict':>17}") print(" " + "─" * 78) for r in results: live_tgt_str = f"{r['live_target']:.5f}" if r['live_target'] else "N/A" print(f" {str(r['signal_time'])[:16]:<17} " f"{r['signal']:>3} " f"{r['z_bt']:>+7.3f} " f"{r['bt_target']:>9.5f} " f"{live_tgt_str:>9} " f"{('YES' if r['target_hit_live'] else 'NO'):>5} " f"{('IN' if r['in_live_session'] else 'OUT'):>5} " f"{('BLOCK' if r['position_blocked'] else 'OK'):>5} " f"{r['verdict']:>17}") # Summary print() print("=" * 80) print(" ATTRIBUTION SUMMARY") print("=" * 80) from collections import Counter verdicts = Counter(r["verdict"] for r in results) print(f"\n Total backtest trades: {len(results)}") print(f" Live would match: {verdicts['WOULD_MATCH']}") print() print(f" Reasons live would have missed:") print(f" Session filter: {verdicts['SESSION_FILTER']}") print(f" Position already open: {verdicts['POSITION_OPEN']}") print(f" Fib target not hit: {verdicts['FIB_NOT_HIT_LIVE']}") # Look more closely at FIB_NOT_HIT cases — show the target difference fib_misses = [r for r in results if r["verdict"] == "FIB_NOT_HIT_LIVE"] if fib_misses: print(f"\n Detailed look at FIB_NOT_HIT cases ({len(fib_misses)} trades):") print(f" Backtest's target uses the H1 signal bar's H/L/C") print(f" Live's target uses the last 15M bar's H/L/C") print() print(f" {'Signal time':<17} {'Sig':>3} {'BT target':>10} {'Live target':>12} {'Diff(pips)':>11}") print(" " + "─" * 60) for r in fib_misses[:30]: if r["live_target"] is None: continue diff_pips = (r["live_target"] - r["bt_target"]) * 10000 # convert to pips print(f" {str(r['signal_time'])[:16]:<17} " f"{r['signal']:>3} " f"{r['bt_target']:>10.5f} " f"{r['live_target']:>12.5f} " f"{diff_pips:>+11.1f}") print() print("=" * 80) print(" WHAT THIS TELLS US") print("=" * 80) pct_match = 100 * verdicts['WOULD_MATCH'] / len(results) if results else 0 pct_session = 100 * verdicts['SESSION_FILTER'] / len(results) if results else 0 pct_pos = 100 * verdicts['POSITION_OPEN'] / len(results) if results else 0 pct_fib = 100 * verdicts['FIB_NOT_HIT_LIVE'] / len(results) if results else 0 print(f" Trade-loss attribution:") print(f" {pct_match:.0f}% would fire correctly in live") print(f" {pct_session:.0f}% lost to session filter mismatch") print(f" {pct_pos:.0f}% lost to position-already-open (live takes earlier signals)") print(f" {pct_fib:.0f}% lost to Fib target calculation mismatch") print() print(f" If we set live's session to match backtest (UTC 7-16) and use H1-bar") print(f" Fib targets, the live monitor would produce approximately:") estimated_live = verdicts['WOULD_MATCH'] + verdicts['POSITION_OPEN'] print(f" {estimated_live} trades over this period ({estimated_live / 6.3:.1f}/month)") print(f" vs current actual live rate: ~1/month") if __name__ == "__main__": main()