From f5521da2876a2c19afc24f370b3258f2be95f81a Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Thu, 2 Apr 2026 10:22:44 +0900 Subject: feat: add parameter optimization script for Asian Session RSI --- scripts/optimize_asian_rsi.py | 77 ++++++++++++++++++++++++------------------- 1 file changed, 44 insertions(+), 33 deletions(-) (limited to 'scripts/optimize_asian_rsi.py') 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] = [] -- cgit v1.2.3