diff options
Diffstat (limited to 'scripts/optimize_asian_rsi.py')
| -rwxr-xr-x | scripts/optimize_asian_rsi.py | 214 |
1 files changed, 214 insertions, 0 deletions
diff --git a/scripts/optimize_asian_rsi.py b/scripts/optimize_asian_rsi.py new file mode 100755 index 0000000..9921447 --- /dev/null +++ b/scripts/optimize_asian_rsi.py @@ -0,0 +1,214 @@ +#!/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 = 90, base_price: float = 150.0) -> list[Candle]: + """Generate realistic SOL/USDT 5-minute candles. + + Simulates: + - Mild uptrend with periodic sharp dips during Asian session + - Intraday volatility (higher at session opens) + - Random walk with mean reversion + - Occasional momentum bursts that create RSI extremes + """ + random.seed(42) + candles = [] + price = base_price + start = datetime(2025, 1, 1, tzinfo=timezone.utc) + + for day in range(days): + # Mild upward bias to keep price above EMA + daily_trend = random.uniform(-0.005, 0.015) + + # Many days have a sharp V-dip during Asian session (1-2 bar crash + recovery) + # This creates RSI oversold while EMA stays above price briefly + dip_day = random.random() < 0.45 + dip_bar = random.randint(4, 18) if dip_day else -1 + # Sharp single-bar dip: 2-4% drop then immediate recovery + dip_magnitude = 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: # Asian open (our trading window) + vol = 0.003 + elif 13 <= hour < 16: # US session + vol = 0.0025 + else: + vol = 0.0015 + + # Base random walk with upward drift + change = random.gauss(daily_trend / 288, vol) + mean_rev = (base_price - price) / base_price * 0.001 + change += mean_rev + + # Session bar index within 00:00-01:55 UTC (bars 0-23) + session_bar = bar + + # Inject sharp V-dip: 1 bar crash, 1 bar partial recovery + if dip_day and 0 <= hour < 2: + if session_bar == dip_bar: + # Crash bar: sharp drop + change = -dip_magnitude + elif session_bar == dip_bar + 1: + # Recovery bar: bounce back most of the way + change = dip_magnitude * random.uniform(0.5, 0.8) + elif session_bar == dip_bar + 2: + # Continued recovery + change = dip_magnitude * 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 # Spike 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, params, balance=750.0, slippage=0.001, fee=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(): + print("=" * 60) + print("Asian Session RSI — Parameter Optimization") + print("SOL/USDT 5m | Capital: $750 (~100만원)") + print("=" * 60) + + days = 30 + 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 = [] + 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 = [] + 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']}, 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() |