向量化回测引擎：从零实现一个可信的日频回测框架

远思为径，码动价值。欢迎来到远思码技术实验室的最新技术实践记录。

回测是量化研究里最容易产生幻觉的一步。同样一段策略逻辑，用不同框架、不同参数假设跑出来的年化收益率可以差出十几个点，而绝大多数差异都藏在三件小事里：成交价、交易限制、成本。

实验室的结论是：先自己写一个能看懂每一行的引擎，再去用现成框架。这篇就是那个引擎，全部代码不到 200 行。

一、引擎的契约

在写代码之前，先把输入输出定死，这是防止自己作弊的第一步：

输入	形状	含义
`close`	T×N	复权收盘价矩阵
`volume`	T×N	成交量矩阵（用于判断可交易性）
`pct_chg`	T×N	当日涨跌幅，用于判断涨跌停
`target_w`	T×N	T 日收盘后决定的下一期目标权重，行和 ≤ 1，余下为现金

输出	含义
`nav`	净值曲线（起点为 1）
`holdings`	每日实际持仓权重、换手、成本明细

关键约定只有一条：target_w 的第 t 行，是从第 t 日收盘之后开始生效的权重。引擎内部自己做 shift(1) 来实现"信号滞后一期"，策略代码里严禁再手动 shift，否则会重复滞后。

二、骨架：收益、漂移、成本

# backtest/engine.py
from __future__ import annotations

import itertools
from dataclasses import dataclass, field

import numpy as np
import pandas as pd


@dataclass
class BacktestConfig:
    fee_rate: float = 0.0003      # 佣金，双边
    stamp_duty: float = 0.0005    # 印花税，仅卖出
    slippage: float = 0.0010      # 单边滑点
    impact_coef: float = 0.0015   # 冲击成本系数（相对换手）
    rf: float = 0.02              # 无风险利率（年化）
    periods: int = 244            # 年化交易日


@dataclass
class BacktestResult:
    nav: pd.Series
    daily_return: pd.Series
    turnover: pd.Series
    cost: pd.Series
    holdings: pd.DataFrame
    metrics: dict = field(default_factory=dict)

核心循环全部用矩阵运算完成，一行 for 都不需要：

def run_backtest(close: pd.DataFrame,
                 volume: pd.DataFrame,
                 target_w: pd.DataFrame,
                 pct_chg: pd.DataFrame | None = None,
                 cfg: BacktestConfig | None = None) -> BacktestResult:
    cfg = cfg or BacktestConfig()
    close, volume = close.align(volume, join="inner")
    target_w = target_w.reindex(index=close.index, columns=close.columns).fillna(0.0)

    ret = close.pct_change().fillna(0.0)            # 第 t 日收益，close_{t-1} -> close_t

    # ---- 1. 可交易性：停牌(volume=0) / 涨跌停 / 缺失价 ----
    tradable = (volume > 0) & close.notna() & (target_w.notna())
    if pct_chg is not None:
        pct = pct_chg.reindex_like(tradable)
        # 主板 ±10%、创业板/科创板 ±20%，用绝对阈值统一近似：≥9.8% 视为涨停不可买
        limit_up = pct >= 9.8
        limit_down = pct <= -9.8
        # 目标权重 > 0 且涨停 -> 买不进；目标权重 < 0 且跌停 -> 卖不出
        can_buy = tradable & ~limit_up
        can_sell = tradable & ~limit_down
        w_prev = target_w.shift(1)
        blocked_buy = (w_prev.isna() | (w_prev <= 0)) & (target_w > 0) & ~can_buy
        blocked_sell = (w_prev > 0) & (target_w <= 0) & ~can_sell
        target_w = target_w.mask(blocked_buy, 0.0).mask(blocked_sell, w_prev)

    # ---- 2. 归一化：不足 1 的部分视为现金 ----
    # 允许做多为主；若行和 > 1 则等比压缩，避免隐式杠杆
    gross = target_w.sum(axis=1).replace(0, np.nan)
    scale = np.where(gross > 1, 1 / gross, 1.0)
    target_w = target_w.mul(pd.Series(scale, index=target_w.index), axis=0).fillna(0.0)

    # ---- 3. 实际持仓 = 上一期目标权重（信号滞后一期） ----
    w_hold = target_w.shift(1).fillna(0.0)

    # ---- 4. 毛收益：今日持仓 × 今日收益 ----
    gross_ret = (w_hold * ret).sum(axis=1)

    # ---- 5. 权重漂移：今日收盘后各资产权重自然变化 ----
    denom = (1 + gross_ret).replace(0, np.nan)
    w_drift = w_hold.mul(1 + ret, axis=0).div(denom, axis=0).fillna(0.0)

    # ---- 6. 换手与成本：今日收盘调仓，相对漂移权重 ----
    delta = target_w - w_drift
    turnover_buy = delta.clip(lower=0).sum(axis=1)      # 买入额占比
    turnover_sell = (-delta).clip(lower=0).sum(axis=1)  # 卖出额占比
    turnover = turnover_buy + turnover_sell

    cost = (turnover_buy * (cfg.fee_rate + cfg.slippage)
            + turnover_sell * (cfg.fee_rate + cfg.slippage + cfg.stamp_duty)
            + turnover * cfg.impact_coef * turnover)     # 冲击成本近似平方项

    port_ret = gross_ret - cost
    nav = (1 + port_ret).cumprod()

    res = BacktestResult(nav=nav, daily_return=port_ret, turnover=turnover,
                         cost=cost, holdings=w_hold)
    res.metrics = performance(nav, port_ret, turnover, cfg)
    return res

