292 lines
9.5 KiB
Python
292 lines
9.5 KiB
Python
"""
|
|
Comparison tests: ferro_ta vs ta (Bukosabino's library) (Priority 5 - requires ta).
|
|
|
|
Secondary cross-check using Bukosabino's ta library. Validates same indicators
|
|
from a second independent implementation. This is shorter (~200 lines) and
|
|
focused on highest-value duplicates.
|
|
|
|
Requirements
|
|
------------
|
|
Install ta before running these tests::
|
|
|
|
pip install ta
|
|
|
|
The tests are automatically skipped when ta is not installed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skip the whole module when ta is not available
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ta = pytest.importorskip(
|
|
"ta", reason="ta library not installed; skipping comparison tests"
|
|
)
|
|
pd = pytest.importorskip("pandas", reason="pandas required for ta")
|
|
|
|
import ferro_ta # noqa: E402
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _valid_mask(*arrays: np.ndarray) -> np.ndarray:
|
|
"""Return boolean mask for positions where *all* arrays are finite."""
|
|
mask = np.ones(len(arrays[0]), dtype=bool)
|
|
for a in arrays:
|
|
mask &= ~np.isnan(a)
|
|
return mask
|
|
|
|
|
|
def _allclose(
|
|
a: np.ndarray, b: np.ndarray, atol: float = 1e-6, tail_fraction: float = 1.0
|
|
) -> bool:
|
|
"""Compare arrays within tolerance, optionally only comparing tail."""
|
|
mask = _valid_mask(a, b)
|
|
if not mask.any():
|
|
return False
|
|
|
|
if tail_fraction < 1.0:
|
|
n = len(a)
|
|
start_idx = int(n * (1 - tail_fraction))
|
|
mask[:start_idx] = False
|
|
|
|
if not mask.any():
|
|
return False
|
|
|
|
return bool(np.allclose(a[mask], b[mask], atol=atol))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Overlap Studies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSMAVsTA:
|
|
"""SMA — Exact match."""
|
|
|
|
def test_sma_exact_match(self, ohlcv_500):
|
|
"""SMA should match ta library exactly."""
|
|
close = ohlcv_500["close"]
|
|
period = 20
|
|
|
|
ft = ferro_ta.SMA(close, timeperiod=period)
|
|
|
|
df = pd.DataFrame({"close": close})
|
|
ta_indicator = ta.trend.SMAIndicator(close=df["close"], window=period)
|
|
ta_result = ta_indicator.sma_indicator().to_numpy()
|
|
|
|
assert _allclose(ft, ta_result, atol=1e-8)
|
|
|
|
|
|
class TestEMAVsTA:
|
|
"""EMA — Tail 30% match."""
|
|
|
|
def test_ema_tail_convergence(self, ohlcv_500):
|
|
"""EMA should converge in tail 30%."""
|
|
close = ohlcv_500["close"]
|
|
period = 20
|
|
|
|
ft = ferro_ta.EMA(close, timeperiod=period)
|
|
|
|
df = pd.DataFrame({"close": close})
|
|
ta_indicator = ta.trend.EMAIndicator(close=df["close"], window=period)
|
|
ta_result = ta_indicator.ema_indicator().to_numpy()
|
|
|
|
assert _allclose(ft, ta_result, atol=1e-4, tail_fraction=0.3)
|
|
|
|
|
|
class TestBBANDSVsTA:
|
|
"""BBANDS — Exact match."""
|
|
|
|
def test_bbands_exact_match(self, ohlcv_500):
|
|
"""Bollinger Bands should match ta library exactly."""
|
|
close = ohlcv_500["close"]
|
|
period = 20
|
|
nbdev = 2.0
|
|
|
|
ft_upper, ft_middle, ft_lower = ferro_ta.BBANDS(
|
|
close, timeperiod=period, nbdevup=nbdev, nbdevdn=nbdev
|
|
)
|
|
|
|
df = pd.DataFrame({"close": close})
|
|
ta_indicator = ta.volatility.BollingerBands(
|
|
close=df["close"], window=period, window_dev=nbdev
|
|
)
|
|
ta_upper = ta_indicator.bollinger_hband().to_numpy()
|
|
ta_middle = ta_indicator.bollinger_mavg().to_numpy()
|
|
ta_lower = ta_indicator.bollinger_lband().to_numpy()
|
|
|
|
assert _allclose(ft_upper, ta_upper, atol=1e-8)
|
|
assert _allclose(ft_middle, ta_middle, atol=1e-8)
|
|
assert _allclose(ft_lower, ta_lower, atol=1e-8)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Momentum Indicators
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRSIVsTA:
|
|
"""RSI — Tail 30% match."""
|
|
|
|
def test_rsi_tail_convergence(self, ohlcv_500):
|
|
"""RSI should converge in tail 30%."""
|
|
close = ohlcv_500["close"]
|
|
period = 14
|
|
|
|
ft = ferro_ta.RSI(close, timeperiod=period)
|
|
|
|
df = pd.DataFrame({"close": close})
|
|
ta_indicator = ta.momentum.RSIIndicator(close=df["close"], window=period)
|
|
ta_result = ta_indicator.rsi().to_numpy()
|
|
|
|
assert _allclose(ft, ta_result, atol=1e-3, tail_fraction=0.3)
|
|
|
|
|
|
class TestMACDVsTA:
|
|
"""MACD — Tail 30% match."""
|
|
|
|
def test_macd_tail_convergence(self, ohlcv_500):
|
|
"""MACD should converge in tail 30%."""
|
|
close = ohlcv_500["close"]
|
|
|
|
ft_macd, ft_signal, ft_hist = ferro_ta.MACD(
|
|
close, fastperiod=12, slowperiod=26, signalperiod=9
|
|
)
|
|
|
|
df = pd.DataFrame({"close": close})
|
|
ta_indicator = ta.trend.MACD(
|
|
close=df["close"], window_slow=26, window_fast=12, window_sign=9
|
|
)
|
|
ta_macd = ta_indicator.macd().to_numpy()
|
|
ta_signal = ta_indicator.macd_signal().to_numpy()
|
|
ta_hist = ta_indicator.macd_diff().to_numpy()
|
|
|
|
assert _allclose(ft_macd, ta_macd, atol=1e-2, tail_fraction=0.3)
|
|
assert _allclose(ft_signal, ta_signal, atol=1e-2, tail_fraction=0.3)
|
|
assert _allclose(ft_hist, ta_hist, atol=1e-2, tail_fraction=0.3)
|
|
|
|
|
|
class TestSTOCHVsTA:
|
|
"""STOCH — Structural validation (algorithms are incompatible with ta library).
|
|
|
|
Note: the ``ta`` library's StochasticOscillator uses simple rolling-mean (SMA)
|
|
smoothing, while ferro_ta follows TA-Lib and applies Wilder's exponential smoothing.
|
|
The two approaches produce values that diverge by up to 30 percentage points, so
|
|
a direct numeric comparison is meaningless. Instead we validate structural
|
|
properties that every correct STOCH implementation must satisfy.
|
|
"""
|
|
|
|
def test_stoch_structural_properties(self, ohlcv_500):
|
|
"""STOCH output satisfies range and warm-up constraints."""
|
|
high = ohlcv_500["high"]
|
|
low = ohlcv_500["low"]
|
|
close = ohlcv_500["close"]
|
|
|
|
ft_slowk, ft_slowd = ferro_ta.STOCH(
|
|
high, low, close, fastk_period=14, slowk_period=3, slowd_period=3
|
|
)
|
|
|
|
# Values in valid region must be within [0, 100]
|
|
valid_k = ft_slowk[np.isfinite(ft_slowk)]
|
|
valid_d = ft_slowd[np.isfinite(ft_slowd)]
|
|
assert len(valid_k) > 0, "STOCH slowk should have valid values"
|
|
assert len(valid_d) > 0, "STOCH slowd should have valid values"
|
|
assert np.all(valid_k >= 0.0) and np.all(valid_k <= 100.0), (
|
|
"STOCH slowk must be in [0, 100]"
|
|
)
|
|
assert np.all(valid_d >= 0.0) and np.all(valid_d <= 100.0), (
|
|
"STOCH slowd must be in [0, 100]"
|
|
)
|
|
|
|
# Warm-up: TA-Lib STOCH NaN count = fastk_period + slowk_period - 1
|
|
expected_nan = (
|
|
14 + 3 + 1 - 1
|
|
) # = fastk_period + slowk_period (TA-Lib convention)
|
|
actual_nan_k = int(np.sum(np.isnan(ft_slowk)))
|
|
assert actual_nan_k == expected_nan, (
|
|
f"STOCH slowk NaN warmup: expected {expected_nan}, got {actual_nan_k}"
|
|
)
|
|
|
|
|
|
class TestWILLRVsTA:
|
|
"""WILLR — Exact match."""
|
|
|
|
def test_willr_exact_match(self, ohlcv_500):
|
|
"""Williams %R should match ta library exactly."""
|
|
high = ohlcv_500["high"]
|
|
low = ohlcv_500["low"]
|
|
close = ohlcv_500["close"]
|
|
period = 14
|
|
|
|
ft = ferro_ta.WILLR(high, low, close, timeperiod=period)
|
|
|
|
df = pd.DataFrame({"high": high, "low": low, "close": close})
|
|
ta_indicator = ta.momentum.WilliamsRIndicator(
|
|
high=df["high"], low=df["low"], close=df["close"], lbp=period
|
|
)
|
|
ta_result = ta_indicator.williams_r().to_numpy()
|
|
|
|
assert _allclose(ft, ta_result, atol=1e-8)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Volatility
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestATRVsTA:
|
|
"""ATR — Tail 30% match."""
|
|
|
|
def test_atr_tail_convergence(self, ohlcv_500):
|
|
"""ATR should converge in tail 30%."""
|
|
high = ohlcv_500["high"]
|
|
low = ohlcv_500["low"]
|
|
close = ohlcv_500["close"]
|
|
period = 14
|
|
|
|
ft = ferro_ta.ATR(high, low, close, timeperiod=period)
|
|
|
|
df = pd.DataFrame({"high": high, "low": low, "close": close})
|
|
ta_indicator = ta.volatility.AverageTrueRange(
|
|
high=df["high"], low=df["low"], close=df["close"], window=period
|
|
)
|
|
ta_result = ta_indicator.average_true_range().to_numpy()
|
|
|
|
assert _allclose(ft, ta_result, atol=1e-2, tail_fraction=0.3)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Volume
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOBVVsTA:
|
|
"""OBV — Incremental match."""
|
|
|
|
def test_obv_incremental_match(self, ohlcv_500):
|
|
"""OBV differences should match."""
|
|
close = ohlcv_500["close"]
|
|
volume = ohlcv_500["volume"]
|
|
|
|
ft = ferro_ta.OBV(close, volume)
|
|
|
|
df = pd.DataFrame({"close": close, "volume": volume})
|
|
ta_indicator = ta.volume.OnBalanceVolumeIndicator(
|
|
close=df["close"], volume=df["volume"]
|
|
)
|
|
ta_result = ta_indicator.on_balance_volume().to_numpy()
|
|
|
|
# Compare differences (OBV can have different starting values)
|
|
ft_diff = np.diff(ft)
|
|
ta_diff = np.diff(ta_result)
|
|
|
|
mask = ~np.isnan(ft_diff) & ~np.isnan(ta_diff)
|
|
assert np.allclose(ft_diff[mask], ta_diff[mask], atol=1e-8)
|