path: root/services/strategy-engine/strategies/asian_session_rsi.py

"""Asian Session RSI Strategy — 한국시간 9:00~11:00 단타.

규칙:
- SOL/USDT 5분봉
- 매수: RSI(14) < 25 + 볼륨 > 평균 + 센티먼트 OK
- 익절: +1.5%, 손절: -0.7%, 시간청산: 11:00 KST (02:00 UTC)
- 하루 최대 3회, 2연패 시 중단
- 센티먼트 필터: Fear & Greed > 80이면 매수 차단, 뉴스 극도 부정이면 차단
"""

from collections import deque
from decimal import Decimal
from datetime import datetime

import pandas as pd

from shared.models import Candle, Signal, OrderSide
from shared.sentiment import SentimentData
from strategies.base import BaseStrategy


class AsianSessionRsiStrategy(BaseStrategy):
    name: str = "asian_session_rsi"

    def __init__(self) -> None:
        super().__init__()
        self._rsi_period: int = 14
        self._rsi_oversold: float = 25.0
        self._rsi_overbought: float = 75.0
        self._quantity: Decimal = Decimal("0.1")
        self._take_profit_pct: float = 1.5
        self._stop_loss_pct: float = 0.7
        # Session: 00:00~02:00 UTC = 09:00~11:00 KST
        self._session_start_utc: int = 0
        self._session_end_utc: int = 2
        self._max_trades_per_day: int = 3
        self._max_consecutive_losses: int = 2
        self._use_sentiment: bool = True
        # Sentiment (updated externally before each session)
        self._sentiment: SentimentData | None = None
        # State
        self._closes: deque[float] = deque(maxlen=200)
        self._volumes: deque[float] = deque(maxlen=50)
        self._today: str | None = None
        self._trades_today: int = 0
        self._consecutive_losses: int = 0
        self._in_position: bool = False
        self._entry_price: float = 0.0

    @property
    def warmup_period(self) -> int:
        return self._rsi_period + 1

    def configure(self, params: dict) -> None:
        self._rsi_period = int(params.get("rsi_period", 14))
        self._rsi_oversold = float(params.get("rsi_oversold", 25.0))
        self._rsi_overbought = float(params.get("rsi_overbought", 75.0))
        self._quantity = Decimal(str(params.get("quantity", "0.1")))
        self._take_profit_pct = float(params.get("take_profit_pct", 1.5))
        self._stop_loss_pct = float(params.get("stop_loss_pct", 0.7))
        self._session_start_utc = int(params.get("session_start_utc", 0))
        self._session_end_utc = int(params.get("session_end_utc", 2))
        self._max_trades_per_day = int(params.get("max_trades_per_day", 3))
        self._max_consecutive_losses = int(params.get("max_consecutive_losses", 2))
        self._use_sentiment = bool(params.get("use_sentiment", True))

        if self._quantity <= 0:
            raise ValueError(f"Quantity must be positive, got {self._quantity}")
        if self._stop_loss_pct <= 0:
            raise ValueError(f"Stop loss must be positive, got {self._stop_loss_pct}")
        if self._take_profit_pct <= 0:
            raise ValueError(f"Take profit must be positive, got {self._take_profit_pct}")

        self._init_filters(
            require_trend=False,
            adx_threshold=25.0,
            min_volume_ratio=0.5,
            atr_stop_multiplier=1.5,
            atr_tp_multiplier=2.0,
        )

    def reset(self) -> None:
        super().reset()
        self._closes.clear()
        self._volumes.clear()
        self._today = None
        self._trades_today = 0
        self._consecutive_losses = 0
        self._in_position = False
        self._entry_price = 0.0
        self._sentiment = None

    def update_sentiment(self, sentiment: SentimentData) -> None:
        """Update sentiment data. Call before each trading session."""
        self._sentiment = sentiment

    def _check_sentiment(self) -> bool:
        """Check if sentiment allows buying. Returns True if OK."""
        if not self._use_sentiment or self._sentiment is None:
            return True  # No sentiment data, allow by default
        return not self._sentiment.should_block

    def _is_session_active(self, dt: datetime) -> bool:
        """Check if current time is within trading session."""
        hour = dt.hour
        if self._session_start_utc <= self._session_end_utc:
            return self._session_start_utc <= hour < self._session_end_utc
        # Wrap around midnight
        return hour >= self._session_start_utc or hour < self._session_end_utc

    def _compute_rsi(self) -> float | None:
        if len(self._closes) < self._rsi_period + 1:
            return None
        series = pd.Series(list(self._closes))
        delta = series.diff()
        gain = delta.clip(lower=0)
        loss = -delta.clip(upper=0)
        avg_gain = gain.ewm(com=self._rsi_period - 1, min_periods=self._rsi_period).mean()
        avg_loss = loss.ewm(com=self._rsi_period - 1, min_periods=self._rsi_period).mean()
        rs = avg_gain / avg_loss.replace(0, float("nan"))
        rsi = 100 - (100 / (1 + rs))
        val = rsi.iloc[-1]
        if pd.isna(val):
            return None
        return float(val)

    def _volume_above_average(self) -> bool:
        if len(self._volumes) < 20:
            return True  # Not enough data, allow
        avg = sum(self._volumes) / len(self._volumes)
        return self._volumes[-1] >= avg

    def on_candle(self, candle: Candle) -> Signal | None:
        self._update_filter_data(candle)

        close = float(candle.close)
        self._closes.append(close)
        self._volumes.append(float(candle.volume))

        # Daily reset
        day = candle.open_time.strftime("%Y-%m-%d")
        if self._today != day:
            self._today = day
            self._trades_today = 0
            # Don't reset consecutive_losses — carries across days

        # Check exit conditions first (if in position)
        if self._in_position:
            pnl_pct = (close - self._entry_price) / self._entry_price * 100

            # Take profit
            if pnl_pct >= self._take_profit_pct:
                self._in_position = False
                self._consecutive_losses = 0
                return self._apply_filters(
                    Signal(
                        strategy=self.name,
                        symbol=candle.symbol,
                        side=OrderSide.SELL,
                        price=candle.close,
                        quantity=self._quantity,
                        conviction=0.9,
                        reason=f"Take profit {pnl_pct:.2f}% >= {self._take_profit_pct}%",
                    )
                )

            # Stop loss
            if pnl_pct <= -self._stop_loss_pct:
                self._in_position = False
                self._consecutive_losses += 1
                return self._apply_filters(
                    Signal(
                        strategy=self.name,
                        symbol=candle.symbol,
                        side=OrderSide.SELL,
                        price=candle.close,
                        quantity=self._quantity,
                        conviction=1.0,
                        reason=f"Stop loss {pnl_pct:.2f}% <= -{self._stop_loss_pct}%",
                    )
                )

            # Time exit: session ended while in position
            if not self._is_session_active(candle.open_time):
                self._in_position = False
                if pnl_pct < 0:
                    self._consecutive_losses += 1
                else:
                    self._consecutive_losses = 0
                return self._apply_filters(
                    Signal(
                        strategy=self.name,
                        symbol=candle.symbol,
                        side=OrderSide.SELL,
                        price=candle.close,
                        quantity=self._quantity,
                        conviction=0.5,
                        reason=f"Time exit (session ended), PnL {pnl_pct:.2f}%",
                    )
                )

            return None  # Still in position, no action

        # Entry conditions
        if not self._is_session_active(candle.open_time):
            return None  # Outside trading hours

        if self._trades_today >= self._max_trades_per_day:
            return None  # Daily limit reached

        if self._consecutive_losses >= self._max_consecutive_losses:
            return None  # Consecutive loss limit

        if not self._check_sentiment():
            return None  # Sentiment blocked (extreme greed or very negative news)

        rsi = self._compute_rsi()
        if rsi is None:
            return None

        if rsi < self._rsi_oversold and self._volume_above_average():
            self._in_position = True
            self._entry_price = close
            self._trades_today += 1

            # Conviction: lower RSI = stronger signal
            conv = min((self._rsi_oversold - rsi) / self._rsi_oversold, 1.0)
            conv = max(conv, 0.3)

            sl = candle.close * (1 - Decimal(str(self._stop_loss_pct / 100)))
            tp = candle.close * (1 + Decimal(str(self._take_profit_pct / 100)))

            return self._apply_filters(
                Signal(
                    strategy=self.name,
                    symbol=candle.symbol,
                    side=OrderSide.BUY,
                    price=candle.close,
                    quantity=self._quantity,
                    conviction=conv,
                    stop_loss=sl,
                    take_profit=tp,
                    reason=f"RSI {rsi:.1f} < {self._rsi_oversold} (session active, vol OK)",
                )
            )

        return None