有几处设计值得展开说：

w_hold = target_w.shift(1) 就是"信号滞后一期"的全部实现。策略在 T 日收盘后生成信号，T+1 日才能享受（或承受）收益。
漂移权重 w_drift 常被忽略。如果你 T 日满仓买入，T+1 日股票涨了 5%，那么 T+1 收盘时它在组合里的权重是 5% 之后的自然权重，不是当初的目标权重。调仓换手必须相对它算，否则每次调仓的换手都会被高估。
冲击成本写成 turnover * coef * turnover，即随换手平方增长。这比线性模型更贴近现实：大额调仓的边际成本上升。

三、绩效指标

def performance(nav: pd.Series, ret: pd.Series, turnover: pd.Series,
                cfg: BacktestConfig) -> dict:
    n = len(nav)
    if n < 2 or nav.iloc[-1] <= 0:
        return {"error": "样本不足或净值归零"}
    ann = nav.iloc[-1] ** (cfg.periods / n) - 1
    vol = ret.std(ddof=1) * np.sqrt(cfg.periods)
    sharpe = (ann - cfg.rf) / vol if vol else np.nan

    dd = nav / nav.cummax() - 1
    mdd = dd.min()
    calmar = ann / abs(mdd) if mdd else np.nan
    # 最大回撤持续天数（净值在水面之下的最长连续区间）
    underwater = (dd < 0).astype(int)
    max_flat = max((len(list(g)) for k, g in itertools.groupby(underwater) if k == 1),
                   default=0)

    win = (ret > 0).mean()
    avg_turnover = turnover.mean()
    return {
        "累计收益": round(float(nav.iloc[-1] - 1), 4),
        "年化收益": round(float(ann), 4),
        "年化波动": round(float(vol), 4),
        "夏普": round(float(sharpe), 3),
        "最大回撤": round(float(mdd), 4),
        "最大回撤天数": int(max_flat),
        "Calmar": round(float(calmar), 3),
        "日胜率": round(float(win), 3),
        "日均换手": round(float(avg_turnover), 4),
        "年化换手": round(float(avg_turnover * cfg.periods), 2),
        "交易日数": int(n),
    }

这里 日均换手 × 244 = 年化换手 是个粗略但足够用的口径。日频策略年化换手超过 30 倍，成本假设就必须认真对待，不能再用"万一三"敷衍过去。

四、把正确性钉死：三个单元测试

引擎写完不测，等于没写。实验室用三个手算案例做回归测试，任何改动都必须过：

