114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
"""Moving Average Stack strategy — trade on alignment of multiple moving averages."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
|
|
from shared.strategies.base import BaseStrategy
|
|
|
|
|
|
class MAStackStrategy(BaseStrategy):
|
|
"""Generate signals based on moving average alignment (stacking).
|
|
|
|
Checks whether the price and four moving averages (EMA-9, EMA-21,
|
|
SMA-50, SMA-200) are aligned in bullish or bearish order. Also
|
|
detects golden/death cross (SMA-50 vs SMA-200).
|
|
|
|
**Buy signal** (LONG):
|
|
Bull alignment score >= 3 (at least 3 of 4 ordering conditions met).
|
|
|
|
**Sell signal** (SHORT):
|
|
Bear alignment score >= 3.
|
|
|
|
Signal strength = ``score / 4.0 + cross_bonus``, clamped to [0, 1].
|
|
Cross bonus = 0.15 for golden cross, -0.15 for death cross (applied
|
|
only when direction agrees).
|
|
"""
|
|
|
|
name: str = "ma_stack"
|
|
|
|
def __init__(self) -> None:
|
|
self._prev_sma_50: dict[str, float] = {}
|
|
self._prev_sma_200: dict[str, float] = {}
|
|
|
|
async def evaluate(
|
|
self,
|
|
ticker: str,
|
|
market: MarketSnapshot,
|
|
sentiment: SentimentContext | None = None,
|
|
) -> TradeSignal | None:
|
|
if (
|
|
market.ema_9 is None
|
|
or market.ema_21 is None
|
|
or market.sma_50 is None
|
|
or market.sma_200 is None
|
|
):
|
|
return None
|
|
|
|
price = market.current_price
|
|
ema_9 = market.ema_9
|
|
ema_21 = market.ema_21
|
|
sma_50 = market.sma_50
|
|
sma_200 = market.sma_200
|
|
|
|
# Count bull alignment: price > ema_9 > ema_21 > sma_50 > sma_200
|
|
bull_score = 0
|
|
if price > ema_9:
|
|
bull_score += 1
|
|
if ema_9 > ema_21:
|
|
bull_score += 1
|
|
if ema_21 > sma_50:
|
|
bull_score += 1
|
|
if sma_50 > sma_200:
|
|
bull_score += 1
|
|
|
|
# Count bear alignment: price < ema_9 < ema_21 < sma_50 < sma_200
|
|
bear_score = 0
|
|
if price < ema_9:
|
|
bear_score += 1
|
|
if ema_9 < ema_21:
|
|
bear_score += 1
|
|
if ema_21 < sma_50:
|
|
bear_score += 1
|
|
if sma_50 < sma_200:
|
|
bear_score += 1
|
|
|
|
# Detect golden/death cross (SMA-50 vs SMA-200).
|
|
cross_bonus = 0.0
|
|
if ticker in self._prev_sma_50:
|
|
prev_50 = self._prev_sma_50[ticker]
|
|
prev_200 = self._prev_sma_200[ticker]
|
|
# Golden cross: SMA-50 crosses above SMA-200.
|
|
if prev_50 <= prev_200 and sma_50 > sma_200:
|
|
cross_bonus = 0.15
|
|
# Death cross: SMA-50 crosses below SMA-200.
|
|
elif prev_50 >= prev_200 and sma_50 < sma_200:
|
|
cross_bonus = -0.15
|
|
|
|
# Update stored state.
|
|
self._prev_sma_50[ticker] = sma_50
|
|
self._prev_sma_200[ticker] = sma_200
|
|
|
|
# Determine direction.
|
|
if bull_score >= 3:
|
|
direction = SignalDirection.LONG
|
|
score = bull_score
|
|
# Only apply positive cross bonus for LONG.
|
|
bonus = max(0.0, cross_bonus)
|
|
elif bear_score >= 3:
|
|
direction = SignalDirection.SHORT
|
|
score = bear_score
|
|
# Only apply negative cross bonus (as positive value) for SHORT.
|
|
bonus = abs(min(0.0, cross_bonus))
|
|
else:
|
|
return None
|
|
|
|
raw_strength = score / 4.0 + bonus
|
|
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),
|
|
)
|