""" ftmo_sequence_simulator_v1.py ============================== Proper trade-sequence FTMO challenge simulator for both finalist candidates. Uses actual trade returns from export_trade_logs_v1.py — NOT average-return smoothing. Every trade is applied individually in sequence so drawdown clustering and losing streaks are preserved. FTMO 100K Challenge Rules Applied ----------------------------------- Account start : $100,000 Profit target : +10% = $110,000 (challenge pass) Max overall DD : -10% = $90,000 (breach — immediate fail) Max daily DD : -5% = $5,000 below start-of-day balance (breach — fail) Min trading days: 4 (tracked, reported) Position Sizing --------------- Fixed-fractional: risk_per_trade = 0.5% of initial account = $500 stop_pct = 0.0025 (0.25%) position_notional = risk / stop = $500 / 0.0025 = $200,000 per trade dollar_pnl = trade_return × $200,000 This means a stop-hit trade loses exactly $500 + spread ($20) = $520. A winner returning +0.1% gains $200. Dollar risk is FIXED (not compounding) — this is conservative and appropriate for a challenge where the primary goal is avoiding breach, not maximising growth. Monte Carlo ----------- 2000 simulations with shuffled trade sequence. Each sim draws from the same return pool in random order. Reports: pass rate, breach rates, median final balance, percentile bands. Output ------ Printed results + CSV saved to data/processed/trades/ Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\ftmo_sequence_simulator_v1.py Run from project root: python src/research/ftmo_sequence_simulator_v1.py """ import pandas as pd import numpy as np from pathlib import Path import sys from datetime import timedelta # ── Path setup ──────────────────────────────────────────────────────────────── 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)) TRADES_DIR = BASE_PATH / "data" / "processed" / "trades" OUTPUT_DIR = TRADES_DIR OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # ── FTMO Parameters ─────────────────────────────────────────────────────────── ACCOUNT_START = 100_000.0 PROFIT_TARGET = ACCOUNT_START * 0.10 # $110,000 MAX_OVERALL_LOSS = ACCOUNT_START * 0.10 # must stay above $90,000 MAX_DAILY_LOSS = ACCOUNT_START * 0.05 # $5,000 per day MIN_TRADING_DAYS = 4 # ── Sizing ──────────────────────────────────────────────────────────────────── RISK_PER_TRADE_PCT = 0.005 # 0.5% of initial RISK_PER_TRADE = ACCOUNT_START * RISK_PER_TRADE_PCT # $500 STOP_PCT = 0.0025 # 0.25% POSITION_NOTIONAL = RISK_PER_TRADE / STOP_PCT # $200,000 # ── Monte Carlo ─────────────────────────────────────────────────────────────── N_SIMS = 2000 RANDOM_SEED = 42 CANDIDATES = ["growth_52h", "smoother_25h"] # ── Core simulation (one pass through a return sequence) ───────────────────── def simulate_ftmo(returns: np.ndarray, exit_dates: pd.Series) -> dict: """ Runs one FTMO challenge simulation through a sequence of trade returns. Parameters ---------- returns : array of trade returns (decimals, e.g. 0.001, -0.0026) exit_dates : Series of trade exit timestamps (same length as returns) Returns ------- dict with outcome, final_balance, max_dd_reached, daily_dd_breach_day, trades_taken, trading_days, peak_balance """ balance = ACCOUNT_START peak_balance = ACCOUNT_START max_dd_reached = 0.0 # worst % drawdown seen (stored as positive) # Daily tracking — keyed by date string daily_start_balance = {} # balance at start of each trading day daily_pnl = {} # cumulative P&L within each day outcome = "INCOMPLETE" daily_dd_breach_day = None total_dd_breach_bal = None trading_days_set = set() trades_taken = 0 for i, (ret, exit_dt) in enumerate(zip(returns, exit_dates)): dollar_pnl = ret * POSITION_NOTIONAL date_key = str(pd.Timestamp(exit_dt).date()) # Initialise day if first trade exiting on this date if date_key not in daily_start_balance: daily_start_balance[date_key] = balance daily_pnl[date_key] = 0.0 # Apply trade balance += dollar_pnl trades_taken += 1 trading_days_set.add(date_key) daily_pnl[date_key] += dollar_pnl # Update peak if balance > peak_balance: peak_balance = balance # Check drawdown metrics current_dd = (peak_balance - balance) / ACCOUNT_START if current_dd > max_dd_reached: max_dd_reached = current_dd # ── Check daily DD breach ───────────────────────────────────────────── # FTMO: cannot lose more than 5% of initial balance in a single day day_loss = daily_pnl[date_key] # negative if losing if day_loss < -MAX_DAILY_LOSS: outcome = "BREACH_DAILY_DD" daily_dd_breach_day = date_key break # ── Check overall DD breach ─────────────────────────────────────────── # Balance must stay above $90,000 if balance <= (ACCOUNT_START - MAX_OVERALL_LOSS): outcome = "BREACH_OVERALL_DD" total_dd_breach_bal = balance break # ── Check profit target ─────────────────────────────────────────────── if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" break # If we exhausted all trades without hitting target or breach if outcome == "INCOMPLETE": if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" elif (peak_balance - balance) / ACCOUNT_START >= 0.10: outcome = "BREACH_OVERALL_DD" return { "outcome" : outcome, "final_balance" : balance, "peak_balance" : peak_balance, "max_dd_reached_pct" : max_dd_reached * 100, "total_pnl" : balance - ACCOUNT_START, "total_pnl_pct" : (balance / ACCOUNT_START - 1) * 100, "trades_taken" : trades_taken, "trading_days" : len(trading_days_set), "daily_dd_breach_day" : daily_dd_breach_day, "total_dd_breach_bal" : total_dd_breach_bal, } # ── Sequential simulation (real historical order) ──────────────────────────── def run_sequential(df: pd.DataFrame, name: str) -> dict: returns = df["return"].values exit_dates = df["exit_time"] result = simulate_ftmo(returns, exit_dates) result["candidate"] = name result["sim_type"] = "SEQUENTIAL" return result # ── Monte Carlo simulation ──────────────────────────────────────────────────── def run_monte_carlo(df: pd.DataFrame, name: str, n_sims: int = N_SIMS) -> pd.DataFrame: """ Shuffles the trade sequence n_sims times and runs the FTMO sim each time. Uses the same set of exit_dates but reassigned to shuffled returns so daily DD tracking remains date-realistic. """ rng = np.random.default_rng(RANDOM_SEED) returns = df["return"].values.copy() exit_dates = df["exit_time"].values.copy() sim_results = [] for sim_i in range(n_sims): # Shuffle returns (keep dates in original order — realistic daily DD) shuffled_returns = rng.permuted(returns) result = simulate_ftmo(shuffled_returns, exit_dates) result["sim_num"] = sim_i sim_results.append(result) mc_df = pd.DataFrame(sim_results) mc_df["candidate"] = name return mc_df # ── Printer helpers ─────────────────────────────────────────────────────────── def print_sequential_result(r: dict): print(f"\n Outcome : {r['outcome']}") print(f" Final balance : ${r['final_balance']:>12,.2f}") print(f" Total P&L : ${r['total_pnl']:>+12,.2f} ({r['total_pnl_pct']:+.2f}%)") print(f" Peak balance : ${r['peak_balance']:>12,.2f}") print(f" Max DD reached : {r['max_dd_reached_pct']:.3f}% (limit 10%)") print(f" Trades taken : {r['trades_taken']}") print(f" Trading days : {r['trading_days']} (need ≥ {MIN_TRADING_DAYS})") if r["outcome"] == "BREACH_DAILY_DD": print(f" Daily DD breach : {r['daily_dd_breach_day']}") elif r["outcome"] == "BREACH_OVERALL_DD": print(f" Breach balance : ${r['total_dd_breach_bal']:,.2f}") def print_mc_summary(mc_df: pd.DataFrame, name: str): outcomes = mc_df["outcome"].value_counts() n = len(mc_df) pass_rate = outcomes.get("PASS", 0) / n * 100 breach_daily = outcomes.get("BREACH_DAILY_DD", 0) / n * 100 breach_total = outcomes.get("BREACH_OVERALL_DD", 0) / n * 100 incomplete = outcomes.get("INCOMPLETE", 0) / n * 100 print(f"\n Monte Carlo Results ({n:,} simulations):") print(f" {'─'*40}") print(f" PASS : {pass_rate:6.2f}% ({outcomes.get('PASS',0):,} sims)") print(f" BREACH — daily DD : {breach_daily:6.2f}% ({outcomes.get('BREACH_DAILY_DD',0):,} sims)") print(f" BREACH — overall DD: {breach_total:6.2f}% ({outcomes.get('BREACH_OVERALL_DD',0):,} sims)") print(f" Incomplete (no edge): {incomplete:6.2f}% ({outcomes.get('INCOMPLETE',0):,} sims)") final_bal = mc_df["final_balance"] dd_reached = mc_df["max_dd_reached_pct"] print(f"\n Final balance distribution:") print(f" 5th pct : ${np.percentile(final_bal, 5):>10,.2f}") print(f" 25th pct : ${np.percentile(final_bal, 25):>10,.2f}") print(f" Median : ${np.percentile(final_bal, 50):>10,.2f}") print(f" 75th pct : ${np.percentile(final_bal, 75):>10,.2f}") print(f" 95th pct : ${np.percentile(final_bal, 95):>10,.2f}") print(f"\n Max drawdown reached distribution (% of account):") print(f" 5th pct : {np.percentile(dd_reached, 5):.3f}%") print(f" 25th pct : {np.percentile(dd_reached, 25):.3f}%") print(f" Median : {np.percentile(dd_reached, 50):.3f}%") print(f" 75th pct : {np.percentile(dd_reached, 75):.3f}%") print(f" 95th pct : {np.percentile(dd_reached, 95):.3f}%") # Danger zone — how many sims came within 2% of breach? danger = (dd_reached > 8.0).sum() print(f"\n Sims reaching >8% DD (within 2% of breach): {danger} ({danger/n*100:.1f}%)") # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 65) print("FTMO SEQUENCE SIMULATOR V1 — 100K Challenge") print(f" Risk per trade : ${RISK_PER_TRADE:,.0f} ({RISK_PER_TRADE_PCT:.1%} of initial)") print(f" Position notional : ${POSITION_NOTIONAL:,.0f}") print(f" Profit target : ${PROFIT_TARGET:,.0f} (+10%)") print(f" Max overall DD : ${MAX_OVERALL_LOSS:,.0f} (-10%)") print(f" Max daily DD : ${MAX_DAILY_LOSS:,.0f} (-5% per day)") print(f" Monte Carlo sims : {N_SIMS:,}") print("=" * 65) all_mc_results = [] seq_results = [] for name in CANDIDATES: csv_path = TRADES_DIR / f"trades_{name}.csv" if not csv_path.exists(): print(f"\n[ERROR] {csv_path} not found.") print(" Run export_trade_logs_v1.py first.") continue df = pd.read_csv(csv_path, parse_dates=["entry_time", "exit_time"]) df = df.sort_values("entry_time").reset_index(drop=True) print(f"\n{'─'*65}") print(f"CANDIDATE: {name.upper()}") print(f" Trades loaded : {len(df)}") print(f" Date range : {df['entry_time'].min().date()} → {df['exit_time'].max().date()}") print(f" Win rate : {(df['return']>0).mean():.4%}") print(f" Avg return : {df['return'].mean():.6f}") print(f" Avg $P&L/trade : ${df['return'].mean()*POSITION_NOTIONAL:+.2f}") # ── Sequential ──────────────────────────────────────────────────────── print(f"\n ── Sequential Simulation (historical order) ──") seq = run_sequential(df, name) print_sequential_result(seq) seq_results.append(seq) # ── Monte Carlo ─────────────────────────────────────────────────────── print(f"\n ── Monte Carlo Simulation ──") mc_df = run_monte_carlo(df, name) print_mc_summary(mc_df, name) all_mc_results.append(mc_df) # Save Monte Carlo results mc_out = OUTPUT_DIR / f"ftmo_mc_{name}.csv" mc_df.to_csv(mc_out, index=False) print(f"\n MC results saved: {mc_out}") # ── Cross-candidate comparison ──────────────────────────────────────────── print(f"\n{'='*65}") print("CROSS-CANDIDATE COMPARISON") print(f"{'='*65}") if len(all_mc_results) == 2: labels = CANDIDATES mc_a = all_mc_results[0] mc_b = all_mc_results[1] def pass_pct(mc): return (mc["outcome"] == "PASS").sum() / len(mc) * 100 def breach_pct(mc): return ((mc["outcome"].str.startswith("BREACH"))).sum() / len(mc) * 100 def median_dd(mc): return mc["max_dd_reached_pct"].median() metrics = [ ("MC Pass rate" , pass_pct), ("MC Breach rate" , breach_pct), ("MC Median max DD %" , median_dd), ] print(f"\n{'Metric':<28}{labels[0]:<22}{labels[1]:<22}") print("-" * 72) for label, fn in metrics: print(f"{label:<28}{fn(mc_a):<22.2f}{fn(mc_b):<22.2f}") # ── Sizing sensitivity note ─────────────────────────────────────────────── print(f"\n{'='*65}") print("SIZING SENSITIVITY NOTE") print(f"{'='*65}") print(f""" Current sizing: ${RISK_PER_TRADE:.0f} per trade (0.5% of $100k) To stress-test: how does pass rate change at different risk levels? Risk $300 (0.3%) → Notional $120k → Safer but slower to hit target Risk $500 (0.5%) → Notional $200k → Current setting (above) Risk $750 (0.75%) → Notional $300k → Faster growth, higher breach risk Risk $1000 (1.0%) → Notional $400k → Aggressive — approach with caution The 23-trade losing streak (growth_52h) at $500 risk = $12,000+ cumulative loss if all on same days. Daily DD limit is the binding constraint, not overall DD — because multiple losses in one day stack dangerously. This is why 25h smoother (max streak 15) is the prop-firm candidate. """) print("Next step: run mae_crossover_analysis_v1.py") if __name__ == "__main__": main()