210 lines
7.3 KiB
Python
210 lines
7.3 KiB
Python
"""Backtest contract tests for weights, NAV, rebalancing, and benchmarks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from quant_engine.backtest import (
|
|
BacktestResult,
|
|
compare_to_benchmark,
|
|
compute_nav_from_weights,
|
|
compute_returns_from_nav,
|
|
rebalance_periodic,
|
|
run_weight_backtest,
|
|
weights_to_long_short,
|
|
)
|
|
|
|
|
|
def test_compute_nav_from_weights_forward_fills_rebalance_weights() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=3, freq="B")
|
|
weights = pd.DataFrame({"A": [0.5], "B": [0.5]}, index=dates[:1])
|
|
returns = pd.DataFrame({"A": [0.10, 0.00, -0.10], "B": [0.00, 0.10, 0.00]}, index=dates)
|
|
|
|
nav = compute_nav_from_weights(weights, returns, initial_capital=100.0)
|
|
|
|
expected = pd.Series([105.0, 110.25, 104.7375], index=dates)
|
|
pd.testing.assert_series_equal(nav, expected)
|
|
|
|
|
|
def test_compute_nav_stays_in_cash_before_first_rebalance() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=3, freq="B")
|
|
weights = pd.DataFrame({"A": [1.0]}, index=dates[1:2])
|
|
returns = pd.DataFrame({"A": [0.50, 0.10, 0.10]}, index=dates)
|
|
|
|
nav = compute_nav_from_weights(weights, returns)
|
|
|
|
pd.testing.assert_series_equal(nav, pd.Series([1.0, 1.1, 1.21], index=dates))
|
|
|
|
|
|
def test_compute_nav_ignores_weight_columns_without_returns() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=2, freq="B")
|
|
weights = pd.DataFrame({"A": [0.5], "MISSING": [0.5]}, index=dates[:1])
|
|
returns = pd.DataFrame({"A": [0.10, 0.10]}, index=dates)
|
|
|
|
nav = compute_nav_from_weights(weights, returns)
|
|
|
|
pd.testing.assert_series_equal(nav, pd.Series([1.05, 1.1025], index=dates))
|
|
|
|
|
|
def test_compute_nav_charges_configured_turnover_cost() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=2, freq="B")
|
|
weights = pd.DataFrame({"A": [1.0]}, index=dates[:1])
|
|
returns = pd.DataFrame({"A": [0.0, 0.0]}, index=dates)
|
|
|
|
nav = compute_nav_from_weights(weights, returns, tc_rate=0.01)
|
|
|
|
pd.testing.assert_series_equal(nav, pd.Series([0.995, 0.995], index=dates))
|
|
|
|
|
|
def test_compute_returns_from_nav_preserves_index_and_sets_initial_zero() -> None:
|
|
nav = pd.Series([100.0, 110.0, 99.0], index=pd.date_range("2026-01-05", periods=3))
|
|
|
|
result = compute_returns_from_nav(nav)
|
|
|
|
pd.testing.assert_series_equal(result, pd.Series([0.0, 0.1, -0.1], index=nav.index))
|
|
|
|
|
|
def test_rebalance_periodic_maps_weekend_to_previous_trading_day() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=5, freq="B")
|
|
target = pd.Series({"A": 0.6, "B": 0.4})
|
|
|
|
result = rebalance_periodic(target, [pd.Timestamp("2026-01-10")], dates)
|
|
|
|
assert result.loc[pd.Timestamp("2026-01-08")].sum() == 0.0
|
|
pd.testing.assert_series_equal(
|
|
result.loc[pd.Timestamp("2026-01-09")], target, check_names=False
|
|
)
|
|
|
|
|
|
def test_rebalance_periodic_accepts_empty_trading_calendar() -> None:
|
|
target = pd.Series({"A": 1.0})
|
|
|
|
result = rebalance_periodic(
|
|
target,
|
|
[pd.Timestamp("2026-01-05")],
|
|
pd.DatetimeIndex([]),
|
|
)
|
|
|
|
assert result.empty
|
|
assert result.columns.tolist() == ["A"]
|
|
|
|
|
|
def test_weights_to_long_short_allocates_each_leg() -> None:
|
|
result = weights_to_long_short(["A", "B"], ["C"], long_weight=0.6, short_weight=0.4)
|
|
|
|
assert result["A"] == pytest.approx(0.3)
|
|
assert result["B"] == pytest.approx(0.3)
|
|
assert result["C"] == pytest.approx(-0.4)
|
|
assert result.sum() == pytest.approx(0.2)
|
|
|
|
|
|
def test_weights_to_long_short_keeps_explicit_universe() -> None:
|
|
result = weights_to_long_short(["A"], [], all_tickers=["A", "B"])
|
|
|
|
pd.testing.assert_series_equal(result, pd.Series({"A": 0.5, "B": 0.0}))
|
|
|
|
|
|
def test_compare_to_benchmark_returns_report_table() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=4, freq="B")
|
|
strategy = pd.Series([1.0, 1.1, 1.0, 1.2], index=dates)
|
|
benchmark = pd.Series([1.0, 1.0, 1.05, 1.1], index=dates)
|
|
|
|
result = compare_to_benchmark(strategy, benchmark)
|
|
|
|
assert result.columns.tolist() == ["策略", "基准"]
|
|
assert result.loc["n_days", "策略"] == 4
|
|
assert result.loc["累计收益", "策略"] == pytest.approx(0.2)
|
|
assert result.loc["累计收益", "基准"] == pytest.approx(0.1)
|
|
|
|
|
|
def test_compare_to_benchmark_rejects_non_overlapping_dates() -> None:
|
|
strategy = pd.Series([1.0], index=[pd.Timestamp("2026-01-05")])
|
|
benchmark = pd.Series([1.0], index=[pd.Timestamp("2026-02-05")])
|
|
|
|
with pytest.raises(ValueError, match="overlapping dates"):
|
|
compare_to_benchmark(strategy, benchmark)
|
|
|
|
|
|
# ── 统一回测结果门面 ──────────────────────────────────────
|
|
|
|
|
|
def test_run_weight_backtest_returns_nav_returns_and_input_snapshot() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=3, freq="B")
|
|
weights = pd.DataFrame({"A": [1.0]}, index=dates[:1])
|
|
stock_returns = pd.DataFrame({"A": [0.10, -0.10, 0.20]}, index=dates)
|
|
|
|
result = run_weight_backtest(weights, stock_returns, initial_capital=100.0)
|
|
|
|
assert isinstance(result, BacktestResult)
|
|
pd.testing.assert_series_equal(
|
|
result.nav,
|
|
pd.Series([110.0, 99.0, 118.8], index=dates),
|
|
)
|
|
pd.testing.assert_series_equal(
|
|
result.returns,
|
|
pd.Series([0.0, -0.1, 0.2], index=dates),
|
|
)
|
|
pd.testing.assert_frame_equal(result.weights, weights)
|
|
|
|
|
|
def test_backtest_result_stats_reuses_standard_metrics_contract() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=3, freq="B")
|
|
result = run_weight_backtest(
|
|
pd.DataFrame({"A": [1.0]}, index=dates[:1]),
|
|
pd.DataFrame({"A": [0.10, -0.10, 0.20]}, index=dates),
|
|
)
|
|
|
|
stats = result.stats(rf=0.02)
|
|
|
|
assert stats["n_days"] == 3
|
|
assert stats["ann_return"] == pytest.approx(
|
|
(1.0 * 0.9 * 1.2) ** (252 / 3) - 1.0
|
|
)
|
|
assert "sharpe" in stats
|
|
assert stats["drawback"] == stats["max_drawdown"]
|
|
|
|
|
|
def test_backtest_result_builds_benchmark_report() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=3, freq="B")
|
|
benchmark = pd.Series([1.0, 1.05, 1.10], index=dates, name="benchmark")
|
|
result = run_weight_backtest(
|
|
pd.DataFrame({"A": [1.0]}, index=dates[:1]),
|
|
pd.DataFrame({"A": [0.10, -0.10, 0.20]}, index=dates),
|
|
benchmark_nav=benchmark,
|
|
)
|
|
|
|
report = result.benchmark_report()
|
|
|
|
assert report.columns.tolist() == ["策略", "基准"]
|
|
assert report.loc["累计收益", "基准"] == pytest.approx(0.10)
|
|
|
|
|
|
def test_backtest_result_requires_benchmark_for_comparison() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=2, freq="B")
|
|
result = run_weight_backtest(
|
|
pd.DataFrame({"A": [1.0]}, index=dates[:1]),
|
|
pd.DataFrame({"A": [0.0, 0.0]}, index=dates),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="benchmark_nav"):
|
|
result.benchmark_report()
|
|
|
|
|
|
def test_backtest_result_isolated_from_mutated_caller_inputs() -> None:
|
|
dates = pd.date_range("2026-01-05", periods=2, freq="B")
|
|
weights = pd.DataFrame({"A": [1.0]}, index=dates[:1])
|
|
benchmark = pd.Series([1.0, 1.1], index=dates)
|
|
result = run_weight_backtest(
|
|
weights,
|
|
pd.DataFrame({"A": [0.0, 0.0]}, index=dates),
|
|
benchmark_nav=benchmark,
|
|
)
|
|
|
|
weights.iloc[0, 0] = 0.0
|
|
benchmark.iloc[1] = 99.0
|
|
|
|
assert result.weights.iloc[0, 0] == 1.0
|
|
assert result.benchmark_nav is not None
|
|
assert result.benchmark_nav.iloc[1] == 1.1
|