diff options
Diffstat (limited to 'services/strategy-engine/strategies/moc_strategy.py')
| -rw-r--r-- | services/strategy-engine/strategies/moc_strategy.py | 81 |
1 files changed, 44 insertions, 37 deletions
diff --git a/services/strategy-engine/strategies/moc_strategy.py b/services/strategy-engine/strategies/moc_strategy.py index bb14e78..7eaa59e 100644 --- a/services/strategy-engine/strategies/moc_strategy.py +++ b/services/strategy-engine/strategies/moc_strategy.py @@ -6,6 +6,7 @@ Rules: - Screening: bullish candle, volume above average, RSI 30-60, positive momentum - Risk: -2% stop loss, max 5 positions, 20% of capital per position """ + from collections import deque from decimal import Decimal from datetime import datetime @@ -33,9 +34,9 @@ class MocStrategy(BaseStrategy): self._min_volume_ratio: float = 1.0 # Volume must be above average # Session times (UTC hours) self._buy_start_utc: int = 19 # 15:00 ET = 19:00 UTC (summer) / 20:00 UTC (winter) - self._buy_end_utc: int = 21 # 16:00 ET = 20:00 UTC / 21:00 UTC + self._buy_end_utc: int = 21 # 16:00 ET = 20:00 UTC / 21:00 UTC self._sell_start_utc: int = 13 # 9:00 ET = 13:00 UTC / 14:00 UTC - self._sell_end_utc: int = 15 # 10:00 ET = 14:00 UTC / 15:00 UTC + self._sell_end_utc: int = 15 # 10:00 ET = 14:00 UTC / 15:00 UTC self._max_positions: int = 5 # State self._closes: deque[float] = deque(maxlen=200) @@ -120,7 +121,7 @@ class MocStrategy(BaseStrategy): def _volume_above_average(self) -> bool: if len(self._volumes) < self._volume_avg_period: return True - avg = sum(list(self._volumes)[-self._volume_avg_period:]) / self._volume_avg_period + avg = sum(list(self._volumes)[-self._volume_avg_period :]) / self._volume_avg_period return avg > 0 and self._volumes[-1] / avg >= self._min_volume_ratio def _positive_momentum(self) -> bool: @@ -153,31 +154,35 @@ class MocStrategy(BaseStrategy): self._sold_today = True conv = 0.8 if pnl_pct > 0 else 0.5 - return self._apply_filters(Signal( - strategy=self.name, - symbol=candle.symbol, - side=OrderSide.SELL, - price=candle.close, - quantity=Decimal(str(self._quantity_pct)), - conviction=conv, - reason=f"MOC sell at open, PnL {pnl_pct:.2f}%", - )) + return self._apply_filters( + Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.SELL, + price=candle.close, + quantity=Decimal(str(self._quantity_pct)), + conviction=conv, + reason=f"MOC sell at open, PnL {pnl_pct:.2f}%", + ) + ) # --- STOP LOSS --- if self._in_position: pnl_pct = (close - self._entry_price) / self._entry_price * 100 if pnl_pct <= -self._stop_loss_pct: self._in_position = False - return self._apply_filters(Signal( - strategy=self.name, - symbol=candle.symbol, - side=OrderSide.SELL, - price=candle.close, - quantity=Decimal(str(self._quantity_pct)), - conviction=1.0, - stop_loss=candle.close, - reason=f"MOC stop loss {pnl_pct:.2f}% <= -{self._stop_loss_pct}%", - )) + return self._apply_filters( + Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.SELL, + price=candle.close, + quantity=Decimal(str(self._quantity_pct)), + conviction=1.0, + stop_loss=candle.close, + reason=f"MOC stop loss {pnl_pct:.2f}% <= -{self._stop_loss_pct}%", + ) + ) # --- BUY LOGIC (near market close) --- if not self._in_position and self._is_buy_window(candle.open_time): @@ -190,11 +195,11 @@ class MocStrategy(BaseStrategy): return None checks = [ - self._rsi_min <= rsi <= self._rsi_max, # RSI in sweet spot - self._is_bullish_candle(candle), # Bullish candle - self._price_above_ema(), # Above EMA (uptrend) - self._volume_above_average(), # Volume confirmation - self._positive_momentum(), # Short-term momentum + self._rsi_min <= rsi <= self._rsi_max, # RSI in sweet spot + self._is_bullish_candle(candle), # Bullish candle + self._price_above_ema(), # Above EMA (uptrend) + self._volume_above_average(), # Volume confirmation + self._positive_momentum(), # Short-term momentum ] if all(checks): @@ -209,15 +214,17 @@ class MocStrategy(BaseStrategy): sl = candle.close * (1 - Decimal(str(self._stop_loss_pct / 100))) - return self._apply_filters(Signal( - strategy=self.name, - symbol=candle.symbol, - side=OrderSide.BUY, - price=candle.close, - quantity=Decimal(str(self._quantity_pct)), - conviction=conv, - stop_loss=sl, - reason=f"MOC buy: RSI={rsi:.1f}, bullish candle, above EMA, vol OK", - )) + return self._apply_filters( + Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.BUY, + price=candle.close, + quantity=Decimal(str(self._quantity_pct)), + conviction=conv, + stop_loss=sl, + reason=f"MOC buy: RSI={rsi:.1f}, bullish candle, above EMA, vol OK", + ) + ) return None |