from collections import deque from decimal import Decimal import pandas as pd from shared.models import Candle, Signal, OrderSide from strategies.base import BaseStrategy class EmaCrossoverStrategy(BaseStrategy): name: str = "ema_crossover" def __init__(self) -> None: self._closes: deque[float] = deque(maxlen=500) self._short_period: int = 9 self._long_period: int = 21 self._quantity: Decimal = Decimal("0.01") self._prev_short_above: bool | None = None @property def warmup_period(self) -> int: return self._long_period def configure(self, params: dict) -> None: self._short_period = int(params.get("short_period", 9)) self._long_period = int(params.get("long_period", 21)) self._quantity = Decimal(str(params.get("quantity", "0.01"))) if self._short_period >= self._long_period: raise ValueError( f"EMA short_period must be < long_period, " f"got short={self._short_period}, long={self._long_period}" ) if self._short_period < 2: raise ValueError(f"EMA short_period must be >= 2, got {self._short_period}") if self._long_period < 2: raise ValueError(f"EMA long_period must be >= 2, got {self._long_period}") if self._quantity <= 0: raise ValueError(f"Quantity must be positive, got {self._quantity}") def reset(self) -> None: self._closes.clear() self._prev_short_above = None def on_candle(self, candle: Candle) -> Signal | None: self._closes.append(float(candle.close)) if len(self._closes) < self._long_period: return None series = pd.Series(list(self._closes)) short_ema = series.ewm(span=self._short_period, adjust=False).mean().iloc[-1] long_ema = series.ewm(span=self._long_period, adjust=False).mean().iloc[-1] short_above = short_ema > long_ema signal = None if self._prev_short_above is not None: if not self._prev_short_above and short_above: signal = Signal( strategy=self.name, symbol=candle.symbol, side=OrderSide.BUY, price=candle.close, quantity=self._quantity, reason=f"Golden Cross: short EMA ({short_ema:.2f}) crossed above long EMA ({long_ema:.2f})", ) elif self._prev_short_above and not short_above: signal = Signal( strategy=self.name, symbol=candle.symbol, side=OrderSide.SELL, price=candle.close, quantity=self._quantity, reason=f"Death Cross: short EMA ({short_ema:.2f}) crossed below long EMA ({long_ema:.2f})", ) self._prev_short_above = short_above return signal