diff options
Diffstat (limited to 'scripts/optimize_asian_rsi.py')
| -rwxr-xr-x | scripts/optimize_asian_rsi.py | 214 |
1 files changed, 0 insertions, 214 deletions
diff --git a/scripts/optimize_asian_rsi.py b/scripts/optimize_asian_rsi.py deleted file mode 100755 index 9921447..0000000 --- a/scripts/optimize_asian_rsi.py +++ /dev/null @@ -1,214 +0,0 @@ -#!/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() |