从零搭建最小回测引擎

为什么建议自己先写一遍

市面上现成的回测框架不少，功能都比自己写的强。但对刚入门的人来说，直接用框架有个隐性代价：你不知道它在哪一步替你做了假设。成交价用的是收盘还是次开、手续费按什么口径扣、限价单未成交时怎么处理、净值是每根更新还是平仓才更新——这些假设每一个都能让年化收益差出十几个百分点。

自己写一遍两百行的最小引擎，是把这些假设逐个显式写出来的过程。写完之后你再用任何框架，都知道该去文档里查哪几件事。这个投入通常是一两天，回报是此后所有回测结论都建立在你理解的地基上。

最小引擎不追求性能和功能，只追求一件事：结果可信。不支持限价单、不支持部分成交、不支持多空同时持仓都没关系；但成交时点、成本口径、净值计算这三处必须严格。

模块一：数据加载

数据层的任务是把交易所返回的原始K线变成一张干净、可信、可复现的表。它看起来最简单，实际上是错误来源排第二的地方（第一是撮合）。要处理的问题有五个，缺一个都会在后面变成难查的怪现象。

第一是时区。所有时间戳统一成 UTC，不要用本地时间。第二是去重。交易所分页拉取时经常在边界重复返回一根K线，重复会让指标的滚动窗口错位。第三是缺失。网络中断或交易所维护会留下空洞，必须显式检测出来并决定策略——跳过还是终止，绝不要静默 ffill 价格。

第四是排序与索引唯一性，取完数据后强制按时间升序并断言索引无重复。第五是缓存，本地存成 parquet 按 symbol 和周期分文件，避免每次调试都重新拉网络，也保证结果可复现。

import pandas as pd
from pathlib import Path

COLS = ["open", "high", "low", "close", "volume"]

def load_ohlcv(path: str, freq: str = "15min") -> pd.DataFrame:
    df = pd.read_parquet(path)

    # 1) 统一 UTC 时间索引
    df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    df = df.set_index("ts").sort_index()

    # 2) 去重: 分页拉取的边界重复
    df = df[~df.index.duplicated(keep="last")]

    # 3) 显式检测缺失, 不做价格 ffill
    full = pd.date_range(df.index[0], df.index[-1], freq=freq, tz="UTC")
    missing = full.difference(df.index)
    if len(missing):
        print(f"[warn] 缺失 {len(missing)} 根K线, 首个: {missing[0]}")

    # 4) 断言基本一致性
    assert df.index.is_monotonic_increasing
    assert df.index.is_unique
    assert (df["high"] >= df["low"]).all()
    assert (df["high"] >= df[["open", "close"]].max(axis=1)).all()
    assert (df["volume"] >= 0).all()

    return df[COLS].astype("float64")

python

模块二：撮合逻辑（最关键的一步）

撮合的唯一职责是回答：给定一个信号，我在什么价格、什么时刻、付多少成本成交。核心规则一句话——用第 t 根K线的收盘数据决策，用第 t+1 根K线的开盘价成交。

为什么必须这样？因为你在 t 根收盘那一瞬才知道收盘价，此时该K线已经结束，你能提交的最早订单会在下一根开始时成交。如果回测里用 t 根的收盘价成交，等于假设你在知道收盘价的同时还能按那个价格买到，这在物理上不成立。

这个错误的影响有多大？对趋势突破类策略，突破根往往是当天涨幅最大的一根，用当根收盘成交相当于每次都在最有利的价格入场。一条日频突破策略因此虚增的年化收益，实测常见在十几到几十个百分点区间，足以把一条亏钱策略包装成漂亮曲线。

止损止盈的处理另有一条规则：同一根K线内若最高价与最低价同时触及止盈和止损，按最坏情况处理，即假定先触发止损。你无法从K线数据知道谁先到，乐观假设会系统性高估收益。

def backtest(df, signal_long, signal_exit,
             init_cash=10_000.0,
             fee_rate=0.0005,        # 单边 0.05%
             slip_rate=0.0005,       # 单边滑点 0.05%
             stop_rate=0.03):
    """最小撮合循环。signal_* 为与 df 同索引的布尔序列。"""
    cash, qty, entry, stop = init_cash, 0.0, 0.0, 0.0
    equity_curve, trades = [], []

    idx = df.index
    for i in range(len(idx) - 1):        # 注意: 到 -1, 因为要用 i+1
        ts, row = idx[i], df.iloc[i]
        nxt = df.iloc[i + 1]
        fill_px = nxt["open"]            # ★ 下一根开盘价成交

        # --- 1) 先处理离场(风控优先于入场) ---
        if qty > 0:
            # 最坏情况假设: 同根内先看止损
            if row["low"] <= stop:
                px = stop * (1 - slip_rate)
                cash += qty * px * (1 - fee_rate)
                trades.append((ts, "stop", entry, px, qty))
                qty, entry, stop = 0.0, 0.0, 0.0
            elif bool(signal_exit.iloc[i]):
                px = fill_px * (1 - slip_rate)
                cash += qty * px * (1 - fee_rate)
                trades.append((ts, "exit", entry, px, qty))
                qty, entry, stop = 0.0, 0.0, 0.0

        # --- 2) 再处理入场 ---
        if qty == 0 and bool(signal_long.iloc[i]):
            px = fill_px * (1 + slip_rate)
            qty = (cash * 0.99) / (px * (1 + fee_rate))
            cash -= qty * px * (1 + fee_rate)
            entry, stop = px, px * (1 - stop_rate)

        # --- 3) 每根都按市价重估净值 (mark-to-market) ---
        equity_curve.append((idx[i + 1], cash + qty * nxt["close"]))

    eq = pd.Series(dict(equity_curve)).sort_index()
    return eq, pd.DataFrame(
        trades, columns=["ts", "reason", "entry", "exit", "qty"])

