stonks-oracle/services/signal_engine/signals/ma_stack.py

"""Moving average stack signal evaluator.

Detects bullish alignment (MA_10 > MA_20 > MA_50 > MA_200) and bearish
alignment (MA_10 < MA_20 < MA_50 < MA_200), producing a signal strength
proportional to the degree of alignment.

Full alignment (4/4 MAs in order) yields strength 1.0, partial alignment
(3/4) yields 0.6, and no alignment returns ``None`` (no signal).
"""

from __future__ import annotations

from services.signal_engine.models import OHLCVBar, SignalDirection, SignalResult
from services.signal_engine.signals.base import compute_sma, validate_lookback

# MA periods used for stack evaluation
MA_PERIODS: list[int] = [10, 20, 50, 200]

# Minimum number of bars required (longest MA period)
MIN_BARS: int = 200

# Strength values
_FULL_ALIGNMENT_STRENGTH: float = 1.0
_PARTIAL_ALIGNMENT_STRENGTH: float = 0.6

# Confidence multiplier (high confidence for clear alignment patterns)
_CONFIDENCE_MULTIPLIER: float = 0.9


class MAStackEvaluator:
    """Moving average stack signal evaluator.

    Satisfies the :class:`~services.signal_engine.signals.base.SignalEvaluator`
    protocol.

    Computes MA_10, MA_20, MA_50, and MA_200 and checks whether they are
    in bullish or bearish order.  Full alignment (all four in strict order)
    produces strength 1.0; partial alignment (any three consecutive in order)
    produces strength 0.6.  When no alignment is detected the evaluator
    returns ``None``.
    """

    # ------------------------------------------------------------------
    # Public API (SignalEvaluator protocol)
    # ------------------------------------------------------------------

    def evaluate(
        self,
        bars: list[OHLCVBar],
        timeframe: str,
    ) -> SignalResult | None:
        """Evaluate moving average stack alignment on *bars*.

        Returns ``None`` when there are fewer than 200 bars (insufficient
        data for MA_200) or when no alignment is detected.
        """
        if not validate_lookback(bars, MIN_BARS):
            return None

        # Compute all four moving averages
        ma_10 = compute_sma(bars, 10)
        ma_20 = compute_sma(bars, 20)
        ma_50 = compute_sma(bars, 50)
        ma_200 = compute_sma(bars, 200)

        # Safety check — compute_sma returns None on insufficient data
        if ma_10 is None or ma_20 is None or ma_50 is None or ma_200 is None:
            return None

        ma_values = [ma_10, ma_20, ma_50, ma_200]

        # Check full bullish alignment: MA_10 > MA_20 > MA_50 > MA_200
        full_bullish = ma_10 > ma_20 > ma_50 > ma_200

        # Check full bearish alignment: MA_10 < MA_20 < MA_50 < MA_200
        full_bearish = ma_10 < ma_20 < ma_50 < ma_200

        if full_bullish:
            return self._build_result(
                direction=SignalDirection.BULLISH,
                strength=_FULL_ALIGNMENT_STRENGTH,
                alignment="full_bullish",
                timeframe=timeframe,
                ma_values=ma_values,
            )

        if full_bearish:
            return self._build_result(
                direction=SignalDirection.BEARISH,
                strength=_FULL_ALIGNMENT_STRENGTH,
                alignment="full_bearish",
                timeframe=timeframe,
                ma_values=ma_values,
            )

        # Check partial alignment (3 out of 4 consecutive MAs in order)
        partial_bullish = self._check_partial_bullish(ma_values)
        partial_bearish = self._check_partial_bearish(ma_values)

        if partial_bullish:
            return self._build_result(
                direction=SignalDirection.BULLISH,
                strength=_PARTIAL_ALIGNMENT_STRENGTH,
                alignment="partial_bullish",
                timeframe=timeframe,
                ma_values=ma_values,
            )

        if partial_bearish:
            return self._build_result(
                direction=SignalDirection.BEARISH,
                strength=_PARTIAL_ALIGNMENT_STRENGTH,
                alignment="partial_bearish",
                timeframe=timeframe,
                ma_values=ma_values,
            )

        # No alignment detected — no signal
        return None

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _check_partial_bullish(ma_values: list[float]) -> bool:
        """Return ``True`` if any 3 consecutive MAs are in bullish order.

        Checks windows [0:3] and [1:4] of the ordered MA list
        (MA_10, MA_20, MA_50, MA_200) for strictly descending values
        (higher MA value = bullish when shorter period > longer period).
        """
        # Window 1: MA_10 > MA_20 > MA_50
        if ma_values[0] > ma_values[1] > ma_values[2]:
            return True
        # Window 2: MA_20 > MA_50 > MA_200
        if ma_values[1] > ma_values[2] > ma_values[3]:
            return True
        return False

    @staticmethod
    def _check_partial_bearish(ma_values: list[float]) -> bool:
        """Return ``True`` if any 3 consecutive MAs are in bearish order.

        Checks windows [0:3] and [1:4] of the ordered MA list
        for strictly ascending values (lower MA value = bearish when
        shorter period < longer period).
        """
        # Window 1: MA_10 < MA_20 < MA_50
        if ma_values[0] < ma_values[1] < ma_values[2]:
            return True
        # Window 2: MA_20 < MA_50 < MA_200
        if ma_values[1] < ma_values[2] < ma_values[3]:
            return True
        return False

    @staticmethod
    def _build_result(
        *,
        direction: SignalDirection,
        strength: float,
        alignment: str,
        timeframe: str,
        ma_values: list[float],
    ) -> SignalResult:
        """Construct a ``SignalResult`` for the MA stack signal."""
        confidence = strength * _CONFIDENCE_MULTIPLIER

        return SignalResult(
            signal_type="ma_stack",
            timeframe=timeframe,
            strength=strength,
            direction=direction,
            confidence=confidence,
            metadata={
                "ma_10": ma_values[0],
                "ma_20": ma_values[1],
                "ma_50": ma_values[2],
                "ma_200": ma_values[3],
                "alignment": alignment,
            },
        )