feat: add parameter optimization script for Asian Session RSI

1 files changed, 44 insertions, 33 deletions

diff --git a/scripts/optimize_asian_rsi.py b/scripts/optimize_asian_rsi.py
index 7cf24d7..209453a 100755
--- a/scripts/optimize_asian_rsi.py
+++ b/scripts/optimize_asian_rsi.py
@@ -1,6 +1,10 @@
 #!/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
 """
 
@@ -8,6 +12,7 @@ import sys
 from pathlib import Path
 from decimal import Decimal
 from datetime import datetime, timedelta, timezone
+from typing import Optional
 import random
 
 # Add paths
@@ -17,19 +22,35 @@ 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 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 realistic SOL/USDT 5-minute candles.
+    """Generate synthetic SOL/USDT 5-minute candles with dip-bounce patterns.
 
-    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)
+    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] = []
@@ -37,49 +58,38 @@ def generate_sol_candles(days: int = 60, base_price: float = 150.0) -> list[Cand
     start = datetime(2025, 1, 1, tzinfo=timezone.utc)
 
     for day in range(days):
-        daily_trend = random.uniform(-0.002, 0.008)
+        daily_trend = random.uniform(-0.003, 0.008)
 
-        # Most days have dip patterns to generate enough signals
+        # ~50 % of days feature a V-dip during the Asian session
         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)
+        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 5-minute bars per day
+        for bar in range(288):  # 288 five-minute bars per day
             dt = start + timedelta(days=day, minutes=bar * 5)
             hour = dt.hour
 
-            # Volatility varies by session
+            # Session-dependent volatility
             if 0 <= hour < 2:
-                vol = 0.002
+                vol = 0.003
             elif 13 <= hour < 16:
-                vol = 0.002
+                vol = 0.0025
             else:
-                vol = 0.001  # Keep off-hours quiet for low ADX
+                vol = 0.0015
 
-            # 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
-
+            # Inject V-dip: sharp single-bar crash then recovery
+            session_bar = bar
             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)
+                    change = dip_pct * random.uniform(0.5, 0.8)
                 elif session_bar == dip_bar + 2:
-                    # Continued recovery
-                    change = dip_pct * random.uniform(0.1, 0.4)
+                    change = dip_pct * random.uniform(0.1, 0.3)
 
             open_p = price
             close_p = price * (1 + change)
@@ -90,7 +100,7 @@ def generate_sol_candles(days: int = 60, base_price: float = 150.0) -> list[Cand
             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
+                    volume *= 2.5
 
             candles.append(
                 Candle(
@@ -118,7 +128,7 @@ def run_backtest(
     fee: float = 0.001,
 ):
     """Run a single backtest with given parameters."""
-    strategy = AsianSessionRsiStrategy()
+    strategy = OptimizableAsianRsi()
     strategy.configure(params)
 
     engine = BacktestEngine(
@@ -140,6 +150,7 @@ def main() -> None:
     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] = []