# tests/test_engine.py
import numpy as np
import pandas as pd
from backtest.engine import BacktestConfig, run_backtest

def _frames(ret_rows):
    """由收益矩阵构造价格矩阵（起点 10 元）。"""
    close = (1 + pd.DataFrame(ret_rows)).cumprod() * 10
    volume = pd.DataFrame(1e6, index=close.index, columns=close.columns)
    return close, volume


def test_single_stock_buy_and_hold():
    # 两天，每天涨 10%。T日收盘建仓 -> T+1 才吃到收益
    close, volume = _frames([[0.10], [0.10]])
    w = pd.DataFrame([[1.0], [1.0]], index=close.index, columns=close.columns)
    res = run_backtest(close, volume, w, cfg=BacktestConfig(fee_rate=0, stamp_duty=0,
                                                            slippage=0, impact_coef=0))
    # 第一天无持仓（w 被 shift），第二天满仓吃 10%
    assert abs(res.nav.iloc[0] - 1.0) < 1e-12
    assert abs(res.nav.iloc[1] - 1.10) < 1e-12


def test_cost_is_charged_on_turnover():
    close, volume = _frames([[0.0], [0.0]])
    w = pd.DataFrame([[1.0], [0.0]], index=close.index, columns=close.columns)
    cfg = BacktestConfig(fee_rate=0.001, stamp_duty=0.001, slippage=0, impact_coef=0)
    res = run_backtest(close, volume, w, cfg=cfg)
    # 第一天：w.shift 后仍为 0 -> 无持仓；第二天持仓 1 后立刻清仓？
    # 实际：hold 恒为 [0, 1]，故第二天收盘换手 = 1，卖出成本 = 1*(佣金0.001+印花税0.001)
    assert res.turnover.iloc[1] == 1.0
    assert abs(res.cost.iloc[1] - 0.002) < 1e-12


def test_weights_never_imply_leverage():
    close, volume = _frames([[0.01], [0.01], [0.01]], )
    w = pd.DataFrame([[0.8], [0.8], [0.8]], index=close.index, columns=close.columns)
    res = run_backtest(close, volume, w, cfg=BacktestConfig(fee_rate=0, stamp_duty=0,
                                                            slippage=0, impact_coef=0))
    assert (res.holdings.sum(axis=1) <= 1 + 1e-9).all()

第二个测试看起来有点绕，但它锁住了一个非常具体的语义：换手发生在"目标权重与漂移权重不一致"的那一天，成本当天计提。这个语义一旦写错，所有策略的成本都会被系统性低估。

五、什么时候该换成事件驱动框架

向量化引擎的假设是：同一时点所有股票都能按收盘价成交，且份额可以无限细分。它在下列场景会失真：

场景	向量化引擎的问题	建议
分钟级 / 盘口撮合	无法表达排队、部分成交	事件驱动（vn.py / backtrader / 自研撮合）
资金量大于标的日成交额的 1%	冲击成本模型过于乐观	引入成交量参与率约束
需要盘中撤单、止损单	无订单生命周期	事件驱动
组合再平衡涉及资金约束（现金不足）	权重假设掩盖了资金不足	加现金约束层

实验室的日常研究 90% 用向量化，因为要跑几千组参数；只有在策略进入仿真阶段，才用事件驱动复核一遍。先用快而粗糙的尺子找方向，再用慢而精确的尺子确认。

六、一个常见误用

有人会把 target_w 直接设成"今天涨得最好的股票"，然后跑出 500% 的年化——因为他在 T 日收盘时用了 T 日的数据去选股，而引擎的 shift(1) 让这份"未来信息"合法化了。

判断方法很简单：把 shift(1) 改成 shift(2)，如果年化收益暴跌，说明策略严重依赖即时信息，值得警惕。

结语

到这里我们有了两样东西：可检验的因子，和可信的尺子。下一篇用它们跑第一个真实策略——双均线与动量轮动的组合，顺带把参数稳健性检验做完。

远思为径，码动价值。回测是用来证伪的，不是用来给自己发奖金的。