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.
297 lines
9.3 KiB
Python
297 lines
9.3 KiB
Python
"""Edge-case tests for ferro_ta indicators.
|
|
|
|
Covers NaN handling, empty arrays, single-element inputs, extreme values,
|
|
constant series, and dtype robustness.
|
|
"""
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from ferro_ta import (
|
|
ATR,
|
|
BBANDS,
|
|
EMA,
|
|
MACD,
|
|
MFI,
|
|
OBV,
|
|
RSI,
|
|
SMA,
|
|
STOCH,
|
|
WMA,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _all_nan(arr):
|
|
"""True if every element is NaN."""
|
|
return np.all(np.isnan(arr))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Empty arrays
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEmptyInput:
|
|
"""All indicators should return an empty array (not crash) for len-0 input."""
|
|
|
|
def test_sma_empty(self):
|
|
result = SMA(np.array([], dtype=np.float64), timeperiod=14)
|
|
assert len(result) == 0
|
|
|
|
def test_ema_empty(self):
|
|
result = EMA(np.array([], dtype=np.float64), timeperiod=14)
|
|
assert len(result) == 0
|
|
|
|
def test_rsi_empty(self):
|
|
result = RSI(np.array([], dtype=np.float64), timeperiod=14)
|
|
assert len(result) == 0
|
|
|
|
def test_bbands_empty(self):
|
|
upper, mid, lower = BBANDS(np.array([], dtype=np.float64), timeperiod=5)
|
|
assert len(upper) == 0
|
|
assert len(mid) == 0
|
|
assert len(lower) == 0
|
|
|
|
def test_macd_empty(self):
|
|
macd, sig, hist = MACD(np.array([], dtype=np.float64))
|
|
assert len(macd) == 0
|
|
|
|
def test_wma_empty(self):
|
|
result = WMA(np.array([], dtype=np.float64), timeperiod=10)
|
|
assert len(result) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Single-element arrays
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSingleElement:
|
|
"""Single-element inputs should produce NaN (insufficient data) without panic."""
|
|
|
|
def test_sma_single(self):
|
|
result = SMA(np.array([42.0]), timeperiod=14)
|
|
assert len(result) == 1
|
|
assert np.isnan(result[0])
|
|
|
|
def test_ema_single(self):
|
|
result = EMA(np.array([42.0]), timeperiod=14)
|
|
assert len(result) == 1
|
|
assert np.isnan(result[0])
|
|
|
|
def test_rsi_single(self):
|
|
result = RSI(np.array([42.0]), timeperiod=14)
|
|
assert len(result) == 1
|
|
assert np.isnan(result[0])
|
|
|
|
def test_sma_period_1_single(self):
|
|
"""SMA(period=1) on a single element should return that element."""
|
|
result = SMA(np.array([42.0]), timeperiod=1)
|
|
assert len(result) == 1
|
|
np.testing.assert_allclose(result[0], 42.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# All-NaN input
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAllNaN:
|
|
"""Indicators fed entirely NaN input should not crash and return all NaN."""
|
|
|
|
@pytest.fixture()
|
|
def nan_50(self):
|
|
return np.full(50, np.nan)
|
|
|
|
def test_sma_all_nan(self, nan_50):
|
|
result = SMA(nan_50, timeperiod=14)
|
|
assert len(result) == 50
|
|
assert _all_nan(result)
|
|
|
|
def test_ema_all_nan(self, nan_50):
|
|
result = EMA(nan_50, timeperiod=14)
|
|
assert len(result) == 50
|
|
assert _all_nan(result)
|
|
|
|
def test_rsi_all_nan(self, nan_50):
|
|
result = RSI(nan_50, timeperiod=14)
|
|
assert len(result) == 50
|
|
assert _all_nan(result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NaN in the middle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNaNInMiddle:
|
|
"""A single NaN in a valid series should propagate but not crash."""
|
|
|
|
def test_sma_nan_mid(self):
|
|
data = np.arange(1.0, 21.0)
|
|
data[10] = np.nan
|
|
result = SMA(data, timeperiod=5)
|
|
assert len(result) == 20
|
|
# Values around the NaN should be NaN
|
|
for i in range(10, min(15, 20)):
|
|
assert np.isnan(result[i])
|
|
|
|
def test_rsi_nan_mid(self):
|
|
data = np.arange(1.0, 31.0)
|
|
data[15] = np.nan
|
|
result = RSI(data, timeperiod=14)
|
|
assert len(result) == 30
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Extreme values
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExtremeValues:
|
|
"""Indicators should not crash on very large or very small values."""
|
|
|
|
def test_sma_large_values(self):
|
|
data = np.full(50, 1e300)
|
|
result = SMA(data, timeperiod=14)
|
|
assert len(result) == 50
|
|
# Non-NaN values should be ~1e300
|
|
valid = result[~np.isnan(result)]
|
|
if len(valid) > 0:
|
|
np.testing.assert_allclose(valid, 1e300, rtol=1e-10)
|
|
|
|
def test_sma_tiny_values(self):
|
|
data = np.full(50, 1e-300)
|
|
result = SMA(data, timeperiod=14)
|
|
assert len(result) == 50
|
|
valid = result[~np.isnan(result)]
|
|
if len(valid) > 0:
|
|
np.testing.assert_allclose(valid, 1e-300, rtol=1e-10)
|
|
|
|
def test_rsi_large_monotone(self):
|
|
"""Monotonically increasing large values -> RSI should approach 100."""
|
|
data = np.linspace(1e10, 2e10, 100)
|
|
result = RSI(data, timeperiod=14)
|
|
valid = result[~np.isnan(result)]
|
|
if len(valid) > 0:
|
|
assert valid[-1] > 90.0 # strongly bullish
|
|
|
|
def test_rsi_zero_change(self):
|
|
"""Constant series -> RSI should be 50 (or NaN in some implementations)."""
|
|
data = np.full(100, 50.0)
|
|
result = RSI(data, timeperiod=14)
|
|
valid = result[~np.isnan(result)]
|
|
# Constant series: no gains, no losses -> typically NaN or 50
|
|
# Just verify no crash and valid range
|
|
for v in valid:
|
|
assert 0.0 <= v <= 100.0 or np.isnan(v)
|
|
|
|
def test_bbands_constant_series(self):
|
|
"""Constant series -> upper == middle == lower (zero std dev)."""
|
|
data = np.full(50, 100.0)
|
|
upper, mid, lower = BBANDS(data, timeperiod=10)
|
|
valid_mask = ~np.isnan(mid)
|
|
np.testing.assert_allclose(upper[valid_mask], mid[valid_mask])
|
|
np.testing.assert_allclose(lower[valid_mask], mid[valid_mask])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Timeperiod edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTimePeriodEdge:
|
|
"""Boundary conditions for the timeperiod parameter."""
|
|
|
|
def test_sma_period_equals_length(self):
|
|
data = np.arange(1.0, 11.0) # 10 elements
|
|
result = SMA(data, timeperiod=10)
|
|
assert len(result) == 10
|
|
# Only last element should be valid
|
|
assert not np.isnan(result[-1])
|
|
np.testing.assert_allclose(result[-1], 5.5)
|
|
|
|
def test_sma_period_exceeds_length(self):
|
|
data = np.arange(1.0, 6.0) # 5 elements
|
|
result = SMA(data, timeperiod=10)
|
|
assert len(result) == 5
|
|
assert _all_nan(result)
|
|
|
|
def test_ema_period_1(self):
|
|
"""EMA with period=1 should return the input itself."""
|
|
data = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
|
|
result = EMA(data, timeperiod=1)
|
|
np.testing.assert_allclose(result, data)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Multi-input indicator edge cases (OHLCV)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOHLCVEdgeCases:
|
|
"""Edge cases for indicators requiring multiple price series."""
|
|
|
|
def test_atr_empty(self):
|
|
empty = np.array([], dtype=np.float64)
|
|
result = ATR(empty, empty, empty, timeperiod=14)
|
|
assert len(result) == 0
|
|
|
|
def test_stoch_empty(self):
|
|
empty = np.array([], dtype=np.float64)
|
|
slowk, slowd = STOCH(empty, empty, empty)
|
|
assert len(slowk) == 0
|
|
assert len(slowd) == 0
|
|
|
|
def test_obv_empty(self):
|
|
empty = np.array([], dtype=np.float64)
|
|
result = OBV(empty, empty)
|
|
assert len(result) == 0
|
|
|
|
def test_atr_single_bar(self):
|
|
h = np.array([10.0])
|
|
l = np.array([9.0])
|
|
c = np.array([9.5])
|
|
result = ATR(h, l, c, timeperiod=14)
|
|
assert len(result) == 1
|
|
assert np.isnan(result[0])
|
|
|
|
def test_mfi_constant_price(self):
|
|
"""Constant price -> no money flow direction -> MFI should be well-defined."""
|
|
n = 50
|
|
h = np.full(n, 100.0)
|
|
l = np.full(n, 100.0)
|
|
c = np.full(n, 100.0)
|
|
v = np.full(n, 1000.0)
|
|
result = MFI(h, l, c, v, timeperiod=14)
|
|
assert len(result) == n
|
|
# Should not crash; values may be NaN or 50
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dtype robustness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDtypeRobustness:
|
|
"""Indicators should accept float32/int inputs and coerce to float64."""
|
|
|
|
def test_sma_float32(self):
|
|
data = np.arange(1.0, 51.0, dtype=np.float32)
|
|
result = SMA(data, timeperiod=14)
|
|
assert len(result) == 50
|
|
|
|
def test_sma_int64(self):
|
|
data = np.arange(1, 51, dtype=np.int64)
|
|
result = SMA(data, timeperiod=14)
|
|
assert len(result) == 50
|
|
|
|
def test_rsi_float32(self):
|
|
data = np.arange(1.0, 51.0, dtype=np.float32)
|
|
result = RSI(data, timeperiod=14)
|
|
assert len(result) == 50
|