"""MACD trend-confirmed momentum, with an optional per-bar stop-loss.
Found while surveying `../backtesting` (issue #35, 2026-08-20): MACD-based
strategies were the best-performing entries in that repo's own grid-search
database. This ports two variants together, since the second is a strict
superset of the first rather than a separate strategy type.
**Core signal (standard 12/26/9 MACD):**
- MACD line = EMA(close, fast) - EMA(close, slow).
- Signal line = EMA(MACD line, signal_period) -- an EMA of a *derived*
series, not of the closes themselves.
- `momentum_positive = MACD > signal AND signal > 0`.
- `momentum_negative = MACD < signal AND signal < 0`.
- Trend filter: an N-period SMA of the close. `trend_upward = close > MA`;
`trend_downward = close < MA`.
- **Buy: momentum_positive AND trend_upward. Sell: momentum_negative AND
trend_downward.** The MA trend filter is a confirmation gate on top of a
bare MACD crossover, not a second independent signal -- dropping it
entirely is a different, out-of-scope variant (issue #35's "Out of scope").
**Optional stop-loss**, a per-bar exit check independent of the MACD/MA
signal above: sell if `close < entry_price * (1 - risk)` OR
`close < previous_close * (1 - risk)`, where `risk` is `stop_loss_pct`. It is
a modifier on this strategy's signal, not a separate strategy type, and it is
checked *before* the MACD/trend logic and can fire even while those
indicators are still warming up -- a held position needs protection from the
first bar it exists, not only once `warmup_bars()` worth of history has
accumulated.
Ships `mode: shadow` (see `config/strategies.yaml`); `Decimal` throughout, no
pandas/numpy crossing into `strategies/`, matching every other rule strategy
in this package.
**Issue #109 (`requirements.md` ยง4.1):** this strategy used to hand-roll its
own EMA/MACD recursion internally rather than going through
`trader/indicators/` at all -- the exact "same indicator, reinvented twice"
gap the issue found. It now calls `trader.indicators.macd`, which delegates
to `talib.MACD` (see that module for the boundary-conversion details). The
underlying algorithm is unchanged (an SMA-seeded EMA-of-EMA-difference, the
same construction the old hand-rolled `_ema`/`_macd_and_signal` used); only
the engine computing it moved. One real, new constraint from that move:
`talib.MACD` requires `fast_period`/`slow_period`/`signal_period` each `>=
2` (a period of 1 is undefined for an EMA difference), enforced in
`__post_init__` below with this strategy's own `ConfigError`, rather than
surfacing as talib's undecorated exception the first time `evaluate()` runs.
"""
from collections.abc import Sequence
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import ClassVar
from trader.domain import Bar, Position
from trader.errors import ConfigError
from trader.indicators import macd as macd_indicator
from trader.indicators import sma
from trader.strategies.base import Action, Signal
__all__ = ["MacdStrategy"]
_ZERO = Decimal(0)
_ONE = Decimal(1)
#: A previous close exists only once there are at least two bars.
_MIN_BARS_FOR_PREVIOUS_CLOSE = 2
#: `trader.indicators.macd` (talib.MACD) requires each period `>= 2` -- an EMA
#: difference is undefined over a single point. Checked in `__post_init__`
#: rather than left to surface as `trader.indicators.macd`'s own `ValueError`
#: (which names the parameter, not this strategy's id) the first time
#: `evaluate()` runs against real bars.
_MIN_MACD_PERIOD = 2
def _to_decimal_param(strategy_id: str, name: str, value: object) -> Decimal:
"""A configured YAML number as `Decimal`, never via `float()`.
Mirrors `_to_decimal_param` in `trailing_stop_floor.py`: `Decimal(str(value))`
is exact where `Decimal(value)` on a float would carry that float's
binary-fraction error into a number that gates a sell decision. `bool` is
rejected explicitly because it is an `int` subclass -- a YAML `true`
would otherwise silently become `Decimal(1)`, a 100% stop.
"""
if isinstance(value, bool):
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r} (bool)."
)
if isinstance(value, Decimal):
candidate = value
elif isinstance(value, (int, float, str)):
try:
candidate = Decimal(str(value))
except InvalidOperation:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r}."
) from None
else:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r}."
)
if not candidate.is_finite():
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a finite number, got {value!r}."
)
return candidate
def _stop_loss_reason(
close: Decimal,
*,
entry_price: Decimal,
entry_floor: Decimal,
below_entry: bool,
previous_close: Decimal | None,
previous_floor: Decimal | None,
below_previous: bool,
) -> str:
"""Name only the condition(s) that actually triggered -- never both by default."""
triggers = []
if below_entry:
triggers.append(f"entry-price floor {entry_floor} (entry {entry_price})")
if below_previous:
triggers.append(
f"previous-close floor {previous_floor} (previous close {previous_close})"
)
return f"Stop-loss: close {close} broke below " + " and ".join(triggers)
[docs]
@dataclass(frozen=True, slots=True)
class MacdStrategy:
"""MACD momentum, confirmed by a moving-average trend filter, with an
optional per-bar stop-loss.
See the module docstring for the buy/sell rule and the stop-loss
semantics.
"""
id: str
fast_period: int = 12
slow_period: int = 26
signal_period: int = 9
# A 50-period SMA trend filter. Deliberately a different, longer window
# than any of the MACD periods above (max 26): the filter's job is to
# confirm an established trend the MACD/signal comparison does not by
# itself guarantee (MACD can be locally positive during a longer
# downtrend's dead-cat bounce), so it needs to look further back than the
# indicator it is gating. 50 is the conventional "intermediate trend"
# lookback this app already uses elsewhere as a comparison baseline
# (`docs/ideas.md`'s trend/relative-strength inventory item).
trend_period: int = 50
# `Decimal | None` at the type level; the registry's generic parameter
# coercion only recognises bare `int`/`float` hints (see
# `registry.build_strategy`), so a configured value reaches here
# unconverted and `__post_init__` below does the real validation and
# `Decimal` conversion -- the same split `TrailingStopFloorStrategy` uses
# for `trail_pct`/`floor_pct`. Left `None` by default: the core MACD
# signal is what issue #35 asks to measure first, and enabling a second,
# independent exit path by default would make a shadow-mode comparison
# against the bare signal impossible to read cleanly. `config/
# strategies.yaml`'s shipped instance opts in explicitly at 2.5%, the
# value the source grid-search used, so the stop-loss path is still
# exercised every cycle in shadow.
stop_loss_pct: Decimal | None = None
# 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 __post_init__(self) -> None:
# Runs unconditionally -- before the `stop_loss_pct is None` early
# return below -- because `stop_loss_pct` is `None` by default
# (issue #109's own MacdStrategy default), and a period below
# talib's floor must be caught for every configured instance, not
# only ones that also opted into a stop-loss.
for period_name in ("fast_period", "slow_period", "signal_period"):
value = getattr(self, period_name)
if value < _MIN_MACD_PERIOD:
raise ConfigError(
f"Strategy {self.id!r}: {period_name} must be >= "
f"{_MIN_MACD_PERIOD} (talib.MACD is undefined for an EMA "
f"difference over a single point), got {value}."
)
if self.stop_loss_pct is None:
return
converted = _to_decimal_param(self.id, "stop_loss_pct", self.stop_loss_pct)
if converted <= 0:
raise ConfigError(
f"Strategy {self.id!r}: stop_loss_pct must be > 0, got "
f"{self.stop_loss_pct!r}."
)
# Frozen + slots still permits assignment via `object.__setattr__` in
# `__post_init__`; this is the one place this dataclass's own fields
# are ever mutated, and only to replace a raw YAML value with its
# validated `Decimal`.
object.__setattr__(self, "stop_loss_pct", converted)
[docs]
def warmup_bars(self) -> int:
# The signal line needs the MACD line defined for `signal_period`
# bars running, and the MACD line first becomes defined at whichever
# of the two price EMAs is slower -- index `slow_period - 1` (the
# slow EMA is always the larger of the two periods in a standard
# configuration; if it were not, this is still a safe, if loose,
# bound). So the signal line's own first defined index is
# `slow_period - 1 + signal_period - 1`, needing
# `slow_period + signal_period - 1` bars. The trend filter needs its
# own `trend_period` bars independently of all of that; take
# whichever bound is larger.
return max(self.trend_period, self.slow_period + self.signal_period - 1)
[docs]
def evaluate(self, bars: Sequence[Bar], position: Position | None) -> Signal:
if not bars:
return Signal(Action.HOLD, "No bars available")
close = bars[-1].close
previous_close = (
bars[-2].close if len(bars) >= _MIN_BARS_FOR_PREVIOUS_CLOSE else None
)
macd_line, signal_line = macd_indicator(
bars, self.fast_period, self.slow_period, self.signal_period
)
trend_ma = sma(bars, self.trend_period)
macd_now = macd_line[-1]
signal_now = signal_line[-1]
trend_now = trend_ma[-1]
indicators: dict[str, Decimal] = {"close": close}
if macd_now is not None:
indicators["macd"] = macd_now
if signal_now is not None:
indicators["signal"] = signal_now
if trend_now is not None:
indicators["trend_ma"] = trend_now
# The stop-loss is a risk-management override, independent of the
# MACD/trend signal below, and is checked first: it must be able to
# exit a held position even while the MACD/trend indicators are
# still warming up, and it must win over whatever the core signal
# would otherwise say.
if position is not None and self.stop_loss_pct is not None:
risk = self.stop_loss_pct
entry_price = position.avg_entry_price
entry_floor = entry_price * (_ONE - risk)
below_entry = close < entry_floor
previous_floor = (
previous_close * (_ONE - risk) if previous_close is not None else None
)
below_previous = previous_floor is not None and close < previous_floor
if below_entry or below_previous:
indicators["stop_loss_pct"] = risk
reason = _stop_loss_reason(
close,
entry_price=entry_price,
entry_floor=entry_floor,
below_entry=below_entry,
previous_close=previous_close,
previous_floor=previous_floor,
below_previous=below_previous,
)
return Signal(Action.SELL, reason, indicators)
if macd_now is None or signal_now is None or trend_now is None:
return Signal(
Action.HOLD, "MACD/trend indicators not yet defined", indicators
)
momentum_positive = macd_now > signal_now > _ZERO
momentum_negative = macd_now < signal_now < _ZERO
trend_upward = close > trend_now
trend_downward = close < trend_now
if position is None and momentum_positive and trend_upward:
return Signal(
Action.BUY,
f"MACD {macd_now} above signal {signal_now} (both positive) and "
f"close {close} above trend MA {trend_now}",
indicators,
)
if position is not None and momentum_negative and trend_downward:
return Signal(
Action.SELL,
f"MACD {macd_now} below signal {signal_now} (both negative) and "
f"close {close} below trend MA {trend_now}",
indicators,
)
# Two structurally different states reach this branch, so the reason
# has to say which -- the same defect already fixed in `rsi_revert`,
# `bollinger_revert` and `ma_crossover`: a reason true in one state
# can be false in the other.
if position is None:
return Signal(
Action.HOLD,
f"Flat, and momentum/trend do not both confirm a buy "
f"(momentum_positive={momentum_positive}, trend_upward={trend_upward})",
indicators,
)
return Signal(
Action.HOLD,
f"Holding, and momentum/trend do not both confirm a sell "
f"(momentum_negative={momentum_negative}, trend_downward={trend_downward})",
indicators,
)