Pratik Bhadane
307beeca02
Prepare the first public 1.0.0 release and finish the remaining CI hardening work. Highlights: - align Python, Rust, WASM, Conda, API, MCP, and docs version metadata to 1.0.0 - promote package metadata to Production/Stable and update stability/versioning docs for the stable series - move the accumulated Unreleased notes into a dated 1.0.0 changelog section and keep a fresh top-level Unreleased block - strengthen the changelog checker so it validates a single top-level Unreleased section - fix the CI/package support mismatch by declaring Python >=3.10 consistently and gating pandas-ta extras to Python 3.12+ - restore Sphinx autodoc compatibility for documented ferro_ta.<module> imports by registering module aliases - make the TA-Lib benchmark guardrail less flaky by checking median and tail-percentile speedups instead of failing on a single mild outlier - switch PyPI publishing to OIDC-only trusted publishing and wire the changelog check into the required CI gate - apply the Ruff-driven cleanup across the Python and test tree and refresh uv/cargo lockfiles Validated locally: - python3 scripts/check_changelog.py - uv run --with ruff ruff check python tests - uv run --with ruff ruff format --check python tests - uv lock --check - sphinx-build -b html docs docs/_build -W --keep-going - build/install the ferro_ta 1.0.0 wheel successfully
2179 lines
75 KiB
Python
2179 lines
75 KiB
Python
"""
|
||
Comparison tests: ferro_ta vs TA-Lib (ta-lib Python wrapper).
|
||
|
||
This module verifies that ferro_ta is a drop-in replacement for TA-Lib by
|
||
comparing the outputs of every shared indicator for:
|
||
|
||
* **Shape compatibility** — same output length and NaN count (or ±1 where a
|
||
documented off-by-one exists).
|
||
* **Value accuracy** — exact match within floating-point tolerance where the
|
||
algorithms are identical; range / convergence checks where initialization
|
||
differs.
|
||
|
||
Known differences are documented next to each test so consumers know what
|
||
to expect when migrating from TA-Lib.
|
||
|
||
Requirements
|
||
------------
|
||
Install ta-lib before running these tests::
|
||
|
||
pip install ta-lib
|
||
|
||
The tests are automatically skipped when ta-lib is not installed, so the
|
||
main CI pipeline never fails because of a missing optional dependency.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Skip the whole module when ta-lib is not available
|
||
# ---------------------------------------------------------------------------
|
||
|
||
talib = pytest.importorskip(
|
||
"talib", reason="ta-lib not installed; skipping comparison tests"
|
||
)
|
||
|
||
import ferro_ta # noqa: E402 (import after potential skip)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shared realistic OHLCV data (500 bars for proper convergence)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
RNG = np.random.default_rng(42)
|
||
|
||
N = 500 # Increased from 100 to 500 for proper EMA/RSI/ATR convergence
|
||
CLOSE = 44.0 + np.cumsum(RNG.standard_normal(N) * 0.5)
|
||
HIGH = CLOSE + RNG.uniform(0.1, 1.0, N)
|
||
LOW = CLOSE - RNG.uniform(0.1, 1.0, N)
|
||
OPEN = CLOSE + RNG.standard_normal(N) * 0.2
|
||
VOLUME = RNG.uniform(500.0, 2000.0, N)
|
||
|
||
# Simple monotonically increasing series used for deterministic checks
|
||
LINEAR = np.arange(1.0, N + 1.0, dtype=np.float64)
|
||
LINEAR_HIGH = LINEAR + 0.5
|
||
LINEAR_LOW = LINEAR - 0.5
|
||
LINEAR_OPEN = LINEAR - 0.2
|
||
LINEAR_VOL = np.ones(N) * 1000.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Minimum fraction of values that must agree in sign for correlated indicators.
|
||
SIGN_AGREEMENT_THRESHOLD = 0.8
|
||
|
||
# Per-pattern candlestick agreement thresholds.
|
||
# Most patterns use 0.80; patterns with known definition differences from TA-Lib
|
||
# use lower thresholds with a documented reason.
|
||
CDL_AGREEMENT_THRESHOLDS: dict[str, float] = {
|
||
# Body/shadow ratio thresholds differ between ferro_ta and TA-Lib
|
||
"CDLHIGHWAVE": 0.65, # Shadow length threshold differs; 69% observed
|
||
"CDLLONGLEGGEDDOJI": 0.70, # Long-leg threshold differs; 75% observed
|
||
"CDLSHORTLINE": 0.20, # Body-size cutoff definition completely differs; 25% observed
|
||
"CDLSPINNINGTOP": 0.75, # Body ratio threshold differs; 78% observed
|
||
"CDLDOJI": 0.85, # Shadow ratio precision differs; 86% observed
|
||
}
|
||
|
||
|
||
def _nan_count(arr: np.ndarray) -> int:
|
||
return int(np.sum(np.isnan(arr)))
|
||
|
||
|
||
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) -> bool:
|
||
mask = _valid_mask(a, b)
|
||
if not mask.any():
|
||
return False
|
||
return bool(np.allclose(a[mask], b[mask], atol=atol))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Overlap Studies
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSMA:
|
||
"""SMA — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.SMA(CLOSE, timeperiod=10)
|
||
ta = talib.SMA(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.SMA(CLOSE, timeperiod=10)
|
||
ta = talib.SMA(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.SMA(CLOSE, timeperiod=5)
|
||
ta = talib.SMA(CLOSE, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
|
||
class TestEMA:
|
||
"""EMA — shape matches; values differ slightly in the warmup region.
|
||
|
||
ferro_ta seeds the EMA from the very first data point using the standard
|
||
recursive formula, while TA-Lib seeds the first EMA value with the SMA
|
||
of the initial ``timeperiod`` bars. After enough bars the two series
|
||
converge. We verify:
|
||
|
||
* Same NaN count (warmup length is identical).
|
||
* After the series converge (last 30 % of the output) values agree.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.EMA(CLOSE, timeperiod=10)
|
||
ta = talib.EMA(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.EMA(CLOSE, timeperiod=5)
|
||
ta = talib.EMA(CLOSE, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_values_converge(self):
|
||
"""After convergence (tail 30%), EMA should be very close with 500 bars."""
|
||
ft = ferro_ta.EMA(CLOSE, timeperiod=5)
|
||
ta = talib.EMA(CLOSE, timeperiod=5)
|
||
# With 500 bars, compare last 30% with tighter tolerance
|
||
tail_start = int(N * 0.7)
|
||
assert np.allclose(
|
||
ft[tail_start:], ta[tail_start:], atol=1e-5
|
||
) # Tightened from 1e-3
|
||
|
||
def test_values_finite_and_reasonable(self):
|
||
ft = ferro_ta.EMA(CLOSE, timeperiod=5)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert finite.min() > 0
|
||
assert finite.max() < 1000
|
||
|
||
|
||
class TestWMA:
|
||
"""WMA — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.WMA(CLOSE, timeperiod=10)
|
||
ta = talib.WMA(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.WMA(CLOSE, timeperiod=10)
|
||
ta = talib.WMA(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestDEMA:
|
||
"""DEMA — shape matches; values differ (EMA-based initialization)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.DEMA(CLOSE, timeperiod=5)
|
||
ta = talib.DEMA(CLOSE, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.DEMA(CLOSE, timeperiod=5)
|
||
ta = talib.DEMA(CLOSE, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_values_converge(self):
|
||
ft = ferro_ta.DEMA(CLOSE, timeperiod=5)
|
||
ta = talib.DEMA(CLOSE, timeperiod=5)
|
||
mid = N // 2
|
||
assert np.allclose(ft[mid:], ta[mid:], atol=1e-2)
|
||
|
||
|
||
class TestTEMA:
|
||
"""TEMA — shape matches; values differ (EMA-based initialization)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.TEMA(CLOSE, timeperiod=5)
|
||
ta = talib.TEMA(CLOSE, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.TEMA(CLOSE, timeperiod=5)
|
||
ta = talib.TEMA(CLOSE, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_values_converge(self):
|
||
ft = ferro_ta.TEMA(CLOSE, timeperiod=5)
|
||
ta = talib.TEMA(CLOSE, timeperiod=5)
|
||
mid = N // 2
|
||
assert np.allclose(ft[mid:], ta[mid:], atol=1e-2)
|
||
|
||
|
||
class TestTRIMA:
|
||
"""TRIMA — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.TRIMA(CLOSE, timeperiod=10)
|
||
ta = talib.TRIMA(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.TRIMA(CLOSE, timeperiod=10)
|
||
ta = talib.TRIMA(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestKAMA:
|
||
"""KAMA — values match after the first bar.
|
||
|
||
TA-Lib marks index ``timeperiod - 1`` as NaN (the last element of the
|
||
seed window), while ferro_ta emits a value there. All subsequent values
|
||
are identical.
|
||
"""
|
||
|
||
def test_values_match_after_seed(self):
|
||
ft = ferro_ta.KAMA(CLOSE, timeperiod=10)
|
||
ta = talib.KAMA(CLOSE, timeperiod=10)
|
||
# Skip the one bar where TA-Lib is still NaN
|
||
start = max(_nan_count(ft), _nan_count(ta)) + 1
|
||
assert np.allclose(ft[start:], ta[start:], atol=1e-8)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.KAMA(CLOSE, timeperiod=10)
|
||
ta = talib.KAMA(CLOSE, timeperiod=10)
|
||
assert len(ft) == len(ta)
|
||
|
||
|
||
class TestT3:
|
||
"""T3 — shape matches; values differ (EMA-based initialization)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.T3(CLOSE, timeperiod=5)
|
||
ta = talib.T3(CLOSE, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.T3(CLOSE, timeperiod=5)
|
||
ta = talib.T3(CLOSE, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_values_converge(self):
|
||
ft = ferro_ta.T3(CLOSE, timeperiod=5)
|
||
ta = talib.T3(CLOSE, timeperiod=5)
|
||
# With 500 bars, use last 30% with tighter tolerance
|
||
tail_start = int(N * 0.7)
|
||
assert np.allclose(
|
||
ft[tail_start:], ta[tail_start:], atol=1e-3
|
||
) # Tightened from 5e-2
|
||
|
||
|
||
class TestBBANDS:
|
||
"""BBANDS — exact match."""
|
||
|
||
def test_values_match(self):
|
||
upper_ft, mid_ft, lower_ft = ferro_ta.BBANDS(
|
||
CLOSE, timeperiod=10, nbdevup=2.0, nbdevdn=2.0
|
||
)
|
||
upper_ta, mid_ta, lower_ta = talib.BBANDS(
|
||
CLOSE, timeperiod=10, nbdevup=2.0, nbdevdn=2.0
|
||
)
|
||
assert _allclose(upper_ft, upper_ta)
|
||
assert _allclose(mid_ft, mid_ta)
|
||
assert _allclose(lower_ft, lower_ta)
|
||
|
||
def test_nan_count_match(self):
|
||
upper_ft, _, _ = ferro_ta.BBANDS(CLOSE, timeperiod=10)
|
||
upper_ta, _, _ = talib.BBANDS(CLOSE, timeperiod=10)
|
||
assert _nan_count(upper_ft) == _nan_count(upper_ta)
|
||
|
||
def test_output_length_match(self):
|
||
upper_ft, _, _ = ferro_ta.BBANDS(CLOSE, timeperiod=5)
|
||
upper_ta, _, _ = talib.BBANDS(CLOSE, timeperiod=5)
|
||
assert len(upper_ft) == len(upper_ta)
|
||
|
||
|
||
class TestMACD:
|
||
"""MACD — shape matches; values differ (EMA-based initialization).
|
||
|
||
The MACD line, signal, and histogram converge after sufficient warmup.
|
||
The histogram relationship (macd - signal) is preserved in both.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft_m, ft_s, ft_h = ferro_ta.MACD(
|
||
CLOSE, fastperiod=3, slowperiod=6, signalperiod=2
|
||
)
|
||
ta_m, ta_s, ta_h = talib.MACD(CLOSE, fastperiod=3, slowperiod=6, signalperiod=2)
|
||
assert _nan_count(ft_m) == _nan_count(ta_m)
|
||
|
||
def test_output_length_match(self):
|
||
ft_m, ft_s, ft_h = ferro_ta.MACD(CLOSE)
|
||
ta_m, ta_s, ta_h = talib.MACD(CLOSE)
|
||
assert len(ft_m) == len(ta_m) == len(CLOSE)
|
||
|
||
def test_histogram_relationship(self):
|
||
"""Histogram = MACD line − signal line (must hold for both libraries)."""
|
||
for fn, lib in [(ferro_ta.MACD, "ferro_ta"), (talib.MACD, "talib")]:
|
||
m, s, h = fn(CLOSE, fastperiod=3, slowperiod=6, signalperiod=2)
|
||
mask = _valid_mask(m, s, h)
|
||
assert np.allclose(h[mask], m[mask] - s[mask], atol=1e-10), (
|
||
f"{lib} histogram mismatch"
|
||
)
|
||
|
||
def test_values_converge(self):
|
||
ft_m, _, _ = ferro_ta.MACD(CLOSE, fastperiod=3, slowperiod=6, signalperiod=2)
|
||
ta_m, _, _ = talib.MACD(CLOSE, fastperiod=3, slowperiod=6, signalperiod=2)
|
||
assert np.allclose(ft_m[-N // 4 :], ta_m[-N // 4 :], atol=1e-2)
|
||
|
||
|
||
class TestMACDFIX:
|
||
"""MACDFIX — shape matches; values differ (EMA-based initialization)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft_m, ft_s, ft_h = ferro_ta.MACDFIX(CLOSE)
|
||
ta_m, ta_s, ta_h = talib.MACDFIX(CLOSE)
|
||
assert _nan_count(ft_m) == _nan_count(ta_m)
|
||
|
||
def test_output_length_match(self):
|
||
ft_m, _, _ = ferro_ta.MACDFIX(CLOSE)
|
||
ta_m, _, _ = talib.MACDFIX(CLOSE)
|
||
assert len(ft_m) == len(ta_m)
|
||
|
||
|
||
class TestSAR:
|
||
"""SAR — same output length; values may differ due to reversal history.
|
||
|
||
Known difference: Parabolic SAR reversal history can diverge from TA-Lib
|
||
due to floating-point accumulation in early bars. Output shape (length,
|
||
NaN count) matches exactly.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.SAR(HIGH, LOW)
|
||
ta = talib.SAR(HIGH, LOW)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.SAR(HIGH, LOW)
|
||
ta = talib.SAR(HIGH, LOW)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_values_positive(self):
|
||
ft = ferro_ta.SAR(HIGH, LOW)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v > 0 for v in finite)
|
||
|
||
def test_correlation_above_threshold(self):
|
||
"""Correlated with TA-Lib even if not exact (same algorithm, different accumulation)."""
|
||
ft = ferro_ta.SAR(HIGH, LOW)
|
||
ta = talib.SAR(HIGH, LOW)
|
||
mask = _valid_mask(ft, ta)
|
||
if mask.sum() >= 5:
|
||
corr = float(np.corrcoef(ft[mask], ta[mask])[0, 1])
|
||
assert corr > 0.90, f"SAR correlation {corr:.3f} < 0.90"
|
||
|
||
|
||
class TestSAREXT:
|
||
"""SAREXT — SAR Extended. Shape must match; values may differ.
|
||
|
||
Known difference: Same as SAR — reversal history from TA-Lib diverges
|
||
due to floating-point accumulation.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.SAREXT(HIGH, LOW)
|
||
ta = talib.SAREXT(HIGH, LOW)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.SAREXT(HIGH, LOW)
|
||
ta = talib.SAREXT(HIGH, LOW)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestMAMA:
|
||
"""MAMA — MESA Adaptive Moving Average.
|
||
|
||
Known difference: TA-Lib C applies slightly different floating-point rounding
|
||
in the adaptive factor clamp. The two series are highly correlated (r > 0.95)
|
||
and values converge after ~100 bars, but differ numerically in early bars.
|
||
Status: ⚠️ Corr.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft_m, ft_f = ferro_ta.MAMA(CLOSE)
|
||
ta_m, ta_f = talib.MAMA(CLOSE)
|
||
assert len(ft_m) == len(ta_m)
|
||
assert len(ft_f) == len(ta_f)
|
||
|
||
def test_nan_count_match(self):
|
||
ft_m, ft_f = ferro_ta.MAMA(CLOSE)
|
||
ta_m, ta_f = talib.MAMA(CLOSE)
|
||
assert _nan_count(ft_m) == _nan_count(ta_m)
|
||
assert _nan_count(ft_f) == _nan_count(ta_f)
|
||
|
||
def test_mama_correlated_with_talib(self):
|
||
"""MAMA should be highly correlated with TA-Lib (r > 0.95)."""
|
||
ft_m, _ = ferro_ta.MAMA(CLOSE)
|
||
ta_m, _ = talib.MAMA(CLOSE)
|
||
mask = _valid_mask(ft_m, ta_m)
|
||
if mask.sum() >= 5:
|
||
corr = float(np.corrcoef(ft_m[mask], ta_m[mask])[0, 1])
|
||
assert corr > 0.95, f"MAMA correlation {corr:.3f} < 0.95"
|
||
|
||
def test_fama_correlated_with_talib(self):
|
||
"""FAMA should be correlated with TA-Lib (r > 0.80)."""
|
||
_, ft_f = ferro_ta.MAMA(CLOSE)
|
||
_, ta_f = talib.MAMA(CLOSE)
|
||
mask = _valid_mask(ft_f, ta_f)
|
||
if mask.sum() >= 5:
|
||
corr = float(np.corrcoef(ft_f[mask], ta_f[mask])[0, 1])
|
||
assert corr > 0.80, f"FAMA correlation {corr:.3f} < 0.80"
|
||
|
||
def test_mama_converges_in_tail(self):
|
||
"""After 100 bars the difference should be small (< 0.5% of price)."""
|
||
long_close = 44.0 + np.cumsum(
|
||
np.random.default_rng(99).standard_normal(200) * 0.5
|
||
)
|
||
ft_m, _ = ferro_ta.MAMA(long_close)
|
||
ta_m, _ = talib.MAMA(long_close)
|
||
mask = _valid_mask(ft_m, ta_m)
|
||
if mask.sum() >= 10:
|
||
tail = np.where(mask)[0][-min(10, mask.sum()) :] # last valid bars
|
||
diff = np.abs(ft_m[tail] - ta_m[tail])
|
||
price_scale = np.abs(ta_m[tail]).mean()
|
||
assert (diff / price_scale).max() < 0.01, (
|
||
f"MAMA tail relative diff: {(diff / price_scale).max():.4f}"
|
||
)
|
||
|
||
|
||
class TestMIDPOINT:
|
||
"""MIDPOINT — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.MIDPOINT(CLOSE, timeperiod=5)
|
||
ta = talib.MIDPOINT(CLOSE, timeperiod=5)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.MIDPOINT(CLOSE, timeperiod=5)
|
||
ta = talib.MIDPOINT(CLOSE, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestMIDPRICE:
|
||
"""MIDPRICE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.MIDPRICE(HIGH, LOW, timeperiod=5)
|
||
ta = talib.MIDPRICE(HIGH, LOW, timeperiod=5)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.MIDPRICE(HIGH, LOW, timeperiod=5)
|
||
ta = talib.MIDPRICE(HIGH, LOW, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Momentum Indicators
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRSI:
|
||
"""RSI — same NaN count and length; values differ due to Wilder smoothing seed.
|
||
|
||
ferro_ta and TA-Lib use slightly different initializations for Wilder's
|
||
smoothed average gain/loss, leading to permanently different RSI values.
|
||
Both libraries produce values in [0, 100] with the same NaN structure.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_range_0_to_100(self):
|
||
for lib_rsi in [ferro_ta.RSI(CLOSE, 14), talib.RSI(CLOSE, 14)]:
|
||
finite = lib_rsi[~np.isnan(lib_rsi)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
def test_values_same_direction(self):
|
||
"""RSI should move in the same direction as TA-Lib (correlation > 0.9)."""
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.9
|
||
|
||
def test_values_converge_in_tail(self):
|
||
"""With 500 bars, RSI should converge in tail 30%."""
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
tail_start = int(N * 0.7)
|
||
mask = _valid_mask(ft[tail_start:], ta[tail_start:])
|
||
if mask.any():
|
||
assert np.allclose(
|
||
ft[tail_start:][mask], ta[tail_start:][mask], atol=1e-3
|
||
) # Added value comparison
|
||
|
||
|
||
class TestMOM:
|
||
"""MOM — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.MOM(CLOSE, timeperiod=10)
|
||
ta = talib.MOM(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.MOM(CLOSE, timeperiod=10)
|
||
ta = talib.MOM(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestROC:
|
||
"""ROC — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.ROC(CLOSE, timeperiod=10)
|
||
ta = talib.ROC(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.ROC(CLOSE, timeperiod=10)
|
||
ta = talib.ROC(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestROCP:
|
||
"""ROCP — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.ROCP(CLOSE, timeperiod=10)
|
||
ta = talib.ROCP(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestROCR:
|
||
"""ROCR — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.ROCR(CLOSE, timeperiod=10)
|
||
ta = talib.ROCR(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestROCR100:
|
||
"""ROCR100 — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.ROCR100(CLOSE, timeperiod=10)
|
||
ta = talib.ROCR100(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestWILLR:
|
||
"""WILLR — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.WILLR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.WILLR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.WILLR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.WILLR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_range_minus100_to_0(self):
|
||
ft = ferro_ta.WILLR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.WILLR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
for arr in [ft, ta]:
|
||
finite = arr[~np.isnan(arr)]
|
||
assert all(-100.0 <= v <= 0.0 for v in finite)
|
||
|
||
|
||
class TestAROON:
|
||
"""AROON — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft_down, ft_up = ferro_ta.AROON(HIGH, LOW, timeperiod=14)
|
||
ta_down, ta_up = talib.AROON(HIGH, LOW, timeperiod=14)
|
||
assert _allclose(ft_down, ta_down) and _allclose(ft_up, ta_up)
|
||
|
||
def test_nan_count_match(self):
|
||
ft_down, ft_up = ferro_ta.AROON(HIGH, LOW, timeperiod=14)
|
||
ta_down, ta_up = talib.AROON(HIGH, LOW, timeperiod=14)
|
||
assert _nan_count(ft_down) == _nan_count(ta_down)
|
||
|
||
def test_range_0_to_100(self):
|
||
ft_down, ft_up = ferro_ta.AROON(HIGH, LOW, timeperiod=14)
|
||
for arr in [ft_down, ft_up]:
|
||
finite = arr[~np.isnan(arr)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
|
||
class TestAROONOSC:
|
||
"""AROONOSC — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.AROONOSC(HIGH, LOW, timeperiod=14)
|
||
ta = talib.AROONOSC(HIGH, LOW, timeperiod=14)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.AROONOSC(HIGH, LOW, timeperiod=14)
|
||
ta = talib.AROONOSC(HIGH, LOW, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestCCI:
|
||
"""CCI — same NaN count and shape; mean-absolute-deviation may differ.
|
||
|
||
TA-Lib divides by 0.015 × MAD computed with the population formula.
|
||
ferro_ta may use a slightly different MAD implementation, producing
|
||
proportionally scaled but directionally identical values.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_same_sign(self):
|
||
"""CCI values should have the same sign as TA-Lib."""
|
||
ft = ferro_ta.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
# Both should agree on whether CCI is positive/negative
|
||
assert (
|
||
np.sum(np.sign(ft[mask]) == np.sign(ta[mask]))
|
||
> SIGN_AGREEMENT_THRESHOLD * mask.sum()
|
||
)
|
||
|
||
def test_values_strongly_correlated(self):
|
||
"""CCI values should be strongly correlated with TA-Lib values."""
|
||
ft = ferro_ta.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
|
||
class TestBOP:
|
||
"""BOP — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.BOP(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.BOP(OPEN, HIGH, LOW, CLOSE)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.BOP(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.BOP(OPEN, HIGH, LOW, CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
|
||
class TestMFI:
|
||
"""MFI — values match on a well-constructed series.
|
||
|
||
MFI (Money Flow Index) is computed from OHLCV and should agree exactly
|
||
when the typical prices and volumes are not degenerate.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.MFI(HIGH, LOW, CLOSE, VOLUME, timeperiod=14)
|
||
ta = talib.MFI(HIGH, LOW, CLOSE, VOLUME, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_range_0_to_100(self):
|
||
ft = ferro_ta.MFI(HIGH, LOW, CLOSE, VOLUME, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.MFI(HIGH, LOW, CLOSE, VOLUME, timeperiod=14)
|
||
ta = talib.MFI(HIGH, LOW, CLOSE, VOLUME, timeperiod=14)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestSTOCHF:
|
||
"""STOCHF — fast %K values match exactly.
|
||
|
||
Note: ferro_ta uses ``fastk_period - 1`` NaNs while TA-Lib uses
|
||
``fastk_period + fastd_period - 2`` NaNs (i.e., it waits for both %K
|
||
and %D to be valid before emitting anything). The overlapping valid
|
||
region is identical.
|
||
"""
|
||
|
||
def test_fastk_values_match(self):
|
||
ft_k, ft_d = ferro_ta.STOCHF(HIGH, LOW, CLOSE, fastk_period=5, fastd_period=3)
|
||
ta_k, ta_d = talib.STOCHF(
|
||
HIGH, LOW, CLOSE, fastk_period=5, fastd_period=3, fastd_matype=0
|
||
)
|
||
assert _allclose(ft_k, ta_k)
|
||
|
||
def test_output_length_match(self):
|
||
ft_k, _ = ferro_ta.STOCHF(HIGH, LOW, CLOSE, fastk_period=5, fastd_period=3)
|
||
ta_k, _ = talib.STOCHF(
|
||
HIGH, LOW, CLOSE, fastk_period=5, fastd_period=3, fastd_matype=0
|
||
)
|
||
assert len(ft_k) == len(ta_k)
|
||
|
||
def test_range_0_to_100(self):
|
||
ft_k, ft_d = ferro_ta.STOCHF(HIGH, LOW, CLOSE, fastk_period=5, fastd_period=3)
|
||
for arr in [ft_k, ft_d]:
|
||
finite = arr[~np.isnan(arr)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
|
||
class TestSTOCH:
|
||
"""STOCH — same shape; slow %K may differ by EMA initialisation."""
|
||
|
||
def test_output_length_match(self):
|
||
ft_k, ft_d = ferro_ta.STOCH(HIGH, LOW, CLOSE)
|
||
ta_k, ta_d = talib.STOCH(HIGH, LOW, CLOSE)
|
||
assert len(ft_k) == len(ta_k)
|
||
|
||
def test_range_0_to_100(self):
|
||
ft_k, ft_d = ferro_ta.STOCH(HIGH, LOW, CLOSE)
|
||
for arr in [ft_k, ft_d]:
|
||
finite = arr[~np.isnan(arr)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
|
||
class TestSTOCHRSI:
|
||
"""STOCHRSI — same length; NaN count may differ by up to 2.
|
||
|
||
The RSI seed difference propagates into StochRSI. ferro_ta emits values
|
||
sooner (fewer NaN) than TA-Lib in some configurations.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft_k, ft_d = ferro_ta.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
ta_k, ta_d = talib.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3, fastd_matype=0
|
||
)
|
||
assert len(ft_k) == len(ta_k)
|
||
|
||
def test_nan_count_within_tolerance(self):
|
||
ft_k, ft_d = ferro_ta.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
ta_k, ta_d = talib.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3, fastd_matype=0
|
||
)
|
||
assert abs(_nan_count(ft_k) - _nan_count(ta_k)) <= 2
|
||
|
||
def test_range_0_to_100(self):
|
||
ft_k, _ = ferro_ta.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
finite = ft_k[~np.isnan(ft_k)]
|
||
# Allow small numerical tolerance for float boundaries
|
||
assert all(-1e-9 <= v <= 100.0 + 1e-9 for v in finite)
|
||
|
||
|
||
class TestAPO:
|
||
"""APO — shape matches; values differ (EMA-based when matype != SMA)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.APO(CLOSE, fastperiod=12, slowperiod=26)
|
||
ta = talib.APO(CLOSE, fastperiod=12, slowperiod=26, matype=0)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.APO(CLOSE, fastperiod=12, slowperiod=26)
|
||
ta = talib.APO(CLOSE, fastperiod=12, slowperiod=26, matype=0)
|
||
assert len(ft) == len(ta)
|
||
|
||
|
||
class TestPPO:
|
||
"""PPO — ferro_ta returns (ppo, signal, histogram); TA-Lib returns only ppo.
|
||
|
||
ferro_ta extends PPO with a signal line and histogram (similar to MACD),
|
||
while TA-Lib's PPO only returns the percentage-difference line. We verify
|
||
the output length and that all three ferro_ta arrays have valid shapes.
|
||
The ppo line converges toward the TA-Lib value after the EMA seed window.
|
||
"""
|
||
|
||
def test_output_is_tuple_of_three(self):
|
||
result = ferro_ta.PPO(CLOSE, fastperiod=12, slowperiod=26)
|
||
assert isinstance(result, tuple) and len(result) == 3
|
||
|
||
def test_output_length_match(self):
|
||
ppo, signal, hist = ferro_ta.PPO(CLOSE, fastperiod=12, slowperiod=26)
|
||
ta = talib.PPO(CLOSE, fastperiod=12, slowperiod=26, matype=0)
|
||
assert len(ppo) == len(ta)
|
||
|
||
def test_all_arrays_same_length(self):
|
||
ppo, signal, hist = ferro_ta.PPO(CLOSE, fastperiod=12, slowperiod=26)
|
||
assert len(ppo) == len(signal) == len(hist) == N
|
||
|
||
def test_ppo_converges_to_talib(self):
|
||
"""PPO line should be strongly correlated with TA-Lib's PPO output.
|
||
|
||
Note: EMA seeding differences mean correlation is ~0.90 for short periods.
|
||
We verify > 0.85 to confirm same signal direction.
|
||
"""
|
||
ppo, _, _ = ferro_ta.PPO(CLOSE, fastperiod=3, slowperiod=6)
|
||
ta = talib.PPO(CLOSE, fastperiod=3, slowperiod=6, matype=0)
|
||
mask = _valid_mask(ppo, ta)
|
||
corr = np.corrcoef(ppo[mask], ta[mask])[0, 1]
|
||
assert corr > 0.85
|
||
|
||
"""CMO — same NaN count and shape; values may differ slightly.
|
||
|
||
Both libraries compute the Chande Momentum Oscillator as
|
||
(sum_up - sum_dn) / (sum_up + sum_dn) × 100, but use different rolling
|
||
window implementations (TA-Lib uses Wilder's smoothing for the gains/
|
||
losses; ferro_ta uses a plain rolling sum). Values are strongly
|
||
correlated but not numerically identical.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.CMO(CLOSE, timeperiod=14)
|
||
ta = talib.CMO(CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.CMO(CLOSE, timeperiod=14)
|
||
ta = talib.CMO(CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_range_minus100_to_100(self):
|
||
ft = ferro_ta.CMO(CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(-100.0 <= v <= 100.0 for v in finite)
|
||
|
||
def test_values_strongly_correlated(self):
|
||
ft = ferro_ta.CMO(CLOSE, timeperiod=14)
|
||
ta = talib.CMO(CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.85
|
||
|
||
|
||
class TestTRIX:
|
||
"""TRIX — shape matches; values differ (triple EMA initialisation)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.TRIX(CLOSE, timeperiod=5)
|
||
ta = talib.TRIX(CLOSE, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.TRIX(CLOSE, timeperiod=5)
|
||
ta = talib.TRIX(CLOSE, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
|
||
class TestULTOSC:
|
||
"""ULTOSC — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.ULTOSC(
|
||
HIGH, LOW, CLOSE, timeperiod1=7, timeperiod2=14, timeperiod3=28
|
||
)
|
||
ta = talib.ULTOSC(
|
||
HIGH, LOW, CLOSE, timeperiod1=7, timeperiod2=14, timeperiod3=28
|
||
)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.ULTOSC(HIGH, LOW, CLOSE)
|
||
ta = talib.ULTOSC(HIGH, LOW, CLOSE)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestADX:
|
||
"""ADX — same shape; values differ on random data (Wilder smoothing seed).
|
||
|
||
On monotonically trending data the values match TA-Lib exactly. On
|
||
random price series the Wilder's smoothing seed for ATR and DM causes
|
||
permanent divergence (values do not converge).
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_range_0_to_100(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
def test_values_strongly_correlated(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
|
||
class TestADXR:
|
||
"""ADXR — same shape (±1 NaN); values differ (Wilder smoothing seed).
|
||
|
||
ADXR = (ADX[t] + ADX[t - timeperiod]) / 2. The ADX values differ from
|
||
TA-Lib due to the Wilder smoothing seed, so ADXR differs too.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_within_one(self):
|
||
ft = ferro_ta.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert abs(_nan_count(ft) - _nan_count(ta)) <= 1
|
||
|
||
def test_values_strongly_correlated(self):
|
||
ft = ferro_ta.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.95
|
||
|
||
|
||
class TestDX:
|
||
"""DX — same NaN count and shape; values differ on random data.
|
||
|
||
DX = |+DI - -DI| / (+DI + -DI) × 100. The +DI and -DI values depend on
|
||
Wilder's smoothed ATR and DM, both of which have different seeds in
|
||
ferro_ta vs TA-Lib. Values are strongly correlated.
|
||
"""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_range_0_to_100(self):
|
||
ft = ferro_ta.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(0.0 <= v <= 100.0 for v in finite)
|
||
|
||
|
||
class TestPLUSDI:
|
||
"""PLUS_DI — same NaN count; values differ on random data (Wilder smoothing)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_non_negative(self):
|
||
ft = ferro_ta.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v >= 0.0 for v in finite)
|
||
|
||
|
||
class TestMINUSDI:
|
||
"""MINUS_DI — same NaN count; values differ on random data (Wilder smoothing)."""
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_non_negative(self):
|
||
ft = ferro_ta.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v >= 0.0 for v in finite)
|
||
|
||
|
||
class TestPLUSDM:
|
||
"""PLUS_DM — values match in the non-degenerate (OHLCV) region."""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
ta = talib.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_non_negative(self):
|
||
ft = ferro_ta.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v >= 0.0 for v in finite)
|
||
|
||
|
||
class TestMINUSDM:
|
||
"""MINUS_DM — same length; NaN count may differ by 1 (Wilder smoothing seed)."""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
ta = talib.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_within_one(self):
|
||
ft = ferro_ta.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
ta = talib.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
assert abs(_nan_count(ft) - _nan_count(ta)) <= 1
|
||
|
||
def test_non_negative(self):
|
||
ft = ferro_ta.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v >= 0.0 for v in finite)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Volume Indicators
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAD:
|
||
"""AD — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.AD(HIGH, LOW, CLOSE, VOLUME)
|
||
ta = talib.AD(HIGH, LOW, CLOSE, VOLUME)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.AD(HIGH, LOW, CLOSE, VOLUME)
|
||
ta = talib.AD(HIGH, LOW, CLOSE, VOLUME)
|
||
assert len(ft) == len(ta)
|
||
|
||
|
||
class TestADOSC:
|
||
"""ADOSC — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.ADOSC(HIGH, LOW, CLOSE, VOLUME, fastperiod=3, slowperiod=10)
|
||
ta = talib.ADOSC(HIGH, LOW, CLOSE, VOLUME, fastperiod=3, slowperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.ADOSC(HIGH, LOW, CLOSE, VOLUME)
|
||
ta = talib.ADOSC(HIGH, LOW, CLOSE, VOLUME)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestOBV:
|
||
"""OBV — values match after the first bar.
|
||
|
||
TA-Lib starts OBV accumulation at the *first* bar (OBV[0] = volume[0] if
|
||
price rose, else -volume[0]). ferro_ta initialises OBV[0] = 0 and applies
|
||
the direction rule from bar 1 onward. All increments are identical; the
|
||
two series differ only by a constant offset equal to the first OBV value.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.OBV(CLOSE, VOLUME)
|
||
ta = talib.OBV(CLOSE, VOLUME)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_increments_match(self):
|
||
"""Day-over-day OBV changes must be identical."""
|
||
ft = ferro_ta.OBV(CLOSE, VOLUME)
|
||
ta = talib.OBV(CLOSE, VOLUME)
|
||
ft_diff = np.diff(ft)
|
||
ta_diff = np.diff(ta)
|
||
assert np.allclose(ft_diff, ta_diff, atol=1e-8)
|
||
|
||
def test_no_nans(self):
|
||
ft = ferro_ta.OBV(CLOSE, VOLUME)
|
||
assert not np.any(np.isnan(ft))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Volatility Indicators
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestATR:
|
||
"""ATR — same length; values differ (different Wilder smoothing seed).
|
||
|
||
TA-Lib uses Wilder's smoothing and marks the very first ATR value (at
|
||
index ``timeperiod``) as NaN. ferro_ta emits a value there. The Wilder
|
||
recursion runs from a different seed, so values do not converge. Both
|
||
produce strongly correlated positive ATR values.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_within_one(self):
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert abs(_nan_count(ft) - _nan_count(ta)) <= 1
|
||
|
||
def test_values_positive(self):
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v > 0 for v in finite)
|
||
|
||
def test_values_strongly_correlated(self):
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.95
|
||
|
||
|
||
class TestNATR:
|
||
"""NATR — same shape tolerance as ATR; values differ (Wilder smoothing seed)."""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_within_one(self):
|
||
ft = ferro_ta.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert abs(_nan_count(ft) - _nan_count(ta)) <= 1
|
||
|
||
def test_values_positive(self):
|
||
ft = ferro_ta.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(v > 0 for v in finite)
|
||
|
||
def test_values_strongly_correlated(self):
|
||
ft = ferro_ta.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.95
|
||
|
||
|
||
class TestTRANGE:
|
||
"""TRANGE — values match.
|
||
|
||
TA-Lib emits NaN at index 0 (no previous close to compute true range).
|
||
ferro_ta emits TRANGE[0] = high[0] − low[0] (high-low only, no prior
|
||
close). From index 1 onward the values are identical.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.TRANGE(HIGH, LOW, CLOSE)
|
||
ta = talib.TRANGE(HIGH, LOW, CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_values_match_after_first(self):
|
||
ft = ferro_ta.TRANGE(HIGH, LOW, CLOSE)
|
||
ta = talib.TRANGE(HIGH, LOW, CLOSE)
|
||
assert np.allclose(ft[1:], ta[1:], atol=1e-8)
|
||
|
||
def test_values_positive(self):
|
||
ft = ferro_ta.TRANGE(HIGH, LOW, CLOSE)
|
||
assert all(v > 0 for v in ft[1:])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Statistical Functions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSTDDEV:
|
||
"""STDDEV — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.STDDEV(CLOSE, timeperiod=10)
|
||
ta = talib.STDDEV(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.STDDEV(CLOSE, timeperiod=10)
|
||
ta = talib.STDDEV(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestVAR:
|
||
"""VAR — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.VAR(CLOSE, timeperiod=10)
|
||
ta = talib.VAR(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestLINEARREG:
|
||
"""LINEARREG — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.LINEARREG(CLOSE, timeperiod=10)
|
||
ta = talib.LINEARREG(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.LINEARREG(CLOSE, timeperiod=10)
|
||
ta = talib.LINEARREG(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestLINEARREGSlope:
|
||
"""LINEARREG_SLOPE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.LINEARREG_SLOPE(CLOSE, timeperiod=10)
|
||
ta = talib.LINEARREG_SLOPE(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestLINEARREGIntercept:
|
||
"""LINEARREG_INTERCEPT — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.LINEARREG_INTERCEPT(CLOSE, timeperiod=10)
|
||
ta = talib.LINEARREG_INTERCEPT(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestLINEARREGAngle:
|
||
"""LINEARREG_ANGLE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.LINEARREG_ANGLE(CLOSE, timeperiod=10)
|
||
ta = talib.LINEARREG_ANGLE(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
|
||
class TestTSF:
|
||
"""TSF — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.TSF(CLOSE, timeperiod=10)
|
||
ta = talib.TSF(CLOSE, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.TSF(CLOSE, timeperiod=10)
|
||
ta = talib.TSF(CLOSE, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestBETA:
|
||
"""BETA — same shape; algorithm differs from TA-Lib.
|
||
|
||
ferro_ta computes a simplified rolling beta (covariance / variance of the
|
||
reference series), while TA-Lib uses the standard CAPM beta estimator.
|
||
Shape compatibility (NaN count, length) is verified; exact value match is
|
||
not expected.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.BETA(CLOSE, HIGH, timeperiod=5)
|
||
ta = talib.BETA(CLOSE, HIGH, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.BETA(CLOSE, HIGH, timeperiod=5)
|
||
ta = talib.BETA(CLOSE, HIGH, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
|
||
class TestCORREL:
|
||
"""CORREL — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.CORREL(CLOSE, HIGH, timeperiod=10)
|
||
ta = talib.CORREL(CLOSE, HIGH, timeperiod=10)
|
||
assert _allclose(ft, ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.CORREL(CLOSE, HIGH, timeperiod=10)
|
||
ta = talib.CORREL(CLOSE, HIGH, timeperiod=10)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_range_minus1_to_1(self):
|
||
ft = ferro_ta.CORREL(CLOSE, HIGH, timeperiod=10)
|
||
finite = ft[~np.isnan(ft)]
|
||
assert all(-1.0 <= v <= 1.0 for v in finite)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Price Transformations
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAVGPRICE:
|
||
"""AVGPRICE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.AVGPRICE(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.AVGPRICE(OPEN, HIGH, LOW, CLOSE)
|
||
assert np.allclose(ft, ta, atol=1e-10)
|
||
|
||
def test_output_length_match(self):
|
||
assert len(ferro_ta.AVGPRICE(OPEN, HIGH, LOW, CLOSE)) == N
|
||
|
||
|
||
class TestMEDPRICE:
|
||
"""MEDPRICE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.MEDPRICE(HIGH, LOW)
|
||
ta = talib.MEDPRICE(HIGH, LOW)
|
||
assert np.allclose(ft, ta, atol=1e-10)
|
||
|
||
|
||
class TestTYPPRICE:
|
||
"""TYPPRICE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.TYPPRICE(HIGH, LOW, CLOSE)
|
||
ta = talib.TYPPRICE(HIGH, LOW, CLOSE)
|
||
assert np.allclose(ft, ta, atol=1e-10)
|
||
|
||
|
||
class TestWCLPRICE:
|
||
"""WCLPRICE — exact match."""
|
||
|
||
def test_values_match(self):
|
||
ft = ferro_ta.WCLPRICE(HIGH, LOW, CLOSE)
|
||
ta = talib.WCLPRICE(HIGH, LOW, CLOSE)
|
||
assert np.allclose(ft, ta, atol=1e-10)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Pattern Recognition
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPatternShapeCompatibility:
|
||
"""Patterns — same output length and dtype; values may differ.
|
||
|
||
Pattern recognition algorithms depend heavily on thresholds and candle
|
||
body/shadow definitions. ferro_ta implements simplified versions of these
|
||
patterns. These tests verify that:
|
||
|
||
* Output length matches TA-Lib.
|
||
* Values are restricted to {-100, 0, 100} (same convention as TA-Lib).
|
||
"""
|
||
|
||
PATTERNS = [
|
||
"CDLDOJI",
|
||
"CDLENGULFING",
|
||
"CDLHAMMER",
|
||
"CDLSHOOTINGSTAR",
|
||
"CDLMARUBOZU",
|
||
"CDLSPINNINGTOP",
|
||
"CDLMORNINGSTAR",
|
||
"CDLEVENINGSTAR",
|
||
"CDL2CROWS",
|
||
# Additional candlestick patterns
|
||
"CDL3BLACKCROWS",
|
||
"CDL3INSIDE",
|
||
"CDL3LINESTRIKE",
|
||
"CDL3OUTSIDE",
|
||
"CDL3STARSINSOUTH",
|
||
"CDL3WHITESOLDIERS",
|
||
"CDLABANDONEDBABY",
|
||
"CDLADVANCEBLOCK",
|
||
"CDLBELTHOLD",
|
||
"CDLBREAKAWAY",
|
||
"CDLCLOSINGMARUBOZU",
|
||
"CDLCONCEALBABYSWALL",
|
||
"CDLCOUNTERATTACK",
|
||
"CDLDARKCLOUDCOVER",
|
||
"CDLDOJISTAR",
|
||
"CDLDRAGONFLYDOJI",
|
||
"CDLGAPSIDESIDEWHITE",
|
||
"CDLGRAVESTONEDOJI",
|
||
"CDLHANGINGMAN",
|
||
"CDLHARAMI",
|
||
"CDLHARAMICROSS",
|
||
"CDLHIGHWAVE",
|
||
"CDLHIKKAKE",
|
||
"CDLHIKKAKEMOD",
|
||
"CDLHOMINGPIGEON",
|
||
"CDLIDENTICAL3CROWS",
|
||
"CDLINNECK",
|
||
"CDLINVERTEDHAMMER",
|
||
"CDLKICKING",
|
||
"CDLKICKINGBYLENGTH",
|
||
"CDLLADDERBOTTOM",
|
||
"CDLLONGLEGGEDDOJI",
|
||
"CDLLONGLINE",
|
||
"CDLMATCHINGLOW",
|
||
"CDLMATHOLD",
|
||
"CDLMORNINGDOJISTAR",
|
||
"CDLEVENINGDOJISTAR",
|
||
"CDLONNECK",
|
||
"CDLPIERCING",
|
||
"CDLRICKSHAWMAN",
|
||
"CDLRISEFALL3METHODS",
|
||
"CDLSEPARATINGLINES",
|
||
"CDLSHORTLINE",
|
||
"CDLSTALLEDPATTERN",
|
||
"CDLSTICKSANDWICH",
|
||
"CDLTAKURI",
|
||
"CDLTASUKIGAP",
|
||
"CDLTHRUSTING",
|
||
"CDLTRISTAR",
|
||
"CDLUNIQUE3RIVER",
|
||
"CDLUPSIDEGAP2CROWS",
|
||
"CDLXSIDEGAP3METHODS",
|
||
]
|
||
|
||
@pytest.mark.parametrize("name", PATTERNS)
|
||
def test_output_length_match(self, name: str):
|
||
ft_fn = getattr(ferro_ta, name)
|
||
ta_fn = getattr(talib, name)
|
||
ft = ft_fn(OPEN, HIGH, LOW, CLOSE)
|
||
ta = ta_fn(OPEN, HIGH, LOW, CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
@pytest.mark.parametrize("name", PATTERNS)
|
||
def test_valid_output_values(self, name: str):
|
||
ft_fn = getattr(ferro_ta, name)
|
||
ft = ft_fn(OPEN, HIGH, LOW, CLOSE)
|
||
assert all(v in (-100, 0, 100) for v in ft), (
|
||
f"{name}: unexpected values {set(ft)}"
|
||
)
|
||
|
||
def test_cdlengulfing_values_match(self):
|
||
"""CDLENGULFING matches TA-Lib exactly on random OHLCV data."""
|
||
ft = ferro_ta.CDLENGULFING(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.CDLENGULFING(OPEN, HIGH, LOW, CLOSE)
|
||
assert np.array_equal(ft, ta)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parity suite additions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestParitySuite:
|
||
"""
|
||
Comprehensive parity validation against TA-Lib.
|
||
|
||
Covers:
|
||
* Large-dataset SMA equivalence (10,000 rows)
|
||
* Strict shape and dtype checks for MACD and BBANDS
|
||
* float32 input handling (should cast safely via _to_f64)
|
||
"""
|
||
|
||
# 10,000-row synthetic OHLCV data
|
||
N_LARGE = 10_000
|
||
_rng = np.random.default_rng(2024)
|
||
CLOSE_LARGE = 100.0 + np.cumsum(_rng.standard_normal(N_LARGE) * 0.5)
|
||
|
||
def test_sma_10k_allclose(self):
|
||
"""SMA on 10,000 rows must match TA-Lib within floating-point tolerance."""
|
||
ft = ferro_ta.SMA(self.CLOSE_LARGE, timeperiod=30)
|
||
ta = talib.SMA(self.CLOSE_LARGE, timeperiod=30)
|
||
assert np.allclose(ft, ta, equal_nan=True), "SMA mismatch on 10k-row dataset"
|
||
|
||
def test_macd_shape_and_dtype(self):
|
||
"""MACD output must have correct shape and float64 dtype."""
|
||
macd_line, signal, hist = ferro_ta.MACD(CLOSE)
|
||
assert macd_line.shape == (N,)
|
||
assert signal.shape == (N,)
|
||
assert hist.shape == (N,)
|
||
assert macd_line.dtype == np.float64
|
||
assert signal.dtype == np.float64
|
||
assert hist.dtype == np.float64
|
||
|
||
def test_bbands_shape_and_dtype(self):
|
||
"""BBANDS output must have correct shape and float64 dtype."""
|
||
upper, middle, lower = ferro_ta.BBANDS(CLOSE, timeperiod=20)
|
||
assert upper.shape == (N,)
|
||
assert middle.shape == (N,)
|
||
assert lower.shape == (N,)
|
||
assert upper.dtype == np.float64
|
||
assert middle.dtype == np.float64
|
||
assert lower.dtype == np.float64
|
||
|
||
def test_float32_input_casts_safely(self):
|
||
"""Passing float32 arrays should cast to float64 silently (no error)."""
|
||
close32 = CLOSE.astype(np.float32)
|
||
# _to_f64 should cast — result must be finite and match float64 version
|
||
result = ferro_ta.SMA(close32, timeperiod=10)
|
||
expected = ferro_ta.SMA(CLOSE, timeperiod=10)
|
||
assert result.dtype == np.float64
|
||
valid = ~np.isnan(result) & ~np.isnan(expected)
|
||
assert np.allclose(result[valid], expected[valid], atol=1e-4)
|
||
|
||
def test_macd_nan_count_vs_talib(self):
|
||
"""MACD NaN counts must agree with TA-Lib (same warmup period)."""
|
||
ft_m, ft_s, ft_h = ferro_ta.MACD(CLOSE)
|
||
ta_m, ta_s, ta_h = talib.MACD(CLOSE)
|
||
assert _nan_count(ft_m) == _nan_count(ta_m)
|
||
assert _nan_count(ft_s) == _nan_count(ta_s)
|
||
|
||
def test_bbands_values_match_talib(self):
|
||
"""BBANDS must match TA-Lib exactly (SMA-based, no EMA seeding issue)."""
|
||
ft_u, ft_m, ft_l = ferro_ta.BBANDS(CLOSE, timeperiod=20)
|
||
ta_u, ta_m, ta_l = talib.BBANDS(CLOSE, timeperiod=20)
|
||
assert _allclose(ft_u, ta_u), "BBANDS upper mismatch"
|
||
assert _allclose(ft_m, ta_m), "BBANDS middle mismatch"
|
||
assert _allclose(ft_l, ta_l), "BBANDS lower mismatch"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Numerical parity — RSI, ATR, NATR, CCI, BETA alignment
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestNumericalParity:
|
||
"""Verify RSI, ATR, NATR, CCI, BETA alignment with TA-Lib."""
|
||
|
||
def test_rsi_output_length_matches(self):
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_rsi_nan_count_matches(self):
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta), (
|
||
f"RSI NaN count: ferro_ta={_nan_count(ft)}, talib={_nan_count(ta)}"
|
||
)
|
||
|
||
def test_rsi_values_allclose(self):
|
||
"""RSI values must match TA-Lib within tolerance after seeding."""
|
||
ft = ferro_ta.RSI(CLOSE, timeperiod=14)
|
||
ta = talib.RSI(CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any(), "No valid bars to compare"
|
||
assert np.allclose(ft[mask], ta[mask], atol=1e-8), (
|
||
f"RSI max diff: {np.abs(ft[mask] - ta[mask]).max()}"
|
||
)
|
||
|
||
def test_atr_output_length_matches(self):
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_atr_nan_count_matches(self):
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta), (
|
||
f"ATR NaN count: ferro_ta={_nan_count(ft)}, talib={_nan_count(ta)}"
|
||
)
|
||
|
||
def test_atr_values_allclose(self):
|
||
"""ATR values must match TA-Lib within tolerance."""
|
||
ft = ferro_ta.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
assert np.allclose(ft[mask], ta[mask], atol=1e-8), (
|
||
f"ATR max diff: {np.abs(ft[mask] - ta[mask]).max()}"
|
||
)
|
||
|
||
def test_natr_values_allclose(self):
|
||
ft = ferro_ta.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.NATR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
assert np.allclose(ft[mask], ta[mask], atol=1e-6)
|
||
|
||
def test_cci_output_length_matches(self):
|
||
ft = ferro_ta.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_cci_values_allclose(self):
|
||
"""CCI values must match TA-Lib exactly."""
|
||
ft = ferro_ta.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.CCI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
assert np.allclose(ft[mask], ta[mask], atol=1e-6), (
|
||
f"CCI max diff: {np.abs(ft[mask] - ta[mask]).max()}"
|
||
)
|
||
|
||
def test_beta_output_length_matches(self):
|
||
ft = ferro_ta.BETA(CLOSE, HIGH, timeperiod=5)
|
||
ta = talib.BETA(CLOSE, HIGH, timeperiod=5)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_beta_nan_count_matches(self):
|
||
ft = ferro_ta.BETA(CLOSE, HIGH, timeperiod=5)
|
||
ta = talib.BETA(CLOSE, HIGH, timeperiod=5)
|
||
assert _nan_count(ft) == _nan_count(ta), (
|
||
f"BETA NaN count: ferro_ta={_nan_count(ft)}, talib={_nan_count(ta)}"
|
||
)
|
||
|
||
def test_beta_values_close_to_talib(self):
|
||
"""BETA values using returns-based regression must be close to TA-Lib."""
|
||
ft = ferro_ta.BETA(CLOSE, HIGH, timeperiod=5)
|
||
ta = talib.BETA(CLOSE, HIGH, timeperiod=5)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
# TA-Lib BETA uses returns-based regression — allow small tolerance
|
||
assert np.allclose(ft[mask], ta[mask], atol=1e-8), (
|
||
f"BETA max diff: {np.abs(ft[mask] - ta[mask]).max()}"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Math operators vs TA-Lib
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMathOperatorsVsTalib:
|
||
"""Verify that math operator shims match TA-Lib exactly."""
|
||
|
||
def test_add_matches_talib(self):
|
||
ft = ferro_ta.ADD(CLOSE, HIGH)
|
||
ta = talib.ADD(CLOSE, HIGH)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_sub_matches_talib(self):
|
||
ft = ferro_ta.SUB(HIGH, LOW)
|
||
ta = talib.SUB(HIGH, LOW)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_mult_matches_talib(self):
|
||
ft = ferro_ta.MULT(CLOSE, VOLUME)
|
||
ta = talib.MULT(CLOSE, VOLUME)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_div_matches_talib(self):
|
||
ft = ferro_ta.DIV(CLOSE, HIGH)
|
||
ta = talib.DIV(CLOSE, HIGH)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_sum_matches_talib(self):
|
||
ft = ferro_ta.SUM(CLOSE, timeperiod=10)
|
||
ta = talib.SUM(CLOSE, timeperiod=10)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_max_matches_talib(self):
|
||
ft = ferro_ta.MAX(CLOSE, timeperiod=10)
|
||
ta = talib.MAX(CLOSE, timeperiod=10)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_min_matches_talib(self):
|
||
ft = ferro_ta.MIN(CLOSE, timeperiod=10)
|
||
ta = talib.MIN(CLOSE, timeperiod=10)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_sin_matches_talib(self):
|
||
ft = ferro_ta.SIN(CLOSE)
|
||
ta = talib.SIN(CLOSE)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_cos_matches_talib(self):
|
||
ft = ferro_ta.COS(CLOSE)
|
||
ta = talib.COS(CLOSE)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_sqrt_matches_talib(self):
|
||
ft = ferro_ta.SQRT(CLOSE)
|
||
ta = talib.SQRT(CLOSE)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_exp_matches_talib(self):
|
||
ft = ferro_ta.EXP(LINEAR)
|
||
ta = talib.EXP(LINEAR)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_ln_matches_talib(self):
|
||
ft = ferro_ta.LN(CLOSE)
|
||
ta = talib.LN(CLOSE)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
def test_log10_matches_talib(self):
|
||
ft = ferro_ta.LOG10(CLOSE)
|
||
ta = talib.LOG10(CLOSE)
|
||
assert np.allclose(ft, ta, equal_nan=True)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# STOCH, STOCHRSI, ADX, DI, DM parity
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDirectionalMovementVsTalib:
|
||
"""Verify ADX, DX, +DI, -DI, +DM, -DM are strongly correlated with TA-Lib.
|
||
|
||
Wilder smoothing seed differs between ferro_ta and TA-Lib, so values are
|
||
not numerically identical but must be strongly correlated.
|
||
"""
|
||
|
||
def test_plus_di_output_length(self):
|
||
ft = ferro_ta.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_plus_di_nan_count(self):
|
||
ft = ferro_ta.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_plus_di_values_strongly_correlated(self):
|
||
ft = ferro_ta.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.PLUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
def test_minus_di_values_strongly_correlated(self):
|
||
ft = ferro_ta.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.MINUS_DI(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
def test_plus_dm_output_length(self):
|
||
ft = ferro_ta.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
ta = talib.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_plus_dm_values_strongly_correlated(self):
|
||
ft = ferro_ta.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
ta = talib.PLUS_DM(HIGH, LOW, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
def test_minus_dm_values_strongly_correlated(self):
|
||
ft = ferro_ta.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
ta = talib.MINUS_DM(HIGH, LOW, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
def test_dx_values_strongly_correlated(self):
|
||
ft = ferro_ta.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.DX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
def test_adx_output_length(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_adx_nan_count(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_adx_values_strongly_correlated(self):
|
||
ft = ferro_ta.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADX(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.99
|
||
|
||
def test_adxr_values_strongly_correlated(self):
|
||
ft = ferro_ta.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
ta = talib.ADXR(HIGH, LOW, CLOSE, timeperiod=14)
|
||
mask = _valid_mask(ft, ta)
|
||
assert mask.any()
|
||
corr = np.corrcoef(ft[mask], ta[mask])[0, 1]
|
||
assert corr > 0.95
|
||
|
||
|
||
class TestSTOCHVsTalib:
|
||
"""Verify STOCH and STOCHRSI match TA-Lib."""
|
||
|
||
def test_stoch_slowk_output_length(self):
|
||
ft_k, _ = ferro_ta.STOCH(HIGH, LOW, CLOSE)
|
||
ta_k, _ = talib.STOCH(HIGH, LOW, CLOSE)
|
||
assert len(ft_k) == len(ta_k)
|
||
|
||
def test_stoch_nan_count_matches(self):
|
||
ft_k, ft_d = ferro_ta.STOCH(HIGH, LOW, CLOSE)
|
||
ta_k, ta_d = talib.STOCH(HIGH, LOW, CLOSE)
|
||
assert _nan_count(ft_k) == _nan_count(ta_k)
|
||
assert _nan_count(ft_d) == _nan_count(ta_d)
|
||
|
||
def test_stoch_values_allclose(self):
|
||
ft_k, ft_d = ferro_ta.STOCH(HIGH, LOW, CLOSE)
|
||
ta_k, ta_d = talib.STOCH(HIGH, LOW, CLOSE)
|
||
mask_k = _valid_mask(ft_k, ta_k)
|
||
mask_d = _valid_mask(ft_d, ta_d)
|
||
assert mask_k.any()
|
||
assert np.allclose(ft_k[mask_k], ta_k[mask_k], atol=1e-8)
|
||
assert np.allclose(ft_d[mask_d], ta_d[mask_d], atol=1e-8)
|
||
|
||
def test_stochrsi_output_length(self):
|
||
ft_k, _ = ferro_ta.STOCHRSI(CLOSE)
|
||
ta_k, _ = talib.STOCHRSI(CLOSE)
|
||
assert len(ft_k) == len(ta_k)
|
||
|
||
def test_stochrsi_nan_count_matches(self):
|
||
ft_k, ft_d = ferro_ta.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
ta_k, ta_d = talib.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
# RSI seed difference can yield ±2 NaN count (see TestSTOCHRSI)
|
||
assert abs(_nan_count(ft_k) - _nan_count(ta_k)) <= 2
|
||
|
||
def test_stochrsi_values_close(self):
|
||
ft_k, ft_d = ferro_ta.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
ta_k, ta_d = talib.STOCHRSI(
|
||
CLOSE, timeperiod=14, fastk_period=5, fastd_period=3
|
||
)
|
||
mask_k = _valid_mask(ft_k, ta_k)
|
||
assert mask_k.any()
|
||
assert np.allclose(ft_k[mask_k], ta_k[mask_k], atol=1e-8)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# MAMA, SAR/SAREXT, and HT_* cycle indicator tests
|
||
#
|
||
# These indicators are documented as ⚠️ Corr or ⚠️ Shape in the README because
|
||
# TA-Lib C uses slightly different floating-point accumulation and clamping
|
||
# order. Tests enforce shape parity and minimum correlation rather than
|
||
# exact allclose.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestHTTrendline:
|
||
"""HT_TRENDLINE — 63-bar lookback; values correlated with TA-Lib.
|
||
|
||
Known difference: Ehlers HT filter — same algorithm and 63-bar lookback;
|
||
values are correlated (r > 0.90) but not numerically identical due to
|
||
different clamp order in TA-Lib C source.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.HT_TRENDLINE(CLOSE)
|
||
ta = talib.HT_TRENDLINE(CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.HT_TRENDLINE(CLOSE)
|
||
ta = talib.HT_TRENDLINE(CLOSE)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_correlated_with_talib(self):
|
||
"""HT_TRENDLINE should be highly correlated with TA-Lib output."""
|
||
ft = ferro_ta.HT_TRENDLINE(CLOSE)
|
||
ta = talib.HT_TRENDLINE(CLOSE)
|
||
mask = _valid_mask(ft, ta)
|
||
if mask.sum() >= 5:
|
||
corr = float(np.corrcoef(ft[mask], ta[mask])[0, 1])
|
||
assert corr > 0.90, f"HT_TRENDLINE correlation {corr:.3f} < 0.90"
|
||
|
||
|
||
class TestHTDCPeriod:
|
||
"""HT_DCPERIOD — 63-bar lookback; shape parity enforced.
|
||
|
||
Known difference: Dominant cycle period values correlated with TA-Lib
|
||
but not exact (same Ehlers algorithm, different floating-point accumulation).
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.HT_DCPERIOD(CLOSE)
|
||
ta = talib.HT_DCPERIOD(CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_within_tolerance(self):
|
||
ft = ferro_ta.HT_DCPERIOD(CLOSE)
|
||
ta = talib.HT_DCPERIOD(CLOSE)
|
||
# ferro_ta uses 63-bar lookback; TA-Lib may use different warmup
|
||
assert abs(_nan_count(ft) - _nan_count(ta)) <= 35
|
||
|
||
def test_period_in_reasonable_range(self):
|
||
"""Period should typically be in [6, 50] for realistic price data."""
|
||
ft = ferro_ta.HT_DCPERIOD(CLOSE)
|
||
valid = ft[~np.isnan(ft)]
|
||
assert valid.min() > 0
|
||
assert valid.max() <= 100.0 # allow some slack
|
||
|
||
|
||
class TestHTDCPhase:
|
||
"""HT_DCPHASE — 63-bar lookback; shape parity enforced."""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.HT_DCPHASE(CLOSE)
|
||
ta = talib.HT_DCPHASE(CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.HT_DCPHASE(CLOSE)
|
||
ta = talib.HT_DCPHASE(CLOSE)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_phase_sign_agreement(self):
|
||
"""DC phase sign should agree with TA-Lib for some valid bars (Ehlers algo diff)."""
|
||
ft = ferro_ta.HT_DCPHASE(CLOSE)
|
||
ta = talib.HT_DCPHASE(CLOSE)
|
||
mask = _valid_mask(ft, ta)
|
||
if mask.sum() >= 5:
|
||
sign_agree = np.mean(np.sign(ft[mask]) == np.sign(ta[mask]))
|
||
# HT indicators use different warmup/accumulation vs TA-Lib
|
||
assert sign_agree >= 0.40, (
|
||
f"HT_DCPHASE sign agreement {sign_agree:.2f} < 0.40"
|
||
)
|
||
|
||
|
||
class TestHTPhasor:
|
||
"""HT_PHASOR — 63-bar lookback; shape parity enforced.
|
||
|
||
Returns (inphase, quadrature). Both components are correlated with TA-Lib.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft_i, ft_q = ferro_ta.HT_PHASOR(CLOSE)
|
||
ta_i, ta_q = talib.HT_PHASOR(CLOSE)
|
||
assert len(ft_i) == len(ta_i)
|
||
assert len(ft_q) == len(ta_q)
|
||
|
||
def test_nan_count_within_tolerance(self):
|
||
ft_i, ft_q = ferro_ta.HT_PHASOR(CLOSE)
|
||
ta_i, ta_q = talib.HT_PHASOR(CLOSE)
|
||
# ferro_ta uses 63-bar lookback; TA-Lib may use different warmup
|
||
assert abs(_nan_count(ft_i) - _nan_count(ta_i)) <= 35
|
||
assert abs(_nan_count(ft_q) - _nan_count(ta_q)) <= 35
|
||
|
||
def test_inphase_sign_agreement(self):
|
||
"""Inphase component sign should agree with TA-Lib for most valid bars."""
|
||
ft_i, _ = ferro_ta.HT_PHASOR(CLOSE)
|
||
ta_i, _ = talib.HT_PHASOR(CLOSE)
|
||
mask = _valid_mask(ft_i, ta_i)
|
||
if mask.sum() >= 5:
|
||
sign_agree = np.mean(np.sign(ft_i[mask]) == np.sign(ta_i[mask]))
|
||
assert sign_agree >= SIGN_AGREEMENT_THRESHOLD
|
||
|
||
|
||
class TestHTSine:
|
||
"""HT_SINE — 63-bar lookback; shape parity enforced.
|
||
|
||
Returns (sine, leadsine). Values in [-1, 1].
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft_s, ft_l = ferro_ta.HT_SINE(CLOSE)
|
||
ta_s, ta_l = talib.HT_SINE(CLOSE)
|
||
assert len(ft_s) == len(ta_s)
|
||
assert len(ft_l) == len(ta_l)
|
||
|
||
def test_nan_count_match(self):
|
||
ft_s, ft_l = ferro_ta.HT_SINE(CLOSE)
|
||
ta_s, ta_l = talib.HT_SINE(CLOSE)
|
||
assert _nan_count(ft_s) == _nan_count(ta_s)
|
||
assert _nan_count(ft_l) == _nan_count(ta_l)
|
||
|
||
def test_sine_range(self):
|
||
"""Sine component should be in [-1.1, 1.1] (allow small numerical overshoot)."""
|
||
ft_s, _ = ferro_ta.HT_SINE(CLOSE)
|
||
valid = ft_s[~np.isnan(ft_s)]
|
||
assert valid.min() >= -1.1
|
||
assert valid.max() <= 1.1
|
||
|
||
|
||
class TestHTTrendMode:
|
||
"""HT_TRENDMODE — 63-bar lookback; values are 0 or 1.
|
||
|
||
Known difference: Boolean output derived from HT_DCPERIOD — may differ
|
||
from TA-Lib in first ~10 valid bars due to the same floating-point diff.
|
||
"""
|
||
|
||
def test_output_length_match(self):
|
||
ft = ferro_ta.HT_TRENDMODE(CLOSE)
|
||
ta = talib.HT_TRENDMODE(CLOSE)
|
||
assert len(ft) == len(ta)
|
||
|
||
def test_nan_count_match(self):
|
||
ft = ferro_ta.HT_TRENDMODE(CLOSE)
|
||
ta = talib.HT_TRENDMODE(CLOSE)
|
||
assert _nan_count(ft) == _nan_count(ta)
|
||
|
||
def test_binary_output(self):
|
||
"""TRENDMODE values must be 0 or 1 (or NaN for warmup)."""
|
||
ft = ferro_ta.HT_TRENDMODE(CLOSE)
|
||
valid = ft[~np.isnan(ft)]
|
||
assert set(valid.astype(int)).issubset({0, 1})
|
||
|
||
def test_sign_agreement_with_talib(self):
|
||
"""Trend mode should agree with TA-Lib for majority of valid bars.
|
||
|
||
Note: HT_TRENDMODE is highly sensitive to Hilbert Transform phase
|
||
accumulator initialization; the two implementations use different
|
||
precision for the adaptive period, so agreement is ~54%. We verify
|
||
> 50% to confirm the indicator is better-than-random.
|
||
"""
|
||
ft = ferro_ta.HT_TRENDMODE(CLOSE)
|
||
ta = talib.HT_TRENDMODE(CLOSE)
|
||
mask = _valid_mask(ft, ta)
|
||
if mask.sum() >= 5:
|
||
agree = np.mean(ft[mask] == ta[mask])
|
||
assert agree >= 0.50, f"HT_TRENDMODE agreement {agree:.2f} < 0.50"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Candlestick Pattern Agreement Tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# List of all candlestick patterns to test
|
||
ALL_CDL_PATTERNS = [
|
||
"CDL2CROWS",
|
||
"CDL3BLACKCROWS",
|
||
"CDL3INSIDE",
|
||
"CDL3LINESTRIKE",
|
||
"CDL3OUTSIDE",
|
||
"CDL3STARSINSOUTH",
|
||
"CDL3WHITESOLDIERS",
|
||
"CDLABANDONEDBABY",
|
||
"CDLADVANCEBLOCK",
|
||
"CDLBELTHOLD",
|
||
"CDLBREAKAWAY",
|
||
"CDLCLOSINGMARUBOZU",
|
||
"CDLCONCEALBABYSWALL",
|
||
"CDLCOUNTERATTACK",
|
||
"CDLDARKCLOUDCOVER",
|
||
"CDLDOJI",
|
||
"CDLDOJISTAR",
|
||
"CDLDRAGONFLYDOJI",
|
||
"CDLENGULFING",
|
||
"CDLEVENINGDOJISTAR",
|
||
"CDLEVENINGSTAR",
|
||
"CDLGAPSIDESIDEWHITE",
|
||
"CDLGRAVESTONEDOJI",
|
||
"CDLHAMMER",
|
||
"CDLHANGINGMAN",
|
||
"CDLHARAMI",
|
||
"CDLHARAMICROSS",
|
||
"CDLHIGHWAVE",
|
||
"CDLHIKKAKE",
|
||
"CDLHIKKAKEMOD",
|
||
"CDLHOMINGPIGEON",
|
||
"CDLIDENTICAL3CROWS",
|
||
"CDLINNECK",
|
||
"CDLINVERTEDHAMMER",
|
||
"CDLKICKING",
|
||
"CDLKICKINGBYLENGTH",
|
||
"CDLLADDERBOTTOM",
|
||
"CDLLONGLEGGEDDOJI",
|
||
"CDLLONGLINE",
|
||
"CDLMARUBOZU",
|
||
"CDLMATCHINGLOW",
|
||
"CDLMATHOLD",
|
||
"CDLMORNINGDOJISTAR",
|
||
"CDLMORNINGSTAR",
|
||
"CDLONNECK",
|
||
"CDLPIERCING",
|
||
"CDLRICKSHAWMAN",
|
||
"CDLRISEFALL3METHODS",
|
||
"CDLSEPARATINGLINES",
|
||
"CDLSHOOTINGSTAR",
|
||
"CDLSHORTLINE",
|
||
"CDLSPINNINGTOP",
|
||
"CDLSTALLEDPATTERN",
|
||
"CDLSTICKSANDWICH",
|
||
"CDLTAKURI",
|
||
"CDLTASUKIGAP",
|
||
"CDLTHRUSTING",
|
||
"CDLTRISTAR",
|
||
"CDLUNIQUE3RIVER",
|
||
"CDLUPSIDEGAP2CROWS",
|
||
"CDLXSIDEGAP3METHODS",
|
||
]
|
||
|
||
|
||
class TestCandlestickPatternAgreement:
|
||
"""Pattern recognition: agreement rate tests.
|
||
|
||
Candlestick patterns may have slightly different threshold parameters
|
||
between implementations. We validate >80% agreement rate for pattern
|
||
detection (non-zero output).
|
||
"""
|
||
|
||
@pytest.mark.parametrize("pattern_name", ALL_CDL_PATTERNS)
|
||
def test_pattern_agreement_rate(self, pattern_name):
|
||
"""Test that pattern agreement rate is > 80%."""
|
||
# Get pattern functions
|
||
ft_func = getattr(ferro_ta, pattern_name, None)
|
||
ta_func = getattr(talib, pattern_name, None)
|
||
|
||
if ft_func is None:
|
||
pytest.skip(f"ferro_ta.{pattern_name} not implemented")
|
||
if ta_func is None:
|
||
pytest.skip(f"talib.{pattern_name} not available")
|
||
|
||
# Compute patterns
|
||
ft = ft_func(OPEN, HIGH, LOW, CLOSE)
|
||
ta = ta_func(OPEN, HIGH, LOW, CLOSE)
|
||
|
||
# Check output length match
|
||
assert len(ft) == len(ta), f"{pattern_name}: length mismatch"
|
||
|
||
# Compute agreement rate (exact match of output values)
|
||
# Patterns typically return 0, ±100, or ±200
|
||
agreement = np.mean(ft == ta)
|
||
|
||
# Use per-pattern threshold (some patterns have known definition differences)
|
||
threshold = CDL_AGREEMENT_THRESHOLDS.get(pattern_name, 0.80)
|
||
assert agreement > threshold, (
|
||
f"{pattern_name}: agreement rate {agreement:.2%} < {threshold:.0%}"
|
||
)
|
||
|
||
def test_pattern_sample_doji(self):
|
||
"""Spot check: CDLDOJI should have high agreement (known: shadow ratio precision differs)."""
|
||
ft = ferro_ta.CDLDOJI(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.CDLDOJI(OPEN, HIGH, LOW, CLOSE)
|
||
|
||
agreement = np.mean(ft == ta)
|
||
# ferro_ta uses slightly different shadow/body ratio threshold; 86% observed
|
||
assert agreement > 0.85
|
||
|
||
def test_pattern_sample_engulfing(self):
|
||
"""Spot check: CDLENGULFING should have high agreement."""
|
||
ft = ferro_ta.CDLENGULFING(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.CDLENGULFING(OPEN, HIGH, LOW, CLOSE)
|
||
|
||
agreement = np.mean(ft == ta)
|
||
assert agreement > 0.80
|
||
|
||
def test_pattern_sample_hammer(self):
|
||
"""Spot check: CDLHAMMER should have high agreement."""
|
||
ft = ferro_ta.CDLHAMMER(OPEN, HIGH, LOW, CLOSE)
|
||
ta = talib.CDLHAMMER(OPEN, HIGH, LOW, CLOSE)
|
||
|
||
agreement = np.mean(ft == ta)
|
||
assert agreement > 0.80
|