#!/usr/bin/env python3 """Backtest and optimize MOC strategy on synthetic US stock data. Usage: python scripts/backtest_moc.py """ import random import sys from datetime import UTC, datetime, timedelta from decimal import Decimal from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "services" / "strategy-engine" / "src")) sys.path.insert(0, str(ROOT / "services" / "strategy-engine")) sys.path.insert(0, str(ROOT / "services" / "backtester" / "src")) sys.path.insert(0, str(ROOT / "shared" / "src")) from backtester.engine import BacktestEngine # noqa: E402 from strategies.moc_strategy import MocStrategy # noqa: E402 from shared.models import Candle # noqa: E402 def generate_stock_candles( symbol: str = "AAPL", days: int = 90, base_price: float = 180.0, daily_drift: float = 0.0003, # Slight upward bias daily_vol: float = 0.015, # 1.5% daily vol ) -> list[Candle]: """Generate realistic US stock intraday candles (5-min bars). Simulates: - Market hours only (14:30-21:00 UTC = 9:30-16:00 ET) - Opening gaps (overnight news effect) - Intraday volatility pattern (higher at open/close) - Volume pattern (U-shaped: high at open, low midday, high at close) """ candles = [] price = base_price start_date = datetime(2025, 1, 2, tzinfo=UTC) # Start on a Thursday trading_day = 0 current_date = start_date while trading_day < days: # Skip weekends if current_date.weekday() >= 5: current_date += timedelta(days=1) continue # Opening gap: overnight news effect gap_pct = random.gauss(daily_drift, daily_vol * 0.5) # Gap is ~50% of daily vol price *= 1 + gap_pct # Generate 78 5-minute bars (9:30-16:00 = 6.5 hours = 78 bars) intraday_bars = 78 for bar in range(intraday_bars): # Time: 14:30 UTC + bar * 5 minutes dt = current_date.replace(hour=14, minute=30) + timedelta(minutes=bar * 5) # Intraday volatility pattern (U-shaped) hour_of_day = bar / intraday_bars if hour_of_day < 0.1: # First 10% of day (opening) vol = daily_vol * 0.003 elif hour_of_day > 0.9: # Last 10% (closing) vol = daily_vol * 0.0025 else: # Middle of day vol = daily_vol * 0.001 # Add daily trend component intraday_drift = daily_drift / intraday_bars change = random.gauss(intraday_drift, vol) open_p = price close_p = price * (1 + change) high_p = max(open_p, close_p) * (1 + abs(random.gauss(0, vol * 0.3))) low_p = min(open_p, close_p) * (1 - abs(random.gauss(0, vol * 0.3))) # Volume pattern (U-shaped) if hour_of_day < 0.1 or hour_of_day > 0.85: volume = random.uniform(500000, 2000000) else: volume = random.uniform(100000, 500000) candles.append( Candle( symbol=symbol, timeframe="5Min", open_time=dt, open=Decimal(str(round(open_p, 2))), high=Decimal(str(round(high_p, 2))), low=Decimal(str(round(low_p, 2))), close=Decimal(str(round(close_p, 2))), volume=Decimal(str(int(volume))), ) ) price = close_p trading_day += 1 current_date += timedelta(days=1) return candles def run_backtest(candles, params, balance=750.0): """Run a single backtest.""" strategy = MocStrategy() strategy.configure(params) engine = BacktestEngine( strategy=strategy, initial_balance=Decimal(str(balance)), slippage_pct=0.0005, # 0.05% slippage (stocks have tighter spreads) taker_fee_pct=0.0, # Alpaca = 0% commission ) return engine.run(candles) def main(): random.seed(42) print("=" * 60) print("MOC Strategy Backtest — US Stocks") print("Capital: $750 (~100만원)") print("=" * 60) # Test across multiple stocks stocks = [ ("AAPL", 180.0, 0.0003, 0.015), ("MSFT", 420.0, 0.0004, 0.014), ("TSLA", 250.0, 0.0001, 0.030), ("NVDA", 800.0, 0.0005, 0.025), ("AMZN", 185.0, 0.0003, 0.018), ] # Parameter grid param_sets = [ { "quantity_pct": 0.2, "stop_loss_pct": sl, "rsi_min": rsi_min, "rsi_max": rsi_max, "ema_period": ema, "volume_avg_period": 20, "min_volume_ratio": 0.8, "buy_start_utc": 19, "buy_end_utc": 21, "sell_start_utc": 14, "sell_end_utc": 15, "max_positions": 5, } for rsi_min in [25, 30, 35] for rsi_max in [55, 60, 65] for sl in [1.5, 2.0, 3.0] for ema in [10, 20] ] print(f"\nParameter combinations: {len(param_sets)}") print(f"Stocks: {[s[0] for s in stocks]}") print("Generating 90 days of 5-min data per stock...\n") # Generate data for each stock all_candles = {} for symbol, base, drift, vol in stocks: all_candles[symbol] = generate_stock_candles( symbol, days=90, base_price=base, daily_drift=drift, daily_vol=vol ) print(f" {symbol}: {len(all_candles[symbol])} candles") # Test each parameter set across all stocks print( f"\nRunning {len(param_sets)} x {len(stocks)} = {len(param_sets) * len(stocks)} backtests..." ) param_results = [] for i, params in enumerate(param_sets): total_profit = Decimal("0") total_trades = 0 total_sharpe = 0.0 stock_details = [] for symbol, _, _, _ in stocks: result = run_backtest(all_candles[symbol], params) total_profit += result.profit total_trades += result.total_trades if result.detailed: total_sharpe += result.detailed.sharpe_ratio stock_details.append((symbol, result)) avg_sharpe = total_sharpe / len(stocks) if stocks else 0 param_results.append((params, total_profit, total_trades, avg_sharpe, stock_details)) if (i + 1) % 18 == 0: print(f" Progress: {i + 1}/{len(param_sets)}") # Sort by average Sharpe param_results.sort(key=lambda x: x[3], reverse=True) print("\n" + "=" * 60) print("TOP 5 PARAMETER SETS (by avg Sharpe across all stocks)") print("=" * 60) for rank, (params, profit, trades, sharpe, details) in enumerate(param_results[:5], 1): print(f"\n#{rank}:") print( f" RSI: {params['rsi_min']}-{params['rsi_max']}," f" SL: {params['stop_loss_pct']}%, EMA: {params['ema_period']}" ) print(f" Total Profit: ${float(profit):.2f}, Trades: {trades}, Avg Sharpe: {sharpe:.3f}") print(" Per stock:") for symbol, result in details: pct = float(result.profit_pct) dd = result.detailed.max_drawdown if result.detailed else 0 print(f" {symbol}: {pct:+.2f}% ({result.total_trades} trades, DD: {dd:.1f}%)") # Best params best = param_results[0] print("\n" + "=" * 60) print("RECOMMENDED PARAMETERS") print("=" * 60) bp = best[0] print(f" rsi_min: {bp['rsi_min']}") print(f" rsi_max: {bp['rsi_max']}") print(f" stop_loss_pct: {bp['stop_loss_pct']}") print(f" ema_period: {bp['ema_period']}") print(f" min_volume_ratio: {bp['min_volume_ratio']}") print(f"\n Avg Sharpe: {best[3]:.3f}") print(f" Total Profit: ${float(best[1]):.2f} across 5 stocks over 90 days") # Worst for comparison print("\n" + "=" * 60) print("WORST 3 PARAMETER SETS") print("=" * 60) for _rank, (params, profit, _trades, sharpe, _) in enumerate(param_results[-3:], 1): print( f" RSI({params['rsi_min']}-{params['rsi_max']})," f" SL={params['stop_loss_pct']}%, EMA={params['ema_period']}" ) print(f" Profit: ${float(profit):.2f}, Sharpe: {sharpe:.3f}") if __name__ == "__main__": main()