#!/usr/bin/env python3 """Optimize Asian Session RSI strategy parameters via grid search. Uses synthetic SOL/USDT 5m candle data with dip-bounce patterns to exercise the strategy's RSI entry/exit logic across different parameter combinations. Usage: python scripts/optimize_asian_rsi.py """ import sys from pathlib import Path from decimal import Decimal from datetime import datetime, timedelta, timezone from typing import Optional 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, Signal # noqa: E402 from backtester.engine import BacktestEngine # noqa: E402 from strategies.asian_session_rsi import AsianSessionRsiStrategy # noqa: E402 class OptimizableAsianRsi(AsianSessionRsiStrategy): """Subclass that bypasses ADX/EMA base-class filters for synthetic data. The base-class ``_apply_filters`` enforces ADX regime and EMA-level checks that are nearly impossible to satisfy on synthetic data (RSI oversold and price-above-EMA are mutually exclusive on random walks). This subclass passes signals through unchanged so that the core RSI entry/exit logic can be optimized independently. """ def _apply_filters(self, signal: Optional[Signal]) -> Optional[Signal]: return signal def _price_above_ema(self) -> bool: # Bypass EMA filter for synthetic data optimisation return True def generate_sol_candles(days: int = 60, base_price: float = 150.0) -> list[Candle]: """Generate synthetic SOL/USDT 5-minute candles with dip-bounce patterns. Creates range-bound price action around *base_price* with periodic sharp dips during the Asian session window (00:00-02:00 UTC) followed by recovery bounces. This exercises the RSI oversold entry logic. """ 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.003, 0.008) # ~50 % of days feature a V-dip during the Asian session dip_day = random.random() < 0.50 dip_bar = random.randint(6, 18) if dip_day else -1 dip_pct = random.uniform(0.015, 0.035) for bar in range(288): # 288 five-minute bars per day dt = start + timedelta(days=day, minutes=bar * 5) hour = dt.hour # Session-dependent volatility if 0 <= hour < 2: vol = 0.003 elif 13 <= hour < 16: vol = 0.0025 else: vol = 0.0015 change = random.gauss(daily_trend / 288, vol) mean_rev = (base_price - price) / base_price * 0.001 change += mean_rev # Inject V-dip: sharp single-bar crash then recovery session_bar = bar if dip_day and 0 <= hour < 2: if session_bar == dip_bar: change = -dip_pct elif session_bar == dip_bar + 1: change = dip_pct * random.uniform(0.5, 0.8) elif session_bar == dip_bar + 2: change = dip_pct * random.uniform(0.1, 0.3) 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 *= 2.5 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 = OptimizableAsianRsi() 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") print("(base-class ADX/EMA filters bypassed for synthetic data)") # 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()