Compare commits

...
3 Commits
Author SHA1 Message Date
ao gong 979166ad16 docs: document unified backtest result workflow
CI / lite (pull_request) Successful in 3s
2026-08-21 21:11:52 +08:00
ao gong cdf41edf1f feat: add unified weight backtest result facade 2026-08-21 21:11:06 +08:00
ao gong 585a635797 test: add red contract for unified backtest result 2026-08-21 21:10:33 +08:00
3 changed files with 148 additions and 10 deletions
+11
View File
@@ -60,6 +60,7 @@ from quant_engine.alpha_factors import alpha_001, alpha_005, ALPHA158_REGISTRY
from quant_engine.execution import ( from quant_engine.execution import (
ExecutionConfig, simulate_with_daily_data, compute_realized_pnl, ExecutionConfig, simulate_with_daily_data, compute_realized_pnl,
) )
from quant_engine.backtest import run_weight_backtest
from quant_engine.indicators import macd, bollinger, kdj from quant_engine.indicators import macd, bollinger, kdj
from quant_engine.data_adapter import ( from quant_engine.data_adapter import (
long_to_wide, wide_to_long, rename_tushare_columns, long_to_wide, wide_to_long, rename_tushare_columns,
@@ -72,6 +73,16 @@ from quant_engine.data_adapter import (
df = load_qtdb_daily(["000001.SZ"], "2024-01-01", with_adj=True) df = load_qtdb_daily(["000001.SZ"], "2024-01-01", with_adj=True)
prices, volumes = prepare_execution_inputs(df) prices, volumes = prepare_execution_inputs(df)
result = simulate_with_daily_data(prices, initial_cash=1_000_000.0) result = simulate_with_daily_data(prices, initial_cash=1_000_000.0)
# 权重回测 → 稳定结果对象 → 绩效/基准分析
backtest = run_weight_backtest(
weights=rebalance_weights,
stock_returns=daily_returns,
initial_capital=1_000_000.0,
benchmark_nav=benchmark_nav,
)
print(backtest.stats())
print(backtest.benchmark_report())
``` ```
## 与 research_results 的关系 ## 与 research_results 的关系
+52 -10
View File
@@ -13,29 +13,28 @@
```python ```python
from quant_engine.backtest import ( from quant_engine.backtest import (
compute_nav_from_weights, # 调仓表 → 净值 rebalance_periodic, # 周期性再平衡
rebalance_table, # 周期性再平衡 run_weight_backtest, # 权重 → 统一结果对象
compare_to_benchmark, # 策略 vs 基准
weights_to_long_short, # 多空组合 weights_to_long_short, # 多空组合
) )
# 1. 调仓表 → 净值 # 调仓表 → 净值、收益、绩效与基准报告
nav = compute_nav_from_weights( rebalance_table = rebalance_periodic(target_weights, rebalance_dates, returns.index)
result = run_weight_backtest(
weights=rebalance_table, # 每周/每月调仓 weights=rebalance_table, # 每周/每月调仓
stock_returns=returns, # 个股日收益 stock_returns=returns, # 个股日收益
initial_capital=1.0, initial_capital=1.0,
benchmark_nav=benchmark_nav,
) )
print(result.stats())
# 2. 跟基准比 print(result.benchmark_report())
result = compare_to_benchmark(nav, benchmark_nav)
print(result.summary())
``` ```
""" """
from __future__ import annotations from __future__ import annotations
from pathlib import Path
from collections.abc import Mapping, Sequence from collections.abc import Mapping, Sequence
from dataclasses import dataclass
import numpy as np import numpy as np
import pandas as pd import pandas as pd
@@ -46,6 +45,26 @@ from quant_engine.metrics import summary as metrics_summary
logger = get_logger(__name__) logger = get_logger(__name__)
@dataclass(frozen=True, slots=True, eq=False)
class BacktestResult:
"""一次权重回测的稳定结果快照。"""
nav: pd.Series
returns: pd.Series
weights: pd.DataFrame
benchmark_nav: pd.Series | None = None
def stats(self, rf: float = 0.0) -> Mapping[str, float]:
"""返回标准绩效指标。"""
return metrics_summary(self.returns, rf)
def benchmark_report(self, rf: float = 0.0) -> pd.DataFrame:
"""返回策略与基准的对比报告。"""
if self.benchmark_nav is None:
raise ValueError("benchmark_nav is required for benchmark comparison")
return compare_to_benchmark(self.nav, self.benchmark_nav, rf)
# ── 调仓表 → 净值 ────────────────────────────────────── # ── 调仓表 → 净值 ──────────────────────────────────────
@@ -113,6 +132,29 @@ def compute_returns_from_nav(nav: pd.Series) -> pd.Series:
return nav.pct_change().fillna(0.0) return nav.pct_change().fillna(0.0)
def run_weight_backtest(
weights: pd.DataFrame,
stock_returns: pd.DataFrame,
initial_capital: float = 1.0,
tc_rate: float = 0.0,
benchmark_nav: pd.Series | None = None,
) -> BacktestResult:
"""执行权重回测并返回隔离于调用方输入的结果快照。"""
weights_snapshot = weights.copy(deep=True)
nav = compute_nav_from_weights(
weights=weights_snapshot,
stock_returns=stock_returns,
initial_capital=initial_capital,
tc_rate=tc_rate,
)
return BacktestResult(
nav=nav,
returns=compute_returns_from_nav(nav),
weights=weights_snapshot,
benchmark_nav=None if benchmark_nav is None else benchmark_nav.copy(deep=True),
)
# ── 调仓工具 ────────────────────────────────────── # ── 调仓工具 ──────────────────────────────────────
+85
View File
@@ -6,10 +6,12 @@ import pandas as pd
import pytest import pytest
from quant_engine.backtest import ( from quant_engine.backtest import (
BacktestResult,
compare_to_benchmark, compare_to_benchmark,
compute_nav_from_weights, compute_nav_from_weights,
compute_returns_from_nav, compute_returns_from_nav,
rebalance_periodic, rebalance_periodic,
run_weight_backtest,
weights_to_long_short, weights_to_long_short,
) )
@@ -122,3 +124,86 @@ def test_compare_to_benchmark_rejects_non_overlapping_dates() -> None:
with pytest.raises(ValueError, match="overlapping dates"): with pytest.raises(ValueError, match="overlapping dates"):
compare_to_benchmark(strategy, benchmark) 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