ferro-ta/benchmarks/test_accuracy.py

"""
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"