82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
"""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),
|
|
)
|