"""MACD crossover strategy — trade on MACD/signal line crossovers.""" from datetime import datetime, timezone from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal from shared.strategies.base import BaseStrategy class MACDCrossoverStrategy(BaseStrategy): """Detect MACD / signal line crossovers for entry signals. Tracks previous MACD and signal values per ticker. On the first call for any ticker the strategy stores state and returns None. **Buy signal** (LONG): Bullish crossover — previous ``macd - signal <= 0`` and current ``macd - signal > 0``. **Sell signal** (SHORT): Bearish crossover — previous ``macd - signal >= 0`` and current ``macd - signal < 0``. Signal strength = ``abs(histogram) / atr`` (or ``abs(histogram) / 2.0`` when ATR is unavailable), clamped to [0, 1]. """ name: str = "macd_crossover" def __init__(self) -> None: self._prev_macd: dict[str, float] = {} self._prev_signal: dict[str, float] = {} async def evaluate( self, ticker: str, market: MarketSnapshot, sentiment: SentimentContext | None = None, ) -> TradeSignal | None: if market.macd is None or market.macd_signal is None: return None macd = market.macd signal = market.macd_signal # First call for this ticker — store state only. if ticker not in self._prev_macd: self._prev_macd[ticker] = macd self._prev_signal[ticker] = signal return None prev_diff = self._prev_macd[ticker] - self._prev_signal[ticker] curr_diff = macd - signal # Update stored state. self._prev_macd[ticker] = macd self._prev_signal[ticker] = signal direction: SignalDirection | None = None if prev_diff <= 0 and curr_diff > 0: direction = SignalDirection.LONG elif prev_diff >= 0 and curr_diff < 0: direction = SignalDirection.SHORT else: return None # Compute strength. histogram = market.macd_histogram if market.macd_histogram is not None else curr_diff if market.atr is not None and market.atr > 0: raw_strength = abs(histogram) / market.atr else: raw_strength = abs(histogram) / 2.0 strength = max(0.0, min(1.0, raw_strength)) return TradeSignal( ticker=ticker, direction=direction, strength=strength, strategy_sources=[self.name], timestamp=datetime.now(tz=timezone.utc), )