path: root/scripts/optimize_asian_rsi.py

#!/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()