循环里有两个容易忽略的顺序问题。一是离场必须先于入场判断，否则同一根上会出现先加仓再止损的诡异序列。二是净值每根都更新，而不是只在平仓时记一笔。只在平仓记账会完全看不见持仓期间的浮亏，最大回撤能被低估一半以上。

python

模块三：绩效统计

有了净值曲线，绩效计算是纯数学。但有两个地方经常算错：年化的频率因子，和最大回撤的定义。

年化收益用几何方式算：总倍数开 (年数) 次方再减一，不要用平均日收益乘 365。年化波动率是收益率序列标准差乘以周期数的平方根，周期数取决于数据频率——加密市场全年无休，日频是 365，4小时频是 365×6=2190，15分钟频是 365×96=35040。用错这个因子，夏普会差出两三倍。

最大回撤是净值相对历史最高点的最大跌幅，不是最高点到最低点的跌幅。还要一并输出回撤持续时间（从创新高到回到新高的天数），这个指标对能否坚持执行策略的影响，往往比回撤幅度本身更大。

import numpy as np

PERIODS = {"1d": 365, "4h": 365 * 6, "1h": 365 * 24,
           "15min": 365 * 96}

def performance(eq: pd.Series, freq="15min", rf=0.0) -> dict:
    ret = eq.pct_change().dropna()
    n = PERIODS[freq]
    years = len(ret) / n

    total = eq.iloc[-1] / eq.iloc[0] - 1
    cagr = (eq.iloc[-1] / eq.iloc[0]) ** (1 / years) - 1 if years > 0 else 0.0
    vol = ret.std(ddof=1) * np.sqrt(n)
    sharpe = (cagr - rf) / vol if vol > 0 else 0.0

    # 下行波动 -> Sortino
    dn = ret[ret < 0].std(ddof=1) * np.sqrt(n)
    sortino = (cagr - rf) / dn if dn > 0 else 0.0

    # 最大回撤: 相对历史最高点
    peak = eq.cummax()
    dd = eq / peak - 1
    mdd = dd.min()

    # 回撤持续期(根数): 未创新高的最长连续段
    under = (dd < 0).astype(int)
    longest, cur = 0, 0
    for v in under:
        cur = cur + 1 if v else 0
        longest = max(longest, cur)

    return {
        "total_return": round(total, 4),
        "cagr": round(cagr, 4),
        "vol": round(vol, 4),
        "sharpe": round(sharpe, 2),
        "sortino": round(sortino, 2),
        "max_drawdown": round(mdd, 4),
        "dd_bars": longest,
        "calmar": round(cagr / abs(mdd), 2) if mdd < 0 else None,
    }

python

三个模块的职责边界

把职责写成表贴在项目 README 里，能避免后期代码互相污染。判断某段逻辑该放哪一层，只问一句：它需要知道账户有多少钱吗？需要就属于撮合层，不需要就属于数据或信号层。

成本参数虽然只是撮合层的两个数字，但它对结论的影响远超其他所有细节，值得单独研究，可以看手续费与滑点建模那篇的具体演算。

引擎跑通之后，下一步不是调参提高收益，而是验证结果的稳定性，方法见Walk-Forward 滚动前推验证。单次全样本回测的数字，无论多漂亮，都不足以支持任何决策。

怎么确认引擎本身没有bug

引擎写完，第一件事不是跑策略，而是用四个已知答案的极端案例做校验。这些测试跑起来只要几秒，但能拦住大多数实现错误。

1. 买入持有测试：信号恒为 True、离场恒为 False、成本设为 0，结果必须精确等于 (末价/首价 - 1)。误差超过千分之一说明净值或仓位算错了。
2. 零信号测试：信号全为 False，净值必须是一条完全水平的直线，最大回撤为 0。
3. 成本单调性测试：手续费从 0 递增到 0.2%，总收益必须严格单调下降。若出现上升，说明成本符号或方向搞反了。
4. 未来函数探测：把成交价从 t+1 开盘改成 t 收盘，收益应明显上升。若两者几乎相同，说明你的信号触发点其实和K线关系不大，或者 shift 写错了位置。

第四项值得展开一句：它不是在鼓励你用当根收盘价，而是把差值当成诊断量。这个差值越大，说明策略越依赖入场时点的精确性，实盘中滑点对它的伤害也越大。差值特别大的策略，实盘落差通常最严重。

四项都过了，你的引擎就可以用来做初筛。之后需要补的功能按优先级排：成本模型细化、多币种支持（见多周期与多币种策略的工程结构）、限价单与部分成交、资金费率。功能可以慢慢加，但前面那三个正确性要求一开始就不能让步。