""" dynamic_exit_v1.py =================== Tests the dynamic exit framework on top of the validated Fibonacci entry. The Problem ----------- Current model: fixed 0.25% stop hits 66% of trades. Winner avg MAE: -0.086% — winners barely move against you before recovering Loser avg MAE: -0.645% — losers move hard and keep going The current stop sits between these two distributions but 66% of trades cross it before recovering. Most stopped trades were winners being killed early. The Solution ------------ Replace the fixed 0.25% stop with two intelligent exit layers: Layer 1 — Tail Risk Stop (wide, rarely triggered) Set at -0.75% — beyond the 99th pct winner MAE (-0.244%) Only triggers on genuine catastrophic moves, not normal noise Expected hit rate: ~5-8% (vs current 66%) Layer 2 — Z-Score Reversal Exit (thesis invalidation) During the hold window, monitor lag_zscore_24h_v3 on 1H bars If z-score crosses zero strongly in OPPOSITE direction to the trade: - Was long (signal=1), z-score now <= -1.5 → thesis invalidated - Was short (signal=-1), z-score now >= +1.5 → thesis invalidated Exit at close of the bar where reversal is confirmed This uses the SAME signal source as the entry — logically consistent Layer 3 — Profit Protection Exit (optional, tested separately) Once trade is up > 0.40% (MFE median), if z-score is moving against position AND price has given back 30% of peak gain → exit Locks in profits on winners without cutting them prematurely Layer 4 — Time Exit (default) If none of the above triggers → hold to 52h and exit Methods Tested -------------- A) Baseline : Fibonacci entry, 0.25% fixed stop, 52h hold (reference) B) Tail stop : Wide 0.75% tail stop only (no z-score exit) C) Z-exit only : Z-score reversal exit + current 0.25% stop D) Combined : Wide 0.75% tail stop + z-score reversal exit E) Full system : Wide stop + z-score exit + profit protection Each method is tested with signal scaling (1x/1.5x/2x) and full Monte Carlo FTMO simulation. Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\dynamic_exit_v1.py Run from project root: python src/research/dynamic_exit_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 import load_eurusd 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) # ── Frozen parameters ───────────────────────────────────────────────────────── THRESHOLD = 2.75 FIB = 0.786 HOLD_HOURS = 52 CURRENT_STOP = 0.0025 # 0.25% — current baseline TAIL_STOP = 0.0075 # 0.75% — wide tail risk stop SPREAD_COST = 0.0001 ALLOWED_HOURS = set(range(7, 17)) # Z-score reversal thresholds ZSCORE_REVERSAL_STRONG = 1.5 # strong reversal — thesis clearly invalidated ZSCORE_REVERSAL_WEAK = 1.0 # weak reversal — used for profit protection # Profit protection trigger PROFIT_PROTECTION_ENTRY = 0.0040 # trade must be up 0.40% before protection activates PROFIT_PROTECTION_GIVEBACK = 0.30 # exit if price gives back 30% of peak gain # FTMO / sizing ACCOUNT_START = 100_000.0 PROFIT_TARGET = ACCOUNT_START * 0.10 MAX_OVERALL = ACCOUNT_START * 0.10 MAX_DAILY = ACCOUNT_START * 0.05 BASE_RISK_PCT = 0.003 BASE_NOTIONAL = (ACCOUNT_START * BASE_RISK_PCT) / CURRENT_STOP # $120,000 ZSCORE_BANDS = [ (2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0), ] N_SIMS = 2000 RANDOM_SEED= 42 YEARS = 22.0 def get_multiplier(z: float) -> float: for lo, hi, mult in ZSCORE_BANDS: if lo <= z < hi: return mult return ZSCORE_BANDS[-1][2] # ── Load and prepare signal data with z-scores on 1H bars ──────────────────── def build_signal_data_with_zscore() -> pd.DataFrame: """ Builds the hourly signal dataframe with lag_zscore_24h_v3 for each bar. This is used during trade holds to monitor z-score reversals. """ df = get_model_ready_spot_lag_v3().copy() df["datetime"] = pd.to_datetime(df["datetime"]) df = df.sort_values("datetime").reset_index(drop=True) return df # ── Core trade simulator with dynamic exits ─────────────────────────────────── def simulate_trade_dynamic( m15 : pd.DataFrame, signal_df : pd.DataFrame, # 1H bars with z-score for exit monitoring entry_time : pd.Timestamp, entry_price : float, signal : int, hold_hours : int = HOLD_HOURS, stop : float = CURRENT_STOP, spread_cost : float = SPREAD_COST, use_zscore_exit : bool = False, use_profit_protect: bool = False, ) -> dict | None: """ Simulates a trade with optional dynamic exit layers. Parameters ---------- use_zscore_exit : If True, exit early if z-score reverses strongly use_profit_protect: If True, exit to protect profits when conditions met """ 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 # Get the z-score data during the hold window signal_window = signal_df[ (signal_df["datetime"] >= entry_time) & (signal_df["datetime"] <= path.iloc[-1]["datetime"]) ].copy() if (use_zscore_exit or use_profit_protect) else pd.DataFrame() exit_price = float(path.iloc[-1]["close"]) exit_reason = "time" stop_hit = False peak_gain = 0.0 # Walk through the path bar by bar to check dynamic exits for bar_i in range(len(path)): bar = path.iloc[bar_i] bar_time= bar["datetime"] # Current unrealised P&L if signal == 1: current_ret = (float(bar["close"]) - entry_price) / entry_price adverse_ret = (float(bar["low"]) - entry_price) / entry_price else: current_ret = (entry_price - float(bar["close"])) / entry_price adverse_ret = -((float(bar["high"]) - entry_price) / entry_price) # Track peak gain if current_ret > peak_gain: peak_gain = current_ret # ── Stop loss check ─────────────────────────────────────────────────── if adverse_ret <= -stop: exit_price = entry_price * (1 - stop * signal) if signal == 1 \ else entry_price * (1 + stop) # Simpler: just use the stop level exit_price = entry_price * (1 - stop) if signal == 1 \ else entry_price * (1 + stop) exit_reason = "stop" stop_hit = True break # ── Z-score reversal exit ───────────────────────────────────────────── if use_zscore_exit and not signal_window.empty: # Find z-score at this bar time (backward merge) z_bars = signal_window[signal_window["datetime"] <= bar_time] if not z_bars.empty: current_z = float(z_bars.iloc[-1]["lag_zscore_24h_v3"]) # For long trades: if z-score reverses strongly negative if signal == 1 and current_z <= -ZSCORE_REVERSAL_STRONG: exit_price = float(bar["close"]) exit_reason = "zscore_reversal" break # For short trades: if z-score reverses strongly positive elif signal == -1 and current_z >= ZSCORE_REVERSAL_STRONG: exit_price = float(bar["close"]) exit_reason = "zscore_reversal" break # ── Profit protection exit ──────────────────────────────────────────── if use_profit_protect and peak_gain >= PROFIT_PROTECTION_ENTRY: # Protection only activates once we have a meaningful profit giveback = (peak_gain - current_ret) / peak_gain if peak_gain > 0 else 0 if giveback >= PROFIT_PROTECTION_GIVEBACK and not signal_window.empty: z_bars = signal_window[signal_window["datetime"] <= bar_time] if not z_bars.empty: current_z = float(z_bars.iloc[-1]["lag_zscore_24h_v3"]) # Require z-score to also be weakening in the exit direction z_weakening = (signal == 1 and current_z < 0) or \ (signal == -1 and current_z > 0) if z_weakening: exit_price = float(bar["close"]) exit_reason = "profit_protect" break else: # Loop completed without early exit — time exit exit_price = float(path.iloc[-1]["close"]) exit_reason = "time" stop_hit = exit_reason == "stop" # Calculate final return if signal == 1: raw_ret = (exit_price - entry_price) / entry_price mae = (path["low"].min() - entry_price) / entry_price mfe = (path["high"].max() - entry_price) / entry_price else: raw_ret = -(exit_price - entry_price) / entry_price mae = -((path["high"].max() - entry_price) / entry_price) mfe = (entry_price - path["low"].min()) / entry_price # Override for clean stop calculation if stop_hit: raw_ret = -stop return { "entry_time" : entry_time, "exit_time" : path.iloc[min(bar_i, len(path)-1)]["datetime"], "signal" : signal, "entry_price" : entry_price, "exit_price" : exit_price, "stop_hit" : stop_hit, "exit_reason" : exit_reason, "mae" : mae, "mfe" : mfe, "return" : raw_ret - spread_cost, } # ── Run one method through the full trade loop ──────────────────────────────── def run_method( name : str, signals : pd.DataFrame, m15 : pd.DataFrame, signal_df : pd.DataFrame, stop : float, use_zscore_exit : bool, use_profit_protect: bool, ) -> pd.DataFrame: print(f" Running {name}...") trades = [] last_exit_time = None for _, sig in 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_dynamic( m15 =m15, signal_df =signal_df, entry_time =entry["entry_time"], entry_price =entry["entry_price"], signal =int(sig["signal"]), stop =stop, use_zscore_exit =use_zscore_exit, use_profit_protect=use_profit_protect, ) 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: print(f" [WARNING] No trades generated") 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) streak = max_streak = 0 for r in df["return"]: if r <= 0: streak += 1 max_streak = max(max_streak, streak) else: streak = 0 df["max_streak_val"] = max_streak return df # ── FTMO Monte Carlo ────────────────────────────────────────────────────────── def run_mc(df: pd.DataFrame) -> dict: rng = np.random.default_rng(RANDOM_SEED) returns = df["return"].values zscores = df["zscore_abs"].values exits = df["exit_time"].values per_yr = len(df) / YEARS results = [] for _ in range(N_SIMS): idx = rng.permutation(len(returns)) sh_ret = returns[idx] sh_mult= np.array([get_multiplier(z) for z in zscores[idx]]) balance = ACCOUNT_START peak = ACCOUNT_START max_dd = 0.0 dpnl = {} outcome = "INCOMPLETE" tdays = set() ttrades = 0 for ret, mult, ex_dt in zip(sh_ret, sh_mult, exits): dk = str(pd.Timestamp(ex_dt).date()) if dk not in dpnl: dpnl[dk] = 0.0 notional = min(BASE_NOTIONAL * mult, BASE_NOTIONAL * 3) pnl = ret * notional balance += pnl ttrades += 1 tdays.add(dk) dpnl[dk] += pnl if balance > peak: peak = balance dd = (peak - balance) / ACCOUNT_START if dd > max_dd: max_dd = dd if dpnl[dk] < -MAX_DAILY: outcome = "BREACH_DAILY"; break if balance <= ACCOUNT_START - MAX_OVERALL: outcome = "BREACH_OVERALL"; break if balance >= ACCOUNT_START + PROFIT_TARGET: outcome = "PASS"; break if outcome == "INCOMPLETE": outcome = ("PASS" if balance >= ACCOUNT_START + PROFIT_TARGET else "BREACH_OVERALL" if (peak - balance) / ACCOUNT_START >= 0.10 else "INCOMPLETE") results.append({"outcome": outcome, "max_dd": max_dd * 100, "trades": ttrades}) mc = pd.DataFrame(results) passing = mc[mc["outcome"] == "PASS"] return { "pass_rate" : (mc["outcome"] == "PASS").mean() * 100, "breach_rate": mc["outcome"].str.startswith("BREACH").mean() * 100, "dd_95" : mc["max_dd"].quantile(0.95), "dd_50" : mc["max_dd"].quantile(0.50), "avg_months" : (passing["trades"].mean() / per_yr * 12 if len(passing) > 0 else np.nan), "med_months" : (passing["trades"].median() / per_yr * 12 if len(passing) > 0 else np.nan), } # ── Print summary ───────────────────────────────────────────────────────────── def summarise(name: str, df: pd.DataFrame, mc: dict): if df.empty: print(f"\n {name}: NO TRADES") return n = len(df) wr = df["win"].mean() avg_ret = df["return"].mean() fin_eq = df["equity"].iloc[-1] max_dd = df["drawdown"].min() streak = int(df["max_streak_val"].iloc[-1]) sh_rate = df["stop_hit"].mean() if "stop_hit" in df.columns else 0 mae = df["mae"].mean() * 100 mfe = df["mfe"].mean() * 100 # Exit reason breakdown if "exit_reason" in df.columns: er = df["exit_reason"].value_counts(normalize=True) * 100 er_str = " ".join(f"{k}: {v:.1f}%" for k, v in er.items()) else: er_str = "N/A" print(f"\n {name}") print(f" {'─'*58}") print(f" Trades : {n:,} ({n/YEARS:.1f}/yr)") print(f" Win rate : {wr:.4%}") print(f" Avg return : {avg_ret:.6f}") print(f" Final equity: {fin_eq:.6f}") print(f" Max DD : {max_dd:.4%}") print(f" Max streak : {streak}") print(f" Stop hit : {sh_rate:.4%}") print(f" Avg MAE : {mae:.3f}%") print(f" Avg MFE : {mfe:.3f}%") print(f" Exit reasons: {er_str}") print(f" MC pass : {mc['pass_rate']:.2f}%") print(f" MC 95pct DD : {mc['dd_95']:.3f}%") print(f" Avg months : {mc['avg_months']:.1f}") print(f" Med months : {mc['med_months']:.1f}") # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 65) print("DYNAMIC EXIT V1 — Exit Framework Testing") print(f" Tail stop: {TAIL_STOP:.2%} | " f"Z-reversal threshold: {ZSCORE_REVERSAL_STRONG}") print(f" Profit protection: {PROFIT_PROTECTION_ENTRY:.2%} gain trigger, " f"{PROFIT_PROTECTION_GIVEBACK:.0%} giveback") print("=" * 65) # Load data print("\nLoading data...") m15 = load_eurusd_15m().sort_values("datetime").reset_index(drop=True) signal_df = build_signal_data_with_zscore() signals = build_frozen_signals( threshold=THRESHOLD, allowed_hours=ALLOWED_HOURS, ) print(f" 15M bars : {len(m15):,}") print(f" Signal df : {len(signal_df):,} bars with z-scores") print(f" Signals : {len(signals)}") # ── Define methods ──────────────────────────────────────────────────────── methods = [ { "name" : "A) Baseline (0.25% stop, time exit)", "stop" : CURRENT_STOP, "use_zscore_exit" : False, "use_profit_protect": False, }, { "name" : "B) Wide tail stop (0.75%, time exit only)", "stop" : TAIL_STOP, "use_zscore_exit" : False, "use_profit_protect": False, }, { "name" : "C) 0.25% stop + Z-score reversal exit", "stop" : CURRENT_STOP, "use_zscore_exit" : True, "use_profit_protect": False, }, { "name" : "D) Wide stop + Z-score reversal exit", "stop" : TAIL_STOP, "use_zscore_exit" : True, "use_profit_protect": False, }, { "name" : "E) Wide stop + Z-exit + Profit protect", "stop" : TAIL_STOP, "use_zscore_exit" : True, "use_profit_protect": True, }, ] # ── Run all methods ─────────────────────────────────────────────────────── print("\nRunning methods...") results = {} for m in methods: df = run_method( name =m["name"], signals =signals, m15 =m15, signal_df =signal_df, stop =m["stop"], use_zscore_exit =m["use_zscore_exit"], use_profit_protect=m["use_profit_protect"], ) if not df.empty: mc = run_mc(df) results[m["name"]] = (df, mc) df.to_csv( TRADES_DIR / f"trades_exit_{chr(65+len(results)-1)}.csv", index=False, ) # ── Print results ───────────────────────────────────────────────────────── print(f"\n{'='*65}") print("RESULTS") print(f"{'='*65}") for name, (df, mc) in results.items(): summarise(name, df, mc) # ── Comparison table ────────────────────────────────────────────────────── print(f"\n{'='*65}") print("COMPARISON TABLE") print(f"{'='*65}") print(f"\n {'Method':<42}{'WR':>7}{'AvgRet':>10}{'FinalEq':>10}" f"{'MaxDD':>8}{'Pass%':>7}{'95DD':>8}{'MedMths':>9}") print(f" {'─'*101}") best_pass = 0 best_name = "" for name, (df, mc) in results.items(): wr = df["win"].mean() avg_ret = df["return"].mean() fin_eq = df["equity"].iloc[-1] max_dd = df["drawdown"].min() flag = " ◄ BEST" if mc["pass_rate"] > best_pass and mc["dd_95"] < 10 else "" if mc["pass_rate"] > best_pass and mc["dd_95"] < 10: best_pass = mc["pass_rate"] best_name = name short_name = name[:40] print(f" {short_name:<42}{wr:>6.2%}{avg_ret:>10.6f}" f"{fin_eq:>10.4f}{max_dd:>7.2%}" f"{mc['pass_rate']:>7.2f}{mc['dd_95']:>8.3f}" f"{mc['med_months']:>9.1f}{flag}") # ── Recommendation ──────────────────────────────────────────────────────── print(f"\n{'='*65}") print("RECOMMENDATION") print(f"{'='*65}") if best_name: best_df, best_mc = results[best_name] baseline_df, baseline_mc = list(results.values())[0] wr_gain = (best_df["win"].mean() - baseline_df["win"].mean()) * 100 ret_gain = best_df["return"].mean() - baseline_df["return"].mean() dd_change = best_mc["dd_95"] - baseline_mc["dd_95"] months_saved= baseline_mc["med_months"] - best_mc["med_months"] print(f""" Best method: {best_name} vs baseline: Win rate change : {wr_gain:+.2f} percentage points Avg return change : {ret_gain:+.6f} 95th pct DD change: {dd_change:+.3f}% Months saved : {months_saved:+.1f} months {'VALID — 95th pct DD within 10% limit' if best_mc['dd_95'] < 10 else 'WARNING — 95th pct DD exceeds 10% limit, needs risk adjustment'} """) else: print("\n No method passed all criteria. Review individual results above.") print(f" Trade logs saved to: {TRADES_DIR}") if __name__ == "__main__": main()