"""向量化 + 事件循环混合的回测引擎。
关键假设（务必知悉，均可在配置中调整）：
- T 日收盘产生信号，T+1 起按新权重计价（无前向偏差）；
- 佣金/印花税/滑点在换手时一次性扣除；
- 持仓期间权重按每日收益自然漂移，不做日内再平衡；
- 波动率目标与回撤降仓为组合层面风控覆盖层。
"""
from __future__ import annotations
from typing import Dict, List, Optional
import numpy as np
import pandas as pd
def build_matrices(data_map: Dict[str, pd.DataFrame]):
"""拼装收盘价 / 成交额矩阵。"""
close = pd.DataFrame({c: df["close"] for c, df in data_map.items()})
amount = pd.DataFrame({c: df["close"] * df["volume"] for c, df in data_map.items()})
close = close.sort_index()
amount = amount.sort_index().reindex(close.index)
return close, amount
class Backtester:
"""多因子选股回测器。"""
def __init__(self, strategy: Dict):
self.cfg = strategy
sel = strategy["selection"]
self.factors = sel["factors"]
self.lookbacks = sel["factor_lookbacks"]
self.max_positions = sel["max_positions"]
self.min_history = sel["min_history"]
port = strategy["portfolio"]
self.rebalance_days = port["rebalance_days"]
self.cash_buffer = port["cash_buffer"]
self.max_weight = port["max_weight"]
self.fwd_days = port["fwd_days"]
rc = strategy["risk_control"]
self.target_vol = rc["target_vol"]
self.vol_lookback = rc["vol_lookback"]
self.max_dd_derisk = rc["max_drawdown_derisk"]
self.de_risk_ratio = rc["de_risk_ratio"]
self.trend_filter = rc["trend_filter"]
self.trend_ma = rc["trend_filter_ma"]
ex = strategy["execution"]
self.cost_buy = ex["commission"] + ex["slippage"]
self.cost_sell = ex["commission"] + ex["stamp_tax_sell"] + ex["slippage"]
# ------------------------------------------------------------------
def run(self, data_map: Dict[str, pd.DataFrame], benchmark_close: Optional[pd.Series] = None) -> Dict:
from .factors import compute_factor_panel, cross_section_z
close, amount = build_matrices(data_map)
dates = close.index
rets = close.pct_change(fill_method=None)
ma = close.rolling(self.trend_ma, min_periods=self.trend_ma // 2).mean()
# 预计算各因子截面 Z-score 面板
z_panels: Dict[str, pd.DataFrame] = {}
for f, w in self.factors.items():
if w <= 0:
continue
panel = compute_factor_panel(data_map, f, self.lookbacks[f], dates)
z_panels[f] = cross_section_z(panel)
score = sum(w * z_panels[f] for f, w in self.factors.items() if w > 0)
start = pd.Timestamp(self.cfg["backtest_window"]["start"])
end = pd.Timestamp(self.cfg["backtest_window"]["end"])
bdates = [d for d in dates if start <= d <= end]
if len(bdates) < 60:
raise ValueError("回测区间太短，请扩大 start/end 或加大数据抓取范围")
equity = 1.0
curve, bench_curve, exposure_hist = [], [], []
weights: Dict[str, float] = {}
cash = 1.0
peak = 1.0
de_risked = False
strat_rets: List[float] = []
rebalance_log: List[Dict] = []
turnover_hist: List[float] = []
prev_target: Dict[str, float] = {}
pending_target: Optional[Dict[str, float]] = None
pending_expo = 1.0
for i, d in enumerate(bdates):
di = dates.get_loc(d)
# ---- 1) 当日收益结算（用昨日收盘确定的持仓） ----
if i > 0:
day_ret = 0.0
for c, w in weights.items():
r = rets.at[d, c]
if not np.isnan(r):
day_ret += w * r
day_ret += cash * 0.0
# 换仓日：结算与成本同日发生（近似：以 T+1 收盘换手计价）
if pending_target is not None:
tgt, expo = pending_target, pending_expo
old_w = {c: w for c, w in weights.items()}
old_cash = cash
# 合并当日收益后的旧权重
denom = 1.0 + day_ret
drifted = {c: w * (1.0 + (rets.at[d, c] if not np.isnan(rets.at[d, c]) else 0.0)) / denom
for c, w in old_w.items()}
drifted_cash = old_cash / denom
buy = sum(max(tgt.get(c, 0.0) - drifted.get(c, 0.0), 0.0) for c in set(tgt) | set(drifted))
sell = sum(max(drifted.get(c, 0.0) - tgt.get(c, 0.0), 0.0) for c in set(tgt) | set(drifted))
cost = buy * self.cost_buy + sell * self.cost_sell
day_ret += -cost
weights = {c: w / (1.0 + 0.0) for c, w in tgt.items()}
cash = 1.0 - sum(weights.values())
turnover_hist.append(buy + sell)
pending_target = None
if day_ret < -0.5:
day_ret = -0.5 # 物理下限保护
equity *= (1.0 + day_ret)
strat_rets.append(day_ret)
peak = max(peak, equity)
# ---- 2) 调仓日：生成新目标权重（T 日收盘，T+1 生效） ----
is_rebalance = (i % self.rebalance_days == 0)
if is_rebalance:
scores = score.loc[d].dropna()
eligible = scores.index.tolist()
# 最小历史长度
hist_ok = [c for c in eligible
if data_map[c].index[0] <= d - pd.Timedelta(days=self.min_history)]
eligible = hist_ok
if self.trend_filter:
ma_d = ma.loc[d]
eligible = [c for c in eligible
if not np.isnan(ma_d.get(c, np.nan)) and close.at[d, c] >= ma_d[c]]
ranked = sorted(eligible, key=lambda c: scores[c], reverse=True)
selected = ranked[: self.max_positions]
expo = 1.0
if len(strat_rets) >= self.vol_lookback:
rv = float(np.std(strat_rets[-self.vol_lookback:], ddof=1) * np.sqrt(252))
if rv > 1e-9:
expo = min(1.0, self.target_vol / rv)
dd = 1.0 - equity / peak
if dd > self.max_dd_derisk:
expo *= self.de_risk_ratio
de_risked = True
elif de_risked and dd < self.max_dd_derisk * 0.5:
de_risked = False
base = (1.0 - self.cash_buffer) * expo
if selected:
w_each = min(self.max_weight, base / len(selected))
tgt = {c: w_each for c in selected}
else:
tgt = {}
pending_target = tgt
pending_expo = 1.0 # expo 已并入 base
exposure_hist.append({"date": str(d.date()), "exposure": base, "de_risked": de_risked})
rebalance_log.append({
"date": str(d.date()),
"n_eligible": len(eligible),
"selected": [{"code": c, "name": "", "score": round(float(scores[c]), 3),
"weight": round(tgt.get(c, 0.0), 4)} for c in selected],
})
curve.append({"date": str(d.date()), "equity": equity})
if benchmark_close is not None:
bv = benchmark_close.reindex([d]).dropna()
bench_curve.append({"date": str(d.date()),
"equity": float(bv.iloc[0]) / float(benchmark_close.dropna().iloc[0]) if len(bv) else np.nan})
curve_df = pd.DataFrame(curve).set_index("date")
curve_df.index = pd.to_datetime(curve_df.index)
bench_df = pd.DataFrame(bench_curve).dropna().set_index("date")
if len(bench_df):
bench_df.index = pd.to_datetime(bench_df.index)
bench_df = bench_df / bench_df["equity"].iloc[0]
# ---- IC 序列（合成因子 vs fwd_days 前向收益，Spearman） ----
fwd = close.shift(-self.fwd_days) / close - 1.0
ic_series = {}
for entry in rebalance_log:
d = pd.Timestamp(entry["date"])
if d not in score.index:
continue
s = score.loc[d].dropna()
f = fwd.loc[d].reindex(s.index).dropna()
common = s.index.intersection(f.index)
if len(common) >= 8:
# Spearman = 秩的 Pearson（避免引入 scipy 依赖）
ic_series[str(d.date())] = float(s[common].rank().corr(f[common].rank()))
results = {
"curve": curve_df,
"benchmark": bench_df,
"strat_rets": pd.Series(strat_rets, index=curve_df.index[1:]) if strat_rets else pd.Series(dtype=float),
"rebalance_log": rebalance_log,
"exposure_hist": exposure_hist,
"turnover_hist": turnover_hist,
"ic_series": ic_series,
"latest_weights": weights,
}
return results