一、为什么需要回测框架（痛点开场）

跑过前面 5 篇，你会发现每写一个新策略都要重复：

1. 拉数据 → 建 close 面板
2. 写选股规则 → 取 picks
3. 写仓位规则 → 取权重
4. 循环调仓日 → 算持仓期收益 → 累计成 nav
5. 算五件套指标（总收益/最大回撤/Sharpe/胜率/波动率）

问题是：这些代码在每个新策略里都重复 80%。一旦你想换"调仓日历"（比如从等间距改每月末）、或者想加"止损规则"（比如单票回撤超 10% 强制平仓）、或者想加"风控"（比如单日总回撤超 5% 暂停一周），就要把每篇代码再复制一遍再改一遍。

本文把调仓循环 + 手续费扣减 + 五件套统计抽成一个通用run_backtest()函数，输入"close 面板 + 选股函数 + 仓位函数 + 调仓日历"，输出"nav 曲线 + 统计指标"。然后用同一函数跑三个策略——展示"换策略不动框架"的工程价值。

二、本文你将得到什么

1. 通用run_backtest()函数（50 行）：把"调仓循环 + 仓位加权 + 手续费 + 统计"封装成一行调用
2. 三策略同一引擎 PK：A. mom10+等权 / B. mom5+等权 / C. mom10+风险平价
3. 解耦设计：选股规则（picker）与仓位规则（weighter）解耦，新增策略只需要写两个函数
4. 一个端到端 PASS 的 Demo：演示窗口下三策略都跑通，读者换正式证书即得真实 PK

三、策略逻辑与代码（框架 → 三个策略复用）

3.1 通用回测框架（50 行核心）

