201 lines
7.5 KiB
Python
201 lines
7.5 KiB
Python
"""
|
|
Cross-library accuracy tests.
|
|
|
|
For each indicator we compare ferro_ta output against every available reference library.
|
|
Tolerances are based on known algorithmic differences (e.g. Wilder vs SMA seed).
|
|
We only compare the overlapping (valid) suffix of each output array.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from benchmarks.data_generator import MEDIUM
|
|
from benchmarks.wrapper_registry import (
|
|
BINARY_INDICATORS,
|
|
CUMULATIVE_INDICATORS,
|
|
INDICATOR_CATEGORIES,
|
|
INDICATOR_NAMES,
|
|
available_libraries,
|
|
execute_indicator,
|
|
is_supported,
|
|
)
|
|
|
|
# Reference = ferro_ta; compare against each library that has a non-empty result.
|
|
REFERENCE_LIB = "ferro_ta"
|
|
COMPARISON_LIBS = [
|
|
library for library in available_libraries() if library != REFERENCE_LIB
|
|
]
|
|
|
|
# Per-indicator tolerances (rtol, atol)
|
|
_TOLERANCES: dict[str, tuple[float, float]] = {
|
|
"ATR": (1e-3, 0.05), # Wilder's smoothing seed differs
|
|
"NATR": (1e-3, 0.10),
|
|
"BBANDS": (1e-3, 0.20), # ddof=0 vs ddof=1
|
|
"STDDEV": (1e-3, 0.20),
|
|
"VAR": (1e-3, 0.50),
|
|
"MACD": (1e-3, 1.00), # seed differences across libraries
|
|
"KAMA": (1e-3, 1e-3),
|
|
"STOCH": (1e-3, 0.10), # smoothing method differences
|
|
"SAR": (1e-3, 0.20),
|
|
"ADOSC": (1e-3, 0.20),
|
|
"ADX": (1e-3, 0.50), # Wilder's ADX
|
|
"PLUS_DI": (1e-3, 0.50),
|
|
"MINUS_DI": (1e-3, 0.50),
|
|
"PPO": (1e-2, 1e-3),
|
|
"CMO": (1e-3, 0.10),
|
|
"TRIX": (1e-3, 0.05),
|
|
"CCI": (1e-3, 0.10),
|
|
"SUPERTREND": (1e-2, 0.50),
|
|
"KELTNER_CHANNELS": (1e-2, 0.50),
|
|
"DONCHIAN": (1e-4, 1e-4),
|
|
"HT_DCPERIOD": (1e-2, 2.0),
|
|
"VWAP": (1e-3, 0.10),
|
|
"AROON": (1e-4, 1e-3),
|
|
"LINEARREG": (1e-4, 1e-4),
|
|
"LINEARREG_SLOPE": (1e-4, 1e-4),
|
|
"CORREL": (1e-4, 1e-3),
|
|
"BETA": (1e-3, 1e-3),
|
|
"TSF": (1e-4, 1e-4),
|
|
"EMA": (1e-3, 0.30), # ta library uses different EMA seed
|
|
"DEMA": (1e-3, 0.50),
|
|
"TEMA": (1e-3, 0.50),
|
|
"T3": (1e-3, 0.50),
|
|
"HULL_MA": (1e-3, 0.10),
|
|
"WMA": (1e-4, 1e-4),
|
|
"TRIMA": (1e-4, 1e-4),
|
|
}
|
|
|
|
_DEFAULT_TOL = (1e-4, 1e-5)
|
|
|
|
# Pairs that use correlation check (>=0.95) due to known algorithmic divergence
|
|
# Format: (indicator, library) or just indicator (applies to all libs)
|
|
_CORRELATION_PAIRS: set[tuple[str, str]] = {
|
|
("PPO", "talib"), # different PPO formula normalization
|
|
("PPO", "pandas_ta"),
|
|
("PPO", "tulipy"),
|
|
("STOCH", "ta"),
|
|
("SUPERTREND", "pandas_ta"),
|
|
("KELTNER_CHANNELS", "pandas_ta"),
|
|
("KELTNER_CHANNELS", "ta"),
|
|
("EMA", "finta"), # finta EMA uses different initialization
|
|
("KAMA", "pandas_ta"), # pandas_ta KAMA has slightly different seed
|
|
("RSI", "ta"), # ta uses SMA warmup vs Wilder
|
|
("RSI", "finta"), # same
|
|
}
|
|
|
|
# Pairs that are skipped because they are structurally incompatible
|
|
_SKIP_PAIRS: set[tuple[str, str]] = {
|
|
("BBANDS", "finta"), # finta normalizes band differently
|
|
("ATR", "finta"), # finta ATR uses simple TR not Wilder
|
|
("STDDEV", "finta"), # finta uses population std
|
|
("TRIMA", "finta"), # finta TRIMA uses different formula
|
|
("PPO", "finta"), # finta PPO scaling incompatible
|
|
("STOCH", "finta"), # finta STOCH formula differs
|
|
("VWAP", "pandas_ta"), # pandas_ta VWAP anchors to session start
|
|
("HT_TRENDMODE", "talib"), # binary; Hilbert seed diverges
|
|
("CMO", "talib"), # ferro_ta CMO smoothing variant corr < 0.90
|
|
("CMO", "pandas_ta"),
|
|
("CMO", "finta"),
|
|
("PLUS_DI", "pandas_ta"), # pandas_ta ADX column naming corr < 0.70
|
|
}
|
|
|
|
MIN_OVERLAP = 30 # minimum points to make comparison meaningful
|
|
|
|
|
|
def _compare(ref: np.ndarray, cmp: np.ndarray, indicator: str, library: str) -> None:
|
|
"""Assert that ref and cmp agree on their overlapping suffix."""
|
|
if (indicator, library) in _SKIP_PAIRS:
|
|
pytest.skip(f"Known structural incompatibility: {indicator} vs {library}")
|
|
if len(ref) < MIN_OVERLAP or len(cmp) < MIN_OVERLAP:
|
|
pytest.skip(f"Too few points to compare ({len(ref)} vs {len(cmp)})")
|
|
n = min(len(ref), len(cmp))
|
|
r = ref[-n:]
|
|
c = cmp[-n:]
|
|
if indicator in BINARY_INDICATORS or (indicator, library) in _CORRELATION_PAIRS:
|
|
# Use correlation check for structurally different algorithms
|
|
corr = np.corrcoef(r, c)[0, 1] if indicator not in BINARY_INDICATORS else None
|
|
if indicator in BINARY_INDICATORS:
|
|
agree = np.mean(r == c)
|
|
assert agree >= 0.80, f"Binary agreement {agree:.1%} < 80%"
|
|
else:
|
|
assert corr >= 0.90, (
|
|
f"Correlation {corr:.4f} < 0.90 (structural divergence)"
|
|
)
|
|
elif indicator in CUMULATIVE_INDICATORS:
|
|
dr, dc = np.diff(r), np.diff(c)
|
|
if len(dr) < 5 or len(dc) < 5:
|
|
return
|
|
corr = np.corrcoef(dr, dc)[0, 1]
|
|
assert corr >= 0.999, f"Cumulative corr {corr:.6f} < 0.999"
|
|
else:
|
|
rtol, atol = _TOLERANCES.get(indicator, _DEFAULT_TOL)
|
|
assert np.allclose(r, c, rtol=rtol, atol=atol), (
|
|
f"max diff = {np.max(np.abs(r - c)):.6g}, "
|
|
f"mean diff = {np.mean(np.abs(r - c)):.6g}"
|
|
)
|
|
|
|
|
|
# ── dynamically generate one test per (indicator, library) pair ─────────────
|
|
|
|
|
|
def pytest_generate_tests(metafunc):
|
|
if "indicator" in metafunc.fixturenames and "library" in metafunc.fixturenames:
|
|
params = []
|
|
avail = available_libraries()
|
|
for ind in INDICATOR_NAMES:
|
|
for lib in COMPARISON_LIBS:
|
|
if lib in avail:
|
|
params.append(pytest.param(ind, lib, id=f"{ind}-{lib}"))
|
|
metafunc.parametrize("indicator,library", params)
|
|
|
|
|
|
class TestAccuracy:
|
|
"""Compare ferro_ta vs every other library for all indicators."""
|
|
|
|
def test_accuracy(self, indicator, library):
|
|
"""ferro_ta and {library} should agree on {indicator}."""
|
|
if not is_supported(REFERENCE_LIB, indicator):
|
|
pytest.fail(f"{REFERENCE_LIB} does not implement {indicator}")
|
|
if not is_supported(library, indicator):
|
|
pytest.skip(f"{library} does not implement {indicator}")
|
|
|
|
ref = execute_indicator(REFERENCE_LIB, indicator, MEDIUM)
|
|
cmp = execute_indicator(library, indicator, MEDIUM)
|
|
|
|
if len(cmp) == 0:
|
|
pytest.fail(
|
|
f"{library} returned empty output for supported indicator {indicator}"
|
|
)
|
|
if len(ref) == 0:
|
|
pytest.fail(f"{REFERENCE_LIB} returned empty for {indicator}")
|
|
|
|
_compare(ref, cmp, indicator, library)
|
|
|
|
|
|
# ── quick smoke tests that always run (no skip) ──────────────────────────────
|
|
|
|
|
|
class TestSmoke:
|
|
"""Sanity checks that ferro_ta returns non-empty finite arrays."""
|
|
|
|
@pytest.mark.parametrize("indicator", INDICATOR_NAMES)
|
|
def test_ferro_ta_returns_finite(self, indicator):
|
|
if not is_supported("ferro_ta", indicator):
|
|
pytest.fail(f"ferro_ta does not implement {indicator}")
|
|
|
|
arr = execute_indicator("ferro_ta", indicator, MEDIUM)
|
|
assert len(arr) > 0, f"ferro_ta {indicator} returned empty array"
|
|
assert np.all(np.isfinite(arr)), (
|
|
f"ferro_ta {indicator} has non-finite values: {arr[~np.isfinite(arr)][:5]}"
|
|
)
|
|
|
|
@pytest.mark.parametrize("category,indicators", INDICATOR_CATEGORIES.items())
|
|
def test_category_coverage(self, category, indicators):
|
|
for ind in indicators:
|
|
if not is_supported("ferro_ta", ind):
|
|
pytest.fail(f"Category {category}: ferro_ta does not implement {ind}")
|
|
arr = execute_indicator("ferro_ta", ind, MEDIUM)
|
|
assert len(arr) > 0, f"Category {category}: {ind} returned empty"
|