Pratik Bhadane
436954138f
Update version numbers across Rust, Python, and documentation files to 1.1.0. Enhance the .gitignore to include macOS dSYM files and plans directory. Introduce new dependencies in the Rust core library and update the README to reflect recent performance benchmarks and backtesting engine capabilities. Add new artifacts to the benchmarks manifest and improve documentation for the backtesting engine API.
388 lines
13 KiB
Python
388 lines
13 KiB
Python
"""Tests for ferro_ta streaming / incremental indicators."""
|
|
|
|
import math
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from ferro_ta import EMA, RSI, SMA
|
|
from ferro_ta.data.streaming import StreamingEMA, StreamingRSI, StreamingSMA
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PRICES = np.array(
|
|
[
|
|
44.34,
|
|
44.09,
|
|
44.15,
|
|
43.61,
|
|
44.33,
|
|
44.83,
|
|
45.10,
|
|
45.15,
|
|
43.61,
|
|
44.33,
|
|
44.83,
|
|
45.10,
|
|
45.15,
|
|
43.61,
|
|
44.33,
|
|
],
|
|
dtype=np.float64,
|
|
)
|
|
|
|
|
|
def _finite(arr: np.ndarray) -> np.ndarray:
|
|
return arr[~np.isnan(arr)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# StreamingSMA
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStreamingSMA:
|
|
def test_basic_values(self):
|
|
"""Feed known values, verify manually computed SMA."""
|
|
sma = StreamingSMA(period=3)
|
|
assert math.isnan(sma.update(1.0))
|
|
assert math.isnan(sma.update(2.0))
|
|
assert math.isclose(sma.update(3.0), 2.0)
|
|
assert math.isclose(sma.update(4.0), 3.0)
|
|
assert math.isclose(sma.update(5.0), 4.0)
|
|
|
|
def test_matches_batch_sma(self):
|
|
"""Streaming SMA final values must match batch SMA on the same data."""
|
|
period = 5
|
|
batch = SMA(PRICES, timeperiod=period)
|
|
sma = StreamingSMA(period=period)
|
|
for i, price in enumerate(PRICES):
|
|
val = sma.update(price)
|
|
if math.isnan(batch[i]):
|
|
assert math.isnan(val), f"Expected NaN at index {i}"
|
|
else:
|
|
assert math.isclose(val, batch[i], rel_tol=1e-10), (
|
|
f"Mismatch at index {i}: streaming={val}, batch={batch[i]}"
|
|
)
|
|
|
|
def test_period_property(self):
|
|
sma = StreamingSMA(period=7)
|
|
assert sma.period == 7
|
|
|
|
def test_warmup_returns_nan(self):
|
|
"""First period-1 updates must return NaN."""
|
|
period = 4
|
|
sma = StreamingSMA(period=period)
|
|
for i in range(period - 1):
|
|
assert math.isnan(sma.update(float(i + 1)))
|
|
# The period-th update should NOT be NaN
|
|
assert not math.isnan(sma.update(float(period)))
|
|
|
|
def test_single_value_period_1(self):
|
|
"""Period=1 means every value is immediately returned."""
|
|
sma = StreamingSMA(period=1)
|
|
assert math.isclose(sma.update(42.0), 42.0)
|
|
assert math.isclose(sma.update(99.0), 99.0)
|
|
|
|
def test_reset(self):
|
|
"""After reset, the indicator should behave as freshly constructed."""
|
|
sma = StreamingSMA(period=3)
|
|
sma.update(10.0)
|
|
sma.update(20.0)
|
|
result_before_reset = sma.update(30.0)
|
|
assert math.isclose(result_before_reset, 20.0)
|
|
|
|
sma.reset()
|
|
# After reset, warmup restarts
|
|
assert math.isnan(sma.update(100.0))
|
|
assert math.isnan(sma.update(200.0))
|
|
assert math.isclose(sma.update(300.0), 200.0)
|
|
|
|
def test_invalid_period_zero(self):
|
|
with pytest.raises(Exception):
|
|
StreamingSMA(period=0)
|
|
|
|
def test_repr(self):
|
|
sma = StreamingSMA(period=5)
|
|
assert "StreamingSMA" in repr(sma)
|
|
assert "5" in repr(sma)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# StreamingEMA
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStreamingEMA:
|
|
def test_basic_seeding(self):
|
|
"""EMA seeds from the first `period` values using their SMA."""
|
|
ema = StreamingEMA(period=3)
|
|
assert math.isnan(ema.update(1.0))
|
|
assert math.isnan(ema.update(2.0))
|
|
# Seed = SMA(1,2,3) = 2.0
|
|
seed = ema.update(3.0)
|
|
assert math.isclose(seed, 2.0)
|
|
|
|
def test_matches_batch_ema(self):
|
|
"""Streaming EMA must match batch EMA on the same data."""
|
|
period = 5
|
|
batch = EMA(PRICES, timeperiod=period)
|
|
ema = StreamingEMA(period=period)
|
|
for i, price in enumerate(PRICES):
|
|
val = ema.update(price)
|
|
if math.isnan(batch[i]):
|
|
assert math.isnan(val), f"Expected NaN at index {i}"
|
|
else:
|
|
assert math.isclose(val, batch[i], rel_tol=1e-10), (
|
|
f"Mismatch at index {i}: streaming={val}, batch={batch[i]}"
|
|
)
|
|
|
|
def test_warmup_returns_nan(self):
|
|
period = 5
|
|
ema = StreamingEMA(period=period)
|
|
for i in range(period - 1):
|
|
assert math.isnan(ema.update(float(i + 1)))
|
|
assert not math.isnan(ema.update(float(period)))
|
|
|
|
def test_ema_differs_from_sma_after_warmup(self):
|
|
"""After warmup, EMA and SMA should diverge for non-constant data."""
|
|
period = 3
|
|
prices = [1.0, 2.0, 3.0, 10.0, 11.0]
|
|
sma = StreamingSMA(period=period)
|
|
ema = StreamingEMA(period=period)
|
|
sma_vals = [sma.update(p) for p in prices]
|
|
ema_vals = [ema.update(p) for p in prices]
|
|
# At the seed point they should match (both are SMA of first 3)
|
|
assert math.isclose(sma_vals[2], ema_vals[2])
|
|
# After the seed they should diverge
|
|
assert not math.isclose(sma_vals[-1], ema_vals[-1], rel_tol=1e-9)
|
|
|
|
def test_reset(self):
|
|
ema = StreamingEMA(period=3)
|
|
for p in [10.0, 20.0, 30.0, 40.0]:
|
|
ema.update(p)
|
|
ema.reset()
|
|
# After reset, warmup restarts
|
|
assert math.isnan(ema.update(1.0))
|
|
assert math.isnan(ema.update(2.0))
|
|
assert math.isclose(ema.update(3.0), 2.0)
|
|
|
|
def test_period_property(self):
|
|
ema = StreamingEMA(period=10)
|
|
assert ema.period == 10
|
|
|
|
def test_invalid_period_zero(self):
|
|
with pytest.raises(Exception):
|
|
StreamingEMA(period=0)
|
|
|
|
def test_single_value_period_1(self):
|
|
ema = StreamingEMA(period=1)
|
|
assert math.isclose(ema.update(42.0), 42.0)
|
|
assert math.isclose(ema.update(50.0), 50.0)
|
|
|
|
def test_repr(self):
|
|
ema = StreamingEMA(period=12)
|
|
assert "StreamingEMA" in repr(ema)
|
|
assert "12" in repr(ema)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# StreamingRSI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStreamingRSI:
|
|
def test_matches_batch_rsi(self):
|
|
"""Streaming RSI must match batch RSI on the same data."""
|
|
period = 5
|
|
batch = RSI(PRICES, timeperiod=period)
|
|
rsi = StreamingRSI(period=period)
|
|
for i, price in enumerate(PRICES):
|
|
val = rsi.update(price)
|
|
if math.isnan(batch[i]):
|
|
assert math.isnan(val), f"Expected NaN at index {i}"
|
|
else:
|
|
assert math.isclose(val, batch[i], rel_tol=1e-8), (
|
|
f"Mismatch at index {i}: streaming={val}, batch={batch[i]}"
|
|
)
|
|
|
|
def test_warmup_returns_nan(self):
|
|
"""RSI needs period+1 bars (1 for first prev, then period deltas)."""
|
|
period = 5
|
|
rsi = StreamingRSI(period=period)
|
|
# First bar: sets prev, returns NaN
|
|
assert math.isnan(rsi.update(50.0))
|
|
# Next period-1 bars: accumulating deltas, returns NaN
|
|
for i in range(period - 1):
|
|
assert math.isnan(rsi.update(50.0 + i))
|
|
# The (period+1)-th bar should produce a value
|
|
assert not math.isnan(rsi.update(55.0))
|
|
|
|
def test_rsi_range(self):
|
|
"""All finite RSI values must be in [0, 100]."""
|
|
rsi = StreamingRSI(period=5)
|
|
for price in PRICES:
|
|
val = rsi.update(price)
|
|
if not math.isnan(val):
|
|
assert 0.0 <= val <= 100.0, f"RSI out of range: {val}"
|
|
|
|
def test_constant_prices(self):
|
|
"""Constant prices produce no gains or losses -- RSI should be 100
|
|
(avg_loss == 0 leads to RS = infinity -> RSI = 100)."""
|
|
rsi = StreamingRSI(period=5)
|
|
results = [rsi.update(50.0) for _ in range(20)]
|
|
finite = [v for v in results if not math.isnan(v)]
|
|
assert len(finite) > 0
|
|
for v in finite:
|
|
assert math.isclose(v, 100.0) or math.isclose(v, 0.0) or (0.0 <= v <= 100.0)
|
|
|
|
def test_monotone_increasing(self):
|
|
"""Monotonically increasing prices should yield RSI = 100."""
|
|
rsi = StreamingRSI(period=3)
|
|
results = [rsi.update(float(i)) for i in range(1, 20)]
|
|
finite = [v for v in results if not math.isnan(v)]
|
|
for v in finite:
|
|
assert math.isclose(v, 100.0), (
|
|
f"Expected RSI=100 for monotone increase, got {v}"
|
|
)
|
|
|
|
def test_monotone_decreasing(self):
|
|
"""Monotonically decreasing prices should yield RSI = 0."""
|
|
rsi = StreamingRSI(period=3)
|
|
results = [rsi.update(float(100 - i)) for i in range(20)]
|
|
finite = [v for v in results if not math.isnan(v)]
|
|
for v in finite:
|
|
assert math.isclose(v, 0.0, abs_tol=1e-10), (
|
|
f"Expected RSI=0 for monotone decrease, got {v}"
|
|
)
|
|
|
|
def test_default_period_14(self):
|
|
rsi = StreamingRSI()
|
|
assert rsi.period == 14
|
|
|
|
def test_reset(self):
|
|
rsi = StreamingRSI(period=3)
|
|
for price in PRICES:
|
|
rsi.update(price)
|
|
rsi.reset()
|
|
# After reset, warmup restarts -- first update should be NaN
|
|
assert math.isnan(rsi.update(50.0))
|
|
|
|
def test_invalid_period_zero(self):
|
|
with pytest.raises(Exception):
|
|
StreamingRSI(period=0)
|
|
|
|
def test_repr(self):
|
|
rsi = StreamingRSI(period=14)
|
|
assert "StreamingRSI" in repr(rsi)
|
|
assert "14" in repr(rsi)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge cases (shared across indicators)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStreamingEdgeCases:
|
|
def test_nan_input_sma(self):
|
|
"""Feeding NaN into SMA should propagate NaN through the window."""
|
|
sma = StreamingSMA(period=3)
|
|
sma.update(1.0)
|
|
sma.update(2.0)
|
|
# Third value is NaN -- the sum will include NaN, producing NaN
|
|
val = sma.update(float("nan"))
|
|
assert math.isnan(val)
|
|
|
|
def test_nan_input_ema(self):
|
|
"""Feeding NaN into EMA should produce NaN output."""
|
|
ema = StreamingEMA(period=3)
|
|
ema.update(1.0)
|
|
ema.update(2.0)
|
|
val = ema.update(float("nan"))
|
|
assert math.isnan(val)
|
|
|
|
def test_nan_input_rsi(self):
|
|
"""Feeding NaN into RSI should produce NaN output."""
|
|
rsi = StreamingRSI(period=3)
|
|
rsi.update(1.0)
|
|
rsi.update(2.0)
|
|
val = rsi.update(float("nan"))
|
|
assert math.isnan(val)
|
|
|
|
def test_single_value_sma(self):
|
|
"""Feeding exactly one value to SMA with period > 1 yields NaN."""
|
|
sma = StreamingSMA(period=5)
|
|
assert math.isnan(sma.update(42.0))
|
|
|
|
def test_single_value_ema(self):
|
|
ema = StreamingEMA(period=5)
|
|
assert math.isnan(ema.update(42.0))
|
|
|
|
def test_single_value_rsi(self):
|
|
rsi = StreamingRSI(period=5)
|
|
assert math.isnan(rsi.update(42.0))
|
|
|
|
def test_large_dataset_sma(self):
|
|
"""Ensure streaming SMA is stable over many updates."""
|
|
period = 20
|
|
sma = StreamingSMA(period=period)
|
|
np.random.seed(42)
|
|
data = np.random.randn(10_000).cumsum() + 100.0
|
|
batch = SMA(data, timeperiod=period)
|
|
for i, price in enumerate(data):
|
|
val = sma.update(price)
|
|
if not math.isnan(batch[i]):
|
|
assert math.isclose(val, batch[i], rel_tol=1e-8), (
|
|
f"Drift at index {i}: streaming={val}, batch={batch[i]}"
|
|
)
|
|
|
|
def test_large_dataset_ema(self):
|
|
"""Ensure streaming EMA is stable over many updates."""
|
|
period = 20
|
|
ema = StreamingEMA(period=period)
|
|
np.random.seed(42)
|
|
data = np.random.randn(10_000).cumsum() + 100.0
|
|
batch = EMA(data, timeperiod=period)
|
|
for i, price in enumerate(data):
|
|
val = ema.update(price)
|
|
if not math.isnan(batch[i]):
|
|
assert math.isclose(val, batch[i], rel_tol=1e-8), (
|
|
f"Drift at index {i}: streaming={val}, batch={batch[i]}"
|
|
)
|
|
|
|
def test_large_dataset_rsi(self):
|
|
"""Ensure streaming RSI is stable over many updates."""
|
|
period = 14
|
|
rsi = StreamingRSI(period=period)
|
|
np.random.seed(42)
|
|
data = np.random.randn(10_000).cumsum() + 100.0
|
|
batch = RSI(data, timeperiod=period)
|
|
for i, price in enumerate(data):
|
|
val = rsi.update(price)
|
|
if not math.isnan(batch[i]):
|
|
assert math.isclose(val, batch[i], rel_tol=1e-6), (
|
|
f"Drift at index {i}: streaming={val}, batch={batch[i]}"
|
|
)
|
|
|
|
def test_reset_then_reuse_matches_fresh_instance(self):
|
|
"""A reset indicator should produce identical output to a new one."""
|
|
period = 5
|
|
data = [10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0]
|
|
|
|
sma_reused = StreamingSMA(period=period)
|
|
for p in [99.0, 98.0, 97.0, 96.0, 95.0]:
|
|
sma_reused.update(p)
|
|
sma_reused.reset()
|
|
|
|
sma_fresh = StreamingSMA(period=period)
|
|
|
|
for p in data:
|
|
v1 = sma_reused.update(p)
|
|
v2 = sma_fresh.update(p)
|
|
if math.isnan(v1):
|
|
assert math.isnan(v2)
|
|
else:
|
|
assert math.isclose(v1, v2, rel_tol=1e-12)
|