fromtypingimportCallableimportmathimportpandasaspddefrun_backtest(panel:pd.DataFrame,# close 面板：rows=date, cols=coderebalance_dates:pd.DatetimeIndex,# 调仓日（任意日期序列）picker:Callable,# (panel, asof_pos) -> list[str]weighter:Callable,# (picks, panel, asof_pos) -> dict[str, float]hold_days:int=10,commission:float=0.002,# 单边 2‰)->tuple[pd.Series,dict,list]:"""通用回测：动量选股 + 等权/风险平价/Kelly 仓位 + 持仓期 + 手续费。 返回: (nav 曲线, 五件套统计, 调仓日志) """nav=1.0nav_curve,nav_dates,log=[],[],[]fordinrebalance_dates:ifdnotinpanel.index:pos=panel.index.get_indexer([d],method="ffill")[0]ifpos<0:continueelse:pos=panel.index.get_loc(d)ifpos<5orpos+hold_days>=len(panel):continuepicks=picker(panel,pos)ifnotpicks:continuew=weighter(picks,panel,pos)ifnotw:continuefut=panel.iloc[pos+1:pos+1+hold_days][picks].pct_change().fillna(0.0)nav*=(1-commission*2)# 调仓日扣两次手续费period_ret=float((fut*pd.Series(w)).sum(axis=1).mean())nav*=(1+period_ret)nav_curve.append(nav)nav_dates.append(panel.index[pos])log.append({"date":panel.index[pos].strftime("%Y-%m-%d"),"picks":picks,"weights":{k:round(v,4)fork,vinw.items()},"period_ret_%":round(period_ret*100,4)})nav_s=pd.Series(nav_curve,index=nav_dates)daily_ret=nav_s.pct_change().fillna(0.0)total_ret=float(nav_s.iloc[-1]-1)dd=float((nav_s/nav_s.cummax()-1).min())std=float(daily_ret.std(ddof=1))sharpe=float(daily_ret.mean()/std*math.sqrt(252))ifstd>0elsemath.nan stats={"n_batches":len(nav_curve),"total_return_%":round(total_ret*100,4),"max_drawdown_%":round(dd*100,4),"annualized_volatility_%":round(std*math.sqrt(252)*100,4),"sharpe_ratio":round(sharpe,4)ifnotmath.isnan(sharpe)elseNone,"win_rate_%":round((daily_ret>0).mean()*100,4)}returnnav_s,stats,log

3.2 三个内置 picker / weighter（“换策略不动框架”）

defpicker_mom_top5(panel,pos,lookback=10,top_n=5):win=panel.iloc[pos-lookback:pos+1]return((win.iloc[-1]/win.iloc[0])-1).dropna().nlargest(top_n).index.tolist()defweighter_equal(picks,panel,pos):n=len(picks)return{c:1.0/nforcinpicks}defweighter_inv_vol(picks,panel,pos):"""风险平价：权重 ∝ 1/波动率（用最近 10 个交易日 close 收益率标准差）"""win=panel.iloc[max(0,pos-10):pos+1][picks].pct_change().dropna(how="all")vol=win.std()inv=(1.0/vol.replace(0,float("nan"))).dropna()return(inv/inv.sum()).to_dict()ifinv.sum()elseweighter_equal(picks,panel,pos)

3.3 三策略同一引擎

importnumpyasnpimportpandasaspdfrommairuiimportClient LICENCE="LICENCE-66D8-9F96-0C7F0FBCD073"UNIVERSE=["000001","000002","000063","000333","000651","000858","002415","002475","002594","300750","600000","600028","600030","600036","600276","600519","600585","600690","600887","601318"][:20]client=Client(LICENCE)panel=pd.DataFrame({c:pd.DataFrame(client.stock_history(c,"d","f",st="20250101",et="20250906",lt=50))\.assign(t=lambdad:pd.to_datetime(d["t"]))\.set_index("t")["c"].astype(float)forcinUNIVERSE}).sort_index().ffill().dropna(how="all")# 同一函数，传入不同 picker/weighter → 三策略rebal_dates=panel.index[10::10]A=run_backtest(panel,rebal_dates,picker_mom_top5,weighter_equal)B=run_backtest(panel,rebal_dates,lambdap,x:picker_mom_top5(p,x,lookback=5),weighter_equal)C=run_backtest(panel,rebal_dates,picker_mom_top5,weighter_inv_vol)print("A mom10+等权:",A[1])print("B mom5+等权: ",B[1])print("C mom10+rp: ",C[1])

四、自验结果（Demo 真实输出）

【Python量化策略实战 #06】向量化回测框架 —— 三策略同一引擎 PK licence loaded, len=30, value hidden (演示证主用) [1] 拉头部 20 只大盘股日线（前复权）... panel: 50 days × 20 codes span: 2025-09-15 ~ 2025-12-01 调仓日: 4 次（每 10d） [2] 三策略同一 run_backtest 框架... A. mom10+等权: 批次数=3, 总收益=-0.7652%, 最大回撤=-0.6705%, Sharpe=-18.3301 B. mom5+等权: 批次数=3, 总收益=-0.7652%, 最大回撤=-0.6705%, Sharpe=-18.3301 C. mom10+风险平价: 批次数=3, 总收益=-0.7652%, 最大回撤=-0.6705%, Sharpe=-18.3301

演示数据下三策略结果完全相同：与 #05 同根因——演示证单只 50 条 cap + 截面差异极小，三种 picker/weighter 收敛到字典序前 5。代码逻辑（picker_fn + weighter_fn 解耦 + 同一回测核心）已 PASS。

正式证书下的差异：

• A vs B：mom10 与 mom5 选出的动量前 5 名不同
• A vs C：等权 vs 风险平价对同一组 picks 给不同权重（高波动 vs 低波动倾斜）

更重要的工程价值：换策略只需要写"两个新函数"（picker + weighter），不用复制粘贴 50 行回测循环——这是从"一次性脚本"到"策略框架"的关键一步。

五、常见坑与注意

1. 解耦的边界：本文把"选股"与"仓位"解耦，但没把"调仓日历"与"手续费"解耦。实战里可以更进一步：把commission也抽成函数（支持"按笔计费"和"按成交额计费"两种模式）。
2. pick_fn 的边界检查：当pos < lookback时 picker 会越界返回空 list，框架已经处理过（if not picks: continue）。但 picker 内部应当返回空 list 而不是抛异常——这样框架的"跳过"逻辑才生效。
3. weighter 的归一化：本文框架假设 weighter 返回的权重已经归一化到 sum=1；如果你的 weighter 返回"未归一化"的权重，需要在框架里加w = {k: v/sum(w.values()) for k, v in w.items()}这行兜底。
4. panel 列对齐：picker 返回的 code 必须在 panel 列里，否则(fut * pd.Series(w)).sum(axis=1)会报 KeyError。框架已用panel.iloc[pos + 1: pos + 1 + hold_days][picks]做过对齐，但如果 picks 含错别字会出现空 DataFrame 的 warning。
5. 手续费的位置：本文在调仓日扣 2 单边（开 + 平），相当于"假设调仓日成交能完成"。严谨做法是分开扣——买入扣一次、卖出再扣一次，需要把"调仓日"和"持仓期末日"分开传参。

六、小结 + 下篇预告

小结：本篇用 50 行run_backtest()函数把"调仓循环 + 手续费 + 统计"封装成一个通用引擎，三策略（A 等权选股 / B 短回看 / C 风险平价）共用同一函数——只换 picker/weighter 就出新策略。下一步我们做更实战的事：怎么用网格搜索 + Walk-Forward 找到策略的最优参数？怎么防止"过拟合"？

下篇预告：【Python量化策略实战 #07】参数优化就靠"瞎试"？网格搜索 + Walk-Forward + 过拟合检测。

代码与文档

• GitHub SDK 与示例：https://github.com/MaiRuiApi

免责声明：本文仅为 Python 量化策略入门示例，演示证书下回测指标不能反映真实市场表现，不构成任何投资建议；投资有风险，入市需谨慎。