From 53cadcf7e34f05f77082e84f0696b56bcbcbae36 Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Thu, 2 Apr 2026 10:26:52 +0900 Subject: refactor: remove all crypto/Binance code, update to US stock symbols --- .../strategies/asian_session_rsi.py | 266 --------------------- 1 file changed, 266 deletions(-) delete mode 100644 services/strategy-engine/strategies/asian_session_rsi.py (limited to 'services/strategy-engine/strategies/asian_session_rsi.py') diff --git a/services/strategy-engine/strategies/asian_session_rsi.py b/services/strategy-engine/strategies/asian_session_rsi.py deleted file mode 100644 index 1874591..0000000 --- a/services/strategy-engine/strategies/asian_session_rsi.py +++ /dev/null @@ -1,266 +0,0 @@ -"""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 - self._ema_period: int = 20 - self._require_bullish_candle: bool = True - self._prev_candle_bullish: bool = False - # 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)) - self._ema_period = int(params.get("ema_period", 20)) - self._require_bullish_candle = bool(params.get("require_bullish_candle", 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 - self._prev_candle_bullish = False - - 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 _price_above_ema(self) -> bool: - """Check if current price is above short-term EMA.""" - if len(self._closes) < self._ema_period: - return True # Not enough data, allow by default - series = pd.Series(list(self._closes)) - ema_val = series.ewm(span=self._ema_period, adjust=False).mean().iloc[-1] - return self._closes[-1] >= ema_val - - 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)) - - # Track candle direction for bullish confirmation - is_bullish = float(candle.close) >= float(candle.open) - - # 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() and self._price_above_ema(): - if self._require_bullish_candle and not is_bullish: - return None # Wait for bullish candle confirmation - 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 -- cgit v1.2.3