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
|