r""" live_vs_backtest_30d.py ======================== Focused empirical test: runs the EXACT backtest signal-firing logic on the last 30 days of data and compares against live monitor's actual trade history. Goal: determine concretely which trades the backtest WOULD have taken that the live monitor missed (and vice versa, if any). No live changes — just investigative output. For EURUSD: Uses combined_candidate_matrix_v1.build_frozen_signals() — the exact function that produced the 4,019-trade backtest. Then iterates signals applying the same find_entry_pullback() + position cooldown logic as real_world_costs_v1.py. For USDJPY: Mirrors the loop in usdjpy_real_costs_v1.py — hourly check on daily-level z-score with session filter. Run on home machine: cd C:\Users\paul_\OneDrive\fx_macro_intraday python src\research\live_vs_backtest_30d.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 # ── Parameters (must match backtest) ────────────────────────────────────────── EU_THRESHOLD = 2.75 EU_FIB = 0.786 EU_HOLD_HOURS = 52 EU_STOP = 0.0025 EU_TP = 0.0020 EU_ZSCORE_EXIT = 1.5 EU_ALLOWED_HOURS = set(range(7, 17)) # backtest's session = UTC 7-16 UJ_THRESHOLD = 2.00 UJ_FIB = 0.786 UJ_HOLD_HOURS = 24 UJ_TP = 0.0070 UJ_SL = 0.0040 UJ_Z_WINDOW = 30 UJ_SESSION_LO = 7 UJ_SESSION_HI = 17 # Anchor on a fixed period where backtest data is fully populated # Default: Sep 2025 - Mar 2026 (~7 months pre-live, same macro regime as now) # Override with --start YYYY-MM-DD --end YYYY-MM-DD if you want a different window DEFAULT_START = "2025-09-01" DEFAULT_END = "2026-03-13" # known backtest data cutoff TRADES_DIR = BASE / "data" / "processed" / "trades" LOG_FILE = BASE / "data" / "logs" / "live_monitor_v2.log" RATES_DIR = BASE / "data" / "raw" / "rates" # ────────────────────────────────────────────────────────────────────────────── # EURUSD: replicate the exact backtest loop, but on the last 30 days only # ────────────────────────────────────────────────────────────────────────────── def run_eurusd_backtest_recent(start_date: pd.Timestamp, end_date: pd.Timestamp): """Run the EURUSD backtest signal-firing logic between start and end.""" print(f"\n{'─' * 70}") print(f" EURUSD backtest simulation: {start_date.date()} → {end_date.date()}") print(f"{'─' * 70}") # Build signals (same as backtest) signals = build_frozen_signals( threshold=EU_THRESHOLD, allowed_hours=EU_ALLOWED_HOURS, ) print(f" Total signal bars in full history: {len(signals)}") # Filter to recent period signals["datetime"] = pd.to_datetime(signals["datetime"]) recent_signals = signals[ (signals["datetime"] >= start_date) & (signals["datetime"] <= end_date) ].copy() print(f" Signal bars in window: {len(recent_signals)}") if len(recent_signals) == 0: print(" No signal bars in this period — nothing to simulate.") return [] # Load 15M bars for entry simulation m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) # Iterate signals, attempt entries trades = [] last_exit_time = None for _, sig in recent_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=EU_FIB, wait_hours=6 ) if entry is None: continue # Simulate trade outcome 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 ret = ((exit_price - entry_price) / entry_price) * signal trades.append({ "signal_time": sig["datetime"], "entry_time": entry["entry_time"], "exit_time": exit_time, "signal": signal, "zscore": float(sig["lag_zscore_24h_v3"]), "entry_price": entry_price, "exit_price": exit_price, "exit_reason": exit_reason, "return": ret, }) print(f" Backtest would have taken {len(trades)} trade(s) in this window") return trades # ────────────────────────────────────────────────────────────────────────────── # USDJPY: replicate usdjpy_real_costs_v1.py loop on last 30 days # ────────────────────────────────────────────────────────────────────────────── def run_usdjpy_backtest_recent(start_date: pd.Timestamp, end_date: pd.Timestamp): """Run USDJPY backtest signal-firing logic between start and end.""" print(f"\n{'─' * 70}") print(f" USDJPY backtest simulation: {start_date.date()} → {end_date.date()}") print(f"{'─' * 70}") # Load rates and compute daily z-score (matches usdjpy_real_costs_v1.py) us2y = pd.read_csv(RATES_DIR / "us2y.csv") us2y.columns = [c.lower() for c in us2y.columns] us2y["date"] = pd.to_datetime(us2y["date"]) us2y["us2y"] = pd.to_numeric(us2y["us2y"], errors="coerce") us2y = us2y[["date", "us2y"]].dropna() jp2y = pd.read_csv(RATES_DIR / "jp2y.csv") jp2y.columns = [c.lower() for c in jp2y.columns] jp2y["date"] = pd.to_datetime(jp2y["date"]) jp2y["jp2y"] = pd.to_numeric(jp2y["jp2y"], errors="coerce") jp2y = jp2y[["date", "jp2y"]].dropna() rates = pd.merge(us2y, jp2y, on="date", how="outer").sort_values("date") rates = rates.set_index("date") rates = rates.reindex( pd.date_range(rates.index.min(), rates.index.max(), freq="D") ).ffill().dropna() rates["spread"] = rates["us2y"] - rates["jp2y"] z_daily = (rates["spread"] - rates["spread"].rolling(UJ_Z_WINDOW).mean()) \ / rates["spread"].rolling(UJ_Z_WINDOW).std() z_hourly = z_daily.reindex( pd.date_range(z_daily.index.min(), z_daily.index.max(), freq="h") ).ffill() # Try to load USDJPY 15M fpath = next(BASE.rglob("USDJPY_15M*.csv"), None) if fpath is None: print(" No USDJPY 15M data found — skipping UJ simulation") return [] df = pd.read_csv(fpath) df.columns = df.columns.str.strip().str.lower() dt_col = next(c for c in df.columns if "date" in c or "time" in c) df[dt_col] = pd.to_datetime(df[dt_col]) m15 = df.rename(columns={dt_col: "datetime"}).set_index("datetime").sort_index() trades = [] last_exit = None for dt in pd.date_range(start_date, end_date, freq="h"): if dt not in z_hourly.index: continue z = float(z_hourly.at[dt]) if abs(z) < UJ_THRESHOLD: continue # Session filter (matches backtest) if not (UJ_SESSION_LO <= (dt.hour + 2) % 24 <= UJ_SESSION_HI): continue if last_exit is not None and dt <= last_exit: continue direction = 1 if z >= UJ_THRESHOLD else -1 z_abs = abs(z) # Pullback from this hour's 15M slice try: slice_15m = m15.loc[dt: dt + pd.Timedelta(hours=1) - pd.Timedelta(minutes=1)] except Exception: continue if slice_15m.empty: continue bh = float(slice_15m["high"].max()) bl = float(slice_15m["low"].min()) bc = float(slice_15m["close"].iloc[-1]) if direction == 1: pull = bc - bl if pull <= 0.05: continue target = bc - UJ_FIB * pull else: pull = bh - bc if pull <= 0.05: continue target = bc + UJ_FIB * pull entry_price = entry_time = None try: entry_window = m15.loc[ dt + pd.Timedelta(minutes=1): dt + pd.Timedelta(hours=6) ] except Exception: continue for edt, ebar in entry_window.iterrows(): if direction == 1 and float(ebar["low"]) <= target: entry_price = target; entry_time = edt; break elif direction == -1 and float(ebar["high"]) >= target: entry_price = target; entry_time = edt; break if entry_price is None: continue # Simulate hold tp_px = entry_price * (1 + UJ_TP) if direction == 1 else entry_price * (1 - UJ_TP) sl_px = entry_price * (1 - UJ_SL) if direction == 1 else entry_price * (1 + UJ_SL) hold_bars = m15.loc[ entry_time + pd.Timedelta(minutes=1): entry_time + pd.Timedelta(hours=UJ_HOLD_HOURS) ] exit_price = None; exit_reason = "hold_expiry"; exit_time = None for hdt, hbar in hold_bars.iterrows(): if direction == 1: if float(hbar["high"]) >= tp_px: exit_price = tp_px; exit_reason = "tp"; exit_time = hdt; break if float(hbar["low"]) <= sl_px: exit_price = sl_px; exit_reason = "stop"; exit_time = hdt; break else: if float(hbar["low"]) <= tp_px: exit_price = tp_px; exit_reason = "tp"; exit_time = hdt; break if float(hbar["high"]) >= sl_px: exit_price = sl_px; exit_reason = "stop"; exit_time = hdt; break if exit_price is None and not hold_bars.empty: exit_price = float(hold_bars["close"].iloc[-1]) exit_time = hold_bars.index[-1] if exit_time is None: continue last_exit = exit_time ret = ((exit_price - entry_price) / entry_price) * direction trades.append({ "signal_time": dt, "entry_time": entry_time, "exit_time": exit_time, "signal": direction, "zscore": z, "entry_price": entry_price, "exit_price": exit_price, "exit_reason": exit_reason, "return": ret, }) print(f" Backtest would have taken {len(trades)} trade(s) in this window") return trades # ────────────────────────────────────────────────────────────────────────────── # Read live monitor log to extract actual live trades # ────────────────────────────────────────────────────────────────────────────── def read_live_trades_from_log(start_date: pd.Timestamp): """Parse the live monitor log to extract trade entries since start_date.""" if not LOG_FILE.exists(): return {"EURUSD": [], "USDJPY": []} import re trades = {"EURUSD": [], "USDJPY": []} pat = re.compile( r"(\d{4}-\d\d-\d\d \d\d:\d\d:\d\d).*\[(\w+)\] (TRADE|ORDER PLACED|SIGNAL DETECTED).*" ) z_pat = re.compile(r"z[=:]\s*([+\-]?\d+\.\d+)") with open(LOG_FILE, encoding="utf-8", errors="replace") as f: for line in f: m = pat.search(line) if not m: continue try: ts = pd.Timestamp(m.group(1)) except Exception: continue if ts < start_date: continue pair = m.group(2) if pair not in trades: continue zm = z_pat.search(line) zval = float(zm.group(1)) if zm else None trades[pair].append({"timestamp": ts, "z": zval, "raw": line.strip()}) return trades # ────────────────────────────────────────────────────────────────────────────── # Main # ────────────────────────────────────────────────────────────────────────────── def main(): import argparse parser = argparse.ArgumentParser(description="Compare backtest vs live for a date range") parser.add_argument("--start", default=DEFAULT_START, help="Start date YYYY-MM-DD") parser.add_argument("--end", default=DEFAULT_END, help="End date YYYY-MM-DD") args = parser.parse_args() start_date = pd.Timestamp(args.start) end_date = pd.Timestamp(args.end) print("=" * 72) print(" BACKTEST SIMULATION — anchored period for fair comparison") print("=" * 72) print(f" Period: {start_date.date()} → {end_date.date()}") days = (end_date - start_date).days print(f" Days: {days} ({days/30.4:.1f} months)") # Run backtest sims eu_trades = run_eurusd_backtest_recent(start_date, end_date) uj_trades = run_usdjpy_backtest_recent(start_date, end_date) # Read live trades from log (still useful if log spans this period) live = read_live_trades_from_log(start_date) print(f"\n{'=' * 72}") print(" EURUSD: backtest would have taken these trades") print(f"{'=' * 72}") if eu_trades: print(f" {'Signal time':<20} {'Entry time':<20} {'Z':>7} {'Sig':>4} {'Exit':>10} {'Ret%':>7}") print(" " + "─" * 70) for t in eu_trades: print(f" {str(t['signal_time'])[:16]:<20} " f"{str(t['entry_time'])[:16]:<20} " f"{t['zscore']:>+7.3f} " f"{'L' if t['signal']==1 else 'S':>4} " f"{t['exit_reason']:>10} " f"{t['return']*100:>+7.3f}") # Aggregates eu_df = pd.DataFrame(eu_trades) print(f"\n EU summary: {len(eu_df)} trades | " f"WR {(eu_df['return']>0).mean()*100:.1f}% | " f"Avg return {eu_df['return'].mean()*100:+.3f}% | " f"Trades/month {len(eu_df)/(days/30.4):.1f}") else: print(" (none)") print(f"\n{'=' * 72}") print(" USDJPY: backtest would have taken these trades") print(f"{'=' * 72}") if uj_trades: print(f" {'Signal time':<20} {'Entry time':<20} {'Z':>7} {'Sig':>4} {'Exit':>10} {'Ret%':>7}") print(" " + "─" * 70) for t in uj_trades: print(f" {str(t['signal_time'])[:16]:<20} " f"{str(t['entry_time'])[:16]:<20} " f"{t['zscore']:>+7.3f} " f"{'L' if t['signal']==1 else 'S':>4} " f"{t['exit_reason']:>10} " f"{t['return']*100:>+7.3f}") uj_df = pd.DataFrame(uj_trades) print(f"\n UJ summary: {len(uj_df)} trades | " f"WR {(uj_df['return']>0).mean()*100:.1f}% | " f"Avg return {uj_df['return'].mean()*100:+.3f}% | " f"Trades/month {len(uj_df)/(days/30.4):.1f}") else: print(" (none)") print(f"\n{'=' * 72}") print(" SUMMARY") print(f"{'=' * 72}") print(f" Window: {start_date.date()} → {end_date.date()} ({days/30.4:.1f} months)") print(f" EU backtest: {len(eu_trades)} trades ({len(eu_trades)/(days/30.4):.1f}/month)") print(f" UJ backtest: {len(uj_trades)} trades ({len(uj_trades)/(days/30.4):.1f}/month)") print(f" Combined: {len(eu_trades)+len(uj_trades)} trades " f"({(len(eu_trades)+len(uj_trades))/(days/30.4):.1f}/month)") print() print(f" Live log events in window:") print(f" EURUSD: {len(live['EURUSD'])}") print(f" USDJPY: {len(live['USDJPY'])}") print() print(" This tells you the rate the backtest produces signals in the same") print(" macro regime as the live deployment period. Compare against what") print(" the live monitor has produced since April 11 to assess the gap.") if __name__ == "__main__": main()