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