ferro-ta/examples/streaming.ipynb

Streaming API — bar-by-bar live trading

The ferro_ta.streaming module provides stateful classes that process data bar-by-bar, suitable for real-time feeds and live trading.

Install:

pip install ferro-ta

In [ ]:

import numpy as np

from ferro_ta.streaming import (
    StreamingATR,
    StreamingBBands,
    StreamingEMA,
    StreamingMACD,
    StreamingRSI,
    StreamingSMA,
)

# Simulate incoming bars
np.random.seed(42)
n = 50
closes = np.cumprod(1 + np.random.randn(n) * 0.01) * 100
highs = closes * 1.005
lows = closes * 0.995

print(f"Simulated {n} bars")

StreamingSMA and StreamingEMA

In [ ]:

sma = StreamingSMA(period=5)
ema = StreamingEMA(period=5)

sma_values = [sma.update(c) for c in closes]
ema_values = [ema.update(c) for c in closes]

print(
    "SMA last 5:", [f"{v:.4f}" if not np.isnan(v) else "NaN" for v in sma_values[-5:]]
)
print(
    "EMA last 5:", [f"{v:.4f}" if not np.isnan(v) else "NaN" for v in ema_values[-5:]]
)

StreamingRSI

In [ ]:

rsi_stream = StreamingRSI(period=14)

rsi_values = [rsi_stream.update(c) for c in closes]
finite = [(i, v) for i, v in enumerate(rsi_values) if not np.isnan(v)]
print(f"First valid RSI at bar {finite[0][0]}: {finite[0][1]:.2f}")
print("RSI last 3:", [f"{v:.2f}" for _, v in finite[-3:]])

StreamingBBands

In [ ]:

bbands = StreamingBBands(period=20)

bb_results = [bbands.update(c) for c in closes]
# Each result is (upper, middle, lower) or (nan, nan, nan) during warmup
valid_bb = [
    (i, u, m, lower) for i, (u, m, lower) in enumerate(bb_results) if not np.isnan(m)
]
if valid_bb:
    i, u, m, lower = valid_bb[-1]
    print(f"Latest Bollinger Bands at bar {i}:")
    print(f"  Upper:  {u:.4f}")
    print(f"  Middle: {m:.4f}")
    print(f"  Lower:  {lower:.4f}")

StreamingMACD

In [ ]:

macd_stream = StreamingMACD(fastperiod=12, slowperiod=26, signalperiod=9)

macd_results = [macd_stream.update(c) for c in closes]
# Each result is (macd_line, signal, histogram)
valid_macd = [
    (i, m, s, h) for i, (m, s, h) in enumerate(macd_results) if not np.isnan(m)
]
if valid_macd:
    i, m, s, h = valid_macd[-1]
    print(f"Latest MACD at bar {i}:")
    print(f"  MACD line:  {m:.6f}")
    print(f"  Signal:     {s:.6f}")
    print(f"  Histogram:  {h:.6f}")

StreamingATR

In [ ]:

atr_stream = StreamingATR(period=14)

atr_values = [atr_stream.update(h, low, c) for h, low, c in zip(highs, lows, closes)]
finite_atr = [v for v in atr_values if not np.isnan(v)]
if finite_atr:
    print(f"Latest ATR: {finite_atr[-1]:.4f}")

Reset and reuse

All streaming classes support reset() to clear internal state and start fresh.

In [ ]:

sma.reset()
print("After reset, SMA(5.0):", sma.update(5.0))  # NaN — warm-up restarted
sma.update(6.0)
sma.update(7.0)
sma.update(8.0)
print("SMA after 4 bars:", sma.update(9.0))  # 7.0 = mean of [5,6,7,8,9]