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.
547 lines
19 KiB
Python
547 lines
19 KiB
Python
"""
|
|
v1.1.0 backtest feature tests.
|
|
|
|
Covers:
|
|
- CommissionModel: total_cost, presets, round-trip JSON, save/load
|
|
- Currency: INR/USD formatting, from_code lookup
|
|
- BacktestEngine: initial_capital, commission_model, trailing_stop, benchmark
|
|
- AdvancedBacktestResult: equity_abs, pnl_abs in trade log, summary fields
|
|
- Volatility-target position sizing
|
|
- Benchmark comparison metrics
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from ferro_ta._ferro_ta import CommissionModel
|
|
|
|
from ferro_ta.analysis.backtest import (
|
|
EUR,
|
|
GBP,
|
|
INR,
|
|
JPY,
|
|
USD,
|
|
USDT,
|
|
BacktestEngine,
|
|
Currency,
|
|
format_currency,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def close_500():
|
|
"""500-bar synthetic close price series."""
|
|
rng = np.random.default_rng(12345)
|
|
return np.cumprod(1.0 + rng.standard_normal(500) * 0.01) * 100.0
|
|
|
|
|
|
@pytest.fixture
|
|
def ohlcv_500(close_500):
|
|
close = close_500
|
|
high = close * 1.005
|
|
low = close * 0.995
|
|
open_ = close * 0.999
|
|
volume = np.full(len(close), 1_000_000.0)
|
|
return open_, high, low, close, volume
|
|
|
|
|
|
# ===========================================================================
|
|
# TestCommissionModel
|
|
# ===========================================================================
|
|
|
|
|
|
class TestCommissionModel:
|
|
def test_zero_model_costs_nothing(self):
|
|
m = CommissionModel.zero()
|
|
assert m.total_cost(100_000, 1, True) == 0.0
|
|
assert m.total_cost(100_000, 1, False) == 0.0
|
|
|
|
def test_flat_per_order(self):
|
|
m = CommissionModel()
|
|
m.flat_per_order = 20.0
|
|
assert m.total_cost(100_000, 1, True) == pytest.approx(20.0)
|
|
assert m.total_cost(100_000, 1, False) == pytest.approx(20.0)
|
|
|
|
def test_max_brokerage_cap(self):
|
|
m = CommissionModel()
|
|
m.flat_per_order = 0.0
|
|
m.rate_of_value = 0.001 # 0.1%
|
|
m.max_brokerage = 20.0
|
|
# 0.1% of 50_000 = 50, capped at 20
|
|
assert m.total_cost(50_000, 1, True) == pytest.approx(20.0)
|
|
# 0.1% of 5_000 = 5, not capped
|
|
assert m.total_cost(5_000, 1, True) == pytest.approx(5.0)
|
|
|
|
def test_stt_buy_side_only(self):
|
|
m = CommissionModel()
|
|
m.stt_rate = 0.001
|
|
m.stt_on_buy = True
|
|
m.stt_on_sell = False
|
|
buy_cost = m.total_cost(100_000, 1, True)
|
|
sell_cost = m.total_cost(100_000, 1, False)
|
|
assert buy_cost == pytest.approx(100.0)
|
|
assert sell_cost == pytest.approx(0.0)
|
|
|
|
def test_stt_sell_side_only(self):
|
|
m = CommissionModel()
|
|
m.stt_rate = 0.00025
|
|
m.stt_on_buy = False
|
|
m.stt_on_sell = True
|
|
buy_cost = m.total_cost(100_000, 1, True)
|
|
sell_cost = m.total_cost(100_000, 1, False)
|
|
assert buy_cost == pytest.approx(0.0)
|
|
assert sell_cost == pytest.approx(25.0)
|
|
|
|
def test_gst_on_brokerage_exchange_not_stt(self):
|
|
m = CommissionModel()
|
|
m.flat_per_order = 20.0
|
|
m.exchange_charges_rate = 0.0001
|
|
m.gst_rate = 0.18
|
|
m.stt_rate = 0.001
|
|
m.stt_on_sell = True
|
|
# GST = 0.18 * (20 + 0.0001 * 100_000) = 0.18 * 30 = 5.4
|
|
# STT = 100 (sell side)
|
|
total = m.total_cost(100_000, 1, False)
|
|
expected_gst = 0.18 * (20.0 + 0.0001 * 100_000)
|
|
assert total == pytest.approx(20.0 + 100.0 + 0.0001 * 100_000 + expected_gst)
|
|
|
|
def test_stamp_duty_buy_only(self):
|
|
m = CommissionModel()
|
|
m.stamp_duty_rate = 0.00015
|
|
buy_cost = m.total_cost(100_000, 1, True)
|
|
sell_cost = m.total_cost(100_000, 1, False)
|
|
assert buy_cost == pytest.approx(15.0)
|
|
assert sell_cost == pytest.approx(0.0)
|
|
|
|
def test_per_lot_charge(self):
|
|
m = CommissionModel()
|
|
m.per_lot = 2.0
|
|
# 5 lots
|
|
assert m.total_cost(50_000, 5, True) == pytest.approx(10.0)
|
|
|
|
def test_cost_fraction(self):
|
|
m = CommissionModel()
|
|
m.flat_per_order = 20.0
|
|
frac = m.cost_fraction(100_000, 1, True, 100_000.0)
|
|
assert frac == pytest.approx(20.0 / 100_000.0)
|
|
|
|
def test_cost_fraction_zero_capital(self):
|
|
m = CommissionModel()
|
|
m.flat_per_order = 20.0
|
|
assert m.cost_fraction(100_000, 1, True, 0.0) == 0.0
|
|
|
|
def test_proportional_preset(self):
|
|
m = CommissionModel.proportional(0.001)
|
|
assert m.total_cost(100_000, 1, True) == pytest.approx(100.0)
|
|
assert m.gst_rate == 0.0
|
|
|
|
def test_repr_contains_key_fields(self):
|
|
m = CommissionModel.equity_delivery_india()
|
|
r = repr(m)
|
|
assert "CommissionModel" in r
|
|
assert "lot_size" in r
|
|
|
|
|
|
class TestCommissionPresets:
|
|
def test_equity_delivery_india_smoke(self):
|
|
m = CommissionModel.equity_delivery_india()
|
|
# Buy ₹1L trade: brokerage cap ₹20, STT ₹100 (both sides)
|
|
cost = m.total_cost(100_000, 1, True)
|
|
assert cost > 0.0
|
|
assert cost < 500.0 # sanity upper bound
|
|
# Brokerage should be capped at ₹20
|
|
assert m.flat_per_order == 0.0
|
|
assert m.max_brokerage == pytest.approx(20.0)
|
|
assert m.stt_on_buy is True
|
|
assert m.stt_on_sell is True
|
|
|
|
def test_equity_intraday_india_smoke(self):
|
|
m = CommissionModel.equity_intraday_india()
|
|
cost_buy = m.total_cost(100_000, 1, True)
|
|
cost_sell = m.total_cost(100_000, 1, False)
|
|
# STT only on sell side for intraday
|
|
assert m.stt_on_buy is False
|
|
assert m.stt_on_sell is True
|
|
assert cost_sell > cost_buy # sell has more cost (STT)
|
|
|
|
def test_futures_india_smoke(self):
|
|
m = CommissionModel.futures_india()
|
|
assert m.flat_per_order == pytest.approx(20.0)
|
|
assert m.stt_on_buy is False
|
|
assert m.stt_on_sell is True
|
|
assert m.lot_size == pytest.approx(25.0)
|
|
|
|
def test_options_india_smoke(self):
|
|
m = CommissionModel.options_india()
|
|
assert m.flat_per_order == pytest.approx(20.0)
|
|
assert m.stt_rate == pytest.approx(0.0015)
|
|
assert m.lot_size == pytest.approx(25.0)
|
|
|
|
|
|
class TestCommissionFix:
|
|
"""The old 'commission_per_trade=20.0' bug would subtract ₹20 from 1.0-normalized
|
|
equity — a 2000% error. The new model correctly computes 0.02% fraction."""
|
|
|
|
def test_flat_20_on_1L_capital_is_tiny_fraction(self):
|
|
m = CommissionModel()
|
|
m.flat_per_order = 20.0
|
|
frac = m.cost_fraction(100_000, 1, True, 100_000.0)
|
|
# ₹20 / ₹100_000 = 0.02%
|
|
assert frac == pytest.approx(20.0 / 100_000.0, rel=1e-6)
|
|
assert frac < 0.01 # definitely not 2000%
|
|
|
|
def test_commission_reduces_equity_vs_no_commission(self):
|
|
rng = np.random.default_rng(99)
|
|
close = np.cumprod(1.0 + rng.standard_normal(200) * 0.01) * 100.0
|
|
m = CommissionModel.equity_intraday_india()
|
|
r_comm = (
|
|
BacktestEngine()
|
|
.with_commission_model(m)
|
|
.with_initial_capital(100_000)
|
|
.run(close, "sma_crossover")
|
|
)
|
|
r_none = (
|
|
BacktestEngine().with_initial_capital(100_000).run(close, "sma_crossover")
|
|
)
|
|
# Commission should reduce final equity (or keep equal if zero trades)
|
|
assert r_comm.final_equity <= r_none.final_equity
|
|
|
|
|
|
class TestCommissionSaveLoad:
|
|
def test_to_json_from_json_round_trip(self):
|
|
m = CommissionModel.equity_delivery_india()
|
|
j = m.to_json()
|
|
m2 = CommissionModel.from_json(j)
|
|
assert m == m2
|
|
assert m2.stt_rate == pytest.approx(m.stt_rate)
|
|
assert m2.lot_size == pytest.approx(m.lot_size)
|
|
assert m2.gst_rate == pytest.approx(m.gst_rate)
|
|
|
|
def test_save_load_round_trip(self):
|
|
m = CommissionModel.futures_india()
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
m.save(path)
|
|
assert os.path.exists(path)
|
|
m2 = CommissionModel.load(path)
|
|
assert m == m2
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_from_json_invalid_raises(self):
|
|
with pytest.raises(Exception):
|
|
CommissionModel.from_json("{invalid json")
|
|
|
|
def test_load_missing_file_raises(self):
|
|
with pytest.raises(Exception):
|
|
CommissionModel.load("/nonexistent/path/commission.json")
|
|
|
|
|
|
# ===========================================================================
|
|
# TestCurrency
|
|
# ===========================================================================
|
|
|
|
|
|
class TestCurrency:
|
|
def test_inr_lakh_grouping(self):
|
|
assert INR.format(123456.78) == "₹1,23,456.78"
|
|
assert INR.format(1000000.0) == "₹10,00,000.00"
|
|
assert INR.format(10000000.0) == "₹1,00,00,000.00"
|
|
assert INR.format(100.0) == "₹100.00"
|
|
assert INR.format(1234.5) == "₹1,234.50"
|
|
|
|
def test_inr_negative(self):
|
|
result = INR.format(-5000.0)
|
|
assert result.startswith("-₹")
|
|
assert "5,000.00" in result
|
|
|
|
def test_usd_standard_grouping(self):
|
|
assert USD.format(1234567.89) == "$1,234,567.89"
|
|
assert USD.format(0.5) == "$0.50"
|
|
assert USD.format(1000.0) == "$1,000.00"
|
|
|
|
def test_jpy_no_decimals(self):
|
|
result = JPY.format(1000000.0)
|
|
assert result == "¥1,000,000"
|
|
|
|
def test_eur_format(self):
|
|
assert "€" in EUR.format(100.0)
|
|
|
|
def test_gbp_format(self):
|
|
assert "£" in GBP.format(100.0)
|
|
|
|
def test_usdt_format(self):
|
|
assert "₮" in USDT.format(100.0)
|
|
|
|
def test_format_currency_helper(self):
|
|
assert format_currency(123456.78) == "₹1,23,456.78"
|
|
assert format_currency(1000.0, USD) == "$1,000.00"
|
|
|
|
def test_currency_immutable(self):
|
|
with pytest.raises(AttributeError):
|
|
INR.code = "USD" # type: ignore[misc]
|
|
|
|
def test_currency_equality(self):
|
|
c1 = Currency.from_code("INR")
|
|
assert c1 == INR
|
|
assert INR != USD
|
|
|
|
def test_currency_hash_usable_in_dict(self):
|
|
d = {INR: 100_000, USD: 100}
|
|
assert d[INR] == 100_000
|
|
|
|
|
|
# ===========================================================================
|
|
# TestInitialCapital
|
|
# ===========================================================================
|
|
|
|
|
|
class TestInitialCapital:
|
|
def test_equity_abs_shape(self, close_500):
|
|
result = (
|
|
BacktestEngine()
|
|
.with_initial_capital(200_000)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
assert result.equity_abs.shape == result.equity.shape
|
|
|
|
def test_equity_abs_is_equity_times_capital(self, close_500):
|
|
capital = 150_000.0
|
|
result = (
|
|
BacktestEngine()
|
|
.with_initial_capital(capital)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
np.testing.assert_allclose(result.equity_abs, result.equity * capital)
|
|
|
|
def test_summary_contains_capital_fields(self, close_500):
|
|
capital = 100_000.0
|
|
result = (
|
|
BacktestEngine()
|
|
.with_initial_capital(capital)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
s = result.summary()
|
|
assert "initial_capital" in s
|
|
assert "final_capital" in s
|
|
assert "absolute_pnl" in s
|
|
assert s["initial_capital"] == pytest.approx(capital)
|
|
assert s["final_capital"] == pytest.approx(result.equity_abs[-1])
|
|
assert s["absolute_pnl"] == pytest.approx(s["final_capital"] - capital)
|
|
|
|
def test_pnl_abs_in_trade_log(self, close_500, ohlcv_500):
|
|
open_, high, low, close, _ = ohlcv_500
|
|
capital = 100_000.0
|
|
result = (
|
|
BacktestEngine()
|
|
.with_initial_capital(capital)
|
|
.with_ohlcv(high=high, low=low, open_=open_)
|
|
.run(close, "sma_crossover")
|
|
)
|
|
if result.trades is not None and len(result.trades) > 0:
|
|
assert "pnl_abs" in result.trades.columns
|
|
np.testing.assert_allclose(
|
|
result.trades["pnl_abs"].values,
|
|
result.trades["pnl_pct"].values * capital,
|
|
)
|
|
|
|
|
|
class TestINRRepr:
|
|
def test_repr_shows_inr_symbol(self, close_500):
|
|
result = (
|
|
BacktestEngine()
|
|
.with_currency(INR)
|
|
.with_initial_capital(100_000)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
r = repr(result)
|
|
assert "₹" in r
|
|
|
|
def test_currency_code_in_summary(self, close_500):
|
|
result = (
|
|
BacktestEngine()
|
|
.with_currency("USD")
|
|
.with_initial_capital(10_000)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
s = result.summary()
|
|
assert s["currency"] == "USD"
|
|
|
|
def test_unknown_currency_raises(self):
|
|
with pytest.raises(Exception, match="Unknown currency"):
|
|
BacktestEngine().with_currency("XYZ")
|
|
|
|
|
|
# ===========================================================================
|
|
# TestVolatilityTargetSizing
|
|
# ===========================================================================
|
|
|
|
|
|
class TestVolatilityTargetSizing:
|
|
def test_vol_target_runs_without_error(self, close_500):
|
|
result = (
|
|
BacktestEngine()
|
|
.with_position_sizing("volatility_target", target_vol=0.10)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
assert len(result.equity) == len(close_500)
|
|
assert np.isfinite(result.final_equity)
|
|
|
|
def test_vol_target_signals_are_scaled(self, close_500):
|
|
# With very low target vol the strategy should have fewer active positions
|
|
result_low = (
|
|
BacktestEngine()
|
|
.with_position_sizing("volatility_target", target_vol=0.01)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
result_high = (
|
|
BacktestEngine()
|
|
.with_position_sizing("volatility_target", target_vol=1.0)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
# Lower vol target → lower absolute position sizes → lower annualised vol
|
|
low_std = float(np.nanstd(result_low.strategy_returns))
|
|
high_std = float(np.nanstd(result_high.strategy_returns))
|
|
assert low_std <= high_std or np.isclose(low_std, high_std, rtol=0.5)
|
|
|
|
|
|
# ===========================================================================
|
|
# TestBenchmark
|
|
# ===========================================================================
|
|
|
|
|
|
class TestBenchmark:
|
|
def test_benchmark_metrics_present(self, close_500):
|
|
rng = np.random.default_rng(77)
|
|
benchmark = np.cumprod(1.0 + rng.standard_normal(500) * 0.008) * 100.0
|
|
result = (
|
|
BacktestEngine().with_benchmark(benchmark).run(close_500, "sma_crossover")
|
|
)
|
|
s = result.summary()
|
|
assert "alpha" in s
|
|
assert "beta" in s
|
|
assert "tracking_error" in s
|
|
assert "information_ratio" in s
|
|
assert "benchmark_cagr" in s
|
|
|
|
def test_identical_strategy_benchmark_has_low_tracking_error(self, close_500):
|
|
# When strategy returns = benchmark returns, tracking error ≈ 0
|
|
# Use the equity as its own benchmark
|
|
result = (
|
|
BacktestEngine().with_benchmark(close_500).run(close_500, "sma_crossover")
|
|
)
|
|
m = result.metrics
|
|
# Beta should be finite
|
|
assert np.isfinite(m.get("beta", float("nan")))
|
|
|
|
def test_benchmark_wrong_length_ignored(self, close_500):
|
|
short_bench = close_500[:100]
|
|
# Should not raise — benchmark mismatch is silently ignored
|
|
result = (
|
|
BacktestEngine().with_benchmark(short_bench).run(close_500, "sma_crossover")
|
|
)
|
|
# alpha should NOT appear (length mismatch)
|
|
assert "alpha" not in result.metrics
|
|
|
|
|
|
# ===========================================================================
|
|
# TestTrailingStop
|
|
# ===========================================================================
|
|
|
|
|
|
class TestTrailingStop:
|
|
def test_trailing_stop_runs(self, ohlcv_500, close_500):
|
|
open_, high, low, close, _ = ohlcv_500
|
|
result = (
|
|
BacktestEngine()
|
|
.with_ohlcv(high=high, low=low, open_=open_)
|
|
.with_trailing_stop(0.02)
|
|
.run(close, "sma_crossover")
|
|
)
|
|
assert len(result.equity) == len(close)
|
|
assert np.isfinite(result.final_equity)
|
|
|
|
def test_trailing_stop_reduces_losses_on_downtrend(self):
|
|
"""Trailing stop should exit longs earlier on a falling market."""
|
|
# Construct a clear downtrend after initial rise
|
|
prices = np.concatenate(
|
|
[
|
|
np.linspace(100, 120, 50), # rise (signal stays long)
|
|
np.linspace(120, 60, 150), # sharp fall
|
|
]
|
|
)
|
|
high = prices * 1.002
|
|
low = prices * 0.998
|
|
open_ = prices * 0.999
|
|
|
|
result_trail = (
|
|
BacktestEngine()
|
|
.with_ohlcv(high=high, low=low, open_=open_)
|
|
.with_trailing_stop(0.03)
|
|
.run(prices, "sma_crossover")
|
|
)
|
|
result_no_trail = (
|
|
BacktestEngine()
|
|
.with_ohlcv(high=high, low=low, open_=open_)
|
|
.run(prices, "sma_crossover")
|
|
)
|
|
# Trailing stop should yield better (or equal) max drawdown
|
|
dd_trail = result_trail.metrics.get("max_drawdown", 0.0)
|
|
dd_no_trail = result_no_trail.metrics.get("max_drawdown", 0.0)
|
|
# max_drawdown is negative; higher value = smaller drawdown
|
|
assert dd_trail >= dd_no_trail - 0.05 # allow 5% tolerance
|
|
|
|
|
|
# ===========================================================================
|
|
# TestBacktestEngineChaining
|
|
# ===========================================================================
|
|
|
|
|
|
class TestBacktestEngineChaining:
|
|
def test_full_chain_runs(self, close_500, ohlcv_500):
|
|
open_, high, low, close, _ = ohlcv_500
|
|
rng = np.random.default_rng(42)
|
|
benchmark = np.cumprod(1.0 + rng.standard_normal(500) * 0.008) * 100.0
|
|
|
|
result = (
|
|
BacktestEngine()
|
|
.with_currency("INR")
|
|
.with_initial_capital(100_000)
|
|
.with_commission_model(CommissionModel.equity_intraday_india())
|
|
.with_trailing_stop(0.02)
|
|
.with_benchmark(benchmark)
|
|
.with_ohlcv(high=high, low=low, open_=open_)
|
|
.run(close, "sma_crossover")
|
|
)
|
|
assert len(result.equity) == len(close)
|
|
assert result.currency == INR
|
|
assert result.initial_capital == pytest.approx(100_000.0)
|
|
assert np.isfinite(result.final_equity)
|
|
|
|
s = result.summary()
|
|
assert s["currency"] == "INR"
|
|
assert "alpha" in s # benchmark was set
|
|
|
|
def test_to_equity_dataframe(self, close_500):
|
|
result = (
|
|
BacktestEngine()
|
|
.with_initial_capital(50_000)
|
|
.run(close_500, "sma_crossover")
|
|
)
|
|
df = result.to_equity_dataframe()
|
|
assert "equity" in df.columns
|
|
assert "equity_abs" in df.columns
|
|
assert "strategy_returns" in df.columns
|
|
assert "drawdown" in df.columns
|
|
assert len(df) == len(close_500)
|
|
np.testing.assert_allclose(df["equity_abs"].values, result.equity_abs)
|