File size: 6,793 Bytes
3339913 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """Engine correctness: alignment, costs, and the no-look-ahead guarantee."""
from __future__ import annotations
import numpy as np
import pandas as pd
import pytest
from algotrader.engine import bars_to_returns, run_backtest
from algotrader.metrics import compute_metrics, infer_periods_per_year, max_drawdown, sharpe_ratio
from algotrader.types import CostModel
def make_bars(n: int = 400, seed: int = 0) -> pd.DataFrame:
rng = np.random.default_rng(seed)
close = 100 * np.exp(np.cumsum(rng.normal(0.0003, 0.01, n)))
index = pd.date_range("2020-01-01", periods=n, freq="B")
return pd.DataFrame(
{
"open": close,
"high": close * 1.005,
"low": close * 0.995,
"close": close,
"volume": 1e6,
},
index=index,
)
class TestNoLookAhead:
"""The one property the whole project rests on."""
def test_signal_earns_the_following_bar_not_its_own(self):
df = make_bars()
asset_ret = bars_to_returns(df)
# A target that knows the *next* bar's direction must be perfect...
clairvoyant = np.sign(asset_ret.shift(-1)).fillna(0.0)
good = run_backtest(df, clairvoyant, costs=CostModel(0, 0, 0))
assert (good.returns.iloc[1:-1] >= -1e-12).all(), "perfect foresight should never lose"
# ...and a target that only knows the *current* bar must not be.
hindsight = np.sign(asset_ret).fillna(0.0)
meh = run_backtest(df, hindsight, costs=CostModel(0, 0, 0))
assert (meh.returns < 0).any(), "same-bar signal must not be risk-free"
assert good.sharpe > meh.sharpe
def test_position_is_target_shifted_by_lag(self):
df = make_bars(200)
target = pd.Series(np.linspace(-1, 1, len(df)), index=df.index)
for lag in (1, 2, 5):
result = run_backtest(df, target, lag=lag)
expected = target.shift(lag).fillna(0.0)
pd.testing.assert_series_equal(result.position, expected, check_names=False)
def test_lag_zero_is_rejected(self):
df = make_bars(120)
with pytest.raises(ValueError, match="lag"):
run_backtest(df, pd.Series(1.0, index=df.index), lag=0)
def test_future_bars_cannot_change_past_equity(self):
"""Truncating the data must not alter the equity curve before the cut."""
df = make_bars(400)
target = pd.Series(np.tile([1.0, -1.0], len(df) // 2), index=df.index)
full = run_backtest(df, target)
cut = run_backtest(df.iloc[:250], target.iloc[:250])
np.testing.assert_allclose(
full.equity.iloc[:250].to_numpy(), cut.equity.to_numpy(), rtol=1e-12
)
class TestCosts:
def test_buy_and_hold_pays_once(self):
df = make_bars(300)
target = pd.Series(1.0, index=df.index)
result = run_backtest(df, target, costs=CostModel(commission_bps=5, slippage_bps=5, short_borrow_bps=0))
# One entry at 10bps one-way, and no further turnover.
assert result.costs.sum() == pytest.approx(10 / 1e4, rel=1e-9)
assert int(result.metrics["n_trades"]) == 1
def test_flipping_every_bar_costs_more_than_holding(self):
df = make_bars(300)
costs = CostModel(commission_bps=5, slippage_bps=5, short_borrow_bps=0)
hold = run_backtest(df, pd.Series(1.0, index=df.index), costs=costs)
flip = run_backtest(df, pd.Series(np.tile([1.0, -1.0], 150), index=df.index), costs=costs)
assert flip.costs.sum() > 100 * hold.costs.sum()
def test_zero_costs_means_gross_equals_net(self):
df = make_bars(200)
target = pd.Series(np.tile([1.0, 0.0], 100), index=df.index)
result = run_backtest(df, target, costs=CostModel(0, 0, 0))
pd.testing.assert_series_equal(result.returns, result.gross_returns, check_names=False)
def test_short_borrow_is_charged_only_on_shorts(self):
df = make_bars(260)
costs = CostModel(commission_bps=0, slippage_bps=0, short_borrow_bps=365)
long_only = run_backtest(df, pd.Series(1.0, index=df.index), costs=costs)
short_only = run_backtest(df, pd.Series(-1.0, index=df.index), costs=costs)
assert long_only.costs.sum() == pytest.approx(0.0, abs=1e-12)
assert short_only.costs.sum() > 0
def test_higher_costs_never_improve_returns(self):
df = make_bars(300)
target = pd.Series(np.tile([1.0, -1.0], 150), index=df.index)
cheap = run_backtest(df, target, costs=CostModel(1, 1, 0))
dear = run_backtest(df, target, costs=CostModel(20, 20, 0))
assert dear.equity.iloc[-1] < cheap.equity.iloc[-1]
class TestConstraints:
def test_shorts_are_clipped_when_disallowed(self):
df = make_bars(150)
target = pd.Series(-1.0, index=df.index)
result = run_backtest(df, target, allow_short=False)
assert (result.target >= 0).all()
assert (result.position >= 0).all()
def test_leverage_is_clipped(self):
df = make_bars(150)
result = run_backtest(df, pd.Series(5.0, index=df.index), max_leverage=1.5)
assert result.target.max() == pytest.approx(1.5)
def test_empty_frame_is_rejected(self):
with pytest.raises(ValueError):
run_backtest(pd.DataFrame(columns=["open", "high", "low", "close", "volume"]), pd.Series(dtype=float))
class TestMetrics:
def test_sharpe_of_constant_returns_is_zero_not_infinite(self):
flat = pd.Series([0.001] * 100)
assert sharpe_ratio(flat, 252) == 0.0
def test_sharpe_scales_with_annualisation(self):
rng = np.random.default_rng(1)
returns = pd.Series(rng.normal(0.001, 0.01, 5000))
assert sharpe_ratio(returns, 252) == pytest.approx(sharpe_ratio(returns, 1) * np.sqrt(252))
def test_max_drawdown_matches_a_hand_worked_example(self):
equity = pd.Series([100.0, 120.0, 60.0, 90.0])
assert max_drawdown(equity) == pytest.approx(-0.5)
def test_buy_and_hold_metrics_match_the_price_series(self):
df = make_bars(500)
result = run_backtest(df, pd.Series(1.0, index=df.index), costs=CostModel(0, 0, 0))
expected = df["close"].iloc[-1] / df["close"].iloc[0] - 1.0
assert result.metrics["total_return"] == pytest.approx(expected, rel=1e-9)
def test_periodicity_inference(self):
daily = pd.date_range("2020-01-01", periods=300, freq="B")
assert 200 <= infer_periods_per_year(daily) <= 300
hourly = pd.date_range("2020-01-01", periods=300, freq="h")
assert infer_periods_per_year(hourly) > 1000
def test_metrics_survive_a_degenerate_series(self):
empty = pd.Series(dtype=float)
out = compute_metrics(empty, pd.Series(dtype=float))
assert "periods_per_year" in out
|