`indicators`¶

Typed aliases for TA-Lib functions. No formula in this project is reimplemented, so none can drift from the reference.

indicators ¶

topstep_backtest.indicators — every indicator is TA-Lib.

TalibIndicator drives any of TA-Lib's ~160 functions bar by bar; the named classes are typed spellings of the common ones. See talib_adapter for the causality, streaming-equals-batch, and bounded-history parity guarantees.

Indicator ¶

Bases: Protocol

The surface SymbolStrategy.use() requires of a registered indicator.

lookback `property` ¶

lookback: int

Bars needed before the indicator is ready.

ready `property` ¶

ready: bool

update ¶

update(bar: Bar) -> None

Source code in src/topstep_backtest/indicators/base.py

def update(self, bar: Bar) -> None: ...

NotReadyError ¶

Bases: Exception

value was read before the indicator had seen lookback bars.

ValueSource ¶

Bases: Protocol

Minimal structural input for Cross: anything exposing a Decimal series.

Runtime-checkable so use() can refuse a Cross over something with no value (notably another Cross) at registration, rather than letting it die with an AttributeError hours into a run — and only once both inner inputs happen to be ready.

lookback `property` ¶

lookback: int

ready `property` ¶

ready: bool

value `property` ¶

value: Decimal

WarmIndicator ¶

Bases: Protocol

An Indicator that also declares where its value stops depending on where the run started.

lookback is where a value EXISTS; history_bars is where it is a pure function of the last history_bars bars and nothing earlier — the ready versus warm distinction. Runtime-checkable because not every Indicator has a buffer of its own: a Cross holds no history, so its warmth is entirely its inputs', and those are registered separately. Consumers that need a warm-start bar count should fall back to lookback for anything that fails this check.

history_bars `property` ¶

history_bars: int

Adx ¶

Adx(period: int = 14, *, history: int | None = None)

Bases: TalibIndicator

Average directional index (TA-Lib ADX).

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int = 14, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("ADX", history=history, timeperiod=period)

Atr ¶

Atr(period: int, *, history: int | None = None)

Bases: TalibIndicator

Wilder's average true range (TA-Lib ATR); lookback == period + 1.

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("ATR", history=history, timeperiod=period)

BBands ¶

BBands(period: int = 20, *, deviations: float = 2.0, history: int | None = None)

Bases: TalibIndicator

Bollinger bands (TA-Lib BBANDS): upperband/middleband/lowerband.

The bands inherit StdDev's naive-variance error (~7e-7 relative at 100,000), so a band EDGE is not an exact number: a Cross against one can flip on that noise, and a "touch" of a band is never exact. Use it as a threshold, not as a price.

Source code in src/topstep_backtest/indicators/library.py

def __init__(
    self,
    period: int = 20,
    *,
    deviations: float = 2.0,
    history: int | None = None,
) -> None:
    _check_period(period)
    super().__init__(
        "BBANDS",
        history=history,
        timeperiod=period,
        nbdevup=deviations,
        nbdevdn=deviations,
    )

Cross ¶

Cross(a: ValueSource, b: ValueSource)

Crossover of two value series: up/down fire only on the crossing bar.

Tracks the sign of a.value - b.value per update: up when the sign goes from <= 0 to > 0, down when it goes from >= 0 to < 0 — an exact a == b touch therefore fires on the bar that resolves it, not on the touch itself. Both are False until the detector holds a previous AND a current comparison with both inputs ready, so lookback = max(a.lookback, b.lookback) + 1.

Note that TA-Lib values are float64, so an exact a == b touch is vanishingly rare in practice — the zero-sign branch is a correctness guarantee, not a common path.

update(bar) ignores the bar and reads a.value / b.value — it REQUIRES both inputs to have been updated for the same bar first. Under SymbolStrategy's registration-order update rule that means use() the inputs BEFORE the Cross that reads them (use() enforces it via inputs, resolving a TalibLine to the indicator that owns it).

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, a: ValueSource, b: ValueSource) -> None:
    self._a = a
    self._b = b
    self._prev_sign: int | None = None
    self._curr_sign: int | None = None
    self._updates = 0
    self._registered = False

inputs `property` ¶

inputs: tuple[ValueSource, ValueSource]

