""" atr_position_sizer_v1.py ========================= Implements ATR-based volatility-adjusted position sizing and re-runs the FTMO challenge simulation to measure the impact on tail risk. The Problem (from FTMO simulator output) ----------------------------------------- At fixed 0.5% risk / $200k notional per trade: - growth_52h : MC 95th pct max DD = 12.3% (above 10% breach limit) - smoother_25h: MC 95th pct max DD = 13.4% (above 10% breach limit) ~25% of Monte Carlo paths reach >8% DD (within 2% of breach) The cause is NOT the stop (0.25% is validated as optimal by MAE analysis). The cause is loser clustering — 69% losers (growth) / 60% losers (smoother) stacking into the same time window, multiplied by fixed $200k notional. The Fix ------- Size positions inversely to current ATR. When EUR/USD is volatile, the notional shrinks automatically, so a losing cluster costs less in dollars. Formula: base_notional = risk_per_trade / stop_pct = $500 / 0.0025 = $200,000 atr_multiplier = atr_baseline / atr_at_entry adj_notional = base_notional * atr_multiplier adj_notional = clipped to [min_notional, max_notional] Where atr_baseline is the median ATR over the full trade history. If ATR is at baseline: position = $200k (unchanged) If ATR is 2x baseline: position = $100k (halved) If ATR is 0.5x baseline: position = $400k (but capped at max) ATR Calculation --------------- We compute the 14-period ATR on 1H EURUSD bars. At each trade entry, we look up the ATR at that bar. This is available from the price data already in your pipeline. Outputs ------- 1. ATR statistics across trade entries 2. Position sizing distribution 3. FTMO re-simulation with ATR sizing vs fixed sizing 4. Monte Carlo comparison — fixed vs ATR-adjusted 5. CSV saved to data/processed/trades/ Place this file in: C:\\Users\\paul_\\OneDrive\\fx_macro_intraday\\src\\research\\atr_position_sizer_v1.py Run from project root: python src/research/atr_position_sizer_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)) from ingestion.price_loader import load_eurusd 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 # $10,000 → $110,000 MAX_OVERALL_LOSS = ACCOUNT_START * 0.10 # stay above $90,000 MAX_DAILY_LOSS = ACCOUNT_START * 0.05 # $5,000 per day # ── Sizing Parameters ───────────────────────────────────────────────────────── RISK_PCT = 0.005 # 0.5% of initial account RISK_DOLLAR = ACCOUNT_START * RISK_PCT # $500 per trade STOP_PCT = 0.0025 # 0.25% stop BASE_NOTIONAL = RISK_DOLLAR / STOP_PCT # $200,000 baseline # ATR sizing caps — prevent extreme sizing in very quiet or very volatile regimes MIN_NOTIONAL = BASE_NOTIONAL * 0.40 # $80,000 (floor — very high vol) MAX_NOTIONAL = BASE_NOTIONAL * 2.00 # $400,000 (cap — very low vol) # ATR period ATR_PERIOD = 14 # Monte Carlo N_SIMS = 2000 RANDOM_SEED = 42 CANDIDATES = ["growth_52h", "smoother_25h"] # ── ATR computation ─────────────────────────────────────────────────────────── def compute_atr(df: pd.DataFrame, period: int = ATR_PERIOD) -> pd.DataFrame: """ Computes ATR on 1H OHLCV DataFrame. True Range = max(H-L, |H-Prev_C|, |L-Prev_C|) ATR = rolling mean of True Range over `period` bars. """ df = df.copy().sort_values("datetime").reset_index(drop=True) prev_close = df["close"].shift(1) tr = pd.concat([ df["high"] - df["low"], (df["high"] - prev_close).abs(), (df["low"] - prev_close).abs(), ], axis=1).max(axis=1) df["atr"] = tr.rolling(period, min_periods=period).mean() df["atr_pct"] = df["atr"] / df["close"] # ATR as % of price return df[["datetime", "close", "atr", "atr_pct"]].copy() # ── Attach ATR to trade log ─────────────────────────────────────────────────── def attach_atr_to_trades( trades: pd.DataFrame, prices: pd.DataFrame, ) -> pd.DataFrame: """ For each trade, looks up the ATR at the closest 1H bar at/before entry_time. Uses merge_asof (backward merge) for efficiency. """ trades = trades.copy().sort_values("entry_time").reset_index(drop=True) atr_df = compute_atr(prices) atr_df = atr_df.dropna(subset=["atr"]).sort_values("datetime") # Backward merge: for each trade entry, find the most recent ATR bar merged = pd.merge_asof( trades.rename(columns={"entry_time": "datetime"}), atr_df[["datetime", "atr", "atr_pct"]], on="datetime", direction="backward", ) merged = merged.rename(columns={"datetime": "entry_time"}) return merged # ── Compute ATR-adjusted notional ───────────────────────────────────────────── def compute_adj_notional( trades : pd.DataFrame, atr_baseline : float, ) -> pd.DataFrame: """ Adds adj_notional column to the trade log. atr_baseline: median ATR % across all trades in the dataset. When current ATR = baseline: adj_notional = BASE_NOTIONAL When current ATR > baseline: adj_notional < BASE_NOTIONAL (shrinks) When current ATR < baseline: adj_notional > BASE_NOTIONAL (grows, capped) """ trades = trades.copy() atr_multiplier = atr_baseline / trades["atr_pct"] adj_notional = BASE_NOTIONAL * atr_multiplier trades["adj_notional"] = adj_notional.clip( lower=MIN_NOTIONAL, upper=MAX_NOTIONAL, ) trades["dollar_pnl_fixed"] = trades["return"] * BASE_NOTIONAL trades["dollar_pnl_atr"] = trades["return"] * trades["adj_notional"] return trades # ── FTMO simulation (dollar P&L version) ───────────────────────────────────── def simulate_ftmo_dollar( dollar_pnl : np.ndarray, exit_dates : pd.Series, ) -> dict: """ Runs FTMO challenge on a dollar P&L sequence (not % returns). Same rules as ftmo_sequence_simulator_v1.py but takes pre-computed $ P&L. """ balance = ACCOUNT_START peak_balance = ACCOUNT_START max_dd_reached = 0.0 daily_start_balance = {} daily_pnl = {} outcome = "INCOMPLETE" daily_dd_breach_day = None 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" daily_dd_breach_day = date_key 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, "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), } # ── Monte Carlo (dollar P&L) ────────────────────────────────────────────────── def run_mc_dollar( dollar_pnl : np.ndarray, exit_dates : pd.Series, label : str, 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) df = pd.DataFrame(results) df["label"] = label return df # ── Print MC summary ────────────────────────────────────────────────────────── def print_mc_compare(mc_fixed: pd.DataFrame, mc_atr: pd.DataFrame): def summarise(mc, label): n = len(mc) pass_r = (mc["outcome"] == "PASS").mean() * 100 breach_r = mc["outcome"].str.startswith("BREACH").mean() * 100 dd_50 = mc["max_dd_reached_pct"].quantile(0.50) dd_75 = mc["max_dd_reached_pct"].quantile(0.75) dd_95 = mc["max_dd_reached_pct"].quantile(0.95) danger = (mc["max_dd_reached_pct"] > 8.0).sum() print(f"\n {label}:") print(f" Pass rate : {pass_r:.2f}%") print(f" Breach rate : {breach_r:.2f}%") print(f" Median max DD : {dd_50:.3f}%") print(f" 75th pct max DD : {dd_75:.3f}%") print(f" 95th pct max DD : {dd_95:.3f}% (limit = 10%)") print(f" Sims >8% DD : {danger} ({danger/n*100:.1f}%)") summarise(mc_fixed, "FIXED sizing ($200k notional)") summarise(mc_atr, "ATR sizing (volatility-adjusted)") # Delta pass_delta = (mc_atr["outcome"] == "PASS").mean()*100 - \ (mc_fixed["outcome"] == "PASS").mean()*100 dd95_delta = mc_atr["max_dd_reached_pct"].quantile(0.95) - \ mc_fixed["max_dd_reached_pct"].quantile(0.95) danger_d = (mc_atr["max_dd_reached_pct"] > 8.0).sum() - \ (mc_fixed["max_dd_reached_pct"] > 8.0).sum() print(f"\n Impact of ATR sizing:") print(f" Pass rate change : {pass_delta:+.2f}%") print(f" 95th pct DD change : {dd95_delta:+.3f}%") print(f" Sims >8% DD change : {danger_d:+d}") # ── Main ────────────────────────────────────────────────────────────────────── def main(): print("=" * 65) print("ATR POSITION SIZER V1") print(f" Base notional : ${BASE_NOTIONAL:,.0f}") print(f" Min notional : ${MIN_NOTIONAL:,.0f} (40% of base)") print(f" Max notional : ${MAX_NOTIONAL:,.0f} (200% of base)") print(f" ATR period : {ATR_PERIOD} bars (1H)") print(f" Monte Carlo : {N_SIMS:,} simulations") print("=" * 65) # Load 1H price data once print("\nLoading 1H price data...") prices = load_eurusd().sort_values("datetime").reset_index(drop=True) print(f" Loaded {len(prices):,} bars " f"({prices['datetime'].min().date()} → {prices['datetime'].max().date()})") 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 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{'─'*65}") print(f"CANDIDATE: {name.upper()}") print(f" Trades: {len(trades)}") # ── Attach ATR ──────────────────────────────────────────────────────── print(" Computing ATR at each entry...") trades = attach_atr_to_trades(trades, prices) missing_atr = trades["atr_pct"].isna().sum() if missing_atr > 0: print(f" [WARNING] {missing_atr} trades missing ATR " f"(early data before ATR warmup) — will use baseline.") atr_baseline = trades["atr_pct"].median() trades["atr_pct"] = trades["atr_pct"].fillna(atr_baseline) atr_baseline = trades["atr_pct"].median() print(f"\n ATR statistics at trade entries (% of price):") print(f" Baseline (median) : {atr_baseline*100:.4f}%") print(f" Mean : {trades['atr_pct'].mean()*100:.4f}%") print(f" 10th pct : {trades['atr_pct'].quantile(0.10)*100:.4f}%") print(f" 25th pct : {trades['atr_pct'].quantile(0.25)*100:.4f}%") print(f" 75th pct : {trades['atr_pct'].quantile(0.75)*100:.4f}%") print(f" 90th pct : {trades['atr_pct'].quantile(0.90)*100:.4f}%") # ── Compute ATR-adjusted notional ───────────────────────────────────── trades = compute_adj_notional(trades, atr_baseline) print(f"\n Adjusted notional distribution:") print(f" Min : ${trades['adj_notional'].min():>12,.0f}") print(f" 10th pct : ${trades['adj_notional'].quantile(0.10):>12,.0f}") print(f" 25th pct : ${trades['adj_notional'].quantile(0.25):>12,.0f}") print(f" Median : ${trades['adj_notional'].quantile(0.50):>12,.0f}") print(f" 75th pct : ${trades['adj_notional'].quantile(0.75):>12,.0f}") print(f" 90th pct : ${trades['adj_notional'].quantile(0.90):>12,.0f}") print(f" Max : ${trades['adj_notional'].max():>12,.0f}") pct_reduced = (trades["adj_notional"] < BASE_NOTIONAL).mean() * 100 pct_at_min = (trades["adj_notional"] == MIN_NOTIONAL).mean() * 100 pct_at_max = (trades["adj_notional"] == MAX_NOTIONAL).mean() * 100 print(f"\n Trades sized below base ($200k): {pct_reduced:.1f}%") print(f" Trades at floor ($80k) : {pct_at_min:.1f}%") print(f" Trades at cap ($400k) : {pct_at_max:.1f}%") # ── Sequential comparison ───────────────────────────────────────────── print(f"\n ── Sequential Simulation Comparison ──") seq_fixed = simulate_ftmo_dollar( trades["dollar_pnl_fixed"].values, trades["exit_time"], ) seq_atr = simulate_ftmo_dollar( trades["dollar_pnl_atr"].values, trades["exit_time"], ) print(f"\n {'Metric':<26}{'Fixed $200k':>14}{'ATR-adjusted':>14}") print(f" {'─'*54}") metrics = [ ("Outcome" , seq_fixed["outcome"], seq_atr["outcome"]), ("Final balance" , f"${seq_fixed['final_balance']:>10,.2f}", f"${seq_atr['final_balance']:>10,.2f}"), ("Total P&L" , f"{seq_fixed['total_pnl_pct']:>+.2f}%", f"{seq_atr['total_pnl_pct']:>+.2f}%"), ("Max DD reached" , f"{seq_fixed['max_dd_reached_pct']:.3f}%", f"{seq_atr['max_dd_reached_pct']:.3f}%"), ("Trades taken" , str(seq_fixed["trades_taken"]), str(seq_atr["trades_taken"])), ] for label, v_fixed, v_atr in metrics: print(f" {label:<26}{str(v_fixed):>14}{str(v_atr):>14}") # ── Monte Carlo comparison ───────────────────────────────────────────── print(f"\n ── Monte Carlo Comparison ({N_SIMS:,} simulations) ──") mc_fixed = run_mc_dollar( trades["dollar_pnl_fixed"].values, trades["exit_time"], "fixed", ) mc_atr = run_mc_dollar( trades["dollar_pnl_atr"].values, trades["exit_time"], "atr", ) print_mc_compare(mc_fixed, mc_atr) # ── Save enriched trade log ─────────────────────────────────────────── out_path = OUTPUT_DIR / f"trades_{name}_atr.csv" save_cols = [ "entry_time", "exit_time", "signal", "entry_price", "exit_price", "stop_hit", "mae", "mfe", "return", "equity", "drawdown", "win", "trade_num", "candidate", "atr_pct", "adj_notional", "dollar_pnl_fixed", "dollar_pnl_atr", ] trades[save_cols].to_csv(out_path, index=False) print(f"\n Saved enriched trade log: {out_path}") print(f"\n{'='*65}") print("NEXT STEPS") print(f"{'='*65}") print(""" If ATR sizing reduces 95th pct DD below 10% and raises pass rate: → ATR sizing is validated. Freeze it. If improvement exists but 95th pct DD still slightly above 10%: → Consider combining ATR sizing + regime filter (Step 5). The regime filter will suppress signals during macro environments where the USD/rates thesis is structurally weak, reducing the density of losing clusters and further improving the DD profile. """) if __name__ == "__main__": main()