Skip to content

refactor(streaming): replace with the v0.8.1 Core implementation #757

Description

@mario4tier

Proposed breaking change. talib.stream would be replaced, not extended — the
current last-value functions go away.

Why

TA-Lib C 0.8.1 has a real streaming API: O(1) cost per bar. talib.stream predates it and works a
different way — it takes the whole array on every call and asks the batch function for a single bar (often O(lookback)).

The propose C 0.8.1 changes bring two things at once:

  • High-performance (adding a bar is a few nanoseconds on modern CPU).
  • streaming output bit-exact with batch processing over the same data.

Shape

from talib import stream

s = stream.SMA(close, timeperiod=30)   # same arguments as talib.SMA
s.value                                # value at the last history bar

for price in feed:
    v = s.update(price)                # one closed bar in, its value out

s.peek(provisional)                    # what update would return; commits nothing
s.copy()                               # independent fork

stream.SMA takes exactly the arguments talib.SMA takes — one returns the
series, the other a handle positioned at its end.

A multi-output function answers with a NamedTuple, its fields named the way the
batch docstring already names them:

m = stream.MACD(close)             # same arguments as talib.MACD

r = m.update(price)
r.macd, r.macdsignal, r.macdhist   # named
macd, signal, hist = r             # and still unpacks like the batch tuple

m.value.macdhist                   # last value again, no recompute

A single-output function returns a bare float, not a 1-tuple.

Warm-Up

At stream creation, an history of at least lookback + 1 bars must be provided, which abstract knows:

need = abstract.Function('RSI', timeperiod=14).lookback + 1   # 15

# ... build 'history' here, with at least 'need' bars

s = stream.RSI(history, timeperiod=14)   # open once
emit(s.value)                            # its value, at the last history bar

for bar in feed:                         # then only updates
    emit(s.update(bar))

With fewer than need bars the open raises talib.InsufficientHistory. You can alternatively design your warm-up to keep re-trying opening until success.

If you already hold the history and want the batch series too, one pass gives
both:

s, rsi = stream.RSI.open_and_fill(history, timeperiod=14)
# rsi is what talib.RSI(history) returns, and s is positioned at its end

open_and_fill is an alternate constructor, the stream created has the same capability (update/peel/value etc...).

Migration

Three names go away: talib.stream.X, talib.stream_X (also exported at top
level), and the stream_* stubs in _ta_lib.pyi.

Details

  • Pickling must raise. A handle is a pointer into the C library and never
    crosses a process boundary; __reduce__ should say so rather than let
    multiprocessing find out.
  • InsufficientHistory wants its own exception class. It is the library's one
    recoverable condition, and _ta_check_success raises a bare Exception today,
    so it cannot be caught narrowly inside a bar loop.
  • pandas/polars. __init__.py wraps every stream function with _wrapper;
    whatever takes the history array needs the same, or DataFrame users lose input
    support they have now.
  • Requires ta-lib C >= 0.8.1, so this lands with the 0.8.1 support work.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

No labels
No labels

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions