#!/usr/bin/env python3 """Monte Carlo strategy simulator for the personal US equity agent. The simulator uses synthetic market regimes, not live prices. It is meant to stress-test decision rules before they are used as portfolio policy. """ from __future__ import annotations import argparse import csv import json import math import random import statistics from dataclasses import dataclass from pathlib import Path START_CAPITAL_KRW = 103_000_000 DEFAULT_REGIMES = { "risk_on": { "basket": (0.018, 0.025), "quality": (0.011, 0.014), "leader": (0.025, 0.030), "soxl": (0.050, 0.070), "soxs": (-0.045, 0.060), }, "selective": { "basket": (0.003, 0.030), "quality": (0.007, 0.016), "leader": (0.016, 0.028), "soxl": (0.012, 0.075), "soxs": (-0.010, 0.065), }, "chop": { "basket": (-0.002, 0.028), "quality": (0.000, 0.014), "leader": (0.002, 0.026), "soxl": (-0.005, 0.075), "soxs": (-0.006, 0.070), }, "selloff": { "basket": (-0.035, 0.035), "quality": (-0.018, 0.022), "leader": (-0.028, 0.035), "soxl": (-0.095, 0.085), "soxs": (0.070, 0.075), }, "rebound": { "basket": (0.030, 0.040), "quality": (0.018, 0.024), "leader": (0.040, 0.045), "soxl": (0.105, 0.105), "soxs": (-0.085, 0.085), }, "crash": { "basket": (-0.075, 0.055), "quality": (-0.042, 0.035), "leader": (-0.060, 0.055), "soxl": (-0.205, 0.140), "soxs": (0.145, 0.130), }, } DEFAULT_TRANSITIONS = { "risk_on": [("risk_on", 0.50), ("selective", 0.25), ("chop", 0.12), ("selloff", 0.06), ("rebound", 0.06), ("crash", 0.01)], "selective": [("risk_on", 0.18), ("selective", 0.42), ("chop", 0.22), ("selloff", 0.10), ("rebound", 0.06), ("crash", 0.02)], "chop": [("risk_on", 0.12), ("selective", 0.20), ("chop", 0.38), ("selloff", 0.18), ("rebound", 0.08), ("crash", 0.04)], "selloff": [("risk_on", 0.03), ("selective", 0.08), ("chop", 0.17), ("selloff", 0.36), ("rebound", 0.25), ("crash", 0.11)], "rebound": [("risk_on", 0.30), ("selective", 0.28), ("chop", 0.20), ("selloff", 0.12), ("rebound", 0.08), ("crash", 0.02)], "crash": [("risk_on", 0.02), ("selective", 0.04), ("chop", 0.14), ("selloff", 0.34), ("rebound", 0.35), ("crash", 0.11)], } @dataclass class StrategyResult: name: str final_equity: list[float] max_drawdown: list[float] max_drawdown_krw: list[float] hit_weekly_quest: list[bool] hit_monthly_quest: list[bool] weekly_quest_count: list[int] monthly_quest_count: list[int] hit_double: list[bool] ended_below_start: list[bool] def pick_regime(rng: random.Random, current: str) -> str: roll = rng.random() cumulative = 0.0 for regime, probability in TRANSITIONS[current]: cumulative += probability if roll <= cumulative: return regime return TRANSITIONS[current][-1][0] def normal_return(rng: random.Random, regimes: dict, regime: str, asset: str) -> float: mean, stdev = regimes[regime][asset] value = rng.gauss(mean, stdev) return max(value, -0.80) def signal_quality(rng: random.Random, regime: str) -> dict[str, bool]: crash_risk = regime in {"selloff", "crash"} and rng.random() < 0.74 false_crash = regime in {"chop", "selective"} and rng.random() < 0.15 risk_on_signal = regime in {"risk_on", "selective", "rebound"} and rng.random() < 0.70 false_risk_on = regime in {"chop", "selloff"} and rng.random() < 0.18 rebound_signal = regime == "rebound" and rng.random() < 0.67 return { "crash_risk": crash_risk or false_crash, "risk_on": risk_on_signal or false_risk_on, "rebound": rebound_signal, "confirmed_breakdown": regime in {"selloff", "crash"} and rng.random() < 0.62, } def weights_for_strategy( strategy: str, regime: str, signal: dict[str, bool], drawdown: float, pending_cash_days: int, equity: float = START_CAPITAL_KRW, ) -> dict[str, float]: """Return weights over basket, quality, leader, soxl, soxs, cash.""" if strategy == "passive_hold": return {"basket": 0.65, "quality": 0.20, "leader": 0.10, "soxl": 0.00, "soxs": 0.00, "cash": 0.05} if strategy == "panic_sell_rebuy": if drawdown < -0.10: return {"basket": 0.10, "quality": 0.10, "leader": 0.00, "soxl": 0.00, "soxs": 0.00, "cash": 0.80} if signal["risk_on"]: return {"basket": 0.55, "quality": 0.20, "leader": 0.15, "soxl": 0.00, "soxs": 0.00, "cash": 0.10} return {"basket": 0.45, "quality": 0.20, "leader": 0.05, "soxl": 0.00, "soxs": 0.00, "cash": 0.30} if strategy == "aggressive_leverage": if signal["confirmed_breakdown"]: return {"basket": 0.25, "quality": 0.05, "leader": 0.05, "soxl": 0.00, "soxs": 0.40, "cash": 0.25} if signal["risk_on"] or signal["rebound"]: return {"basket": 0.45, "quality": 0.10, "leader": 0.15, "soxl": 0.25, "soxs": 0.00, "cash": 0.05} return {"basket": 0.45, "quality": 0.10, "leader": 0.10, "soxl": 0.15, "soxs": 0.00, "cash": 0.20} if strategy == "turbo_soxl_soxs_switch": if regime == "crash": return {"basket": 0.00, "quality": 0.10, "leader": 0.00, "soxl": 0.00, "soxs": 0.55 if pending_cash_days == 0 else 0.25, "cash": 0.35 if pending_cash_days == 0 else 0.65} if signal["confirmed_breakdown"]: return {"basket": 0.05, "quality": 0.10, "leader": 0.00, "soxl": 0.00, "soxs": 0.52 if pending_cash_days == 0 else 0.22, "cash": 0.33 if pending_cash_days == 0 else 0.63} if signal["rebound"]: return {"basket": 0.08, "quality": 0.08, "leader": 0.24, "soxl": 0.55, "soxs": 0.00, "cash": 0.05} if signal["risk_on"]: return {"basket": 0.12, "quality": 0.10, "leader": 0.23, "soxl": 0.48, "soxs": 0.00, "cash": 0.07} return {"basket": 0.18, "quality": 0.16, "leader": 0.08, "soxl": 0.10, "soxs": 0.00, "cash": 0.48} if strategy == "selective_leverage_burst": if regime == "crash" or signal["confirmed_breakdown"]: return {"basket": 0.00, "quality": 0.18, "leader": 0.00, "soxl": 0.00, "soxs": 0.35 if pending_cash_days == 0 else 0.00, "cash": 0.47 if pending_cash_days == 0 else 0.82} if signal["rebound"]: return {"basket": 0.10, "quality": 0.18, "leader": 0.25, "soxl": 0.37, "soxs": 0.00, "cash": 0.10} if signal["risk_on"] and drawdown > -0.08: return {"basket": 0.12, "quality": 0.20, "leader": 0.28, "soxl": 0.30, "soxs": 0.00, "cash": 0.10} return {"basket": 0.08, "quality": 0.32, "leader": 0.08, "soxl": 0.00, "soxs": 0.00, "cash": 0.52} if strategy == "compounding_tactical_ladder": profit_buffer = max(0.0, equity / START_CAPITAL_KRW - 1.0) turbo = 1.0 if profit_buffer >= 1.00: turbo = 1.45 elif profit_buffer >= 0.50: turbo = 1.30 elif profit_buffer >= 0.25: turbo = 1.18 if drawdown < -0.18: return {"basket": 0.00, "quality": 0.20, "leader": 0.00, "soxl": 0.00, "soxs": 0.00, "cash": 0.80} if drawdown < -0.10: return {"basket": 0.04, "quality": 0.26, "leader": 0.00, "soxl": 0.00, "soxs": 0.10 if signal["confirmed_breakdown"] and pending_cash_days == 0 else 0.00, "cash": 0.60 if signal["confirmed_breakdown"] and pending_cash_days == 0 else 0.70} if regime == "crash" or signal["confirmed_breakdown"]: soxs_weight = min(0.52, 0.30 * turbo) if pending_cash_days == 0 else 0.00 return {"basket": 0.00, "quality": 0.18, "leader": 0.00, "soxl": 0.00, "soxs": soxs_weight, "cash": 0.82 - soxs_weight} if signal["rebound"]: soxl_weight = min(0.62, 0.38 * turbo) return {"basket": 0.06, "quality": 0.12, "leader": 0.22, "soxl": soxl_weight, "soxs": 0.00, "cash": 0.60 - soxl_weight} if signal["risk_on"]: soxl_weight = min(0.50, 0.28 * turbo) return {"basket": 0.10, "quality": 0.18, "leader": 0.24, "soxl": soxl_weight, "soxs": 0.00, "cash": 0.48 - soxl_weight} return {"basket": 0.08, "quality": 0.28, "leader": 0.08, "soxl": 0.00, "soxs": 0.00, "cash": 0.56} if strategy == "guarded_aggressive_barbell": if regime == "crash" or (signal["crash_risk"] and drawdown < -0.04): return {"basket": 0.05, "quality": 0.30, "leader": 0.00, "soxl": 0.00, "soxs": 0.18 if pending_cash_days == 0 else 0.00, "cash": 0.47 if pending_cash_days == 0 else 0.65} if signal["confirmed_breakdown"]: return {"basket": 0.10, "quality": 0.32, "leader": 0.00, "soxl": 0.00, "soxs": 0.15 if pending_cash_days == 0 else 0.00, "cash": 0.43 if pending_cash_days == 0 else 0.58} if signal["rebound"]: return {"basket": 0.22, "quality": 0.25, "leader": 0.25, "soxl": 0.18, "soxs": 0.00, "cash": 0.10} if signal["risk_on"]: return {"basket": 0.20, "quality": 0.30, "leader": 0.27, "soxl": 0.15, "soxs": 0.00, "cash": 0.08} return {"basket": 0.16, "quality": 0.34, "leader": 0.10, "soxl": 0.03, "soxs": 0.00, "cash": 0.37} if strategy == "defensive_trend": if signal["crash_risk"] or drawdown < -0.08: return {"basket": 0.15, "quality": 0.25, "leader": 0.00, "soxl": 0.00, "soxs": 0.10 if pending_cash_days == 0 else 0.00, "cash": 0.50 if pending_cash_days == 0 else 0.60} if signal["risk_on"]: return {"basket": 0.35, "quality": 0.35, "leader": 0.15, "soxl": 0.05, "soxs": 0.00, "cash": 0.10} return {"basket": 0.25, "quality": 0.35, "leader": 0.05, "soxl": 0.00, "soxs": 0.00, "cash": 0.35} if strategy == "adaptive_barbell": if regime == "crash" or (signal["crash_risk"] and drawdown < -0.03): return {"basket": 0.08, "quality": 0.25, "leader": 0.00, "soxl": 0.00, "soxs": 0.12 if pending_cash_days == 0 else 0.00, "cash": 0.55 if pending_cash_days == 0 else 0.67} if signal["confirmed_breakdown"]: return {"basket": 0.12, "quality": 0.30, "leader": 0.00, "soxl": 0.00, "soxs": 0.08 if pending_cash_days == 0 else 0.00, "cash": 0.50 if pending_cash_days == 0 else 0.58} if signal["rebound"]: return {"basket": 0.22, "quality": 0.28, "leader": 0.25, "soxl": 0.12, "soxs": 0.00, "cash": 0.13} if signal["risk_on"]: return {"basket": 0.22, "quality": 0.32, "leader": 0.24, "soxl": 0.10, "soxs": 0.00, "cash": 0.12} return {"basket": 0.18, "quality": 0.34, "leader": 0.08, "soxl": 0.00, "soxs": 0.00, "cash": 0.40} raise ValueError(f"unknown strategy: {strategy}") def pick_regime_from_transitions(rng: random.Random, transitions: dict, current: str) -> str: roll = rng.random() cumulative = 0.0 for regime, probability in transitions[current]: cumulative += probability if roll <= cumulative: return regime return transitions[current][-1][0] def simulate_one(strategy: str, rng: random.Random, days: int, regimes: dict, transitions: dict) -> tuple[float, float, float, bool, bool, int, int, bool, bool]: equity = START_CAPITAL_KRW high = equity max_dd = 0.0 max_dd_krw = 0.0 monthly_start = equity week_start = equity hit_weekly = False hit_monthly = False weekly_count = 0 monthly_count = 0 hit_double = False regime = rng.choice(["risk_on", "selective", "chop", "selloff"]) pending_cash_days = 0 for day in range(days): regime = pick_regime_from_transitions(rng, transitions, regime) signal = signal_quality(rng, regime) drawdown = equity / high - 1.0 if signal["crash_risk"] and strategy in {"defensive_trend", "adaptive_barbell", "guarded_aggressive_barbell"}: pending_cash_days = max(pending_cash_days, rng.choice([0, 1, 2])) else: pending_cash_days = max(0, pending_cash_days - 1) weights = weights_for_strategy(strategy, regime, signal, drawdown, pending_cash_days, equity) daily_return = 0.0 for asset, weight in weights.items(): if asset == "cash": asset_return = 0.00005 else: asset_return = normal_return(rng, regimes, regime, asset) daily_return += weight * asset_return if strategy == "adaptive_barbell": # Represents target/stop discipline and no forced all-in rotations. daily_return = max(daily_return, -0.045) elif strategy == "guarded_aggressive_barbell": daily_return = max(daily_return, -0.065) elif strategy == "defensive_trend": daily_return = max(daily_return, -0.055) elif strategy == "aggressive_leverage": daily_return = max(daily_return, -0.155) elif strategy == "turbo_soxl_soxs_switch": daily_return = max(daily_return, -0.220) elif strategy == "selective_leverage_burst": daily_return = max(daily_return, -0.130) elif strategy == "compounding_tactical_ladder": daily_return = max(daily_return, -0.145) else: daily_return = max(daily_return, -0.10) equity *= 1.0 + daily_return high = max(high, equity) max_dd = min(max_dd, equity / high - 1.0) max_dd_krw = min(max_dd_krw, equity - high) if equity >= week_start + 2_000_000: hit_weekly = True if equity >= monthly_start + 10_000_000: hit_monthly = True if equity >= START_CAPITAL_KRW * 2: hit_double = True if day % 5 == 4: if equity >= week_start + 2_000_000: weekly_count += 1 week_start = equity if day % 21 == 20: if equity >= monthly_start + 10_000_000: monthly_count += 1 monthly_start = equity return equity, max_dd, max_dd_krw, hit_weekly, hit_monthly, weekly_count, monthly_count, hit_double, equity < START_CAPITAL_KRW def percentile(values: list[float], p: float) -> float: if not values: return float("nan") ordered = sorted(values) k = (len(ordered) - 1) * p low = math.floor(k) high = math.ceil(k) if low == high: return ordered[int(k)] return ordered[low] * (high - k) + ordered[high] * (k - low) def summarize(result: StrategyResult) -> dict[str, float | str]: finals = result.final_equity profits = [(value - START_CAPITAL_KRW) for value in finals] returns = [(value / START_CAPITAL_KRW - 1.0) for value in finals] dds = result.max_drawdown dd_krw = result.max_drawdown_krw mean_return = statistics.mean(returns) median_return = statistics.median(returns) p5_return = percentile(returns, 0.05) p95_return = percentile(returns, 0.95) mean_profit = statistics.mean(profits) median_profit = statistics.median(profits) p5_profit = percentile(profits, 0.05) p95_profit = percentile(profits, 0.95) median_dd = statistics.median(dds) p5_dd = percentile(dds, 0.05) median_dd_krw = statistics.median(dd_krw) p5_dd_krw = percentile(dd_krw, 0.05) hit_double = sum(result.hit_double) / len(result.hit_double) below_start = sum(result.ended_below_start) / len(result.ended_below_start) weekly = sum(result.hit_weekly_quest) / len(result.hit_weekly_quest) monthly = sum(result.hit_monthly_quest) / len(result.hit_monthly_quest) avg_weekly_count = statistics.mean(result.weekly_quest_count) avg_monthly_count = statistics.mean(result.monthly_quest_count) # Challenge-first KRW score: # - prioritize realized account growth in KRW, not average return optics; # - reward weekly/monthly/x2 challenge clearing; # - heavily penalize strategies that require large painful drawdowns. score = ( median_profit + mean_profit * 0.20 + p5_profit * 0.75 + p95_profit * 0.05 + avg_weekly_count * 2_000_000 + avg_monthly_count * 10_000_000 + hit_double * 35_000_000 + p5_dd_krw * 1.15 + median_dd_krw * 0.55 - below_start * 60_000_000 ) # Disqualify strategies that look profitable but demand too much pain. disqualified = False disqualification_reason = "" if p5_dd_krw < -30_000_000: disqualified = True disqualification_reason = "p5_drawdown_exceeds_30m_krw" if below_start > 0.05: disqualified = True disqualification_reason = "below_start_probability_exceeds_5pct" if p5_profit < 0: disqualified = True disqualification_reason = "p5_net_profit_negative" qualified_score = score if not disqualified else score - 1_000_000_000 return { "strategy": result.name, "mean_profit_krw": mean_profit, "median_profit_krw": median_profit, "p5_profit_krw": p5_profit, "p95_profit_krw": p95_profit, "mean_return_pct": mean_return * 100, "median_return_pct": median_return * 100, "p5_return_pct": p5_return * 100, "p95_return_pct": p95_return * 100, "median_final_krw": statistics.median(finals), "p5_final_krw": percentile(finals, 0.05), "p95_final_krw": percentile(finals, 0.95), "median_max_drawdown_pct": median_dd * 100, "p5_max_drawdown_pct": p5_dd * 100, "median_max_drawdown_krw": median_dd_krw, "p5_max_drawdown_krw": p5_dd_krw, "prob_weekly_quest_hit_pct": weekly * 100, "prob_monthly_quest_hit_pct": monthly * 100, "avg_weekly_quest_count": avg_weekly_count, "avg_monthly_quest_count": avg_monthly_count, "prob_x2_hit_pct": hit_double * 100, "prob_below_start_pct": below_start * 100, "challenge_net_profit_score_krw": score, "qualified_score_krw": qualified_score, "disqualified": "yes" if disqualified else "no", "disqualification_reason": disqualification_reason, } def load_regime_config(path: str | None) -> tuple[dict, dict, str]: if not path: return DEFAULT_REGIMES, DEFAULT_TRANSITIONS, "synthetic_default" payload = json.loads(Path(path).read_text()) return payload["regimes"], payload["transitions"], path def run_simulations(iterations: int, days: int, seed: int, regime_config: str | None = None) -> tuple[list[dict[str, float | str]], dict[str, float | str], str]: strategies = [ "passive_hold", "panic_sell_rebuy", "aggressive_leverage", "turbo_soxl_soxs_switch", "selective_leverage_burst", "compounding_tactical_ladder", "guarded_aggressive_barbell", "defensive_trend", "adaptive_barbell", ] rng = random.Random(seed) regimes, transitions, source = load_regime_config(regime_config) results: list[StrategyResult] = [] for strategy in strategies: result = StrategyResult(strategy, [], [], [], [], [], [], [], [], []) for _ in range(iterations): final, dd, dd_krw, weekly, monthly, weekly_count, monthly_count, double, below = simulate_one(strategy, rng, days, regimes, transitions) result.final_equity.append(final) result.max_drawdown.append(dd) result.max_drawdown_krw.append(dd_krw) result.hit_weekly_quest.append(weekly) result.hit_monthly_quest.append(monthly) result.weekly_quest_count.append(weekly_count) result.monthly_quest_count.append(monthly_count) result.hit_double.append(double) result.ended_below_start.append(below) results.append(result) summaries = [summarize(result) for result in results] summaries.sort(key=lambda row: float(row["qualified_score_krw"]), reverse=True) return summaries, summaries[0], source def write_csv(path: Path, rows: list[dict[str, float | str]]) -> None: fields = list(rows[0].keys()) with path.open("w", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(rows) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--iterations", type=int, default=1000) parser.add_argument("--days", type=int, default=252) parser.add_argument("--seed", type=int, default=20260606) parser.add_argument("--regime-config", default=None) parser.add_argument("--json", action="store_true") args = parser.parse_args() if args.iterations < 100: raise SystemExit("--iterations must be at least 100") if args.days < 252: raise SystemExit("--days must be at least 252 for the adopted account-growth simulation standard") summaries, best, regime_source = run_simulations(args.iterations, args.days, args.seed, args.regime_config) out_dir = Path(__file__).resolve().parent write_csv(out_dir / "strategy_monte_carlo_results.csv", summaries) payload = { "iterations": args.iterations, "days": args.days, "seed": args.seed, "start_capital_krw": START_CAPITAL_KRW, "regime_source": regime_source, "best_strategy": best, "summaries": summaries, } if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) else: for row in summaries: print( f"{row['strategy']}: median_profit={row['median_profit_krw']:,.0f} " f"p5_profit={row['p5_profit_krw']:,.0f} " f"p5_mdd={row['p5_max_drawdown_krw']:,.0f} " f"x2={row['prob_x2_hit_pct']:.1f}% " f"qualified={row['disqualified'] == 'no'} " f"score={row['qualified_score_krw']:,.0f}" ) print(f"BEST={best['strategy']}") return 0 if __name__ == "__main__": raise SystemExit(main())