"""向量化 + 事件循环混合的回测引擎。 关键假设（务必知悉，均可在配置中调整）： - 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