""" event_stress_clustered_v1.py (v2 — adversarial) ================================================= Event stress test with DIRECTIONAL BIAS during stress bursts. This is the killer addition: During burst periods (3-5 trades): force adverse drift of 0.10%-0.25% against entry BEFORE TP/SL is reached. This simulates news continuation — not just bad fills in a normal market, but bad fills in a HOSTILE market. Adverse drift mechanics: drift d% is applied against position direction at entry. - TP trades: need to travel TP_dist + d% to win. If d > SL_dist → forced stop. - SL trades: stop out at SL_dist + d% loss (extra slippage on loss). - Drift flips some TP trades to SL trades (adverse news continuation). Monte Carlo: 10,000 simulations per scenario. Scenarios: Normal | Moderate | Severe | Extreme | Catastrophic Pass (prop firm risk desk standard): Severe stress: Sharpe > 2.0, MaxDD 95th pct < 5%, FTMO pass > 90% """ import sys, warnings from pathlib import Path import numpy as np import pandas as pd warnings.filterwarnings('ignore') BASE = Path(__file__).resolve().parents[2] EU_FILE = BASE / 'data/processed/trades/trades_real_costs.csv' UJ_FILE = BASE / 'data/processed/usdjpy_trades_real_costs.csv' ACCOUNT = 100_000 N_SIM = 10_000 # Per-pair parameters PARAMS = { 'EURUSD': dict(tp_pct=0.0020, sl_pct=0.0025, pip=0.0001, pip_val=10.0, base_notional=300_000), 'USDJPY': dict(tp_pct=0.0070, sl_pct=0.0040, pip=0.010, pip_val=6.67, base_notional=300_000), } np.random.seed(42) def load_trades(path, pair): df = pd.read_csv(path) df.columns = [c.lower().strip() for c in df.columns] for col in ['dollar_pnl_real','pnl_real','pnl','net_pnl']: if col in df.columns: df['pnl'] = pd.to_numeric(df[col], errors='coerce'); break for col in ['entry_time','date','entry_date']: if col in df.columns: df['date'] = pd.to_datetime(df[col], errors='coerce'); break df = df.dropna(subset=['pnl','date']).sort_values('date').reset_index(drop=True) if df['pnl'].abs().median() < 1.0: df['pnl'] *= ACCOUNT p = PARAMS[pair] # Infer exit type: TP vs SL vs other tp_dollar = p['base_notional'] * p['tp_pct'] sl_dollar = p['base_notional'] * p['sl_pct'] threshold = sl_dollar * 0.5 df['exit'] = np.where(df['pnl'] > threshold, 'tp', np.where(df['pnl'] < -threshold, 'sl', 'other')) # Estimate lot size (accounts for z-score scaling) df['lots'] = (df['pnl'].abs() / (p['tp_pct'] * p['base_notional'] / 0.01 + 1e-9)).clip(0.01, 10) return df def apply_scenario(pnl_arr, exit_arr, lots_arr, pair, burst_mask, entry_slip, stop_slip, spread_add, delay_pips, adverse_drift_pct): """ Apply full adversarial scenario to burst-masked trades. adverse_drift_pct: price moves this % AGAINST position before TP/SL. - Converts TP trades to SL if drift > SL distance - Reduces TP profit by drift amount on surviving TP trades - Worsens SL loss by drift amount """ p = PARAMS[pair] pip_val = p['pip_val'] result = pnl_arr.copy().astype(float) for i in range(len(result)): if not burst_mask[i]: continue l = max(lots_arr[i], 0.01) ex = exit_arr[i] notional = l * p['base_notional'] / 0.01 # crude notional # ── Execution friction costs ─────────────────────────────────────────── fric = (entry_slip + spread_add + delay_pips) * pip_val * l result[i] -= fric # ── Adverse directional drift (THE NEW ADDITION) ────────────────────── drift_pct = adverse_drift_pct if drift_pct > 0: drift_dollar = drift_pct * notional if ex == 'tp': remaining_to_sl = p['sl_pct'] - drift_pct if remaining_to_sl <= 0: # Drift blew through the stop → forced SL with extra slippage result[i] = -(p['sl_pct'] + drift_pct + stop_slip * p['pip']) * notional else: # Trade still lives but TP is now harder to reach # Effective TP return is reduced by the drift result[i] -= drift_dollar elif ex == 'sl': # SL trade gets even worse: price has already moved against result[i] -= drift_dollar + stop_slip * pip_val * l else: result[i] -= drift_dollar * 0.5 # z-exit or other elif ex == 'sl': # Non-drift SL: just stop slippage result[i] -= stop_slip * pip_val * l return result def daily_sharpe(pnl, dates): dfx = pd.DataFrame({'d': pd.to_datetime(dates).dt.date, 'p': pnl}) daily = dfx.groupby('d')['p'].sum() idx = pd.date_range(daily.index.min(), daily.index.max(), freq='B') daily = daily.reindex(idx, fill_value=0) ret = daily / ACCOUNT return (ret.mean() / ret.std() * np.sqrt(252)) if ret.std() > 0 else 0.0 def max_dd(pnl): eq = np.cumsum(pnl) + ACCOUNT pk = np.maximum.accumulate(eq) return ((eq - pk) / pk * 100).min() def ftmo_pass_rate(pnl, n_sim=2_000, target=0.10, dd_total=0.10, dd_daily=0.05): n, passes = len(pnl), 0 tgt = ACCOUNT * target lim = ACCOUNT * dd_total dlim = ACCOUNT * dd_daily for _ in range(n_sim): seq = pnl[np.random.randint(0, n, n)] eq = ACCOUNT; pk = ACCOUNT; ok = False; breached = False daily_floor = ACCOUNT for p in seq: eq += p pk = max(pk, eq) if (pk - eq) > lim or (daily_floor - eq) > dlim: breached = True; break if eq >= ACCOUNT + tgt: ok = True; break if p > 0: # rough daily reset proxy daily_floor = eq if ok and not breached: passes += 1 return passes / n_sim # ──────────────────────────────────────────────────────────────────────────── # Scenarios: entry_slip, stop_slip, spread_add, delay_pips, # adverse_drift_pct, burst_size, burst_freq, label # ──────────────────────────────────────────────────────────────────────────── SCENARIOS = [ (0.5, 0.8, 0.0, 0.0, 0.000, 0, 0.00, 'Baseline (validated real costs)'), (1.0, 1.5, 1.0, 2.0, 0.001, 3, 0.05, 'Moderate — 3-trade burst, friction only'), (2.0, 3.0, 3.0, 5.0, 0.001, 4, 0.10, 'Moderate+ — burst + 0.10% adverse drift'), (2.0, 3.0, 3.0, 5.0, 0.0015,4, 0.10, 'Severe — burst + 0.15% adverse drift'), (3.0, 5.0, 5.0, 8.0, 0.002, 5, 0.15, 'Extreme — burst + 0.20% adverse drift'), (5.0, 8.0, 10.0, 12.0, 0.0025,5, 0.20, 'Catastrophic — NFP/CPI worst case + 0.25% drift'), ] print("=" * 80) print(" EVENT STRESS — CLUSTERED BURST + DIRECTIONAL ADVERSE DRIFT") print(" 10,000 Monte Carlo simulations per scenario") print("=" * 80) print(""" Adverse drift: price moves AGAINST position by d% before TP/SL is reached. This converts TP trades to forced SL trades when drift > SL distance. Distinguishes "bad fills in normal market" from "bad fills in hostile market". """) for pair in ['EURUSD', 'USDJPY']: path = EU_FILE if pair == 'EURUSD' else UJ_FILE df = load_trades(path, pair) pnl = df['pnl'].values exit_= df['exit'].values lots = df['lots'].values dates= df['date'] n = len(df) p = PARAMS[pair] print(f"\n{'─'*80}") print(f" {pair} ({n} trades | TP {p['tp_pct']*100:.2f}% | SL {p['sl_pct']*100:.2f}%)") print(f" Exit split: TP={( df['exit']=='tp').mean()*100:.1f}% " f"SL={( df['exit']=='sl').mean()*100:.1f}%") print(f"{'─'*80}") print(f" {'Scenario':>48} {'Sharpe':>7} {'DD 95th':>8} " f"{'FTMO':>7} {'vs base':>10} {'Pass?':>6}") print(f" {'─'*48} {'─'*7} {'─'*8} {'─'*7} {'─'*10} {'─'*6}") base_sh = base_dd = None for (es, ss, sa, dp, drift, bsz, bfreq, label) in SCENARIOS: sharpes, maxdds = [], [] pnl_stressed_all = [] for _ in range(N_SIM): # Build burst mask mask = np.zeros(n, dtype=bool) if bsz > 0 and bfreq > 0: n_bursts = max(1, int(n * bfreq / bsz)) starts = np.random.choice(n - bsz, size=n_bursts, replace=False) for s in starts: mask[s:s+bsz] = True stressed = apply_scenario(pnl, exit_, lots, pair, mask, es, ss, sa, dp, drift) sharpes.append(daily_sharpe(stressed, dates)) maxdds.append(max_dd(stressed)) pnl_stressed_all.append(stressed) sh = float(np.median(sharpes)) dd95 = float(np.percentile(maxdds, 95)) # FTMO on median-stressed equity median_stressed = np.median(pnl_stressed_all, axis=0) fpass = ftmo_pass_rate(median_stressed) if base_sh is None: base_sh, base_dd = sh, dd95 vs = f"Sh:{sh-base_sh:+.2f} DD:{dd95-base_dd:+.2f}%" ok = '✅' if (sh > 2.0 and abs(dd95) < 5.0 and fpass > 0.90) else '❌' print(f" {ok} {label:>46} {sh:7.2f} {dd95:7.2f}% " f"{fpass:6.1%} {vs:>10}") # Full distribution for Severe scenario severes_sh, severes_dd = [], [] for _ in range(N_SIM): es,ss,sa,dp,drift,bsz,bfreq,_ = SCENARIOS[3] # Severe mask = np.zeros(n, dtype=bool) n_bursts = max(1, int(n * bfreq / bsz)) starts = np.random.choice(n - bsz, size=n_bursts, replace=False) for s in starts: mask[s:s+bsz] = True stressed = apply_scenario(pnl, exit_, lots, pair, mask, es,ss,sa,dp,drift) severes_sh.append(daily_sharpe(stressed, dates)) severes_dd.append(max_dd(stressed)) sv_sh = np.array(severes_sh) sv_dd = np.array(severes_dd) print(f"\n Severe scenario distribution (10,000 simulations):") print(f" Sharpe: p5={np.percentile(sv_sh,5):.2f} " f"p25={np.percentile(sv_sh,25):.2f} " f"med={np.median(sv_sh):.2f} " f"p75={np.percentile(sv_sh,75):.2f} " f"p95={np.percentile(sv_sh,95):.2f}") print(f" MaxDD: p5={np.percentile(sv_dd,5):.2f}% " f"p25={np.percentile(sv_dd,25):.2f}% " f"med={np.median(sv_dd):.2f}% " f"p95={np.percentile(sv_dd,95):.2f}%") print(f" % sims Sharpe > 2.0: {(sv_sh > 2.0).mean()*100:.1f}%") print(f" % sims MaxDD > 5.0%: {(sv_dd < -5.0).mean()*100:.1f}%") print(f"\n Drift impact (Moderate vs Moderate+, same friction, +0.10% drift):") print(f" Shows exactly how much Sharpe adverse directional move costs.") print(f"\n{'='*80}") print(" VERDICT") print(f"{'='*80}") print(""" What to look for: Moderate (friction only) → shows execution friction cost alone Moderate+ (+ 0.10% drift) → adds news continuation on top Difference in Sharpe between these two = pure directional drift cost Pass criteria (prop firm risk desk): Severe: Sharpe > 2.0, MaxDD 95th pct < 5%, FTMO pass > 90% If passes Severe → model survives realistic FOMC/CPI events If fails Extreme → expected; those are flash-crash scenarios Key: the adverse drift shows what happens when the market moves WITH the news rather than reverting. This is the real tail risk, not just slippage. """)