The (a, b) sources read on update, for registration-order checks.

lookback `property` ¶

lookback: int

ready `property` ¶

ready: bool

up `property` ¶

up: bool

down `property` ¶

down: bool

mark_registered ¶

mark_registered() -> None

Called by use(). A Cross nobody registered never updates, so its up/down would read False forever and the strategy would take zero trades in silence — up/down raise instead once that is provable.

Source code in src/topstep_backtest/indicators/library.py

def mark_registered(self) -> None:
    """Called by ``use()``. A Cross nobody registered never updates, so its
    ``up``/``down`` would read False forever and the strategy would take zero
    trades in silence — ``up``/``down`` raise instead once that is provable."""
    self._registered = True

update ¶

update(bar: Bar) -> None

Source code in src/topstep_backtest/indicators/library.py

def update(self, bar: Bar) -> None:
    self._updates += 1
    if not (self._a.ready and self._b.ready):
        return
    diff = self._a.value - self._b.value
    self._prev_sign = self._curr_sign
    self._curr_sign = 1 if diff > 0 else -1 if diff < 0 else 0

Ema ¶

Ema(period: int, *, history: int | None = None)

Bases: TalibIndicator

Exponential moving average of closes (TA-Lib EMA).

Seeded with the SMA of the first period closes, then k = 2/(period+1) recursion — TA-Lib's convention.

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("EMA", history=history, timeperiod=period)

Highest ¶

Highest(period: int, *, history: int | None = None)

Bases: TalibIndicator

Rolling maximum of bar.high (TA-Lib MAX redirected onto highs).

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("MAX", price="high", history=history, timeperiod=period)

Lowest ¶

Lowest(period: int, *, history: int | None = None)

Bases: TalibIndicator

Rolling minimum of bar.low (TA-Lib MIN redirected onto lows).

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("MIN", price="low", history=history, timeperiod=period)

Macd ¶

Macd(fast: int = 12, slow: int = 26, signal: int = 9, *, history: int | None = None)

Bases: TalibIndicator

MACD (TA-Lib MACD): outputs macd, macdsignal, macdhist.

value is the MACD line; cross the lines with Cross(macd.line("macd"), macd.line("macdsignal")).

Source code in src/topstep_backtest/indicators/library.py

def __init__(
    self,
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
    *,
    history: int | None = None,
) -> None:
    for period in (fast, slow, signal):
        _check_period(period)
    super().__init__(
        "MACD", history=history, fastperiod=fast, slowperiod=slow, signalperiod=signal
    )

Obv ¶

Obv(*, history: int | None = None)

Bases: TalibIndicator

On-balance volume (TA-Lib OBV).

OBV accumulates without decay, so — unlike every other indicator here — its LEVEL depends on where accumulation started and is therefore windowed to history_bars like everything else. Use its slope or divergence, never the absolute level; an inception-anchored level could not be reproduced live in any case.

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, *, history: int | None = None) -> None:
    super().__init__("OBV", history=history)

Rsi ¶

Rsi(period: int, *, history: int | None = None)

Bases: TalibIndicator

Wilder's relative strength index (TA-Lib RSI); lookback == period + 1.

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("RSI", history=history, timeperiod=period)

Sma ¶

Sma(period: int, *, history: int | None = None)

Bases: TalibIndicator

Simple moving average of closes (TA-Lib SMA).

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("SMA", history=history, timeperiod=period)

StdDev ¶

StdDev(period: int, *, history: int | None = None)

Bases: TalibIndicator

POPULATION standard deviation of closes (TA-Lib STDDEV, nbdev=1).

TA-Lib evaluates the NAIVE E[x^2] - E[x]^2 form, which cancels badly once the mean is large relative to the spread — at futures price levels the relative error reaches ~2e-9 at 5,000 and ~7e-7 at 100,000. Harmless for a threshold read; do not treat a value derived from it as exact.

Source code in src/topstep_backtest/indicators/library.py

def __init__(self, period: int, *, history: int | None = None) -> None:
    _check_period(period)
    super().__init__("STDDEV", history=history, timeperiod=period, nbdev=1.0)

Stoch ¶

Stoch(fastk: int = 5, slowk: int = 3, slowd: int = 3, *, history: int | None = None)

