Implementation:TA Lib Ta lib python Stream Function API

Knowledge Sources	ta-lib-python TA-Lib C Library
Domains	Real_Time_Processing, Technical_Analysis
Last Updated	2026-02-09 22:00 GMT

Overview

Concrete tool for computing the latest indicator value as a scalar via streaming Cython functions (stream_SMA, stream_RSI, stream_BBANDS, etc.) provided by the talib.stream module.

Description

The streaming functions are Cython wrappers that call the same TA-Lib C functions but only request computation for the last index. They are defined in talib/_stream.pxi (5350 lines, auto-generated) and exposed through talib/stream.py and talib.__init__ (as talib.stream_SMA, etc.).

Each streaming function has the same parameters as its Function API counterpart but returns a scalar instead of an array.

Usage

Import from talib.stream or call via talib.stream_SMA, talib.stream_RSI, etc.

Code Reference

Source Location

Repository: ta-lib-python
File: talib/_stream.pxi
Lines: L4660-4685 (stream_SMA), L4505-4530 (stream_RSI), L507-541 (stream_BBANDS)
Module: talib/stream.py (L1-6, module-level name mapping)

Signature

# Single-output streaming functions
def stream_SMA(real: np.ndarray, timeperiod: int = 30) -> float:
    """Returns latest SMA value as scalar"""

def stream_RSI(real: np.ndarray, timeperiod: int = 14) -> float:
    """Returns latest RSI value as scalar"""

# Multi-output streaming functions
def stream_BBANDS(real: np.ndarray, timeperiod: int = 5,
                  nbdevup: float = 2.0, nbdevdn: float = 2.0,
                  matype: int = 0) -> Tuple[float, float, float]:
    """Returns (upperband, middleband, lowerband) as scalar tuple"""

Import

from talib import stream

# Or via talib module (with _wrapper applied for pandas/polars support)
import talib
# talib.stream_SMA, talib.stream_RSI, etc.

I/O Contract

Inputs

Name	Type	Required	Description
real	np.ndarray (float64)	Yes	Price data buffer (full history)
timeperiod	int	No	Lookback window (default varies per function)
high, low, close	np.ndarray (float64)	Some indicators	Additional price series for multi-input indicators

Outputs

Name	Type	Description
scalar_result	float	Single latest indicator value (NaN if insufficient data)
scalar_tuple	Tuple[float, ...]	Multiple latest values for multi-output indicators

Usage Examples

Streaming SMA

import numpy as np
from talib import stream

data = np.random.random(100)
latest_sma = stream.SMA(data, timeperiod=20)
print(f"Latest SMA: {latest_sma}")  # Single float value

Streaming BBANDS

upper, middle, lower = stream.BBANDS(data, timeperiod=20, nbdevup=2, nbdevdn=2)
print(f"Upper: {upper}, Middle: {middle}, Lower: {lower}")

Real-Time Event Loop

import numpy as np
from talib import stream

buffer = list(np.random.random(50))  # Initial history

def on_new_candle(close_price):
    buffer.append(close_price)
    data = np.array(buffer, dtype=float)
    sma = stream.SMA(data, timeperiod=20)
    rsi = stream.RSI(data, timeperiod=14)
    print(f"SMA: {sma:.4f}, RSI: {rsi:.2f}")

Related Pages

Implements Principle

Principle:TA_Lib_Ta_lib_python_Streaming_Indicator_Computation

Requires Environment

Uses Heuristic

Heuristic:TA_Lib_Ta_lib_python_Unstable_Period_Configuration

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment