"""Backtest contract tests for weights, NAV, rebalancing, and benchmarks.""" from __future__ import annotations import pandas as pd import pytest from quant_engine.backtest import ( compare_to_benchmark, compute_nav_from_weights, compute_returns_from_nav, rebalance_periodic, 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)