Source code for trader.strategies.ma_crossover

"""Trend following on a moving-average crossover.

Buy the golden cross (short SMA crosses above long SMA); sell the death cross
(short SMA crosses below long SMA). Unlike the level-based strategies
(`rsi_revert`, `bollinger_revert`), the signal here is a *transition*, so
detecting it needs the current bar's SMAs and the prior bar's.
"""

from collections.abc import Sequence
from dataclasses import dataclass
from typing import ClassVar

from trader.domain import Bar, Position
from trader.indicators import sma
from trader.strategies.base import Action, Signal

__all__ = ["MaCrossoverStrategy"]

#: A crossover is a transition, so detecting one needs the current bar's SMAs
#: and the prior bar's — two bars, at minimum.
_MIN_BARS_TO_DETECT_A_CROSSOVER = 2




@dataclass(frozen=True, slots=True)
class MaCrossoverStrategy:
    """Buy when the short SMA crosses above the long SMA; sell on the reverse cross.

    A crossover is a change in which average leads, not a level either sits at,
    so both the current bar and the bar before it need a defined short and long
    SMA before a cross can be detected.
    """

    id: str
    short_period: int = 10
    long_period: int = 30

    # See TurtleStrategy: `Strategy.backtestable`'s Protocol default is not
    # structurally inherited, so the attribute must actually exist here.
    # `ClassVar` keeps it out of the generated `__init__`/`__repr__`/`__eq__`.
    backtestable: ClassVar[bool] = True



    def warmup_bars(self) -> int:
        # `sma`'s first defined value sits at index period - 1. Detecting a
        # crossover compares that value against the one before it, so the
        # longer window needs period + 1 bars before a signal can fire.
        return max(self.short_period, self.long_period) + 1




    def evaluate(self, bars: Sequence[Bar], position: Position | None) -> Signal:
        if len(bars) < _MIN_BARS_TO_DETECT_A_CROSSOVER:
            return Signal(Action.HOLD, "Moving averages not yet defined")

        short_series = sma(bars, self.short_period)
        long_series = sma(bars, self.long_period)
        short_now, long_now = short_series[-1], long_series[-1]
        short_prev, long_prev = short_series[-2], long_series[-2]
        if (
            short_now is None
            or long_now is None
            or short_prev is None
            or long_prev is None
        ):
            return Signal(Action.HOLD, "Moving averages not yet defined")

        indicators = {
            "close": bars[-1].close,
            "short_sma": short_now,
            "long_sma": long_now,
        }

        # A crossover is the sign change of (short - long) between the prior bar
        # and this one, not a level either average sits at. `<=`/`>=` on the
        # prior side (rather than strict `<`/`>`) means a cross away from exact
        # equality still counts, which matters at the boundary bar itself.
        bullish_cross = short_prev <= long_prev and short_now > long_now
        bearish_cross = short_prev >= long_prev and short_now < long_now

        if position is None and bullish_cross:
            return Signal(
                Action.BUY,
                f"Short SMA {short_now} crossed above long SMA {long_now}",
                indicators,
            )
        if position is not None and bearish_cross:
            return Signal(
                Action.SELL,
                f"Short SMA {short_now} crossed below long SMA {long_now}",
                indicators,
            )
        # Two structurally different states reach this branch, so the reason has
        # to say which — the same defect `rsi_revert` and `bollinger_revert`
        # shipped and fixed: a reason that is true only in one of the two states
        # it is reached from is a false claim in the other.
        if position is None:
            return Signal(
                Action.HOLD,
                f"Flat, and short SMA {short_now} has not crossed above long SMA "
                f"{long_now}",
                indicators,
            )
        return Signal(
            Action.HOLD,
            f"Holding, and short SMA {short_now} has not crossed below long SMA "
            f"{long_now}",
            indicators,
        )