"""Technical indicators as pure functions over `list[Bar]`.
Every function returns a list the same length as its input, with `None` for
positions where the indicator is not yet defined. Index-for-index alignment
means a strategy can read `series[i]` alongside `bars[i]` without offset
arithmetic, which is where this kind of code usually goes wrong.
`requirements.md` ยง4.1 (issue #109): this module must not hand-roll its own
indicator math. Every function below delegates the actual computation to
`talib` (TA-Lib 0.7.1, already installed via Homebrew on this machine and
shipped as a prebuilt PyPI wheel for macOS arm64 -- no C build step). `talib`
works in `numpy.float64`; every function here converts `Bar.close`/`.high`/
`.low` to `float` at its own INPUT boundary, calls the matching `talib`
function, and converts the result back to `Decimal` (or, for a dimensionless
value like RSI, plain `float`) at its own OUTPUT boundary -- the same
convert-at-the-boundary pattern this codebase already uses for `brokers/` and
`marketdata/` adapters. No `talib` float ever reaches a caller.
`ema()` is the one deliberate exception, and stays hand-rolled -- see its own
docstring for why: `talib.EMA` cannot reproduce its "defined from index 0"
contract without changing what `AlligatorStrategy` can safely assume about
`warmup_bars()`, the same "no standard off-the-shelf equivalent for this
specific convention" carve-out `research/strategies.py` uses for PSAR,
the Markov signal and the VWAP scalp.
Price-valued outputs are `Decimal` (they are money). Dimensionless ones such as
RSI are `float`.
"""
import math
from collections.abc import Sequence
from dataclasses import dataclass
from decimal import Decimal
import numpy as np
import talib
from trader.domain import Bar
__all__ = [
"BollingerBands",
"atr",
"bollinger",
"donchian_high",
"donchian_low",
"ema",
"macd",
"rsi",
"sma",
"stdev",
]
def _period_label(name: str) -> str:
"""`name`, with the word "period" appended unless it already ends in one.
Every plain function passes its own bare name (`"sma"` ->
`"sma period"`); `macd`'s three distinct parameters pass a name that
already reads as a complete noun phrase (`"macd fast_period"`), which
would otherwise double up as "macd fast_period period must be...".
"""
return name if name.endswith("period") else f"{name} period"
def _require_positive_period(period: int, name: str) -> None:
"""Reject a non-positive period.
Negative values are the reason this exists: Python's negative indexing
would let them write into valid positions and return a well-formed list of
meaningless numbers, which is worse than raising.
"""
if period < 1:
raise ValueError(f"{_period_label(name)} must be >= 1, got {period}")
#: talib's own floor for RSI/BBANDS/MACD: a period of 1 is mathematically
#: undefined for an oscillator, a band, or a moving-average *difference* (a
#: single point has no spread and no history to smooth against), and talib
#: enforces this itself with a bare `Bad Parameter` error that names neither
#: this module nor the strategy that triggered it -- confirmed empirically
#: 2026-09-01 against this installed talib 0.7.1 (`TA_RSI`/`TA_BBANDS`/
#: `TA_MACD` all raise `TA_BAD_PARAM` for `timeperiod=1`; `SMA`/`EMA`/`ATR`/
#: `MAX`/`MIN` do not). Caught here, with the same named-`ValueError`
#: contract every other guard in this module already uses, rather than
#: surfacing as talib's own undecorated exception.
_MIN_OSCILLATOR_PERIOD = 2
def _require_period_at_least_two(period: int, name: str) -> None:
_require_positive_period(period, name)
if period < _MIN_OSCILLATOR_PERIOD:
raise ValueError(
f"{_period_label(name)} must be >= {_MIN_OSCILLATOR_PERIOD}, got {period}"
)
def _to_float_array(values: Sequence[Decimal]) -> np.ndarray:
"""`Decimal` closes/highs/lows, converted to `float64` at talib's input
boundary -- the same direction every price this codebase hands to a
float-only library already crosses (`tearsheet/report.py`,
`research/data.py`)."""
return np.array([float(v) for v in values], dtype=np.float64)
def _decimal_or_none(value: float) -> Decimal | None:
"""A talib output value, converted back to `Decimal` at the boundary.
`Decimal(str(value))` rather than `Decimal(value)` -- the same reasoning
`macd.py`'s own `_to_decimal_param` and this codebase's other
float-to-Decimal boundary conversions already use: `Decimal(value)` on a
`float` carries that float's binary-fraction error into a number that can
reach an order-placement decision; `str()` first gives the shortest
decimal repr Python already computed for display, so the exact same
value round-trips back through `Decimal` without inventing extra digits.
`nan` is talib's convention for "not yet defined" (its own lookback
period) -- translated to `None` at this same boundary, the convention
every function in this module already used before this rewrite.
"""
if math.isnan(value):
return None
return Decimal(str(value))
def _float_or_none(value: float) -> float | None:
"""A dimensionless talib output (RSI), `None` for talib's `nan`."""
if math.isnan(value):
return None
return float(value)
[docs]
def sma(bars: Sequence[Bar], period: int) -> list[Decimal | None]:
"""Simple moving average of the close, via `talib.SMA`."""
_require_positive_period(period, "sma")
if not bars:
return []
closes = _to_float_array([b.close for b in bars])
result = talib.SMA(closes, timeperiod=period)
return [_decimal_or_none(v) for v in result]
[docs]
def stdev(bars: Sequence[Bar], period: int) -> list[Decimal | None]:
"""Population standard deviation of the close, via `talib.STDDEV`.
`nbdev=1` is talib's own default multiplier -- the Bollinger convention
(population, not sample) this function has always documented; verified
2026-09-01 against `talib.STDDEV([1,2,3,4,5], timeperiod=5, nbdev=1) ==
sqrt(2)`, the same population-variance value this module's own hand-rolled
version always produced for that fixture.
"""
_require_positive_period(period, "stdev")
if not bars:
return []
closes = _to_float_array([b.close for b in bars])
result = talib.STDDEV(closes, timeperiod=period, nbdev=1)
return [_decimal_or_none(v) for v in result]
[docs]
def ema(bars: Sequence[Bar], period: int) -> list[Decimal]:
"""Exponential moving average of the close, `alpha = 2 / (period + 1)`.
**Deliberately still hand-rolled** -- the one exception to this module's
"delegate to talib" rule, same status as `research/strategies.py`'s PSAR,
Markov and VWAP-scalp families: not because talib lacks an EMA (it does
not), but because talib's `EMA` cannot reproduce this function's contract
without a real behavioural change downstream.
Confirmed empirically 2026-09-01: `talib.EMA` -- even under
`talib.set_compatibility(1)` (Metastock mode, which seeds the recursion
from the first raw value exactly as this function does, rather than an
SMA of the first `period` values) -- still reports `nan` for the first
`period - 1` indices, because talib's *lookback* is fixed at `period - 1`
independent of the compatibility mode; only the seed differs. This
function's contract, which `AlligatorStrategy` depends on, is that `ema`
is **defined from index 0** (`ewm(adjust=False)`'s convention, no leading
`None` run at all -- see `test_ema_is_defined_from_the_first_bar_unlike_
sma`). Adopting talib's leading-`None` run here would silently understate
`AlligatorStrategy.warmup_bars()` by `period - 1` bars for jaw/teeth/lips
each (their `warmup_bars()` computation assumes `ema` has no warmup of its
own beyond `period + shift`), which is exactly the class of behavioural
regression the issue's own carve-out for Alligator's "custom fan logic"
was written to avoid touching. talib has no parameter that reports the
values it already computes internally for those leading indices, so there
is no way to get talib's own math AND this contract at once.
Stdlib-Decimal recursion, unchanged: seeded with the first close
(`out[0] = bars[0].close`) and then `out[i] = (1 - alpha) * out[i - 1] +
alpha * close[i]`, matching pandas' `ewm(span=period, adjust=False)
.mean()`. A caller that wants a "trust this only once it has had time to
converge" guarantee (as `AlligatorStrategy` does) enforces that itself;
`ema` reports what pandas would report at every index.
"""
_require_positive_period(period, "ema")
if not bars:
return []
alpha = Decimal(2) / Decimal(period + 1)
out: list[Decimal] = [bars[0].close]
for bar in bars[1:]:
out.append((Decimal(1) - alpha) * out[-1] + alpha * bar.close)
return out
[docs]
def rsi(bars: Sequence[Bar], period: int) -> list[float | None]:
"""Relative Strength Index, via `talib.RSI` (Wilder's smoothing).
One deliberate value change from this module's old hand-rolled version:
a perfectly flat run (zero gains AND zero losses) now reads `0.0`, not a
"neutral" `50.0` -- that was this module's own hand-picked convention for
the undefined `0/0` case, and talib's real, shipped convention differs.
Confirmed empirically 2026-09-01: `talib.RSI([7,7,7,7], timeperiod=2)[-1]
== 0.0`. Every other case this module's tests pinned (monotonic up ->
100.0, monotonic down -> 0.0, and the hand-derived Wilder-seeded 50.0/75.0
two-step example) matched talib exactly, unchanged.
`period` must be `>= 2`: talib rejects `timeperiod=1` outright (RSI is
undefined over a single point -- there is no prior close to diff against
within the window).
"""
_require_period_at_least_two(period, "rsi")
if not bars:
return []
closes = _to_float_array([b.close for b in bars])
result = talib.RSI(closes, timeperiod=period)
return [_float_or_none(v) for v in result]
[docs]
@dataclass(frozen=True, slots=True)
class BollingerBands:
"""Upper, middle, and lower band. All price-valued, so all `Decimal`."""
upper: Decimal
middle: Decimal
lower: Decimal
def _shift_forward_one(values: np.ndarray) -> np.ndarray:
"""`values`, shifted so `out[i] is values[i - 1]` and `out[0]` is `nan`.
The Donchian convention this module has always used: a channel "ending at
i-1, excluding bar i" so `close[i] > donchian_high[i]` is a meaningful
breakout test rather than a channel that already contains bar i's own
high. `talib.MAX`/`talib.MIN` report the trailing window INCLUDING the
current bar (the standard convention), so this shift is what recovers the
excluding-today shape -- the same `.shift(1)` `research/strategies.py`
already applies to its own `rolling(...).max()` breakout level, now
applied to talib's equivalent instead of pandas'.
"""
out = np.roll(values, 1)
out[0] = np.nan
return out
[docs]
def donchian_high(bars: Sequence[Bar], period: int) -> list[Decimal | None]:
"""Highest high of the `period` bars ENDING AT i-1, excluding bar i.
Via `talib.MAX`, shifted forward one bar -- see `_shift_forward_one`.
"""
_require_positive_period(period, "donchian_high")
if not bars:
return []
highs = _to_float_array([b.high for b in bars])
raw = talib.MAX(highs, timeperiod=period)
shifted = _shift_forward_one(raw)
return [_decimal_or_none(v) for v in shifted]
[docs]
def donchian_low(bars: Sequence[Bar], period: int) -> list[Decimal | None]:
"""Lowest low of the `period` bars ending at i-1, excluding bar i.
Via `talib.MIN`, shifted forward one bar -- see `_shift_forward_one`.
"""
_require_positive_period(period, "donchian_low")
if not bars:
return []
lows = _to_float_array([b.low for b in bars])
raw = talib.MIN(lows, timeperiod=period)
shifted = _shift_forward_one(raw)
return [_decimal_or_none(v) for v in shifted]
[docs]
def atr(bars: Sequence[Bar], period: int) -> list[Decimal | None]:
"""Average True Range, via `talib.ATR` (Wilder's smoothing, issue #63;
delegated to talib, issue #109).
Confirmed empirically 2026-09-01: `talib.ATR` reproduces this module's
old hand-rolled Wilder ATR byte-for-byte on every fixture in
`tests/indicators/test_atr.py` -- same seeding (a simple average of the
first `period` true ranges), same first-defined index (`period`), same
true-range formula (`max(high-low, |high-prev_close|,
|low-prev_close|)`). No hardcoded test value needed to change for this
function.
"""
_require_positive_period(period, "atr")
if not bars:
return []
highs = _to_float_array([b.high for b in bars])
lows = _to_float_array([b.low for b in bars])
closes = _to_float_array([b.close for b in bars])
result = talib.ATR(highs, lows, closes, timeperiod=period)
return [_decimal_or_none(v) for v in result]
[docs]
def bollinger(
bars: Sequence[Bar], period: int, num_std: int
) -> list[BollingerBands | None]:
"""Bollinger bands: SMA of the close plus/minus `num_std` population
stdevs, via `talib.BBANDS` (`matype=0` -- a simple moving average middle
band, matching this function's own always-SMA convention; `nbdevup` /
`nbdevdn` both `num_std`, matching this function's own single-parameter
symmetric-band shape).
`period` must be `>= 2`, the same talib floor `rsi` documents: a single
point has no spread to measure a deviation against.
"""
_require_period_at_least_two(period, "bollinger")
if not bars:
return []
closes = _to_float_array([b.close for b in bars])
upper, middle, lower = talib.BBANDS(
closes,
timeperiod=period,
nbdevup=float(num_std),
nbdevdn=float(num_std),
matype=0,
)
out: list[BollingerBands | None] = []
for u, m, low_v in zip(upper, middle, lower, strict=True):
if math.isnan(u) or math.isnan(m) or math.isnan(low_v):
out.append(None)
continue
out.append(
BollingerBands(
upper=Decimal(str(u)),
middle=Decimal(str(m)),
lower=Decimal(str(low_v)),
)
)
return out
[docs]
def macd(
bars: Sequence[Bar], fast_period: int, slow_period: int, signal_period: int
) -> tuple[list[Decimal | None], list[Decimal | None]]:
"""The MACD line and its signal line, via `talib.MACD` (issue #109).
Added for `MacdStrategy` (`src/trader/strategies/macd.py`), which
previously hand-rolled its own EMA/MACD recursion internally rather than
going through this module at all. `talib.MACD` computes exactly the two
series `MacdStrategy` needs -- `macd = EMA(close, fast) - EMA(close,
slow)`, `signal = EMA(macd, signal_period)` -- with the same
SMA-seeded-EMA convention `MacdStrategy`'s own former hand-rolled `_ema`
helper used, so the underlying algorithm is unchanged; only the engine
computing it is.
Both `macd_line` and `signal_line` are Decimal (a MACD value is a
difference of two prices -- money) and index-aligned with `bars`, `None`
before each series is defined. `fast_period`/`slow_period`/
`signal_period` must each be `>= 2` -- talib's own floor, the same
reasoning `rsi`/`bollinger` document: an EMA "difference" needs at least
two periods' worth of smoothing to mean anything.
"""
_require_period_at_least_two(fast_period, "macd fast_period")
_require_period_at_least_two(slow_period, "macd slow_period")
_require_period_at_least_two(signal_period, "macd signal_period")
if not bars:
return [], []
closes = _to_float_array([b.close for b in bars])
macd_line, signal_line, _hist = talib.MACD(
closes,
fastperiod=fast_period,
slowperiod=slow_period,
signalperiod=signal_period,
)
return (
[_decimal_or_none(v) for v in macd_line],
[_decimal_or_none(v) for v in signal_line],
)