"""Asian Session RSI Strategy — 한국시간 9:00~11:00 단타. 규칙: - SOL/USDT 5분봉 - 매수: RSI(14) < 25 + 볼륨 > 평균 - 익절: +1.5%, 손절: -0.7%, 시간청산: 11:00 KST (02:00 UTC) - 하루 최대 3회, 2연패 시 중단 """ from collections import deque from decimal import Decimal from datetime import datetime import pandas as pd from shared.models import Candle, Signal, OrderSide 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 # 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)) 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 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 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