""" risk_calibration_v1.py ======================= Finds the optimal risk percentage for FTMO challenge compliance. Problem identified from atr_position_sizer_v1.py -------------------------------------------------- ATR sizing made results worse because it symmetrically reduced sizing on winners and losers alike. With 60-69% loser rate, winners need to be at full size to compound past losses. The real lever is base risk % itself: - Higher risk % → faster to profit target, higher breach probability - Lower risk % → slower to profit target, lower breach probability - Optimal → highest pass rate while keeping 95th pct DD < 10% Method ------ For each risk % in the test ladder: - Compute notional = (account × risk%) / stop_pct - Compute dollar P&L per trade = return × notional - Run 2,000 Monte Carlo simulations - Record pass rate, breach rate, 95th pct max DD, median trades to pass The optimal risk % is the highest value where: 1. 95th pct max DD < 10.0% (hard FTMO limit) 2. Pass rate is maximised Output ------ Full table across risk % ladder for both candidates Recommendation with reasoning CSV saved to data/processed/trades/ Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\risk_calibration_v1.py Run from project root: python src/research/risk_calibration_v1.py """ import pandas as pd import numpy as np from pathlib import Path import sys # ── 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 Rules ──────────────────────────────────────────────────────────────── ACCOUNT_START = 100_000.0 PROFIT_TARGET = ACCOUNT_START * 0.10 # +$10,000 MAX_OVERALL_LOSS = ACCOUNT_START * 0.10 # stay above $90,000 MAX_DAILY_LOSS = ACCOUNT_START * 0.05 # $5,000 per day STOP_PCT = 0.0025 SPREAD_COST_PCT = 0.0001 # ── Risk ladder to test ─────────────────────────────────────────────────────── # Expressed as % of account (0.003 = 0.3%, etc.) RISK_LADDER = [0.0020, 0.0025, 0.0030, 0.0035, 0.0040, 0.0045, 0.0050, 0.0060, 0.0075, 0.0100] N_SIMS = 2000 RANDOM_SEED = 42 CANDIDATES = ["growth_52h", "smoother_25h"] # Target: 95th pct max DD must stay below this DD_HARD_LIMIT = 10.0 # ── FTMO simulation (dollar P&L) ────────────────────────────────────────────── def simulate_ftmo_dollar( dollar_pnl : np.ndarray, exit_dates : pd.Series, ) -> dict: balance = ACCOUNT_START peak_balance = ACCOUNT_START max_dd_reached = 0.0 daily_start_balance = {} daily_pnl = {} outcome = "INCOMPLETE" trading_days_set = set() trades_taken = 0 for pnl, exit_dt in zip(dollar_pnl, exit_dates): date_key = str(pd.Timestamp(exit_dt).date()) if date_key not in daily_start_balance: daily_start_balance[date_key] = balance daily_pnl[date_key] = 0.0 balance += pnl trades_taken += 1 trading_days_set.add(date_key) daily_pnl[date_key] += pnl if balance > peak_balance: peak_balance = balance current_dd = (peak_balance - balance) / ACCOUNT_START if current_dd > max_dd_reached: max_dd_reached = current_dd if daily_pnl[date_key] < -MAX_DAILY_LOSS: outcome = "BREACH_DAILY_DD" break if balance <= (ACCOUNT_START - MAX_OVERALL_LOSS): outcome = "BREACH_OVERALL_DD" break if balance >= (ACCOUNT_START + PROFIT_TARGET): outcome = "PASS" break 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, "max_dd_reached_pct" : max_dd_reached * 100, "trades_taken" : trades_taken, "trading_days" : len(trading_days_set), } # ── Monte Carlo ─────────────────────────────────────────────────────────────── def run_mc( dollar_pnl : np.ndarray, exit_dates : pd.Series, n_sims : int = N_SIMS, ) -> pd.DataFrame: rng = np.random.default_rng(RANDOM_SEED) results = [] for i in range(n_sims): shuffled = rng.permuted(dollar_pnl) r = simulate_ftmo_dollar(shuffled, exit_dates) r["sim"] = i results.append(r) return pd.DataFrame(results) # ── Run calibration for one candidate ──────────────────────────────────────── def calibrate_risk(df: pd.DataFrame, name: str) -> pd.DataFrame: returns = df["return"].values exit_dates = df["exit_time"] rows = [] for risk_pct in RISK_LADDER: notional = (ACCOUNT_START * risk_pct) / STOP_PCT dollar_pnl = returns * notional mc = run_mc(dollar_pnl, exit_dates) pass_rate = (mc["outcome"] == "PASS").mean() * 100 breach_rate = mc["outcome"].str.startswith("BREACH").mean() * 100 daily_breach = (mc["outcome"] == "BREACH_DAILY_DD").mean() * 100 total_breach = (mc["outcome"] == "BREACH_OVERALL_DD").mean() * 100 incomplete = (mc["outcome"] == "INCOMPLETE").mean() * 100 dd_50 = mc["max_dd_reached_pct"].quantile(0.50) dd_75 = mc["max_dd_reached_pct"].quantile(0.75) dd_90 = mc["max_dd_reached_pct"].quantile(0.90) dd_95 = mc["max_dd_reached_pct"].quantile(0.95) dd_99 = mc["max_dd_reached_pct"].quantile(0.99) danger = (mc["max_dd_reached_pct"] > 8.0).sum() # Trades to pass — only among passing sims passing = mc[mc["outcome"] == "PASS"] avg_trades_to_pass = passing["trades_taken"].mean() if len(passing) > 0 else np.nan rows.append({ "risk_pct" : risk_pct, "risk_pct_label" : f"{risk_pct*100:.2f}%", "notional" : notional, "pass_rate" : pass_rate, "breach_rate" : breach_rate, "daily_breach_rate" : daily_breach, "total_breach_rate" : total_breach, "incomplete_rate" : incomplete, "dd_50" : dd_50, "dd_75" : dd_75, "dd_90" : dd_90, "dd_95" : dd_95, "dd_99" : dd_99, "danger_count" : danger, "danger_pct" : danger / N_SIMS * 100, "avg_trades_to_pass": avg_trades_to_pass, "candidate" : name, }) print(f" {risk_pct*100:.2f}% notional ${notional:>8,.0f} " f"pass {pass_rate:6.2f}% 95pct DD {dd_95:6.3f}%" f"{' ✓ WITHIN LIMIT' if dd_95 < DD_HARD_LIMIT else ' ✗ BREACH RISK'}") return pd.DataFrame(rows) # ── Find optimal risk % ──────────────────────────────────────────────────────── def find_optimal(df: pd.DataFrame) -> dict: """ Returns the highest risk % where 95th pct DD < 10%. This maximises growth potential while staying within FTMO hard limit. """ within_limit = df[df["dd_95"] < DD_HARD_LIMIT] if within_limit.empty: return {"found": False, "row": None} # Among rows within limit, find highest pass rate best_by_pass = within_limit.loc[within_limit["pass_rate"].idxmax()] # Also find the highest risk% within limit highest_safe = within_limit.loc[within_limit["risk_pct"].idxmax()] return { "found" : True, "best_pass_rate" : best_by_pass, "highest_safe_risk": highest_safe, } # ── Print full results table ────────────────────────────────────────────────── def print_table(df: pd.DataFrame, name: str): print(f"\n Full calibration table — {name}:") print(f"\n {'Risk%':<8}{'Notional':>10} {'Pass%':>7} {'Breach%':>8} " f"{'DD_50':>7} {'DD_75':>7} {'DD_90':>7} {'DD_95':>7} " f"{'Danger%':>8} {'AvgTrades':>10}") print(f" {'─'*95}") for _, r in df.iterrows(): limit_flag = "✓" if r["dd_95"] < DD_HARD_LIMIT else "✗" curr_flag = " ◄" if abs(r["risk_pct"] - 0.005) < 0.0001 else " " print(f" {r['risk_pct_label']:<8}{r['notional']:>10,.0f} " f"{r['pass_rate']:>7.2f} {r['breach_rate']:>8.2f} " f"{r['dd_50']:>7.3f} {r['dd_75']:>7.3f} " f"{r['dd_90']:>7.3f} {r['dd_95']:>7.3f} " f"{r['danger_pct']:>8.1f} " f"{r['avg_trades_to_pass']:>10.0f} {limit_flag}{curr_flag}") # ── Sequential check at chosen risk % ──────────────────────────────────────── def sequential_check( df : pd.DataFrame, risk_pct : float, name : str, ): notional = (ACCOUNT_START * risk_pct) / STOP_PCT dollar_pnl = df["return"].values * notional exit_dates = df["exit_time"] result = simulate_ftmo_dollar(dollar_pnl, exit_dates) print(f"\n Sequential check at {risk_pct*100:.2f}% risk (${notional:,.0f} notional):") print(f" Outcome : {result['outcome']}") print(f" Final bal : ${result['final_balance']:>12,.2f}") print(f" P&L : ${result['final_balance']-ACCOUNT_START:>+10,.2f} " f"({(result['final_balance']/ACCOUNT_START-1)*100:+.2f}%)") print(f" Max DD : {result['max_dd_reached_pct']:.3f}%") print(f" Trades taken : {result['trades_taken']}") print(f" Trading days : {result['trading_days']}") # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 70) print("RISK CALIBRATION V1") print(f" FTMO 100K | Profit target +10% | Max DD -10% | Daily DD -5%") print(f" Stop: {STOP_PCT:.2%} | {N_SIMS:,} Monte Carlo sims | 95th pct DD limit: {DD_HARD_LIMIT}%") print("=" * 70) all_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. Run export_trade_logs_v1.py first.") continue trades = pd.read_csv(csv_path, parse_dates=["entry_time", "exit_time"]) trades = trades.sort_values("entry_time").reset_index(drop=True) print(f"\n{'─'*70}") print(f"CANDIDATE: {name.upper()}") print(f" Trades: {len(trades)} | " f"Win rate: {(trades['return']>0).mean():.4%} | " f"Avg ret: {trades['return'].mean():.6f}") print(f"\n Running {N_SIMS} MC sims at each risk level...") print(f" {'Risk%':<8} {'Notional':>10} {'Pass':>6} " f"{'95pct DD':>9} {'Status'}") print(f" {'─'*55}") result_df = calibrate_risk(trades, name) all_results[name] = result_df # Print full table print_table(result_df, name) # Find optimal opt = find_optimal(result_df) print(f"\n ── Recommendation ──") if opt["found"]: best = opt["best_pass_rate"] high = opt["highest_safe_risk"] print(f"\n Best pass rate within DD limit:") print(f" Risk % : {best['risk_pct_label']}") print(f" Notional : ${best['notional']:,.0f}") print(f" Pass rate : {best['pass_rate']:.2f}%") print(f" 95th pct DD : {best['dd_95']:.3f}% (limit {DD_HARD_LIMIT}%)") print(f" Avg trades to pass: {best['avg_trades_to_pass']:.0f}") if high["risk_pct"] != best["risk_pct"]: print(f"\n Highest risk% within DD limit:") print(f" Risk % : {high['risk_pct_label']}") print(f" Notional : ${high['notional']:,.0f}") print(f" Pass rate : {high['pass_rate']:.2f}%") print(f" 95th pct DD : {high['dd_95']:.3f}%") print(f" Avg trades to pass: {high['avg_trades_to_pass']:.0f}") # Sequential validation at recommended risk rec_risk = float(best["risk_pct"]) sequential_check(trades, rec_risk, name) else: print(f"\n [WARNING] No risk level found where 95th pct DD < {DD_HARD_LIMIT}%") print(f" Minimum 95th pct DD achieved: " f"{result_df['dd_95'].min():.3f}% at " f"{result_df.loc[result_df['dd_95'].idxmin(), 'risk_pct_label']}") # Save results out_path = OUTPUT_DIR / f"risk_calibration_{name}.csv" result_df.to_csv(out_path, index=False) print(f"\n Saved: {out_path}") # ── Cross-candidate comparison at key risk levels ───────────────────────── if len(all_results) == 2: print(f"\n{'='*70}") print("CROSS-CANDIDATE COMPARISON AT KEY RISK LEVELS") print(f"{'='*70}") for risk_pct in [0.003, 0.0035, 0.004, 0.005]: label = f"{risk_pct*100:.2f}%" print(f"\n Risk {label} (notional ${(ACCOUNT_START*risk_pct/STOP_PCT):,.0f}):") print(f" {'Metric':<28}{'growth_52h':>14}{'smoother_25h':>14}") print(f" {'─'*56}") for metric, col in [ ("Pass rate", "pass_rate"), ("95th pct max DD", "dd_95"), ("Danger >8% DD", "danger_pct"), ("Avg trades to pass", "avg_trades_to_pass"), ]: vals = [] for name in CANDIDATES: if name in all_results: row = all_results[name] match = row[abs(row["risk_pct"] - risk_pct) < 0.00001] if not match.empty: v = match.iloc[0][col] vals.append(f"{v:.2f}") else: vals.append("N/A") else: vals.append("N/A") print(f" {metric:<28}{vals[0]:>14}{vals[1]:>14}") print(f"\n{'='*70}") print("INTERPRETATION") print(f"{'='*70}") print(f""" The optimal risk % is the HIGHEST value where 95th pct max DD < 10%. Why 95th pct and not median? In FTMO you only get one attempt per challenge. The 95th pct represents a 1-in-20 bad-luck sequence. You need this to be survivable — not just the average case. Why not use the lowest risk %? Lower risk = smaller wins. With 60-69% losers, you need enough win size to exceed losses. Too low a risk % = more "incomplete" sims where you run out of trades before hitting the profit target. The sweet spot: highest risk % where 95th pct DD stays below 10%. This maximises your probability of passing while keeping the tail risk within the hard FTMO boundary. Next step: freeze the optimal risk % and build the live execution framework using these validated parameters. """) if __name__ == "__main__": main()