ferro-ta/tests/integration/test_vs_ta.py

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