Bases: TalibIndicator

Slow stochastic (TA-Lib STOCH): outputs slowk and slowd.

Source code in src/topstep_backtest/indicators/library.py

def __init__(
    self,
    fastk: int = 5,
    slowk: int = 3,
    slowd: int = 3,
    *,
    history: int | None = None,
) -> None:
    for period in (fastk, slowk, slowd):
        _check_period(period)
    super().__init__(
        "STOCH",
        history=history,
        fastk_period=fastk,
        slowk_period=slowk,
        slowd_period=slowd,
    )

TalibIndicator ¶

TalibIndicator(name: str, *, price: str | None = None, history: int | None = None, **params: int | float)

Any TA-Lib function, driven bar by bar.

TalibIndicator("RSI", timeperiod=14) is the whole API; the named classes in library.py are thin, typed spellings of it. Parameters are TA-Lib's own (timeperiod, fastperiod, nbdevup, matype, …), validated against the function's declared signature AND run past TA-Lib once at construction, so a typo or an out-of-range value raises here rather than silently using a default or dying mid-backtest.

price redirects a single-series function onto another bar field — TalibIndicator("MAX", price="high", timeperiod=20) is a rolling high. Functions that declare their own multi-field inputs (ATR wants high/low/ close) reject it.

Thread-safety. talib's Function keeps its configured parameters in a threading.local, so a Function configured on one thread silently reverts to TA-Lib's DEFAULTS on another — an EMA(50) built on the main thread would quietly compute EMA(30) inside a worker, with no error. This adapter therefore treats the Function object as stateless: parameters are held here as plain data and passed on EVERY call, and a redirected price is applied by choosing which bar field fills the input slot rather than by mutating input_names. Instances are safe to build on one thread and drive on another (though a single instance is still not safe to drive from two threads at once — it has per-bar state).

Source code in src/topstep_backtest/indicators/talib_adapter.py

def __init__(
    self,
    name: str,
    *,
    price: str | None = None,
    history: int | None = None,
    **params: int | float,
) -> None:
    self._name = name.upper()
    func = _resolve_function(self._name)
    # Assigning `parameters` is what makes `lookback` reflect OUR periods —
    # a fresh Function reports the DEFAULT period's lookback. We then keep
    # the values here as well and re-pass them on every call, because that
    # assignment is thread-local (see the class docstring).
    self._params = _coerce_params(func, self._name, params)
    if self._params:
        func.parameters = self._params
    self._func = func
    self._lookback = func.lookback + 1
    self._slots = _input_slots(func, self._name, price)
    self._window = _resolve_window(self._name, func, self._lookback, history)
    _probe(func, self._name, self._params, self._slots)
    self._buffers = {key: _RingBuffer(self._window) for key, _ in self._slots}
    self._latest: dict[str, float] = {}
    self._seen = 0
    self._dirty = True

lookback `property` ¶

lookback: int

