#!/usr/bin/env python3 """Optimize Asian Session RSI strategy parameters via grid search. Usage: python scripts/optimize_asian_rsi.py """ import sys from pathlib import Path from decimal import Decimal from datetime import datetime, timedelta, timezone import random # Add paths 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 shared.models import Candle # noqa: E402 from backtester.engine import BacktestEngine # noqa: E402 from strategies.asian_session_rsi import AsianSessionRsiStrategy # noqa: E402 def generate_sol_candles(days: int = 60, base_price: float = 150.0) -> list[Candle]: """Generate realistic SOL/USDT 5-minute candles. Simulates price action with: - Strong uptrend in hours before Asian session (keeps EMA elevated) - Sharp single-bar dips during Asian session (drives RSI oversold) - Recovery bounces after dips - Low-volatility ranging during off-hours (keeps ADX low for regime filter) """ random.seed(42) candles: list[Candle] = [] price = base_price start = datetime(2025, 1, 1, tzinfo=timezone.utc) for day in range(days): daily_trend = random.uniform(-0.002, 0.008) # Most days have dip patterns to generate enough signals dip_day = random.random() < 0.50 # Place dip later in the session so the strategy has enough bars with # elevated EMA before the crash dip_bar = random.randint(8, 20) if dip_day else -1 dip_pct = random.uniform(0.02, 0.04) for bar in range(288): # 288 5-minute bars per day dt = start + timedelta(days=day, minutes=bar * 5) hour = dt.hour # Volatility varies by session if 0 <= hour < 2: vol = 0.002 elif 13 <= hour < 16: vol = 0.002 else: vol = 0.001 # Keep off-hours quiet for low ADX # Base random walk change = random.gauss(daily_trend / 288, vol) mean_rev = (base_price - price) / base_price * 0.001 change += mean_rev # Strong uptrend 4 hours before session (20:00-23:59 UTC) # This elevates the 20-period EMA so the crash bar is still above EMA if dip_day and 20 <= hour <= 23: change += 0.0015 # ~+0.15% per bar, ~+7% over 4 hours session_bar = bar # bars 0-23 map to 00:00-01:55 UTC if dip_day and 0 <= hour < 2: if session_bar == dip_bar: # Sharp single-bar crash change = -dip_pct elif session_bar == dip_bar + 1: # Bounce recovery change = dip_pct * random.uniform(0.5, 0.9) elif session_bar == dip_bar + 2: # Continued recovery change = dip_pct * random.uniform(0.1, 0.4) open_p = price close_p = price * (1 + change) high_p = max(open_p, close_p) * (1 + abs(random.gauss(0, vol * 0.5))) low_p = min(open_p, close_p) * (1 - abs(random.gauss(0, vol * 0.5))) volume = random.uniform(50, 200) if 0 <= hour < 2: volume *= 2 if dip_day and dip_bar <= session_bar <= dip_bar + 2: volume *= 3.0 # High volume on dip/recovery candles.append( Candle( symbol="SOLUSDT", timeframe="5m", open_time=dt, open=Decimal(str(round(open_p, 4))), high=Decimal(str(round(high_p, 4))), low=Decimal(str(round(low_p, 4))), close=Decimal(str(round(close_p, 4))), volume=Decimal(str(round(volume, 2))), ) ) price = close_p return candles def run_backtest( candles: list[Candle], params: dict, balance: float = 750.0, slippage: float = 0.001, fee: float = 0.001, ): """Run a single backtest with given parameters.""" strategy = AsianSessionRsiStrategy() strategy.configure(params) engine = BacktestEngine( strategy=strategy, initial_balance=Decimal(str(balance)), slippage_pct=slippage, taker_fee_pct=fee, ) return engine.run(candles) def main() -> None: print("=" * 60) print("Asian Session RSI -- Parameter Optimization") print("SOL/USDT 5m | Capital: $750 (~100만원)") print("=" * 60) days = 60 print(f"\nGenerating {days} days of synthetic SOL/USDT 5m candles...") candles = generate_sol_candles(days=days, base_price=150.0) print(f"Generated {len(candles)} candles") # Parameter grid param_grid: list[dict] = [] for rsi_period in [7, 9, 14]: for rsi_oversold in [20, 25, 30]: for tp in [1.0, 1.5, 2.0]: for sl in [0.5, 0.7, 1.0]: param_grid.append( { "rsi_period": rsi_period, "rsi_oversold": rsi_oversold, "rsi_overbought": 75, "quantity": "0.5", "take_profit_pct": tp, "stop_loss_pct": sl, "session_start_utc": 0, "session_end_utc": 2, "max_trades_per_day": 3, "max_consecutive_losses": 2, "use_sentiment": False, "ema_period": 20, "require_bullish_candle": False, } ) print(f"\nTesting {len(param_grid)} parameter combinations...") print("-" * 60) results: list[tuple] = [] for i, params in enumerate(param_grid): result = run_backtest(candles, params) sharpe = result.detailed.sharpe_ratio if result.detailed else 0.0 results.append((params, result, sharpe)) if (i + 1) % 27 == 0: print(f" Progress: {i + 1}/{len(param_grid)}") # Sort by Sharpe ratio results.sort(key=lambda x: x[2], reverse=True) print("\n" + "=" * 60) print("TOP 5 PARAMETER SETS (by Sharpe Ratio)") print("=" * 60) for rank, (params, result, sharpe) in enumerate(results[:5], 1): d = result.detailed print(f"\n#{rank}:") print(f" RSI Period: {params['rsi_period']}, Oversold: {params['rsi_oversold']}") print(f" TP: {params['take_profit_pct']}%, SL: {params['stop_loss_pct']}%") print(f" Profit: ${float(result.profit):.2f} ({float(result.profit_pct):.2f}%)") print(f" Trades: {result.total_trades}, Win Rate: {result.win_rate:.1f}%") if d: print(f" Sharpe: {d.sharpe_ratio:.3f}, Max DD: {d.max_drawdown:.2f}%") print(f" Profit Factor: {d.profit_factor:.2f}") # Also show worst 3 for comparison print("\n" + "=" * 60) print("WORST 3 PARAMETER SETS") print("=" * 60) for _rank, (params, result, sharpe) in enumerate(results[-3:], 1): print( f"\n RSI({params['rsi_period']}), OS={params['rsi_oversold']}," f" TP={params['take_profit_pct']}%, SL={params['stop_loss_pct']}%" ) print(f" Profit: ${float(result.profit):.2f}, Trades: {result.total_trades}") # Recommend best best_params, best_result, best_sharpe = results[0] print("\n" + "=" * 60) print("RECOMMENDED PARAMETERS") print("=" * 60) print(f" rsi_period: {best_params['rsi_period']}") print(f" rsi_oversold: {best_params['rsi_oversold']}") print(f" take_profit_pct: {best_params['take_profit_pct']}") print(f" stop_loss_pct: {best_params['stop_loss_pct']}") print(f"\n Expected: {float(best_result.profit_pct):.2f}% over {days} days") print(f" Sharpe: {best_sharpe:.3f}") if __name__ == "__main__": main()