Pratik Bhadane
436954138f
Update version numbers across Rust, Python, and documentation files to 1.1.0. Enhance the .gitignore to include macOS dSYM files and plans directory. Introduce new dependencies in the Rust core library and update the README to reflect recent performance benchmarks and backtesting engine capabilities. Add new artifacts to the benchmarks manifest and improve documentation for the backtesting engine API.
1061 lines
34 KiB
Python
1061 lines
34 KiB
Python
"""Tests for exceptions, backtest, registry, release playbook, GPU backend, WASM."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
import ferro_ta
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exception model & validation
|
|
# ---------------------------------------------------------------------------
|
|
from ferro_ta.core.exceptions import (
|
|
FerroTAError,
|
|
FerroTAInputError,
|
|
FerroTAValueError,
|
|
check_equal_length,
|
|
check_finite,
|
|
check_timeperiod,
|
|
)
|
|
|
|
|
|
class TestExceptionHierarchy:
|
|
"""FerroTAError hierarchy and isinstance relationships."""
|
|
|
|
def test_ferro_ta_error_is_exception(self):
|
|
assert issubclass(FerroTAError, Exception)
|
|
|
|
def test_value_error_is_base_and_value_error(self):
|
|
assert issubclass(FerroTAValueError, FerroTAError)
|
|
assert issubclass(FerroTAValueError, ValueError)
|
|
|
|
def test_input_error_is_base_and_value_error(self):
|
|
assert issubclass(FerroTAInputError, FerroTAError)
|
|
assert issubclass(FerroTAInputError, ValueError)
|
|
|
|
def test_exported_from_ferro_ta(self):
|
|
assert ferro_ta.FerroTAError is FerroTAError
|
|
assert ferro_ta.FerroTAValueError is FerroTAValueError
|
|
assert ferro_ta.FerroTAInputError is FerroTAInputError
|
|
|
|
|
|
class TestCheckTimeperiod:
|
|
"""check_timeperiod raises FerroTAValueError with clear message."""
|
|
|
|
def test_valid_timeperiod_does_not_raise(self):
|
|
check_timeperiod(1)
|
|
check_timeperiod(14)
|
|
check_timeperiod(100)
|
|
|
|
def test_zero_raises_ferro_ta_value_error(self):
|
|
with pytest.raises(FerroTAValueError, match="timeperiod must be >= 1, got 0"):
|
|
check_timeperiod(0)
|
|
|
|
def test_negative_raises_ferro_ta_value_error(self):
|
|
with pytest.raises(FerroTAValueError) as exc_info:
|
|
check_timeperiod(-5, name="timeperiod")
|
|
assert "timeperiod" in str(exc_info.value)
|
|
assert "-5" in str(exc_info.value)
|
|
|
|
def test_custom_name_in_message(self):
|
|
with pytest.raises(FerroTAValueError, match="fastperiod"):
|
|
check_timeperiod(0, name="fastperiod")
|
|
|
|
def test_custom_minimum(self):
|
|
with pytest.raises(FerroTAValueError, match=">= 2"):
|
|
check_timeperiod(1, minimum=2)
|
|
|
|
|
|
class TestCheckEqualLength:
|
|
"""check_equal_length raises FerroTAInputError for mismatched arrays."""
|
|
|
|
def test_equal_lengths_pass(self):
|
|
a = np.array([1.0, 2.0, 3.0])
|
|
b = np.array([4.0, 5.0, 6.0])
|
|
check_equal_length(open=a, close=b) # no exception
|
|
|
|
def test_mismatched_lengths_raise(self):
|
|
a = np.array([1.0, 2.0, 3.0])
|
|
b = np.array([4.0, 5.0])
|
|
with pytest.raises(FerroTAInputError) as exc_info:
|
|
check_equal_length(open=a, close=b)
|
|
# message must mention the lengths
|
|
msg = str(exc_info.value)
|
|
assert "3" in msg
|
|
assert "2" in msg
|
|
|
|
def test_three_arrays_all_different(self):
|
|
with pytest.raises(FerroTAInputError):
|
|
check_equal_length(
|
|
open=np.array([1.0]),
|
|
high=np.array([1.0, 2.0]),
|
|
close=np.array([1.0, 2.0, 3.0]),
|
|
)
|
|
|
|
|
|
class TestCheckFinite:
|
|
"""check_finite raises FerroTAInputError for NaN/Inf."""
|
|
|
|
def test_all_finite_passes(self):
|
|
check_finite(np.array([1.0, 2.0, 3.0]))
|
|
|
|
def test_nan_raises(self):
|
|
with pytest.raises(FerroTAInputError, match="NaN or Inf"):
|
|
check_finite(np.array([1.0, float("nan"), 3.0]))
|
|
|
|
def test_inf_raises(self):
|
|
with pytest.raises(FerroTAInputError, match="NaN or Inf"):
|
|
check_finite(np.array([1.0, float("inf"), 3.0]))
|
|
|
|
def test_name_in_message(self):
|
|
with pytest.raises(FerroTAInputError, match="myarray"):
|
|
check_finite(np.array([float("nan")]), name="myarray")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backtesting utilities
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ferro_ta.analysis.backtest import (
|
|
BacktestResult,
|
|
backtest,
|
|
macd_crossover_strategy,
|
|
rsi_strategy,
|
|
sma_crossover_strategy,
|
|
)
|
|
|
|
|
|
def _make_close(n: int = 50, seed: int = 42) -> np.ndarray:
|
|
rng = np.random.default_rng(seed)
|
|
returns = rng.normal(0.001, 0.01, n)
|
|
return np.cumprod(1 + returns) * 100.0
|
|
|
|
|
|
class TestRsiStrategy:
|
|
"""rsi_strategy returns correct signal arrays."""
|
|
|
|
def test_output_shape(self):
|
|
close = _make_close(50)
|
|
signals = rsi_strategy(close, timeperiod=5)
|
|
assert signals.shape == close.shape
|
|
|
|
def test_only_valid_signal_values(self):
|
|
close = _make_close(50)
|
|
signals = rsi_strategy(close, timeperiod=5)
|
|
finite = signals[np.isfinite(signals)]
|
|
assert set(finite).issubset({-1.0, 0.0, 1.0})
|
|
|
|
def test_nan_during_warmup(self):
|
|
close = _make_close(20)
|
|
signals = rsi_strategy(close, timeperiod=5)
|
|
# First 5 values should be NaN (RSI warm-up)
|
|
assert np.all(np.isnan(signals[:5]))
|
|
|
|
def test_invalid_timeperiod(self):
|
|
with pytest.raises(FerroTAValueError):
|
|
rsi_strategy(_make_close(10), timeperiod=0)
|
|
|
|
|
|
class TestSmaCrossoverStrategy:
|
|
"""sma_crossover_strategy returns signals when fast < slow."""
|
|
|
|
def test_output_shape(self):
|
|
close = _make_close(60)
|
|
signals = sma_crossover_strategy(close, fast=5, slow=20)
|
|
assert signals.shape == close.shape
|
|
|
|
def test_only_valid_signal_values(self):
|
|
close = _make_close(60)
|
|
signals = sma_crossover_strategy(close, fast=5, slow=20)
|
|
finite = signals[np.isfinite(signals)]
|
|
assert set(finite).issubset({-1.0, 1.0})
|
|
|
|
def test_fast_must_be_less_than_slow(self):
|
|
with pytest.raises(FerroTAValueError):
|
|
sma_crossover_strategy(_make_close(60), fast=20, slow=10)
|
|
|
|
|
|
class TestMacdCrossoverStrategy:
|
|
"""macd_crossover_strategy returns signals from MACD line vs signal line."""
|
|
|
|
def test_output_shape(self):
|
|
close = _make_close(100)
|
|
signals = macd_crossover_strategy(
|
|
close, fastperiod=12, slowperiod=26, signalperiod=9
|
|
)
|
|
assert signals.shape == close.shape
|
|
|
|
def test_only_valid_signal_values(self):
|
|
close = _make_close(100)
|
|
signals = macd_crossover_strategy(
|
|
close, fastperiod=12, slowperiod=26, signalperiod=9
|
|
)
|
|
finite = signals[np.isfinite(signals)]
|
|
assert set(finite).issubset({-1.0, 1.0})
|
|
|
|
def test_fastperiod_must_be_less_than_slowperiod(self):
|
|
with pytest.raises(FerroTAValueError):
|
|
macd_crossover_strategy(_make_close(60), fastperiod=26, slowperiod=12)
|
|
|
|
|
|
class TestBacktest:
|
|
"""backtest() produces correct BacktestResult."""
|
|
|
|
def test_rsi_strategy_runs(self):
|
|
close = _make_close(100)
|
|
result = backtest(close, strategy="rsi_30_70", timeperiod=5)
|
|
assert isinstance(result, BacktestResult)
|
|
|
|
def test_output_lengths_match_input(self):
|
|
close = _make_close(80)
|
|
result = backtest(close, strategy="rsi_30_70", timeperiod=5)
|
|
n = len(close)
|
|
assert len(result.signals) == n
|
|
assert len(result.positions) == n
|
|
assert len(result.equity) == n
|
|
|
|
def test_equity_starts_near_one(self):
|
|
close = _make_close(50)
|
|
result = backtest(close, strategy="rsi_30_70", timeperiod=5)
|
|
assert abs(result.equity[0] - 1.0) < 0.01
|
|
|
|
def test_sma_crossover_strategy_runs(self):
|
|
close = _make_close(80)
|
|
result = backtest(close, strategy="sma_crossover", fast=5, slow=20)
|
|
assert isinstance(result, BacktestResult)
|
|
assert result.n_trades >= 0
|
|
|
|
def test_custom_callable_strategy(self):
|
|
def my_strategy(close, **_):
|
|
signals = np.zeros(len(close))
|
|
signals[len(close) // 2 :] = 1.0
|
|
return signals
|
|
|
|
close = _make_close(40)
|
|
result = backtest(close, strategy=my_strategy)
|
|
assert isinstance(result, BacktestResult)
|
|
assert len(result.signals) == len(close)
|
|
|
|
def test_unknown_strategy_raises(self):
|
|
with pytest.raises(FerroTAValueError, match="Unknown strategy"):
|
|
backtest(_make_close(30), strategy="nonexistent")
|
|
|
|
def test_too_short_input_raises(self):
|
|
with pytest.raises(FerroTAInputError):
|
|
backtest(np.array([1.0]))
|
|
|
|
def test_non_1d_input_raises(self):
|
|
with pytest.raises(FerroTAInputError):
|
|
backtest(np.array([[1.0, 2.0], [3.0, 4.0]]))
|
|
|
|
def test_n_trades_is_integer(self):
|
|
close = _make_close(60)
|
|
result = backtest(close, strategy="sma_crossover", fast=5, slow=15)
|
|
assert isinstance(result.n_trades, int)
|
|
assert result.n_trades >= 0
|
|
|
|
def test_macd_crossover_strategy_runs(self):
|
|
close = _make_close(100)
|
|
result = backtest(
|
|
close,
|
|
strategy="macd_crossover",
|
|
fastperiod=12,
|
|
slowperiod=26,
|
|
signalperiod=9,
|
|
)
|
|
assert isinstance(result, BacktestResult)
|
|
assert len(result.equity) == len(close)
|
|
|
|
def test_commission_reduces_equity(self):
|
|
close = _make_close(80)
|
|
result_no_comm = backtest(close, strategy="sma_crossover", fast=5, slow=20)
|
|
result_with_comm = backtest(
|
|
close,
|
|
strategy="sma_crossover",
|
|
fast=5,
|
|
slow=20,
|
|
commission_per_trade=0.01,
|
|
)
|
|
assert result_with_comm.final_equity <= result_no_comm.final_equity
|
|
assert result_with_comm.final_equity < result_no_comm.final_equity or (
|
|
result_no_comm.n_trades == 0
|
|
)
|
|
|
|
def test_slippage_reduces_equity(self):
|
|
close = _make_close(80)
|
|
result_no_slip = backtest(close, strategy="sma_crossover", fast=5, slow=20)
|
|
result_with_slip = backtest(
|
|
close,
|
|
strategy="sma_crossover",
|
|
fast=5,
|
|
slow=20,
|
|
slippage_bps=10.0,
|
|
)
|
|
assert result_with_slip.final_equity <= result_no_slip.final_equity
|
|
assert result_with_slip.final_equity < result_no_slip.final_equity or (
|
|
result_no_slip.n_trades == 0
|
|
)
|
|
|
|
def test_commission_matches_reference_loop(self):
|
|
from ferro_ta._ferro_ta import CommissionModel
|
|
|
|
from ferro_ta.analysis.backtest import BacktestEngine
|
|
|
|
close = np.array([100.0, 102.0, 101.0, 104.0, 103.0, 105.0], dtype=np.float64)
|
|
raw_signals = np.array([0.0, 1.0, 1.0, -1.0, -1.0, 0.0], dtype=np.float64)
|
|
|
|
def strategy(_, **__):
|
|
return raw_signals
|
|
|
|
initial_capital = 100_000.0
|
|
cm = CommissionModel.proportional(0.001) # 0.1% proportional commission
|
|
|
|
result = (
|
|
BacktestEngine()
|
|
.with_commission_model(cm)
|
|
.with_initial_capital(initial_capital)
|
|
.run(close, strategy=strategy)
|
|
)
|
|
|
|
expected_positions = np.array(
|
|
[0.0, 0.0, 1.0, 1.0, -1.0, -1.0], dtype=np.float64
|
|
)
|
|
np.testing.assert_allclose(result.positions, expected_positions)
|
|
# With commission, final equity should be less than without
|
|
result_no_comm = (
|
|
BacktestEngine()
|
|
.with_initial_capital(initial_capital)
|
|
.run(close, strategy=strategy)
|
|
)
|
|
assert result.final_equity <= result_no_comm.final_equity
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plugin / Registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ferro_ta.core.registry import (
|
|
FerroTARegistryError,
|
|
get,
|
|
list_indicators,
|
|
register,
|
|
run,
|
|
unregister,
|
|
)
|
|
|
|
|
|
class TestRegistry:
|
|
"""Registry: register, get, run, unregister, list_indicators."""
|
|
|
|
def test_builtins_registered(self):
|
|
names = list_indicators()
|
|
assert "SMA" in names
|
|
assert "RSI" in names
|
|
assert "EMA" in names
|
|
assert "ATR" in names
|
|
|
|
def test_run_builtin_sma(self):
|
|
close = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
|
|
result = run("SMA", close, timeperiod=3)
|
|
# SMA(3) of [1,2,3,4,5]: valid at indices 2,3,4
|
|
assert result.shape == (5,)
|
|
assert np.isnan(result[0])
|
|
assert abs(float(result[2]) - 2.0) < 1e-8
|
|
|
|
def test_run_builtin_rsi(self):
|
|
close = np.array(
|
|
[
|
|
44.34,
|
|
44.09,
|
|
44.15,
|
|
43.61,
|
|
44.33,
|
|
44.83,
|
|
45.10,
|
|
45.15,
|
|
43.61,
|
|
44.33,
|
|
44.83,
|
|
45.10,
|
|
45.15,
|
|
43.61,
|
|
44.33,
|
|
]
|
|
)
|
|
result = run("RSI", close, timeperiod=14)
|
|
assert result.shape == (15,)
|
|
|
|
def test_get_returns_callable(self):
|
|
fn = get("EMA")
|
|
assert callable(fn)
|
|
|
|
def test_register_custom_indicator(self):
|
|
def DOUBLE_SMA(close, timeperiod=5):
|
|
return close * 2.0
|
|
|
|
register("DOUBLE_SMA", DOUBLE_SMA)
|
|
try:
|
|
close = np.array([1.0, 2.0, 3.0])
|
|
result = run("DOUBLE_SMA", close, timeperiod=2)
|
|
np.testing.assert_array_equal(result, np.array([2.0, 4.0, 6.0]))
|
|
finally:
|
|
unregister("DOUBLE_SMA")
|
|
|
|
def test_unregister_removes_indicator(self):
|
|
def TEMP_IND(close):
|
|
return close
|
|
|
|
register("TEMP_IND", TEMP_IND)
|
|
assert "TEMP_IND" in list_indicators()
|
|
unregister("TEMP_IND")
|
|
assert "TEMP_IND" not in list_indicators()
|
|
|
|
def test_unknown_indicator_raises(self):
|
|
with pytest.raises(FerroTARegistryError):
|
|
get("UNKNOWN_INDICATOR_XYZ")
|
|
|
|
def test_run_unknown_indicator_raises(self):
|
|
with pytest.raises(FerroTARegistryError):
|
|
run("NO_SUCH_IND", np.array([1.0, 2.0]))
|
|
|
|
def test_unregister_unknown_raises(self):
|
|
with pytest.raises(FerroTARegistryError):
|
|
unregister("NEVER_REGISTERED")
|
|
|
|
def test_register_non_callable_raises(self):
|
|
with pytest.raises(TypeError):
|
|
register("BAD", 42) # type: ignore[arg-type]
|
|
|
|
def test_list_indicators_is_sorted(self):
|
|
names = list_indicators()
|
|
assert names == sorted(names)
|
|
|
|
def test_all_builtins_are_callable(self):
|
|
for name in list_indicators():
|
|
fn = get(name)
|
|
assert callable(fn), f"{name} is not callable"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New Extended Indicators (KELTNER_CHANNELS, HULL_MA,
|
|
# CHANDELIER_EXIT, VWMA, CHOPPINESS_INDEX)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ferro_ta import (
|
|
CHANDELIER_EXIT,
|
|
CHOPPINESS_INDEX,
|
|
HULL_MA,
|
|
KELTNER_CHANNELS,
|
|
VWMA,
|
|
)
|
|
|
|
_N = 30
|
|
_C = np.cumsum(np.ones(_N)) + 40.0
|
|
_H = _C + 0.5
|
|
_L = _C - 0.5
|
|
_V = np.full(_N, 1_000_000.0)
|
|
|
|
|
|
class TestKeltnerChannels:
|
|
def test_output_shapes(self):
|
|
u, m, lo = KELTNER_CHANNELS(_H, _L, _C, timeperiod=5, atr_period=3)
|
|
assert len(u) == len(m) == len(lo) == _N
|
|
|
|
def test_upper_gt_middle_gt_lower(self):
|
|
u, m, lo = KELTNER_CHANNELS(_H, _L, _C, timeperiod=5, atr_period=3)
|
|
valid = ~np.isnan(u)
|
|
assert np.all(u[valid] > m[valid])
|
|
assert np.all(m[valid] > lo[valid])
|
|
|
|
|
|
class TestHullMA:
|
|
def test_output_length(self):
|
|
hull = HULL_MA(_C, timeperiod=4)
|
|
assert len(hull) == _N
|
|
|
|
def test_leading_nans(self):
|
|
hull = HULL_MA(_C, timeperiod=4)
|
|
assert int(np.sum(np.isnan(hull))) >= 1
|
|
|
|
def test_finite_after_warmup(self):
|
|
hull = HULL_MA(_C, timeperiod=4)
|
|
assert np.all(np.isfinite(hull[~np.isnan(hull)]))
|
|
|
|
|
|
class TestChandelierExit:
|
|
def test_output_shapes(self):
|
|
le, se = CHANDELIER_EXIT(_H, _L, _C, timeperiod=5, multiplier=2.0)
|
|
assert len(le) == len(se) == _N
|
|
|
|
def test_long_lt_high_short_gt_low(self):
|
|
le, se = CHANDELIER_EXIT(_H, _L, _C, timeperiod=5, multiplier=2.0)
|
|
# Both outputs should have valid values after warmup
|
|
valid_le = ~np.isnan(le)
|
|
valid_se = ~np.isnan(se)
|
|
assert valid_le.any()
|
|
assert valid_se.any()
|
|
# Long exit must be finite and positive
|
|
assert np.all(np.isfinite(le[valid_le]))
|
|
assert np.all(le[valid_le] > 0.0)
|
|
# Short exit must be finite and positive
|
|
assert np.all(np.isfinite(se[valid_se]))
|
|
assert np.all(se[valid_se] > 0.0)
|
|
|
|
|
|
class TestVWMA:
|
|
def test_output_length(self):
|
|
v = VWMA(_C, _V, timeperiod=5)
|
|
assert len(v) == _N
|
|
|
|
def test_leading_nans(self):
|
|
v = VWMA(_C, _V, timeperiod=5)
|
|
assert int(np.sum(np.isnan(v))) == 4
|
|
|
|
def test_uniform_volume_equals_sma(self):
|
|
"""With uniform volume, VWMA equals SMA."""
|
|
from ferro_ta import SMA
|
|
|
|
c = np.arange(1.0, 21.0)
|
|
v = np.ones(20)
|
|
vwma = VWMA(c, v, timeperiod=5)
|
|
sma = SMA(c, timeperiod=5)
|
|
valid = ~np.isnan(vwma) & ~np.isnan(sma)
|
|
assert np.allclose(vwma[valid], sma[valid], rtol=1e-9)
|
|
|
|
|
|
class TestChoppinessIndex:
|
|
def test_output_length(self):
|
|
ci = CHOPPINESS_INDEX(_H, _L, _C, timeperiod=5)
|
|
assert len(ci) == _N
|
|
|
|
def test_range_0_to_100(self):
|
|
ci = CHOPPINESS_INDEX(_H, _L, _C, timeperiod=5)
|
|
valid = ci[~np.isnan(ci)]
|
|
if len(valid) > 0:
|
|
assert np.all(valid >= 0.0)
|
|
assert np.all(valid <= 100.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Batch execution API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ferro_ta import EMA, RSI, SMA
|
|
from ferro_ta.data.batch import (
|
|
batch_apply,
|
|
batch_atr,
|
|
batch_ema,
|
|
batch_rsi,
|
|
batch_sma,
|
|
)
|
|
|
|
|
|
class TestBatchSMA:
|
|
C2D = np.random.default_rng(7).random((50, 3)) + 50.0
|
|
C1D = C2D[:, 0]
|
|
|
|
def test_output_shape_2d(self):
|
|
result = batch_sma(self.C2D, timeperiod=10)
|
|
assert result.shape == (50, 3)
|
|
|
|
def test_output_shape_1d_unchanged(self):
|
|
"""1-D input should return 1-D (backward compatible)."""
|
|
result = batch_sma(self.C1D, timeperiod=10)
|
|
assert result.ndim == 1
|
|
assert len(result) == 50
|
|
|
|
def test_column_matches_single_series(self):
|
|
"""Each column of batch_sma must match single-series SMA."""
|
|
result = batch_sma(self.C2D, timeperiod=10)
|
|
for j in range(3):
|
|
expected = SMA(self.C2D[:, j], timeperiod=10)
|
|
assert np.allclose(result[:, j], expected, equal_nan=True)
|
|
|
|
|
|
class TestBatchEMA:
|
|
C2D = np.random.default_rng(8).random((50, 4)) + 40.0
|
|
|
|
def test_output_shape(self):
|
|
result = batch_ema(self.C2D, timeperiod=5)
|
|
assert result.shape == (50, 4)
|
|
|
|
def test_column_matches_single_series(self):
|
|
result = batch_ema(self.C2D, timeperiod=5)
|
|
for j in range(4):
|
|
expected = EMA(self.C2D[:, j], timeperiod=5)
|
|
assert np.allclose(result[:, j], expected, equal_nan=True)
|
|
|
|
|
|
class TestBatchRSI:
|
|
C2D = np.random.default_rng(9).random((50, 2)) + 45.0
|
|
|
|
def test_output_shape(self):
|
|
result = batch_rsi(self.C2D, timeperiod=14)
|
|
assert result.shape == (50, 2)
|
|
|
|
def test_values_in_range(self):
|
|
result = batch_rsi(self.C2D, timeperiod=14)
|
|
valid = result[~np.isnan(result)]
|
|
if len(valid) > 0:
|
|
assert valid.min() >= 0.0
|
|
assert valid.max() <= 100.0
|
|
|
|
def test_column_matches_single_series(self):
|
|
result = batch_rsi(self.C2D, timeperiod=14)
|
|
for j in range(2):
|
|
expected = RSI(self.C2D[:, j], timeperiod=14)
|
|
assert np.allclose(result[:, j], expected, equal_nan=True)
|
|
|
|
|
|
class TestBatchApply:
|
|
C2D = np.random.default_rng(11).random((40, 3)) + 50.0
|
|
|
|
def test_custom_fn(self):
|
|
"""batch_apply should delegate to any single-series function."""
|
|
from ferro_ta import BBANDS
|
|
|
|
def mid(c, **kw):
|
|
return BBANDS(c, **kw)[1]
|
|
|
|
result = batch_apply(self.C2D, mid, timeperiod=5)
|
|
assert result.shape == (40, 3)
|
|
|
|
def test_3d_raises(self):
|
|
with pytest.raises(ValueError, match="1-D or 2-D"):
|
|
batch_apply(np.zeros((5, 5, 5)), SMA, timeperiod=3)
|
|
|
|
def test_sma_fastpath_matches_batch_sma(self):
|
|
from ferro_ta.data.batch import batch_sma
|
|
|
|
fast = batch_apply(self.C2D, SMA, timeperiod=10)
|
|
direct = batch_sma(self.C2D, timeperiod=10)
|
|
assert np.allclose(fast, direct, equal_nan=True)
|
|
|
|
|
|
class TestBatchShapeValidation:
|
|
def test_batch_atr_shape_mismatch_raises(self):
|
|
high = np.ones((5, 2), dtype=np.float64)
|
|
low = np.ones((4, 2), dtype=np.float64)
|
|
close = np.ones((5, 2), dtype=np.float64)
|
|
with pytest.raises(ValueError, match="shape"):
|
|
batch_atr(high, low, close, timeperiod=3)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Release playbook and version consistency
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import os
|
|
import re
|
|
import runpy
|
|
import subprocess
|
|
|
|
try:
|
|
import tomllib # Python 3.11+
|
|
except ImportError:
|
|
try:
|
|
import tomli as tomllib # type: ignore[no-redef] # fallback for Python < 3.11
|
|
except ImportError:
|
|
tomllib = None # type: ignore[assignment]
|
|
|
|
|
|
def _read_cargo_version() -> str:
|
|
"""Extract version from root Cargo.toml."""
|
|
if tomllib is None:
|
|
raise ImportError("tomllib/tomli not available")
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
cargo_toml = os.path.join(root, "Cargo.toml")
|
|
with open(cargo_toml, "rb") as f:
|
|
data = tomllib.load(f)
|
|
return data["package"]["version"]
|
|
|
|
|
|
def _read_pyproject_version() -> str:
|
|
"""Extract version from pyproject.toml."""
|
|
if tomllib is None:
|
|
raise ImportError("tomllib/tomli not available")
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
pyproject_toml = os.path.join(root, "pyproject.toml")
|
|
with open(pyproject_toml, "rb") as f:
|
|
data = tomllib.load(f)
|
|
return data["project"]["version"]
|
|
|
|
|
|
def _read_conda_version() -> str:
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
conda_meta = os.path.join(root, "conda", "meta.yaml")
|
|
text = open(conda_meta).read()
|
|
match = re.search(r'{% set version = "([^"]+)" %}', text)
|
|
if not match:
|
|
raise ValueError("Could not find conda version")
|
|
return match.group(1)
|
|
|
|
|
|
def _read_docs_release() -> str:
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
conf_py = os.path.join(root, "docs", "conf.py")
|
|
old_env = os.environ.pop("FERRO_TA_VERSION", None)
|
|
try:
|
|
data = runpy.run_path(conf_py)
|
|
return data["release"]
|
|
finally:
|
|
if old_env is not None:
|
|
os.environ["FERRO_TA_VERSION"] = old_env
|
|
|
|
|
|
def _run_bump_version_check() -> subprocess.CompletedProcess[str]:
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
return subprocess.run(
|
|
["python3", "scripts/bump_version.py", "--check"],
|
|
cwd=root,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
class TestVersionConsistency:
|
|
"""Public version strings should stay aligned with the package version."""
|
|
|
|
def test_versions_match(self):
|
|
try:
|
|
cargo_ver = _read_cargo_version()
|
|
pyproject_ver = _read_pyproject_version()
|
|
except Exception:
|
|
pytest.skip("tomllib unavailable or files not found")
|
|
assert cargo_ver == pyproject_ver, (
|
|
f"Version mismatch: Cargo.toml={cargo_ver!r}, "
|
|
f"pyproject.toml={pyproject_ver!r}"
|
|
)
|
|
|
|
def test_package_version_matches_project_version(self):
|
|
cargo_ver = _read_cargo_version()
|
|
assert ferro_ta.__version__ == cargo_ver
|
|
|
|
def test_conda_version_matches_project_version(self):
|
|
cargo_ver = _read_cargo_version()
|
|
conda_ver = _read_conda_version()
|
|
assert conda_ver == cargo_ver
|
|
|
|
def test_docs_release_matches_project_version(self):
|
|
cargo_ver = _read_cargo_version()
|
|
docs_release = _read_docs_release()
|
|
assert docs_release == cargo_ver
|
|
|
|
def test_docs_changelog_mentions_current_version(self):
|
|
cargo_ver = _read_cargo_version()
|
|
root = os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
changelog_rst = os.path.join(root, "docs", "changelog.rst")
|
|
text = open(changelog_rst).read()
|
|
assert cargo_ver in text
|
|
|
|
def test_api_version_matches_project_version(self):
|
|
cargo_ver = _read_cargo_version()
|
|
try:
|
|
from api.main import app
|
|
except Exception:
|
|
pytest.skip("api/main.py not importable")
|
|
assert app.version == cargo_ver
|
|
|
|
def test_bump_version_check_passes(self):
|
|
result = _run_bump_version_check()
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
|
|
def test_release_md_exists(self):
|
|
"""RELEASE.md must exist in the repository root."""
|
|
root = os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
release_md = os.path.join(root, "RELEASE.md")
|
|
assert os.path.isfile(release_md), "RELEASE.md not found"
|
|
|
|
def test_release_md_has_key_sections(self):
|
|
"""RELEASE.md must mention tagging and PyPI."""
|
|
root = os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
release_md = os.path.join(root, "RELEASE.md")
|
|
if not os.path.isfile(release_md):
|
|
pytest.skip("RELEASE.md not found")
|
|
text = open(release_md).read()
|
|
assert "git tag" in text or "tag" in text.lower()
|
|
assert "pypi" in text.lower() or "PyPI" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GPU backend (PyTorch, CPU fallback always available)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ferro_ta.tools.gpu import ema as gpu_ema
|
|
from ferro_ta.tools.gpu import rsi as gpu_rsi
|
|
from ferro_ta.tools.gpu import sma as gpu_sma # noqa: E402
|
|
|
|
CLOSE_15 = np.array(
|
|
[
|
|
44.34,
|
|
44.09,
|
|
44.15,
|
|
43.61,
|
|
44.33,
|
|
44.83,
|
|
45.10,
|
|
45.15,
|
|
43.61,
|
|
44.33,
|
|
44.83,
|
|
45.10,
|
|
45.15,
|
|
43.61,
|
|
44.33,
|
|
]
|
|
)
|
|
|
|
|
|
class TestGPUCPUFallback:
|
|
"""GPU module falls back to CPU when CuPy is not available."""
|
|
|
|
def test_sma_cpu_fallback_length(self):
|
|
result = gpu_sma(CLOSE_15, timeperiod=5)
|
|
assert len(result) == len(CLOSE_15)
|
|
|
|
def test_sma_cpu_fallback_values(self):
|
|
from ferro_ta import SMA
|
|
|
|
result = gpu_sma(CLOSE_15, timeperiod=5)
|
|
expected = SMA(CLOSE_15, timeperiod=5)
|
|
np.testing.assert_allclose(result, expected, equal_nan=True)
|
|
|
|
def test_ema_cpu_fallback_values(self):
|
|
from ferro_ta import EMA
|
|
|
|
result = gpu_ema(CLOSE_15, timeperiod=5)
|
|
expected = EMA(CLOSE_15, timeperiod=5)
|
|
np.testing.assert_allclose(result, expected, equal_nan=True)
|
|
|
|
def test_rsi_cpu_fallback_values(self):
|
|
from ferro_ta import RSI
|
|
|
|
result = gpu_rsi(CLOSE_15, timeperiod=5)
|
|
expected = RSI(CLOSE_15, timeperiod=5)
|
|
np.testing.assert_allclose(result, expected, equal_nan=True)
|
|
|
|
def test_sma_returns_numpy_for_numpy_input(self):
|
|
result = gpu_sma(CLOSE_15, timeperiod=5)
|
|
assert isinstance(result, np.ndarray)
|
|
|
|
def test_rsi_finite_values_in_range(self):
|
|
result = gpu_rsi(CLOSE_15, timeperiod=5)
|
|
finite = result[np.isfinite(result)]
|
|
assert len(finite) > 0
|
|
assert np.all(finite >= 0.0)
|
|
assert np.all(finite <= 100.0)
|
|
|
|
def test_gpu_module_all_exports(self):
|
|
from ferro_ta.tools import gpu as gpu_mod
|
|
|
|
for name in gpu_mod.__all__:
|
|
assert callable(getattr(gpu_mod, name))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Indicator pipeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ferro_ta import BBANDS # noqa: E402 (already imported)
|
|
from ferro_ta.tools.pipeline import Pipeline, make_pipeline # noqa: E402
|
|
|
|
CLOSE_20 = np.random.default_rng(99).random(20) * 100 + 50
|
|
|
|
|
|
class TestPipeline:
|
|
"""Tests for ferro_ta.pipeline.Pipeline."""
|
|
|
|
def test_pipeline_run_returns_dict(self):
|
|
pipe = Pipeline().add("sma5", SMA, timeperiod=5)
|
|
result = pipe.run(CLOSE_20)
|
|
assert isinstance(result, dict)
|
|
assert "sma5" in result
|
|
|
|
def test_pipeline_result_length_matches_input(self):
|
|
pipe = Pipeline().add("sma5", SMA, timeperiod=5)
|
|
result = pipe.run(CLOSE_20)
|
|
assert len(result["sma5"]) == len(CLOSE_20)
|
|
|
|
def test_pipeline_multiple_steps(self):
|
|
pipe = (
|
|
Pipeline()
|
|
.add("sma5", SMA, timeperiod=5)
|
|
.add("ema5", EMA, timeperiod=5)
|
|
.add("rsi7", RSI, timeperiod=7)
|
|
)
|
|
result = pipe.run(CLOSE_20)
|
|
assert set(result.keys()) == {"sma5", "ema5", "rsi7"}
|
|
|
|
def test_pipeline_multi_output_with_output_keys(self):
|
|
pipe = Pipeline().add(
|
|
"bb",
|
|
BBANDS,
|
|
timeperiod=5,
|
|
nbdevup=2.0,
|
|
nbdevdn=2.0,
|
|
output_keys=["upper", "mid", "lower"],
|
|
)
|
|
result = pipe.run(CLOSE_20)
|
|
assert "upper" in result
|
|
assert "mid" in result
|
|
assert "lower" in result
|
|
assert "bb" not in result
|
|
|
|
def test_pipeline_multi_output_without_output_keys(self):
|
|
pipe = Pipeline().add("bb", BBANDS, timeperiod=5, nbdevup=2.0, nbdevdn=2.0)
|
|
result = pipe.run(CLOSE_20)
|
|
# Should auto-name as bb_0, bb_1, bb_2
|
|
assert "bb_0" in result
|
|
assert "bb_1" in result
|
|
assert "bb_2" in result
|
|
|
|
def test_pipeline_remove_step(self):
|
|
pipe = Pipeline().add("sma5", SMA, timeperiod=5).add("ema5", EMA, timeperiod=5)
|
|
pipe.remove("sma5")
|
|
assert pipe.steps() == ["ema5"]
|
|
|
|
def test_pipeline_len(self):
|
|
pipe = Pipeline().add("sma5", SMA, timeperiod=5).add("ema5", EMA, timeperiod=5)
|
|
assert len(pipe) == 2
|
|
|
|
def test_pipeline_duplicate_name_raises(self):
|
|
pipe = Pipeline().add("sma5", SMA, timeperiod=5)
|
|
with pytest.raises(ValueError, match="sma5"):
|
|
pipe.add("sma5", SMA, timeperiod=10)
|
|
|
|
def test_make_pipeline_factory(self):
|
|
pipe = make_pipeline(
|
|
sma5=(SMA, {"timeperiod": 5}),
|
|
rsi7=(RSI, {"timeperiod": 7}),
|
|
)
|
|
result = pipe.run(CLOSE_20)
|
|
assert "sma5" in result
|
|
assert "rsi7" in result
|
|
|
|
def test_pipeline_sma_values_match_direct_call(self):
|
|
pipe = Pipeline().add("sma5", SMA, timeperiod=5)
|
|
result = pipe.run(CLOSE_20)
|
|
direct = SMA(CLOSE_20, timeperiod=5)
|
|
np.testing.assert_allclose(result["sma5"], direct, equal_nan=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Polars integration (skipped if polars not installed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPolarsIntegration:
|
|
"""Transparent polars.Series support via polars_wrap."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def skip_if_no_polars(self):
|
|
pytest.importorskip("polars")
|
|
|
|
def test_sma_returns_polars_series(self):
|
|
import polars as pl
|
|
|
|
s = pl.Series("close", CLOSE_20.tolist())
|
|
result = SMA(s, timeperiod=5)
|
|
assert isinstance(result, pl.Series)
|
|
|
|
def test_sma_values_match_numpy(self):
|
|
import polars as pl
|
|
|
|
s = pl.Series("close", CLOSE_20.tolist())
|
|
result = SMA(s, timeperiod=5)
|
|
expected = SMA(CLOSE_20, timeperiod=5)
|
|
np.testing.assert_allclose(result.to_numpy(), expected, equal_nan=True)
|
|
|
|
def test_rsi_returns_polars_series(self):
|
|
import polars as pl
|
|
|
|
s = pl.Series("close", CLOSE_20.tolist())
|
|
result = RSI(s, timeperiod=5)
|
|
assert isinstance(result, pl.Series)
|
|
|
|
def test_numpy_input_still_returns_numpy(self):
|
|
result = SMA(CLOSE_20, timeperiod=5)
|
|
assert isinstance(result, np.ndarray)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration defaults
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import ferro_ta.core.config as ftconfig # noqa: E402
|
|
|
|
|
|
class TestConfig:
|
|
"""Tests for ferro_ta.config module."""
|
|
|
|
def setup_method(self):
|
|
"""Reset config state before each test."""
|
|
ftconfig.reset()
|
|
|
|
def teardown_method(self):
|
|
"""Clean up after each test."""
|
|
ftconfig.reset()
|
|
|
|
def test_set_and_get_default(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
assert ftconfig.get_default("timeperiod") == 20
|
|
|
|
def test_get_default_fallback(self):
|
|
assert ftconfig.get_default("nonexistent") is None
|
|
assert ftconfig.get_default("nonexistent", -1) == -1
|
|
|
|
def test_reset_single_key(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
ftconfig.reset("timeperiod")
|
|
assert ftconfig.get_default("timeperiod") is None
|
|
|
|
def test_reset_all(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
ftconfig.set_default("RSI.timeperiod", 14)
|
|
ftconfig.reset()
|
|
assert ftconfig.list_defaults() == {}
|
|
|
|
def test_list_defaults(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
ftconfig.set_default("RSI.timeperiod", 14)
|
|
defaults = ftconfig.list_defaults()
|
|
assert defaults == {"timeperiod": 20, "RSI.timeperiod": 14}
|
|
|
|
def test_get_defaults_for_indicator(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
ftconfig.set_default("RSI.timeperiod", 14)
|
|
rsi_defaults = ftconfig.get_defaults_for("RSI")
|
|
assert rsi_defaults == {"timeperiod": 14}
|
|
sma_defaults = ftconfig.get_defaults_for("SMA")
|
|
assert sma_defaults == {"timeperiod": 20}
|
|
|
|
def test_config_context_manager(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
with ftconfig.Config(timeperiod=5):
|
|
assert ftconfig.get_default("timeperiod") == 5
|
|
assert ftconfig.get_default("timeperiod") == 20
|
|
|
|
def test_config_context_manager_restores_on_exception(self):
|
|
ftconfig.set_default("timeperiod", 20)
|
|
try:
|
|
with ftconfig.Config(timeperiod=5):
|
|
raise RuntimeError("test error")
|
|
except RuntimeError:
|
|
pass
|
|
assert ftconfig.get_default("timeperiod") == 20
|
|
|
|
def test_config_context_manager_new_key_removed_on_exit(self):
|
|
# Key doesn't exist before context
|
|
assert ftconfig.get_default("nbdevup") is None
|
|
with ftconfig.Config(nbdevup=2.5):
|
|
assert ftconfig.get_default("nbdevup") == 2.5
|
|
assert ftconfig.get_default("nbdevup") is None
|