Bars needed before the first value exists (TA-Lib's lookback + 1).

history_bars `property` ¶

history_bars: int

Bars retained. Preload this many live for bit-exact sim/live parity.

warm `property` ¶

warm: bool

True once the buffer is full, i.e. from the first PARITY-exact bar.

ready says a value exists; warm says the value no longer depends on where this run started. Gate on this when sim and live must agree bit-for-bit (see the module docstring).

outputs `property` ¶

outputs: tuple[str, ...]

This function's output names, TA-Lib's order (value is the first).

function_name `property` ¶

function_name: str

The wrapped TA-Lib function, upper-cased (e.g. "MACD").

ready `property` ¶

ready: bool

value `property` ¶

value: Decimal

The primary (first) output for the most recent bar.

get ¶

get(output: str) -> Decimal

A named output for the most recent bar.

Source code in src/topstep_backtest/indicators/talib_adapter.py

def get(self, output: str) -> Decimal:
    """A named output for the most recent bar."""
    if output not in self._func.output_names:
        raise KeyError(f"{self!r} has no output {output!r}; it has {self.outputs}")
    if self._seen < self._lookback:
        raise NotReadyError(f"{self!r} needs {self._lookback} bars, has seen {self._seen}")
    self._compute()
    if not self._latest:
        raise NotReadyError(
            f"{self!r} has seen {self._seen} bars but produced no finite value for this "
            f"bar — degenerate input for this function, not warmup"
        )
    return _to_decimal(self._latest[output])

line ¶

line(output: str) -> TalibLine

A Cross-compatible view of one named output.

Source code in src/topstep_backtest/indicators/talib_adapter.py

def line(self, output: str) -> TalibLine:
    """A ``Cross``-compatible view of one named output."""
    if output not in self._func.output_names:
        raise KeyError(f"{self!r} has no output {output!r}; it has {self.outputs}")
    return TalibLine(self, output)

lines ¶

lines() -> Iterator[TalibLine]

A view per output, in TA-Lib's order.

Source code in src/topstep_backtest/indicators/talib_adapter.py

def lines(self) -> Iterator[TalibLine]:
    """A view per output, in TA-Lib's order."""
    return (TalibLine(self, name) for name in self._func.output_names)

update ¶

update(bar: Bar) -> None

Source code in src/topstep_backtest/indicators/talib_adapter.py

def update(self, bar: Bar) -> None:
    for key, field in self._slots:
        self._buffers[key].append(_bar_field(bar, field))
    self._seen += 1
    self._dirty = True

TalibLine ¶

TalibLine(owner: TalibIndicator, name: str)

One named output of a multi-output indicator, usable wherever a single value series is (notably as a Cross input).

A line does NOT update itself — it reads whatever its owner last computed, so registering the OWNER with use() is what keeps it current. SymbolStrategy.use() resolves a Cross input through owner for exactly this reason.

Deliberately NO update method: that keeps a line from satisfying Indicator, so use(macd.line("macd")) — which would register something that can never advance and would gate the strategy forever — is a type error, and use() rejects it at runtime too. Register the owner.

Source code in src/topstep_backtest/indicators/talib_adapter.py

def __init__(self, owner: TalibIndicator, name: str) -> None:
    self._owner = owner
    self._name = name

owner `property` ¶

owner: TalibIndicator

The indicator that computes this line (what use() registers).

name `property` ¶

name: str

This line's TA-Lib output name (e.g. "macdsignal").

lookback `property` ¶

lookback: int

ready `property` ¶

ready: bool

warm `property` ¶

warm: bool

value `property` ¶

value: Decimal

talib_function_names ¶

talib_function_names() -> tuple[str, ...]

Every TA-Lib function name TalibIndicator can wrap, sorted.

Source code in src/topstep_backtest/indicators/talib_adapter.py

def talib_function_names() -> tuple[str, ...]:
    """Every TA-Lib function name ``TalibIndicator`` can wrap, sorted."""
    import talib

    return tuple(sorted(cast("list[str]", talib.get_functions())))

indicators¶

indicators ¶

Indicator ¶

lookback property ¶

ready property ¶

update ¶

NotReadyError ¶

ValueSource ¶

lookback property ¶

ready property ¶

value property ¶

WarmIndicator ¶

history_bars property ¶

Adx ¶

Atr ¶

BBands ¶

Cross ¶

inputs property ¶

lookback property ¶

ready property ¶

up property ¶

down property ¶

mark_registered ¶

update ¶

Ema ¶

Highest ¶

Lowest ¶

Macd ¶

Obv ¶

Rsi ¶

Sma ¶

StdDev ¶

Stoch ¶

TalibIndicator ¶

lookback property ¶

history_bars property ¶

warm property ¶

outputs property ¶

function_name property ¶

ready property ¶

value property ¶

get ¶

line ¶

lines ¶

update ¶

TalibLine ¶

owner property ¶

name property ¶

lookback property ¶

ready property ¶

warm property ¶

value property ¶

talib_function_names ¶

`indicators`¶

lookback `property` ¶

ready `property` ¶

lookback `property` ¶

ready `property` ¶

value `property` ¶

history_bars `property` ¶

inputs `property` ¶

lookback `property` ¶

ready `property` ¶

up `property` ¶

down `property` ¶

lookback `property` ¶

history_bars `property` ¶

warm `property` ¶

outputs `property` ¶

function_name `property` ¶

ready `property` ¶

value `property` ¶

owner `property` ¶

name `property` ¶

lookback `property` ¶

ready `property` ¶

warm `property` ¶

value `property` ¶