Pratik Bhadane
71b6343e92
Refresh the benchmark and performance surface across the repo. This updates the benchmark wrappers and helper scripts, regenerates the checked-in benchmark and perf-contract artifacts, and folds in the related roadmap, compatibility, and example notebook changes that belong with this performance-focused pass. Harden the Python CI and local pre-push flow so the same checks pass reliably in both places. The workflow and pre-push script now use module-safe uv typecheck invocations, the Python test environment installs the optional MCP dependency needed by the MCP server tests, and one-off root benchmark outputs are ignored to keep the repo clean. Align local tooling with the current project configuration by updating the Ruff pre-commit hook, tightening the API typing and MCP server helpers, and refreshing the lockfile to pick up the audited PyJWT fix while preserving the rest of the staged source changes.
1520 lines
50 KiB
Python
1520 lines
50 KiB
Python
"""Tests for alerts, crypto helpers, chunked processing,
|
|
regime detection, performance attribution, and dashboard helpers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import runpy
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Synthetic helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
RNG = np.random.default_rng(31415)
|
|
|
|
|
|
def _make_close(n: int = 200) -> np.ndarray:
|
|
return np.cumprod(1 + RNG.normal(0, 0.01, n)) * 100.0
|
|
|
|
|
|
def _make_ohlcv(n: int = 200):
|
|
close = _make_close(n)
|
|
open_ = close * RNG.uniform(0.995, 1.005, n)
|
|
high = np.maximum(close, open_) + RNG.uniform(0, 0.5, n)
|
|
low = np.minimum(close, open_) - RNG.uniform(0, 0.5, n)
|
|
volume = RNG.uniform(500, 5000, n)
|
|
return open_, high, low, close, volume
|
|
|
|
|
|
# ===========================================================================
|
|
# Alerts
|
|
# ===========================================================================
|
|
|
|
|
|
class TestAlertsLowLevel:
|
|
"""Tests for low-level alert condition functions."""
|
|
|
|
def test_check_threshold_cross_above(self):
|
|
from ferro_ta.tools.alerts import check_threshold
|
|
|
|
series = np.array([20.0, 25.0, 30.0, 35.0, 28.0])
|
|
mask = check_threshold(series, level=29.0, direction=1)
|
|
# Cross above fires when series was <= level and now > level.
|
|
# Bar 0: no prior bar, always 0.
|
|
# Bar 2: prev=25 <= 29, curr=30 > 29 → fires
|
|
assert mask[0] == 0
|
|
assert mask[2] == 1
|
|
assert mask[1] == 0
|
|
assert mask[3] == 0 # 30 > 29 previously, so no new crossing
|
|
|
|
def test_check_threshold_cross_below(self):
|
|
from ferro_ta.tools.alerts import check_threshold
|
|
|
|
series = np.array([70.0, 65.0, 28.0, 25.0, 35.0])
|
|
mask = check_threshold(series, level=30.0, direction=-1)
|
|
# index 2: 65 >= 30 → 28 < 30: cross below
|
|
assert mask[2] == 1
|
|
assert mask[0] == 0
|
|
assert mask[4] == 0 # 25 < 30 already, 35 > 30 is not a cross-below
|
|
|
|
def test_check_threshold_invalid_direction(self):
|
|
from ferro_ta.tools.alerts import check_threshold
|
|
|
|
with pytest.raises(Exception):
|
|
check_threshold(np.array([1.0, 2.0]), level=1.5, direction=0)
|
|
|
|
def test_check_cross_bullish(self):
|
|
from ferro_ta.tools.alerts import check_cross
|
|
|
|
fast = np.array([10.0, 12.0, 15.0, 14.0, 16.0])
|
|
slow = np.array([13.0, 13.0, 13.0, 13.0, 13.0])
|
|
mask = check_cross(fast, slow)
|
|
# fast crosses above slow at index 2 (12 <= 13 → 15 > 13)
|
|
assert mask[2] == 1 # bullish
|
|
assert mask[0] == 0
|
|
|
|
def test_check_cross_bearish(self):
|
|
from ferro_ta.tools.alerts import check_cross
|
|
|
|
fast = np.array([15.0, 15.0, 12.0, 11.0])
|
|
slow = np.array([13.0, 13.0, 13.0, 13.0])
|
|
mask = check_cross(fast, slow)
|
|
# fast crosses below slow at index 2 (15 >= 13 → 12 < 13)
|
|
assert mask[2] == -1 # bearish
|
|
|
|
def test_check_cross_length_mismatch_raises(self):
|
|
from ferro_ta.tools.alerts import check_cross
|
|
|
|
with pytest.raises(Exception):
|
|
check_cross(np.array([1.0, 2.0]), np.array([1.0, 2.0, 3.0]))
|
|
|
|
def test_collect_alert_bars(self):
|
|
from ferro_ta.tools.alerts import collect_alert_bars
|
|
|
|
mask = np.array([0, 1, 0, 0, 1, -1], dtype=np.int8)
|
|
bars = collect_alert_bars(mask)
|
|
assert list(bars) == [1, 4, 5]
|
|
|
|
def test_collect_alert_bars_empty(self):
|
|
from ferro_ta.tools.alerts import collect_alert_bars
|
|
|
|
mask = np.zeros(10, dtype=np.int8)
|
|
bars = collect_alert_bars(mask)
|
|
assert len(bars) == 0
|
|
|
|
|
|
class TestAlertManager:
|
|
"""Tests for the AlertManager class."""
|
|
|
|
def test_run_backtest_returns_list(self):
|
|
from ferro_ta import RSI
|
|
from ferro_ta.tools.alerts import AlertManager
|
|
|
|
close = _make_close(200)
|
|
rsi = np.asarray(RSI(close, timeperiod=14), dtype=np.float64)
|
|
am = AlertManager(symbol="TEST")
|
|
am.add_threshold_condition("rsi_os", rsi, level=30.0, direction=-1)
|
|
events = am.run_backtest()
|
|
assert isinstance(events, list)
|
|
|
|
def test_backtest_no_external_calls_by_default(self):
|
|
"""Backtest mode must not invoke callback unless force_live=True."""
|
|
from ferro_ta import SMA
|
|
from ferro_ta.tools.alerts import AlertManager
|
|
|
|
close = _make_close(100)
|
|
sma10 = np.asarray(SMA(close, timeperiod=10), dtype=np.float64)
|
|
sma30 = np.asarray(SMA(close, timeperiod=30), dtype=np.float64)
|
|
|
|
called = []
|
|
|
|
def cb(ev):
|
|
called.append(ev)
|
|
|
|
am = AlertManager()
|
|
am.add_cross_condition("sma_x", sma10, sma30, callback=cb)
|
|
events = am.run_backtest() # default live=False
|
|
assert len(called) == 0, "callback must not fire in backtest mode"
|
|
assert isinstance(events, list)
|
|
|
|
def test_backtest_force_live_invokes_callback(self):
|
|
from ferro_ta import RSI
|
|
from ferro_ta.tools.alerts import AlertManager
|
|
|
|
close = _make_close(300)
|
|
rsi = np.asarray(RSI(close, timeperiod=14), dtype=np.float64)
|
|
|
|
fired = []
|
|
|
|
def cb(ev):
|
|
fired.append(ev)
|
|
|
|
am = AlertManager()
|
|
am.add_threshold_condition("rsi_os", rsi, level=30.0, direction=-1, callback=cb)
|
|
events = am.run_backtest(force_live=True)
|
|
assert len(fired) == len(events)
|
|
|
|
def test_event_payload_contains_symbol(self):
|
|
from ferro_ta import RSI
|
|
from ferro_ta.tools.alerts import AlertManager
|
|
|
|
close = _make_close(200)
|
|
rsi = np.asarray(RSI(close, timeperiod=14), dtype=np.float64)
|
|
am = AlertManager(symbol="BTCUSD")
|
|
am.add_threshold_condition("rsi_os", rsi, level=30.0, direction=-1)
|
|
events = am.run_backtest()
|
|
for ev in events:
|
|
assert ev.payload.get("symbol") == "BTCUSD"
|
|
|
|
def test_event_bar_index_valid(self):
|
|
from ferro_ta import SMA
|
|
from ferro_ta.tools.alerts import AlertManager
|
|
|
|
close = _make_close(100)
|
|
sma5 = np.asarray(SMA(close, timeperiod=5), dtype=np.float64)
|
|
sma20 = np.asarray(SMA(close, timeperiod=20), dtype=np.float64)
|
|
am = AlertManager()
|
|
am.add_cross_condition("x", sma5, sma20)
|
|
events = am.run_backtest()
|
|
for ev in events:
|
|
assert 0 <= ev.bar_index < len(close)
|
|
|
|
def test_alert_event_to_dict(self):
|
|
from ferro_ta.tools.alerts import AlertEvent
|
|
|
|
ev = AlertEvent("my_cond", 42, value=27.5, payload={"symbol": "X"})
|
|
d = ev.to_dict()
|
|
assert d["condition_id"] == "my_cond"
|
|
assert d["bar_index"] == 42
|
|
assert d["symbol"] == "X"
|
|
|
|
|
|
# ===========================================================================
|
|
# Crypto helpers
|
|
# ===========================================================================
|
|
|
|
|
|
class TestCryptoFunding:
|
|
def test_funding_pnl_shape(self):
|
|
from ferro_ta.analysis.crypto import funding_pnl
|
|
|
|
pos = np.ones(100)
|
|
rate = RNG.normal(0, 0.0001, 100)
|
|
pnl = funding_pnl(pos, rate)
|
|
assert pnl.shape == (100,)
|
|
|
|
def test_funding_pnl_cumulative(self):
|
|
from ferro_ta.analysis.crypto import funding_pnl
|
|
|
|
pos = np.ones(5)
|
|
rate = np.array([0.0001, 0.0002, -0.0001, 0.0001, 0.0001])
|
|
pnl = funding_pnl(pos, rate)
|
|
expected = np.cumsum(-pos * rate)
|
|
np.testing.assert_allclose(pnl, expected)
|
|
|
|
def test_funding_pnl_long_pays_positive_rate(self):
|
|
"""Long position should pay (negative PnL) when funding rate > 0."""
|
|
from ferro_ta.analysis.crypto import funding_pnl
|
|
|
|
pos = np.ones(1)
|
|
rate = np.array([0.001]) # positive rate → long pays
|
|
pnl = funding_pnl(pos, rate)
|
|
assert pnl[0] < 0
|
|
|
|
def test_funding_pnl_short_receives_positive_rate(self):
|
|
"""Short position should receive (positive PnL) when funding rate > 0."""
|
|
from ferro_ta.analysis.crypto import funding_pnl
|
|
|
|
pos = np.array([-1.0])
|
|
rate = np.array([0.001])
|
|
pnl = funding_pnl(pos, rate)
|
|
assert pnl[0] > 0
|
|
|
|
def test_funding_pnl_length_mismatch_raises(self):
|
|
from ferro_ta.analysis.crypto import funding_pnl
|
|
|
|
with pytest.raises(Exception):
|
|
funding_pnl(np.ones(5), np.ones(4))
|
|
|
|
|
|
class TestCryptoBarLabels:
|
|
def test_continuous_bar_labels_shape(self):
|
|
from ferro_ta.analysis.crypto import continuous_bar_labels
|
|
|
|
labels = continuous_bar_labels(10, 3)
|
|
assert labels.shape == (10,)
|
|
|
|
def test_continuous_bar_labels_values(self):
|
|
from ferro_ta.analysis.crypto import continuous_bar_labels
|
|
|
|
labels = continuous_bar_labels(10, 3)
|
|
expected = [0, 0, 0, 1, 1, 1, 2, 2, 2, 3]
|
|
np.testing.assert_array_equal(labels, expected)
|
|
|
|
def test_continuous_bar_labels_period_one(self):
|
|
from ferro_ta.analysis.crypto import continuous_bar_labels
|
|
|
|
labels = continuous_bar_labels(5, 1)
|
|
np.testing.assert_array_equal(labels, [0, 1, 2, 3, 4])
|
|
|
|
def test_session_boundaries_daily(self):
|
|
from ferro_ta.analysis.crypto import session_boundaries
|
|
|
|
NS_PER_HOUR = np.int64(3_600_000_000_000)
|
|
# Use a UTC midnight timestamp as base: 1_699_920_000 seconds = Nov 14, 2023 00:00:00 UTC
|
|
base = np.int64(1_699_920_000) * np.int64(1_000_000_000) # midnight UTC
|
|
# 48 hourly bars = 2 full days
|
|
ts = base + np.arange(48, dtype=np.int64) * NS_PER_HOUR
|
|
bounds = session_boundaries(ts)
|
|
assert bounds[0] == 0 # first bar always included
|
|
# Should have exactly 2 boundaries (day 0 and day 1)
|
|
assert len(bounds) == 2
|
|
assert bounds[1] == 24 # second day starts at bar 24
|
|
|
|
|
|
class TestResampleContinuous:
|
|
def test_resample_continuous_shape(self):
|
|
from ferro_ta.analysis.crypto import resample_continuous
|
|
|
|
o, h, l, c, v = _make_ohlcv(100)
|
|
ro, rh, rl, rc, rv = resample_continuous((o, h, l, c, v), period_bars=5)
|
|
assert len(rc) == 20 # 100 / 5
|
|
|
|
def test_resample_continuous_high_ge_low(self):
|
|
from ferro_ta.analysis.crypto import resample_continuous
|
|
|
|
o, h, l, c, v = _make_ohlcv(100)
|
|
_, rh, rl, _, _ = resample_continuous((o, h, l, c, v), period_bars=5)
|
|
assert np.all(rh >= rl)
|
|
|
|
def test_resample_continuous_invalid_period_raises(self):
|
|
from ferro_ta.analysis.crypto import resample_continuous
|
|
|
|
o, h, l, c, v = _make_ohlcv(10)
|
|
with pytest.raises(ValueError):
|
|
resample_continuous((o, h, l, c, v), period_bars=0)
|
|
|
|
|
|
# ===========================================================================
|
|
# Chunked processing
|
|
# ===========================================================================
|
|
|
|
|
|
class TestChunked:
|
|
def test_make_chunk_ranges_shape(self):
|
|
from ferro_ta.data.chunked import make_chunk_ranges
|
|
|
|
ranges = make_chunk_ranges(100, 30, 10)
|
|
assert ranges.ndim == 2
|
|
assert ranges.shape[1] == 2
|
|
|
|
def test_make_chunk_ranges_coverage(self):
|
|
"""All input indices must be covered by some range."""
|
|
from ferro_ta.data.chunked import make_chunk_ranges
|
|
|
|
n = 97
|
|
ranges = make_chunk_ranges(n, 20, 5)
|
|
covered = set()
|
|
for start, end in ranges:
|
|
covered.update(range(int(start), int(end)))
|
|
assert 0 in covered
|
|
assert (n - 1) in covered
|
|
|
|
def test_trim_overlap_basic(self):
|
|
from ferro_ta.data.chunked import trim_overlap
|
|
|
|
arr = np.arange(10, dtype=np.float64)
|
|
trimmed = trim_overlap(arr, overlap=3)
|
|
np.testing.assert_array_equal(trimmed, arr[3:])
|
|
|
|
def test_trim_overlap_zero(self):
|
|
from ferro_ta.data.chunked import trim_overlap
|
|
|
|
arr = np.arange(5, dtype=np.float64)
|
|
trimmed = trim_overlap(arr, overlap=0)
|
|
np.testing.assert_array_equal(trimmed, arr)
|
|
|
|
def test_stitch_chunks_basic(self):
|
|
from ferro_ta.data.chunked import stitch_chunks
|
|
|
|
a = np.array([1.0, 2.0, 3.0])
|
|
b = np.array([4.0, 5.0])
|
|
result = stitch_chunks([a, b])
|
|
np.testing.assert_array_equal(result, [1, 2, 3, 4, 5])
|
|
|
|
def test_chunk_apply_sma_matches_full(self):
|
|
"""chunk_apply(SMA, …) should produce the same result as SMA on the full series."""
|
|
from ferro_ta import SMA
|
|
from ferro_ta.data.chunked import chunk_apply
|
|
|
|
close = _make_close(500)
|
|
full_out = np.asarray(SMA(close, timeperiod=20), dtype=np.float64)
|
|
chunked_out = chunk_apply(SMA, close, chunk_size=100, overlap=30, timeperiod=20)
|
|
|
|
# Compare non-NaN region
|
|
valid = ~np.isnan(full_out)
|
|
np.testing.assert_allclose(
|
|
chunked_out[valid],
|
|
full_out[valid],
|
|
rtol=1e-10,
|
|
err_msg="chunk_apply SMA must match full SMA for non-NaN bars",
|
|
)
|
|
|
|
def test_chunk_apply_output_length(self):
|
|
from ferro_ta import EMA
|
|
from ferro_ta.data.chunked import chunk_apply
|
|
|
|
close = _make_close(300)
|
|
out = chunk_apply(EMA, close, chunk_size=80, overlap=20, timeperiod=10)
|
|
assert len(out) == len(close)
|
|
|
|
|
|
# ===========================================================================
|
|
# Regime detection
|
|
# ===========================================================================
|
|
|
|
|
|
class TestRegimeDetection:
|
|
def test_regime_adx_shape(self):
|
|
from ferro_ta import ADX
|
|
from ferro_ta.analysis.regime import regime_adx
|
|
|
|
o, h, l, c, v = _make_ohlcv(200)
|
|
adx = np.asarray(ADX(h, l, c, timeperiod=14), dtype=np.float64)
|
|
labels = regime_adx(adx, threshold=25.0)
|
|
assert labels.shape == (200,)
|
|
|
|
def test_regime_adx_values_valid(self):
|
|
from ferro_ta import ADX
|
|
from ferro_ta.analysis.regime import regime_adx
|
|
|
|
o, h, l, c, v = _make_ohlcv(200)
|
|
adx = np.asarray(ADX(h, l, c, timeperiod=14), dtype=np.float64)
|
|
labels = regime_adx(adx, threshold=25.0)
|
|
# Values must be -1, 0, or 1
|
|
assert set(labels).issubset({-1, 0, 1})
|
|
|
|
def test_regime_adx_nan_bars_are_minus_one(self):
|
|
from ferro_ta.analysis.regime import regime_adx
|
|
|
|
adx = np.full(20, np.nan)
|
|
adx[15:] = 30.0 # last 5 are trend
|
|
labels = regime_adx(adx, threshold=25.0)
|
|
assert np.all(labels[:15] == -1)
|
|
assert np.all(labels[15:] == 1)
|
|
|
|
def test_regime_combined_shape(self):
|
|
from ferro_ta import ADX, ATR
|
|
from ferro_ta.analysis.regime import regime_combined
|
|
|
|
o, h, l, c, v = _make_ohlcv(200)
|
|
adx = np.asarray(ADX(h, l, c, timeperiod=14), dtype=np.float64)
|
|
atr = np.asarray(ATR(h, l, c, timeperiod=14), dtype=np.float64)
|
|
labels = regime_combined(
|
|
adx, atr, c, adx_threshold=25.0, atr_pct_threshold=0.005
|
|
)
|
|
assert labels.shape == (200,)
|
|
assert set(labels).issubset({-1, 0, 1})
|
|
|
|
def test_regime_high_level_adx(self):
|
|
from ferro_ta.analysis.regime import regime
|
|
|
|
o, h, l, c, v = _make_ohlcv(200)
|
|
labels = regime((o, h, l, c, v), method="adx", adx_threshold=25.0)
|
|
assert labels.shape == (200,)
|
|
assert set(labels).issubset({-1, 0, 1})
|
|
|
|
def test_regime_high_level_combined(self):
|
|
from ferro_ta.analysis.regime import regime
|
|
|
|
o, h, l, c, v = _make_ohlcv(200)
|
|
labels = regime((o, h, l, c, v), method="combined")
|
|
assert labels.shape == (200,)
|
|
|
|
def test_regime_unknown_method_raises(self):
|
|
from ferro_ta.analysis.regime import regime
|
|
|
|
o, h, l, c, v = _make_ohlcv(50)
|
|
with pytest.raises(ValueError):
|
|
regime((o, h, l, c, v), method="unknown")
|
|
|
|
|
|
class TestStructuralBreaks:
|
|
def test_detect_breaks_cusum_shape(self):
|
|
from ferro_ta.analysis.regime import detect_breaks_cusum
|
|
|
|
series = _make_close(200)
|
|
mask = detect_breaks_cusum(series, window=20, threshold=3.0, slack=0.5)
|
|
assert mask.shape == (200,)
|
|
|
|
def test_detect_breaks_cusum_fires_near_break(self):
|
|
"""CUSUM should detect a level shift."""
|
|
from ferro_ta.analysis.regime import detect_breaks_cusum
|
|
|
|
rng = np.random.default_rng(99)
|
|
s1 = rng.normal(0, 1, 100)
|
|
s2 = rng.normal(10, 1, 100) # large level shift
|
|
series = np.concatenate([s1, s2])
|
|
mask = detect_breaks_cusum(series, window=20, threshold=2.0, slack=0.3)
|
|
# Should fire somewhere near the shift
|
|
assert mask[100:130].any()
|
|
|
|
def test_rolling_variance_break_shape(self):
|
|
from ferro_ta.analysis.regime import rolling_variance_break
|
|
|
|
series = _make_close(200)
|
|
mask = rolling_variance_break(
|
|
series, short_window=10, long_window=50, threshold=2.0
|
|
)
|
|
assert mask.shape == (200,)
|
|
|
|
def test_structural_breaks_cusum(self):
|
|
from ferro_ta.analysis.regime import structural_breaks
|
|
|
|
series = _make_close(200)
|
|
mask = structural_breaks(series, method="cusum")
|
|
assert mask.shape == (200,)
|
|
|
|
def test_structural_breaks_variance(self):
|
|
from ferro_ta.analysis.regime import structural_breaks
|
|
|
|
series = _make_close(200)
|
|
mask = structural_breaks(series, method="variance")
|
|
assert mask.shape == (200,)
|
|
|
|
def test_structural_breaks_unknown_method_raises(self):
|
|
from ferro_ta.analysis.regime import structural_breaks
|
|
|
|
with pytest.raises(ValueError):
|
|
structural_breaks(_make_close(50), method="xyz")
|
|
|
|
|
|
# ===========================================================================
|
|
# Performance attribution
|
|
# ===========================================================================
|
|
|
|
|
|
class TestTradeStats:
|
|
def test_basic_stats(self):
|
|
from ferro_ta.analysis.attribution import trade_stats
|
|
|
|
pnl = np.array([10.0, -5.0, 8.0, -3.0, 15.0, -2.0])
|
|
hold = np.array([5.0, 3.0, 7.0, 2.0, 10.0, 1.0])
|
|
ts = trade_stats(pnl, hold)
|
|
assert ts.n_trades == 6
|
|
assert abs(ts.win_rate - 0.5) < 1e-10 # 3 wins out of 6
|
|
assert ts.avg_win > 0
|
|
assert ts.avg_loss < 0
|
|
assert ts.profit_factor > 0
|
|
assert ts.avg_hold_bars == pytest.approx(4.67, abs=0.01)
|
|
|
|
def test_all_wins(self):
|
|
from ferro_ta.analysis.attribution import trade_stats
|
|
|
|
pnl = np.array([5.0, 10.0, 3.0])
|
|
ts = trade_stats(pnl)
|
|
assert ts.win_rate == 1.0
|
|
assert ts.avg_loss == 0.0
|
|
assert ts.profit_factor == float("inf")
|
|
|
|
def test_all_losses(self):
|
|
from ferro_ta.analysis.attribution import trade_stats
|
|
|
|
pnl = np.array([-5.0, -3.0])
|
|
ts = trade_stats(pnl)
|
|
assert ts.win_rate == 0.0
|
|
assert ts.avg_win == 0.0
|
|
assert ts.profit_factor == 0.0
|
|
|
|
def test_empty_raises(self):
|
|
from ferro_ta.analysis.attribution import trade_stats
|
|
|
|
with pytest.raises(Exception):
|
|
trade_stats(np.array([]))
|
|
|
|
def test_to_dict(self):
|
|
from ferro_ta.analysis.attribution import trade_stats
|
|
|
|
pnl = np.array([1.0, -1.0])
|
|
ts = trade_stats(pnl)
|
|
d = ts.to_dict()
|
|
assert "win_rate" in d
|
|
assert "profit_factor" in d
|
|
|
|
|
|
class TestFromBacktest:
|
|
def test_from_backtest_returns_arrays(self):
|
|
from ferro_ta.analysis.attribution import from_backtest
|
|
from ferro_ta.analysis.backtest import backtest
|
|
|
|
close = _make_close(200)
|
|
result = backtest(close, strategy="rsi_30_70")
|
|
pnl, hold = from_backtest(result)
|
|
assert isinstance(pnl, np.ndarray)
|
|
assert isinstance(hold, np.ndarray)
|
|
assert len(pnl) == len(hold)
|
|
# n_trades counts position *changes* (entries + exits);
|
|
# from_backtest counts round-trips (position runs), so len(pnl) <= n_trades
|
|
assert len(pnl) <= result.n_trades
|
|
# Each hold duration should be >= 1
|
|
if len(hold) > 0:
|
|
assert np.all(hold >= 1)
|
|
|
|
def test_from_backtest_no_trades(self):
|
|
from ferro_ta.analysis.attribution import from_backtest
|
|
from ferro_ta.analysis.backtest import BacktestResult
|
|
|
|
n = 50
|
|
result = BacktestResult(
|
|
signals=np.zeros(n),
|
|
positions=np.zeros(n),
|
|
bar_returns=np.zeros(n),
|
|
strategy_returns=np.zeros(n),
|
|
equity=np.ones(n),
|
|
)
|
|
pnl, hold = from_backtest(result)
|
|
assert len(pnl) == 0
|
|
|
|
|
|
class TestAttribution:
|
|
def test_attribution_by_signal_basic(self):
|
|
from ferro_ta.analysis.attribution import attribution_by_signal
|
|
|
|
ret = np.array([0.01, 0.02, -0.01, 0.03, -0.02])
|
|
labels = np.array([0, 0, 1, 1, -1], dtype=np.int64)
|
|
contrib = attribution_by_signal(ret, labels)
|
|
assert isinstance(contrib, dict)
|
|
assert "signal_0" in contrib
|
|
assert "signal_1" in contrib
|
|
assert abs(contrib["signal_0"] - 0.03) < 1e-10 # 0.01 + 0.02
|
|
assert abs(contrib["signal_1"] - 0.02) < 1e-10 # -0.01 + 0.03
|
|
|
|
def test_attribution_by_month_returns_dict(self):
|
|
from ferro_ta.analysis.attribution import attribution_by_month
|
|
|
|
ret = RNG.normal(0, 0.01, 252)
|
|
contrib = attribution_by_month(ret)
|
|
assert isinstance(contrib, dict)
|
|
assert len(contrib) > 0
|
|
|
|
def test_attribution_by_month_sum_close_to_total(self):
|
|
"""Sum of monthly contributions should approximate total strategy return."""
|
|
from ferro_ta.analysis.attribution import attribution_by_month
|
|
|
|
ret = RNG.normal(0, 0.01, 252)
|
|
contrib = attribution_by_month(ret)
|
|
total_monthly = sum(contrib.values())
|
|
total_direct = float(np.sum(ret))
|
|
assert abs(total_monthly - total_direct) < 1e-8
|
|
|
|
|
|
# ===========================================================================
|
|
# Dashboard (smoke tests, no display)
|
|
# ===========================================================================
|
|
|
|
|
|
class TestDashboard:
|
|
def test_streamlit_app_import(self):
|
|
"""Module should import without errors even if streamlit not installed."""
|
|
try:
|
|
from ferro_ta.tools import dashboard # noqa: F401
|
|
except ImportError:
|
|
pytest.skip("dashboard module not importable")
|
|
|
|
def test_indicator_widget_raises_without_ipywidgets(self, monkeypatch):
|
|
from ferro_ta import SMA
|
|
from ferro_ta.tools.dashboard import indicator_widget
|
|
|
|
close = _make_close(50)
|
|
# If ipywidgets not installed, should raise ImportError
|
|
import sys
|
|
|
|
fake_modules = dict(sys.modules)
|
|
fake_modules["ipywidgets"] = None # type: ignore[assignment]
|
|
fake_modules["matplotlib"] = None # type: ignore[assignment]
|
|
fake_modules["matplotlib.pyplot"] = None # type: ignore[assignment]
|
|
monkeypatch.setattr(sys, "modules", fake_modules)
|
|
with pytest.raises((ImportError, TypeError)):
|
|
indicator_widget(close, SMA, "timeperiod", range(5, 10))
|
|
|
|
|
|
# ===========================================================================
|
|
# Web API (unit test with TestClient if fastapi is available)
|
|
# ===========================================================================
|
|
|
|
|
|
class TestWebAPI:
|
|
@pytest.fixture(scope="class")
|
|
def client(self):
|
|
try:
|
|
from fastapi.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi not installed")
|
|
import os
|
|
import sys
|
|
|
|
# Insert project root so that `api.main` is importable
|
|
project_root = os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
if project_root not in sys.path:
|
|
sys.path.insert(0, project_root)
|
|
try:
|
|
from api.main import app
|
|
except ImportError:
|
|
pytest.skip("api/main.py not importable")
|
|
return TestClient(app)
|
|
|
|
def test_health(self, client):
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
|
|
def test_sma_endpoint(self, client):
|
|
close = list(np.linspace(100, 110, 30))
|
|
resp = client.post("/indicators/sma", json={"close": close, "timeperiod": 5})
|
|
assert resp.status_code == 200
|
|
result = resp.json()["result"]
|
|
assert len(result) == 30
|
|
assert result[0] is None # warm-up is null
|
|
|
|
def test_ema_endpoint(self, client):
|
|
close = list(np.linspace(100, 110, 30))
|
|
resp = client.post("/indicators/ema", json={"close": close, "timeperiod": 5})
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["result"]) == 30
|
|
|
|
def test_rsi_endpoint(self, client):
|
|
close = list(np.linspace(100, 110, 30))
|
|
resp = client.post("/indicators/rsi", json={"close": close, "timeperiod": 14})
|
|
assert resp.status_code == 200
|
|
|
|
def test_macd_endpoint(self, client):
|
|
close = list(np.linspace(100, 120, 60))
|
|
resp = client.post("/indicators/macd", json={"close": close})
|
|
assert resp.status_code == 200
|
|
keys = resp.json()["result"].keys()
|
|
assert {"macd", "signal", "hist"} == set(keys)
|
|
|
|
def test_bbands_endpoint(self, client):
|
|
close = list(np.linspace(100, 110, 30))
|
|
resp = client.post("/indicators/bbands", json={"close": close, "timeperiod": 5})
|
|
assert resp.status_code == 200
|
|
keys = resp.json()["result"].keys()
|
|
assert {"upper", "middle", "lower"} == set(keys)
|
|
|
|
def test_backtest_endpoint(self, client):
|
|
close = list(
|
|
np.cumprod(1 + np.random.default_rng(0).normal(0, 0.01, 100)) * 100
|
|
)
|
|
resp = client.post(
|
|
"/backtest",
|
|
json={"close": close, "strategy": "rsi_30_70"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "final_equity" in body
|
|
assert "n_trades" in body
|
|
|
|
def test_unknown_strategy_returns_422(self, client):
|
|
close = list(np.linspace(100, 110, 30))
|
|
resp = client.post(
|
|
"/backtest",
|
|
json={"close": close, "strategy": "no_such_strategy"},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
def test_too_short_series_returns_422(self, client):
|
|
resp = client.post("/indicators/sma", json={"close": [100.0], "timeperiod": 5})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
# ===========================================================================
|
|
# Benchmark suite sanity
|
|
# ===========================================================================
|
|
|
|
|
|
class TestBenchmarkSuite:
|
|
def test_canonical_fixture_exists(self):
|
|
import pathlib
|
|
|
|
fixture = (
|
|
pathlib.Path(__file__).parent.parent.parent
|
|
/ "benchmarks"
|
|
/ "fixtures"
|
|
/ "canonical_ohlcv.npz"
|
|
)
|
|
assert fixture.exists(), f"Canonical fixture not found: {fixture}"
|
|
|
|
def test_canonical_fixture_loadable(self):
|
|
import pathlib
|
|
|
|
fixture = (
|
|
pathlib.Path(__file__).parent.parent.parent
|
|
/ "benchmarks"
|
|
/ "fixtures"
|
|
/ "canonical_ohlcv.npz"
|
|
)
|
|
if not fixture.exists():
|
|
pytest.skip("Canonical fixture not found")
|
|
data = np.load(fixture)
|
|
for key in ["open", "high", "low", "close", "volume"]:
|
|
assert key in data.files, f"Missing key '{key}' in fixture"
|
|
assert len(data["close"]) == 2000
|
|
|
|
def test_benchmark_indicators_run(self):
|
|
import pathlib
|
|
|
|
fixture = (
|
|
pathlib.Path(__file__).parent.parent.parent
|
|
/ "benchmarks"
|
|
/ "fixtures"
|
|
/ "canonical_ohlcv.npz"
|
|
)
|
|
if not fixture.exists():
|
|
pytest.skip("Canonical fixture not found")
|
|
import ferro_ta as ft
|
|
|
|
data = np.load(fixture)
|
|
close = data["close"]
|
|
high = data["high"]
|
|
low = data["low"]
|
|
|
|
out_sma = np.asarray(ft.SMA(close, timeperiod=20))
|
|
out_rsi = np.asarray(ft.RSI(close, timeperiod=14))
|
|
out_atr = np.asarray(ft.ATR(high, low, close, timeperiod=14))
|
|
|
|
assert len(out_sma) == len(close)
|
|
assert len(out_rsi) == len(close)
|
|
assert len(out_atr) == len(close)
|
|
# Last value should be finite
|
|
assert np.isfinite(out_sma[-1])
|
|
assert np.isfinite(out_rsi[-1])
|
|
assert np.isfinite(out_atr[-1])
|
|
|
|
|
|
# ===========================================================================
|
|
# Options / IV helpers
|
|
# ===========================================================================
|
|
|
|
|
|
class TestIVRank:
|
|
def test_basic_shape(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
iv = _make_close(100)
|
|
result = iv_rank(iv, window=20)
|
|
assert result.shape == (100,)
|
|
|
|
def test_warmup_nan(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
iv = _make_close(50)
|
|
result = iv_rank(iv, window=10)
|
|
assert np.all(np.isnan(result[:9]))
|
|
assert not np.isnan(result[9])
|
|
|
|
def test_values_in_0_1(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
iv = _make_close(100)
|
|
result = iv_rank(iv, window=20)
|
|
valid = result[~np.isnan(result)]
|
|
assert np.all(valid >= 0.0)
|
|
assert np.all(valid <= 1.0)
|
|
|
|
def test_max_value_is_1(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
# The maximum of a window should produce rank = 1
|
|
iv = np.array([10.0, 20.0, 30.0, 40.0, 50.0])
|
|
result = iv_rank(iv, window=5)
|
|
assert result[4] == pytest.approx(1.0)
|
|
|
|
def test_min_value_is_0(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
iv = np.array([50.0, 40.0, 30.0, 20.0, 10.0])
|
|
result = iv_rank(iv, window=5)
|
|
assert result[4] == pytest.approx(0.0)
|
|
|
|
def test_empty_raises(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
with pytest.raises(Exception):
|
|
iv_rank(np.array([]), window=5)
|
|
|
|
def test_window_1(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
iv = np.array([10.0, 20.0, 30.0])
|
|
result = iv_rank(iv, window=1)
|
|
# With window=1, all values are equal to min=max, so rank=0
|
|
assert np.all(result == 0.0)
|
|
|
|
def test_invalid_window_raises(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
with pytest.raises(Exception):
|
|
iv_rank(np.array([1.0, 2.0]), window=0)
|
|
|
|
def test_flat_series(self):
|
|
from ferro_ta.analysis.options import iv_rank
|
|
|
|
iv = np.ones(30) * 25.0
|
|
result = iv_rank(iv, window=10)
|
|
valid = result[~np.isnan(result)]
|
|
assert np.all(valid == 0.0)
|
|
|
|
|
|
class TestIVPercentile:
|
|
def test_basic_shape(self):
|
|
from ferro_ta.analysis.options import iv_percentile
|
|
|
|
iv = _make_close(100)
|
|
result = iv_percentile(iv, window=20)
|
|
assert result.shape == (100,)
|
|
|
|
def test_warmup_nan(self):
|
|
from ferro_ta.analysis.options import iv_percentile
|
|
|
|
iv = _make_close(50)
|
|
result = iv_percentile(iv, window=10)
|
|
assert np.all(np.isnan(result[:9]))
|
|
|
|
def test_values_in_0_1(self):
|
|
from ferro_ta.analysis.options import iv_percentile
|
|
|
|
iv = _make_close(100)
|
|
result = iv_percentile(iv, window=20)
|
|
valid = result[~np.isnan(result)]
|
|
assert np.all(valid >= 0.0)
|
|
assert np.all(valid <= 1.0)
|
|
|
|
def test_empty_raises(self):
|
|
from ferro_ta.analysis.options import iv_percentile
|
|
|
|
with pytest.raises(Exception):
|
|
iv_percentile(np.array([]), window=5)
|
|
|
|
def test_known_value(self):
|
|
from ferro_ta.analysis.options import iv_percentile
|
|
|
|
iv = np.array([10.0, 20.0, 30.0, 15.0, 22.0])
|
|
result = iv_percentile(iv, window=3)
|
|
# At index 2: window=[10,20,30], current=30. All 3 <= 30 → 3/3 = 1.0
|
|
assert result[2] == pytest.approx(1.0)
|
|
# At index 3: window=[20,30,15], current=15. Only 15 <= 15 → 1/3
|
|
assert result[3] == pytest.approx(1.0 / 3.0)
|
|
|
|
|
|
class TestIVZScore:
|
|
def test_basic_shape(self):
|
|
from ferro_ta.analysis.options import iv_zscore
|
|
|
|
iv = _make_close(100)
|
|
result = iv_zscore(iv, window=20)
|
|
assert result.shape == (100,)
|
|
|
|
def test_warmup_nan(self):
|
|
from ferro_ta.analysis.options import iv_zscore
|
|
|
|
iv = _make_close(50)
|
|
result = iv_zscore(iv, window=10)
|
|
assert np.all(np.isnan(result[:9]))
|
|
|
|
def test_flat_is_nan(self):
|
|
from ferro_ta.analysis.options import iv_zscore
|
|
|
|
# Flat series has std=0, so z-score should be NaN
|
|
iv = np.ones(30) * 20.0
|
|
result = iv_zscore(iv, window=10)
|
|
valid = result[~np.isnan(result)]
|
|
assert len(valid) == 0 or np.all(np.isnan(valid))
|
|
|
|
def test_empty_raises(self):
|
|
from ferro_ta.analysis.options import iv_zscore
|
|
|
|
with pytest.raises(Exception):
|
|
iv_zscore(np.array([]), window=5)
|
|
|
|
def test_known_value(self):
|
|
from ferro_ta.analysis.options import iv_zscore
|
|
|
|
iv = np.array([10.0, 20.0, 30.0])
|
|
result = iv_zscore(iv, window=3)
|
|
# mean=20, std=std([10,20,30],ddof=0)=8.165...
|
|
expected = (30.0 - 20.0) / np.std([10.0, 20.0, 30.0], ddof=0)
|
|
assert result[2] == pytest.approx(expected, rel=1e-6)
|
|
|
|
|
|
# ===========================================================================
|
|
# Agentic tools and workflow
|
|
# ===========================================================================
|
|
|
|
|
|
class TestComputeIndicator:
|
|
def test_sma_basic(self):
|
|
from ferro_ta.tools import compute_indicator
|
|
|
|
close = np.linspace(100, 110, 20)
|
|
result = compute_indicator("SMA", close, timeperiod=5)
|
|
assert isinstance(result, np.ndarray)
|
|
assert result.shape == (20,)
|
|
|
|
def test_rsi_basic(self):
|
|
from ferro_ta.tools import compute_indicator
|
|
|
|
close = _make_close(100)
|
|
result = compute_indicator("RSI", close, timeperiod=14)
|
|
assert isinstance(result, np.ndarray)
|
|
assert result.shape == (100,)
|
|
|
|
def test_bbands_multi_output(self):
|
|
from ferro_ta.tools import compute_indicator
|
|
|
|
close = _make_close(50)
|
|
result = compute_indicator("BBANDS", close, timeperiod=10)
|
|
assert isinstance(result, dict)
|
|
assert "upper" in result
|
|
assert "middle" in result
|
|
assert "lower" in result
|
|
|
|
def test_macd_multi_output(self):
|
|
from ferro_ta.tools import compute_indicator
|
|
|
|
close = _make_close(100)
|
|
result = compute_indicator(
|
|
"MACD", close, fastperiod=5, slowperiod=10, signalperiod=3
|
|
)
|
|
assert isinstance(result, dict)
|
|
assert "macd" in result
|
|
assert "signal" in result
|
|
assert "hist" in result
|
|
|
|
def test_unknown_indicator_raises(self):
|
|
from ferro_ta.tools import compute_indicator
|
|
|
|
with pytest.raises(Exception):
|
|
compute_indicator("NO_SUCH_INDICATOR", np.ones(20))
|
|
|
|
|
|
class TestRunBacktest:
|
|
def test_basic_result_shape(self):
|
|
from ferro_ta.tools import run_backtest
|
|
|
|
close = _make_close(200)
|
|
summary = run_backtest("rsi_30_70", close)
|
|
assert isinstance(summary, dict)
|
|
assert "final_equity" in summary
|
|
assert "n_trades" in summary
|
|
assert "n_bars" in summary
|
|
assert "equity" in summary
|
|
assert "signals" in summary
|
|
assert "max_drawdown" in summary
|
|
assert summary["n_bars"] == 200
|
|
assert isinstance(summary["final_equity"], float)
|
|
|
|
def test_equity_list(self):
|
|
from ferro_ta.tools import run_backtest
|
|
|
|
close = _make_close(100)
|
|
summary = run_backtest("rsi_30_70", close)
|
|
assert isinstance(summary["equity"], list)
|
|
assert len(summary["equity"]) == 100
|
|
|
|
def test_sma_crossover_strategy(self):
|
|
from ferro_ta.tools import run_backtest
|
|
|
|
close = _make_close(200)
|
|
summary = run_backtest("sma_crossover", close, fast=5, slow=20)
|
|
assert "final_equity" in summary
|
|
|
|
def test_macd_crossover_strategy(self):
|
|
from ferro_ta.tools import run_backtest
|
|
|
|
close = _make_close(200)
|
|
summary = run_backtest("macd_crossover", close)
|
|
assert "final_equity" in summary
|
|
|
|
def test_unknown_strategy_raises(self):
|
|
from ferro_ta.tools import run_backtest
|
|
|
|
with pytest.raises(Exception):
|
|
run_backtest("no_such_strategy", _make_close(100))
|
|
|
|
def test_max_drawdown_non_negative(self):
|
|
from ferro_ta.tools import run_backtest
|
|
|
|
close = _make_close(200)
|
|
summary = run_backtest("rsi_30_70", close)
|
|
assert summary["max_drawdown"] >= 0.0
|
|
|
|
|
|
class TestListIndicators:
|
|
def test_returns_list(self):
|
|
from ferro_ta.tools import list_indicators
|
|
|
|
names = list_indicators()
|
|
assert isinstance(names, list)
|
|
assert len(names) > 0
|
|
|
|
def test_contains_sma_rsi(self):
|
|
from ferro_ta.tools import list_indicators
|
|
|
|
names = list_indicators()
|
|
assert "SMA" in names
|
|
assert "RSI" in names
|
|
|
|
def test_sorted(self):
|
|
from ferro_ta.tools import list_indicators
|
|
|
|
names = list_indicators()
|
|
assert names == sorted(names)
|
|
|
|
|
|
class TestDescribeIndicator:
|
|
def test_returns_string(self):
|
|
from ferro_ta.tools import describe_indicator
|
|
|
|
desc = describe_indicator("SMA")
|
|
assert isinstance(desc, str)
|
|
assert len(desc) > 0
|
|
|
|
def test_unknown_raises(self):
|
|
from ferro_ta.tools import describe_indicator
|
|
|
|
with pytest.raises(Exception):
|
|
describe_indicator("NO_SUCH_INDICATOR")
|
|
|
|
|
|
class TestWorkflow:
|
|
def test_basic_indicators(self):
|
|
from ferro_ta.tools.workflow import Workflow
|
|
|
|
close = _make_close(200)
|
|
result = (
|
|
Workflow()
|
|
.add_indicator("sma_20", "SMA", timeperiod=20)
|
|
.add_indicator("rsi_14", "RSI", timeperiod=14)
|
|
.run(close)
|
|
)
|
|
assert "sma_20" in result
|
|
assert "rsi_14" in result
|
|
assert result["sma_20"].shape == (200,)
|
|
assert result["rsi_14"].shape == (200,)
|
|
|
|
def test_with_strategy(self):
|
|
from ferro_ta.tools.workflow import Workflow
|
|
|
|
close = _make_close(200)
|
|
result = (
|
|
Workflow()
|
|
.add_indicator("rsi_14", "RSI", timeperiod=14)
|
|
.add_strategy("rsi_30_70")
|
|
.run(close)
|
|
)
|
|
assert "backtest" in result
|
|
assert "final_equity" in result["backtest"]
|
|
|
|
def test_with_alert(self):
|
|
from ferro_ta.tools.workflow import Workflow
|
|
|
|
close = _make_close(200)
|
|
result = (
|
|
Workflow()
|
|
.add_indicator("rsi_14", "RSI", timeperiod=14)
|
|
.add_alert("rsi_14", level=30.0, direction=-1)
|
|
.run(close)
|
|
)
|
|
assert "rsi_14" in result
|
|
# Alert key should be present
|
|
alert_keys = [k for k in result if k.startswith("alert_")]
|
|
assert len(alert_keys) > 0
|
|
|
|
def test_empty_workflow(self):
|
|
from ferro_ta.tools.workflow import Workflow
|
|
|
|
close = _make_close(50)
|
|
result = Workflow().run(close)
|
|
assert isinstance(result, dict)
|
|
assert len(result) == 0
|
|
|
|
def test_multi_output_indicator(self):
|
|
from ferro_ta.tools.workflow import Workflow
|
|
|
|
close = _make_close(100)
|
|
result = Workflow().add_indicator("bb", "BBANDS", timeperiod=10).run(close)
|
|
assert "bb" in result
|
|
# BBANDS returns dict from compute_indicator
|
|
assert isinstance(result["bb"], dict)
|
|
|
|
|
|
class TestRunPipeline:
|
|
def test_basic_pipeline(self):
|
|
from ferro_ta.tools.workflow import run_pipeline
|
|
|
|
close = _make_close(200)
|
|
result = run_pipeline(
|
|
close,
|
|
indicators={
|
|
"sma_20": {"name": "SMA", "timeperiod": 20},
|
|
"rsi_14": {"name": "RSI", "timeperiod": 14},
|
|
},
|
|
)
|
|
assert "sma_20" in result
|
|
assert "rsi_14" in result
|
|
|
|
def test_with_strategy(self):
|
|
from ferro_ta.tools.workflow import run_pipeline
|
|
|
|
close = _make_close(200)
|
|
result = run_pipeline(
|
|
close,
|
|
indicators={"rsi_14": {"name": "RSI", "timeperiod": 14}},
|
|
strategy="rsi_30_70",
|
|
)
|
|
assert "backtest" in result
|
|
|
|
def test_no_indicators(self):
|
|
from ferro_ta.tools.workflow import run_pipeline
|
|
|
|
close = _make_close(100)
|
|
result = run_pipeline(close)
|
|
assert isinstance(result, dict)
|
|
|
|
def test_with_alert(self):
|
|
from ferro_ta.tools.workflow import run_pipeline
|
|
|
|
close = _make_close(200)
|
|
result = run_pipeline(
|
|
close,
|
|
indicators={"rsi_14": {"name": "RSI", "timeperiod": 14}},
|
|
alert_indicator="rsi_14",
|
|
alert_level=30.0,
|
|
alert_direction=-1,
|
|
)
|
|
assert "rsi_14" in result
|
|
alert_keys = [k for k in result if k.startswith("alert_")]
|
|
assert len(alert_keys) > 0
|
|
|
|
|
|
# ===========================================================================
|
|
# MCP server
|
|
# ===========================================================================
|
|
|
|
|
|
class TestMCPListTools:
|
|
def test_list_tools_returns_dict(self):
|
|
from ferro_ta.mcp import handle_list_tools
|
|
|
|
result = handle_list_tools()
|
|
assert isinstance(result, dict)
|
|
assert "tools" in result
|
|
|
|
def test_list_tools_has_required_tools(self):
|
|
from ferro_ta.mcp import handle_list_tools
|
|
|
|
result = handle_list_tools()
|
|
names = [t["name"] for t in result["tools"]]
|
|
assert len(names) > 250
|
|
for expected in (
|
|
"sma",
|
|
"ema",
|
|
"rsi",
|
|
"macd",
|
|
"backtest",
|
|
"SMA",
|
|
"compute_indicator",
|
|
"about",
|
|
"check_cross",
|
|
"TickAggregator",
|
|
"call_instance_method",
|
|
"call_stored_callable",
|
|
"delete_instance",
|
|
):
|
|
assert expected in names, f"Expected tool '{expected}' not found"
|
|
|
|
def test_each_tool_has_schema(self):
|
|
from ferro_ta.mcp import handle_list_tools
|
|
|
|
result = handle_list_tools()
|
|
for tool in result["tools"]:
|
|
assert "name" in tool
|
|
assert "description" in tool
|
|
assert "inputSchema" in tool
|
|
|
|
|
|
class TestMCPCallTool:
|
|
def test_sma_call(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(np.linspace(100, 110, 30))
|
|
result = handle_call_tool("sma", {"close": close, "timeperiod": 5})
|
|
assert "content" in result
|
|
import json
|
|
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert len(payload) == 30
|
|
|
|
def test_ema_call(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(np.linspace(100, 110, 30))
|
|
result = handle_call_tool("ema", {"close": close, "timeperiod": 5})
|
|
assert "content" in result
|
|
|
|
def test_rsi_call(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(_make_close(50))
|
|
result = handle_call_tool("rsi", {"close": close, "timeperiod": 14})
|
|
assert "content" in result
|
|
|
|
def test_macd_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(_make_close(100))
|
|
result = handle_call_tool("macd", {"close": close})
|
|
assert "content" in result
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert "macd" in payload
|
|
|
|
def test_backtest_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(_make_close(200))
|
|
result = handle_call_tool("backtest", {"close": close, "strategy": "rsi_30_70"})
|
|
assert "content" in result
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert "final_equity" in payload
|
|
assert "n_trades" in payload
|
|
|
|
def test_top_level_sma_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(np.linspace(100, 110, 30))
|
|
result = handle_call_tool("SMA", {"close": close, "timeperiod": 5})
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert len(payload) == 30
|
|
|
|
def test_compute_indicator_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(_make_close(100))
|
|
result = handle_call_tool(
|
|
"compute_indicator",
|
|
{
|
|
"name": "MACD",
|
|
"args": [close],
|
|
},
|
|
)
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert "macd" in payload
|
|
|
|
def test_about_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
result = handle_call_tool("about", {})
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert payload["indicator_count"] >= 200
|
|
assert payload["method_count"] >= 400
|
|
|
|
def test_check_cross_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
result = handle_call_tool(
|
|
"check_cross",
|
|
{
|
|
"fast": [1.0, 2.0, 3.0, 2.0, 1.0],
|
|
"slow": [2.0, 2.0, 2.0, 2.0, 2.0],
|
|
},
|
|
)
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert len(payload) == 5
|
|
|
|
def test_list_indicators_call(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
result = handle_call_tool("list_indicators", {})
|
|
assert "content" in result
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert isinstance(payload, list)
|
|
assert "SMA" in payload
|
|
|
|
def test_describe_indicator_call(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
result = handle_call_tool("describe_indicator", {"name": "SMA"})
|
|
assert "content" in result
|
|
text = result["content"][0]["text"]
|
|
assert isinstance(text, str)
|
|
assert len(text) > 0
|
|
|
|
def test_unknown_tool_returns_error(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
result = handle_call_tool("no_such_tool", {})
|
|
assert result.get("isError") is True
|
|
|
|
def test_tool_error_handling(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
# Pass an invalid series to trigger an error
|
|
result = handle_call_tool("sma", {"close": [], "timeperiod": 5})
|
|
# Should return error content, not raise
|
|
assert "content" in result or "isError" in result
|
|
|
|
def test_backtest_unknown_strategy(self):
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
close = list(_make_close(100))
|
|
result = handle_call_tool(
|
|
"backtest", {"close": close, "strategy": "no_strategy"}
|
|
)
|
|
assert result.get("isError") is True or "content" in result
|
|
|
|
def test_tick_aggregator_instance_lifecycle(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
created = json.loads(
|
|
handle_call_tool("TickAggregator", {"rule": "tick:2"})["content"][0]["text"]
|
|
)
|
|
instance_id = created["instance_id"]
|
|
|
|
described = json.loads(
|
|
handle_call_tool("describe_instance", {"instance_id": instance_id})[
|
|
"content"
|
|
][0]["text"]
|
|
)
|
|
method_names = [item["name"] for item in described["methods"]]
|
|
assert "aggregate" in method_names
|
|
|
|
aggregated = json.loads(
|
|
handle_call_tool(
|
|
"call_instance_method",
|
|
{
|
|
"instance_id": instance_id,
|
|
"method": "aggregate",
|
|
"args": [
|
|
{
|
|
"price": [1.0, 2.0, 3.0, 4.0],
|
|
"size": [1.0, 1.0, 1.0, 1.0],
|
|
}
|
|
],
|
|
},
|
|
)["content"][0]["text"]
|
|
)
|
|
assert "open" in aggregated
|
|
assert "close" in aggregated
|
|
|
|
deleted = json.loads(
|
|
handle_call_tool("delete_instance", {"instance_id": instance_id})[
|
|
"content"
|
|
][0]["text"]
|
|
)
|
|
assert deleted["deleted"] is True
|
|
|
|
def test_stored_callable_can_be_invoked(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
wrapped = json.loads(
|
|
handle_call_tool("traced", {"func": {"callable": "SMA"}})["content"][0][
|
|
"text"
|
|
]
|
|
)
|
|
instance_id = wrapped["instance_id"]
|
|
|
|
called = json.loads(
|
|
handle_call_tool(
|
|
"call_stored_callable",
|
|
{
|
|
"instance_id": instance_id,
|
|
"args": [[1.0, 2.0, 3.0, 4.0, 5.0]],
|
|
"kwargs": {"timeperiod": 3},
|
|
},
|
|
)["content"][0]["text"]
|
|
)
|
|
assert len(called) == 5
|
|
|
|
handle_call_tool("delete_instance", {"instance_id": instance_id})
|
|
|
|
def test_benchmark_accepts_callable_reference(self):
|
|
import json
|
|
|
|
from ferro_ta.mcp import handle_call_tool
|
|
|
|
result = handle_call_tool(
|
|
"benchmark",
|
|
{
|
|
"func": {"callable": "SMA"},
|
|
"args": [[1.0, 2.0, 3.0, 4.0, 5.0]],
|
|
"kwargs": {"timeperiod": 3},
|
|
"n": 2,
|
|
"warmup": 0,
|
|
},
|
|
)
|
|
payload = json.loads(result["content"][0]["text"])
|
|
assert payload["n"] == 2.0
|
|
assert "mean_ms" in payload
|
|
|
|
|
|
class TestMCPServer:
|
|
def test_create_server_requires_mcp_dependency(self, monkeypatch):
|
|
import ferro_ta.mcp as mcp_mod
|
|
|
|
real_import_module = importlib.import_module
|
|
|
|
def fake_import_module(name, package=None):
|
|
if name.startswith("mcp"):
|
|
raise ImportError("No module named 'mcp'")
|
|
return real_import_module(name, package)
|
|
|
|
mcp_mod.create_server.cache_clear()
|
|
monkeypatch.setattr(importlib, "import_module", fake_import_module)
|
|
|
|
with pytest.raises(RuntimeError, match='pip install "ferro-ta\\[mcp\\]"'):
|
|
mcp_mod.create_server()
|
|
|
|
def test_main_entrypoint_invokes_run_server(self, monkeypatch):
|
|
import ferro_ta.mcp as mcp_mod
|
|
|
|
calls: list[str] = []
|
|
|
|
monkeypatch.setattr(mcp_mod, "run_server", lambda: calls.append("called"))
|
|
runpy.run_module("ferro_ta.mcp.__main__", run_name="__main__")
|
|
|
|
assert calls == ["called"]
|
|
|
|
def test_create_server_registers_generated_tools(self):
|
|
import ferro_ta.mcp as mcp_mod
|
|
|
|
server = mcp_mod.create_server()
|
|
tool_names = [tool.name for tool in server._tool_manager.list_tools()]
|
|
|
|
assert "SMA" in tool_names
|
|
assert "TickAggregator" in tool_names
|
|
assert "call_instance_method" in tool_names
|