Implementation:TA Lib Ta lib python Stream Buffer Pattern

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

Overview

Concrete pattern for managing a rolling numpy array buffer to feed streaming indicator functions in real-time applications.

Description

This is a Pattern Doc documenting the user-side buffer management pattern for streaming API usage. TA-Lib streaming functions are stateless — they take a full numpy array buffer and return a single scalar value. Users must manage the buffer themselves.

Usage

Use this pattern when integrating TA-Lib streaming functions into event-driven or tick-by-tick trading systems.

Code Reference

Source Location

• Repository: ta-lib-python
• File: N/A (user-side pattern, not a library API)

Signature

# User-implemented buffer management pattern
import numpy as np

class PriceBuffer:
    def __init__(self, max_size=1000):
        self.data = []
        self.max_size = max_size

    def append(self, price):
        self.data.append(price)
        if len(self.data) > self.max_size:
            self.data = self.data[-self.max_size:]

    def as_array(self):
        return np.array(self.data, dtype=float)

Import

import numpy as np
import talib
from talib import stream

I/O Contract

Inputs

Name	Type	Required	Description
price_tick	float	Yes	New price value to append to buffer

Outputs

Name	Type	Description
buffer_array	np.ndarray (float64)	Accumulated price data ready for streaming function

Usage Examples

Basic Buffer Pattern

import numpy as np
from talib import stream

# Simulate accumulating price data
buffer = []
for new_price in [44.0, 44.3, 44.1, 43.6, 44.3, 44.8, 45.1, 44.9, 45.2, 45.0]:
    buffer.append(new_price)
    data = np.array(buffer, dtype=float)

    if len(buffer) >= 5:  # Minimum for SMA(5)
        latest_sma = stream.SMA(data, timeperiod=5)
        print(f"Price: {new_price:.1f}, SMA(5): {latest_sma:.2f}")
    else:
        print(f"Price: {new_price:.1f}, SMA(5): insufficient data")

Related Pages

Implements Principle

• Principle:TA_Lib_Ta_lib_python_Streaming_Data_Buffering