Backtesting 框架（搭建专业回测系统）

日期: 2026-08-02 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #回测 #vectorbt #backtrader #lookahead

---

今日目标

---

一、理论与模型

1.1 两种回测范式

1.2 回测必须避免的 7 大偏差

1. Look-ahead bias：用未来信息决策
2. Survivorship bias：只用现存资产
3. Selection bias：只挑表现好的回测期
4. Overfitting：过拟合参数
5. Sizing bias：忽略容量限制
6. Cost bias：低估交易成本
7. Path dependence：忽略回撤路径影响

1.3 Look-ahead bias 的隐形来源

显性：用 t+1 收盘价决策 t 仓位 隐性：

• 用全样本估计参数生成 in-sample 信号
• 用全样本分位数做阈值
• 用 forward-fill 处理缺失值（看到未来）
• 计算 vol 时用未来数据
• 数据修正后的"清洁"价格（survivorship 数据）

严格规则：信号 $s_t$ 只能用 ${x_\tau : \tau \le t}$，仓位 $w_t$ 应用到 $r_{t+1}$。

1.4 撮合模型

简化层级：

滑点函数（square-root law）：

$$ \text{slippage} \approx \sigma_d \sqrt{\frac{Q}{V}} $$

• $\sigma_d$ 日波动率
• $Q$ 订单量
• $V$ 日均成交量

1.5 Walk-forward 优化

[Train 1][Test 1]
        [Train 2][Test 2]
                [Train 3][Test 3]

避免在 test 上调参。每段：用 train 选最优参数，在 test 上验证。

Anchor walk-forward：训练窗口固定起点，逐渐扩大 Rolling walk-forward：训练窗口滚动等长

1.6 过拟合检测

Probabilistic Sharpe Ratio (PSR)

$$ \text{PSR}(\text{SR}^) = \Phi\left(\frac{(\hat{\text{SR}} - \text{SR}^) \sqrt{n - 1}}{\sqrt{1 - \hat\gamma_3 \hat{\text{SR}} + \frac{\hat\gamma_4 - 1}{4} \hat{\text{SR}}^2}}\right) $$

• $\hat{\text{SR}}$：估计的 Sharpe
• $n$：样本数
• $\gamma_3, \gamma_4$：偏度、峰度
• 测试 SR > SR* 的概率

Deflated Sharpe Ratio (DSR)

对 N 次试错做校正：

$$ \text{DSR} = \text{PSR}(\text{SR}^_N), \quad \text{SR}^_N = \sqrt{V[\text{SRs}]} \left((1 - \gamma_E) \Phi^{-1}\left(1 - \frac{1}{N}\right) + \gamma_E \Phi^{-1}\left(1 - \frac{1}{N e}\right)\right) $$

DSR > 0.95 才能宣称策略有效。

---

二、直觉与陷阱

陷阱 1：在测试集调参（隐性）

即使没显式调参，反复在 out-of-sample 看效果，根据效果改主体逻辑也是过拟合。 解法：研究者预先冻结测试集，"final report" 只看一次。

陷阱 2：滑点估计的乐观

加密策略经典坑：

• 高频策略回测假设 0 slippage → 实盘 -50% 收益
• 山寨币用主流币的 spread 估计 → 严重高估 capacity

解法：分品种估滑点；前 5% 收益最优时段排除（often 异常 spike）。

陷阱 3：dropna 的隐藏 look-ahead

df.dropna()  # 删除任何含 NaN 的行

这等于"未来已知该资产存在数据"。正确做法：用 forward-fill 但限制天数；或在 t 时刻用 t 之前的可用数据。

陷阱 4：成交量限制被忽略

回测下单 100 BTC，假设全成交。但当时市场 5 分钟成交量可能只有 50 BTC。 解法：order size 限制为 ADV × 1%（保守）或 5%（激进）。

陷阱 5：rebalance 的成本下沉

月度 vs 周度 rebalance：

• 月度：低 turnover 但 capture 低
• 周度：高 turnover 高成本
• 找最优 rebalance 频率（不是越频越好）

陷阱 6：稳定币假设崩溃

回测把 USDC 当 1 美元处理。2023-03-11 USDC 跌到 0.87。如果策略大量持 USDC 且回测忽略，回测高估收益。

---

三、代码实现

3.1 完整回测框架（模块化）

# backtest_v1/__init__.py
"""
Modular Backtesting Framework
- Layer 1: Data
- Layer 2: Signal
- Layer 3: Execution
- Layer 4: Risk
- Layer 5: Analysis
"""

# backtest_v1/data.py
import pandas as pd
import numpy as np
from typing import Dict, List, Optional


class DataHandler:
    """数据层：处理 OHLCV，确保 point-in-time"""

    def __init__(self, prices: pd.DataFrame, volumes: pd.DataFrame = None):
        """
        prices: index = datetime, columns = symbols
        volumes: 同上
        """
        self.prices = prices.sort_index()
        self.volumes = volumes.sort_index() if volumes is not None else None
        self._validate()

    def _validate(self):
        assert self.prices.index.is_monotonic_increasing, "Index must be sorted"
        assert not self.prices.index.duplicated().any(), "Duplicate timestamps"

    def get_returns(self) -> pd.DataFrame:
        return self.prices.pct_change()

    def get_log_returns(self) -> pd.DataFrame:
        return np.log(self.prices / self.prices.shift(1))

    def slice(self, start, end) -> 'DataHandler':
        return DataHandler(
            self.prices.loc[start:end],
            self.volumes.loc[start:end] if self.volumes is not None else None
        )

    def assert_no_lookahead(self, signal: pd.DataFrame):
        """断言 signal 不含未来信息（基于索引对齐）"""
        assert signal.index.equals(self.prices.index), \
            "Signal index must match data index"


# backtest_v1/signal.py
from abc import ABC, abstractmethod


class Signal(ABC):
    """信号生成基类"""

    @abstractmethod
    def generate(self, data: DataHandler) -> pd.DataFrame:
        """返回 [-1, 1] 范围内的信号矩阵"""
        pass


class TSMOMSignal(Signal):
    def __init__(self, lookback: int = 90):
        self.lookback = lookback

    def generate(self, data: DataHandler) -> pd.DataFrame:
        ret = data.get_returns()
        cum = (1 + ret).rolling(self.lookback).apply(np.prod) - 1
        return np.sign(cum)


class MeanReversionSignal(Signal):
    def __init__(self, lookback: int = 30, entry_z: float = 2.0):
        self.lookback = lookback
        self.entry_z = entry_z

    def generate(self, data: DataHandler) -> pd.DataFrame:
        prices = data.prices
        log_p = np.log(prices)
        mu = log_p.rolling(self.lookback).mean()
        sigma = log_p.rolling(self.lookback).std()
        z = (log_p - mu) / sigma
        sig = pd.DataFrame(0.0, index=prices.index, columns=prices.columns)
        sig[z > self.entry_z] = -1
        sig[z < -self.entry_z] = 1
        return sig


# backtest_v1/execution.py
class ExecutionModel:
    """撮合层：模拟订单执行"""

    def __init__(self, fee_bps: float = 5,
                  slippage_model: str = 'linear',
                  slippage_bps: float = 5,
                  max_participation: float = 0.05):
        """
        slippage_model: 'none', 'linear', 'sqrt'
        max_participation: 单笔最大占 ADV 比例
        """
        self.fee_bps = fee_bps
        self.slippage_model = slippage_model
        self.slippage_bps = slippage_bps
        self.max_participation = max_participation

    def execute(self, target_weights: pd.DataFrame,
                 prices: pd.DataFrame,
                 volumes: pd.DataFrame = None) -> Dict:
        """
        计算实际交易、成本、滑点
        关键：使用 t 时刻的 weight 应用到 t+1 的收益
        """
        # 上一周期持仓
        prev_weights = target_weights.shift(1).fillna(0)
        # 周期内交易（变化量）
        delta = (target_weights - prev_weights).abs()

        # 成交量约束（仅当有 volume 数据）
        if volumes is not None and self.max_participation > 0:
            adv = volumes.rolling(20).mean()
            max_size_pct = self.max_participation
            actual_delta = delta.copy()
            # 简化处理：记录但不强制限制（实盘要分笔）
            # actual_delta = ... (省略复杂逻辑)
        else:
            actual_delta = delta

        # 成本
        fee_cost = actual_delta * (self.fee_bps / 10_000)

        # 滑点
        if self.slippage_model == 'linear':
            slip_cost = actual_delta * (self.slippage_bps / 10_000)
        elif self.slippage_model == 'sqrt' and volumes is not None:
            adv = volumes.rolling(20).mean()
            participation = actual_delta / (adv + 1e-10)
            slip_cost = actual_delta * 0.5 * np.sqrt(participation.fillna(0))
        else:
            slip_cost = pd.DataFrame(0.0, index=delta.index, columns=delta.columns)

        total_cost = fee_cost + slip_cost
        return {
            'actual_weights': prev_weights + (actual_delta * np.sign(target_weights - prev_weights)),
            'turnover': delta,
            'fee_cost': fee_cost,
            'slip_cost': slip_cost,
            'total_cost': total_cost,
        }


# backtest_v1/risk.py
class RiskManager:
    """风控层：仓位限制、止损、暴露管理"""

    def __init__(self, max_position: float = 0.2,
                  max_gross_exposure: float = 1.0,
                  max_net_exposure: float = 0.5,
                  vol_target: Optional[float] = None,
                  stop_loss: Optional[float] = None):
        self.max_position = max_position
        self.max_gross = max_gross_exposure
        self.max_net = max_net_exposure
        self.vol_target = vol_target
        self.stop_loss = stop_loss

    def apply(self, weights: pd.DataFrame, returns: pd.DataFrame) -> pd.DataFrame:
        """应用风控"""
        w = weights.copy()

        # 单仓限制
        w = w.clip(-self.max_position, self.max_position)

        # 总暴露限制
        gross = w.abs().sum(axis=1)
        scale_g = (self.max_gross / gross).clip(upper=1).fillna(1)
        w = w.mul(scale_g, axis=0)

        # 净暴露限制
        net = w.sum(axis=1).abs()
        scale_n = (self.max_net / net).clip(upper=1).fillna(1)
        w = w.mul(scale_n, axis=0)

        # 波动率目标化
        if self.vol_target is not None:
            port_ret = (w.shift(1) * returns).sum(axis=1)
            rolling_vol = port_ret.rolling(30).std() * np.sqrt(365)
            scale_v = (self.vol_target / rolling_vol).clip(upper=3).fillna(1)
            w = w.mul(scale_v, axis=0)

        return w


# backtest_v1/engine.py
class BacktestEngine:
    """主引擎：组装各层"""

    def __init__(self, data: DataHandler, signal: Signal,
                  execution: ExecutionModel, risk: RiskManager):
        self.data = data
        self.signal = signal
        self.execution = execution
        self.risk = risk

    def run(self, initial_capital: float = 100_000) -> Dict:
        # Step 1: 生成信号
        raw_signal = self.signal.generate(self.data)
        self.data.assert_no_lookahead(raw_signal)

        # Step 2: 风控调整
        weights = self.risk.apply(raw_signal, self.data.get_returns())

        # Step 3: 执行
        exec_result = self.execution.execute(
            weights, self.data.prices, self.data.volumes)

        # Step 4: 计算 PnL
        returns = self.data.get_returns()
        # 关键：weight 在 t 决策，应用到 t+1 收益
        gross_ret = (exec_result['actual_weights'].shift(1) * returns).sum(axis=1)
        cost = exec_result['total_cost'].sum(axis=1)
        net_ret = gross_ret - cost

        equity = initial_capital * (1 + net_ret).cumprod()

        return {
            'weights': exec_result['actual_weights'],
            'turnover': exec_result['turnover'],
            'gross_ret': gross_ret,
            'cost': cost,
            'net_ret': net_ret,
            'equity': equity,
        }


# backtest_v1/analysis.py
from scipy import stats


class PerformanceAnalyzer:
    """分析层：完整指标"""

    @staticmethod
    def metrics(returns: pd.Series, freq: int = 365) -> Dict:
        r = returns.dropna()
        if len(r) < 2:
            return {}

        ann_ret = (1 + r.mean()) ** freq - 1
        ann_vol = r.std() * np.sqrt(freq)
        sharpe = ann_ret / ann_vol if ann_vol > 0 else 0

        # Downside
        downside = r[r < 0]
        sortino = ann_ret / (downside.std() * np.sqrt(freq)) if len(downside) > 0 else np.inf

        # Drawdown
        eq = (1 + r).cumprod()
        dd = (eq / eq.cummax() - 1).min()
        calmar = ann_ret / abs(dd) if dd < 0 else np.inf

        # Skew/Kurt
        skew = r.skew()
        kurt = r.kurtosis()

        # PSR
        sr_obs = r.mean() / r.std() if r.std() > 0 else 0
        n = len(r)
        psr_denom = np.sqrt(1 - skew * sr_obs + ((kurt - 1) / 4) * sr_obs**2)
        psr = stats.norm.cdf((sr_obs - 0) * np.sqrt(n - 1) / psr_denom) if psr_denom > 0 else 0.5

        return {
            'ann_return': ann_ret,
            'ann_vol': ann_vol,
            'sharpe': sharpe,
            'sortino': sortino,
            'max_drawdown': dd,
            'calmar': calmar,
            'skew': skew,
            'kurt': kurt,
            'psr': psr,
        }

    @staticmethod
    def deflated_sharpe(sr_obs: float, sr_history: List[float],
                         n_obs: int, skew: float, kurt: float) -> float:
        """N 次试错的 Deflated Sharpe Ratio"""
        N = len(sr_history)
        if N < 2:
            return sr_obs

        sr_var = np.var(sr_history)
        gamma = 0.5772  # Euler-Mascheroni
        e = np.e

        # Expected max SR under null
        sr_star = np.sqrt(sr_var) * (
            (1 - gamma) * stats.norm.ppf(1 - 1/N) +
            gamma * stats.norm.ppf(1 - 1/(N * e))
        )

        denom = np.sqrt(1 - skew * sr_obs + ((kurt - 1) / 4) * sr_obs**2)
        dsr = stats.norm.cdf((sr_obs - sr_star) * np.sqrt(n_obs - 1) / denom)
        return dsr


# backtest_v1/walk_forward.py
def walk_forward_validate(data: DataHandler, signal_class, exec_model,
                           risk_model, train_window: int = 365,
                           test_window: int = 90, param_grid: Dict = None) -> pd.DataFrame:
    """Walk-forward 验证"""
    n = len(data.prices)
    results = []
    start = train_window

    while start + test_window <= n:
        train_data = data.slice(data.prices.index[start - train_window],
                                  data.prices.index[start - 1])
        test_data = data.slice(data.prices.index[start],
                                 data.prices.index[start + test_window - 1])

        # 简化：用固定参数（实战中在 train 上选参数）
        if param_grid:
            # 网格搜索
            best_sr = -np.inf
            best_params = None
            for params in param_grid:
                sig = signal_class(**params)
                eng = BacktestEngine(train_data, sig, exec_model, risk_model)
                res = eng.run()
                sr = PerformanceAnalyzer.metrics(res['net_ret']).get('sharpe', 0)
                if sr > best_sr:
                    best_sr = sr
                    best_params = params
            sig = signal_class(**best_params)
        else:
            sig = signal_class()

        # 在 test 上验证
        eng = BacktestEngine(test_data, sig, exec_model, risk_model)
        test_res = eng.run()
        test_metrics = PerformanceAnalyzer.metrics(test_res['net_ret'])
        test_metrics['period_start'] = test_data.prices.index[0]
        test_metrics['period_end'] = test_data.prices.index[-1]
        results.append(test_metrics)

        start += test_window

    return pd.DataFrame(results)


# Main
if __name__ == '__main__':
    import requests
    def fetch(s, d=730):
        r = requests.get('https://api.binance.com/api/v3/klines',
                         params={'symbol': s, 'interval': '1d', 'limit': d})
        df = pd.DataFrame(r.json(), columns=['ot','o','h','l','c','v','ct',
                                              'qav','t','tb','tq','i'])
        df['ct'] = pd.to_datetime(df['ct'], unit='ms')
        df['c'] = df['c'].astype(float)
        df['v'] = df['v'].astype(float)
        return df.set_index('ct')[['c', 'v']]

    syms = ['BTCUSDT', 'ETHUSDT', 'SOLUSDT']
    data_dict = {s: fetch(s) for s in syms}
    prices = pd.DataFrame({s: data_dict[s]['c'] for s in syms})
    volumes = pd.DataFrame({s: data_dict[s]['v'] for s in syms})

    data = DataHandler(prices, volumes)
    signal = TSMOMSignal(lookback=60)
    exec_model = ExecutionModel(fee_bps=5, slippage_bps=10)
    risk = RiskManager(max_position=0.5, vol_target=0.20)

    engine = BacktestEngine(data, signal, exec_model, risk)
    result = engine.run()

    metrics = PerformanceAnalyzer.metrics(result['net_ret'])
    print("Backtest Results:")
    for k, v in metrics.items():
        print(f"  {k}: {v:.4f}")

    # Walk-forward
    print("\nWalk-Forward Validation:")
    wf = walk_forward_validate(data, TSMOMSignal, exec_model, risk,
                                train_window=300, test_window=60,
                                param_grid=[{'lookback': lb} for lb in [30, 60, 90]])
    print(wf[['period_start', 'sharpe', 'max_drawdown']])

---

四、真实数据/案例

案例 1：Renaissance 的回测哲学

文艺复兴的回测原则（公开访谈）：

• 至少 20 年样本，分 4 段 5 年验证一致性
• 任何单段 Sharpe < 0.7 整体策略不上
• 严格区分 in-sample 和 out-of-sample
• 上线前用 6 个月真实小资金 paper trading

案例 2：QuantConnect 平台一项研究

QC 测试 1000+ 用户提交策略：

• 73% 在 in-sample 表现优秀
• 仅 12% 在 out-of-sample Sharpe > 0.5
• 仅 3% 在实盘维持 6 个月以上
• DSR 测试（Bailey-Lopez de Prado）能淘汰 80% in-sample 优秀但实际无效的策略

案例 3：BitMEX 数据用错的著名错误

某加密对冲基金 2019 用 BitMEX 永续合约数据回测 mean reversion 策略，结论 Sharpe = 3.5。 错误：BitMEX 永续 mark price 来自现货 index，与实际成交价有 50ms-2s 延迟。 纠正：用真实 trade price 回测，Sharpe = 0.4。

案例 4：Survivorship bias 毁掉的"百倍策略"

简单回测：选 2017 年 top 100 加密资产做 momentum，结果 5 年年化 100%+。 问题：100 个里 60+ 已归零或退市，只看现存的就高估收益。 正确：用历史 ranking 快照（point-in-time），结果年化 ≈ 25%。

---

五、CEX vs DEX 策略差异

DeFi 回测特殊考量：

1. AMM 滑点精确建模：用 Curve / Uniswap V3 公式，不是固定 bps
2. Pending tx 重组：被 reorg 的交易回滚
3. MEV bot 抢跑：你的策略信号公开则被抢
4. Liquidity migration：池子流动性突然枯竭

---

六、风险管理

6.1 回测可信度检查清单

• No look-ahead（信号严格 t-1 之前）
• 包含足够 regime（牛 + 熊 + 横盘）
• Out-of-sample 至少占 30%
• DSR > 0.95
• 多参数稳定（参数 ±20% 仍 SR > 0.5）
• 多市场稳定（BTC/ETH/SOL 都有效）
• 滑点保守估计（× 1.5）
• Capacity 不超过 ADV × 5%

6.2 回测信心区间

不要只报 Sharpe = 1.5，要报 95% CI：

$$ \text{SE}(\hat{\text{SR}}) \approx \sqrt{\frac{1 + 0.5 \hat{\text{SR}}^2}{n}} $$

n = 252 时 SR = 1.5 的 95% CI 大约 [1.2, 1.8]

---

七、关键速查

框架架构图

┌──────────┐    ┌──────────┐    ┌──────────┐
│   Data   │ -> │  Signal  │ -> │   Risk   │
└──────────┘    └──────────┘    └──────────┘
                                       │
                                       v
┌──────────┐    ┌──────────┐    ┌──────────┐
│ Analysis │ <- │   PnL    │ <- │ Execute  │
└──────────┘    └──────────┘    └──────────┘

必备指标

---

八、面试题

Q1：你回测出 Sharpe = 2.0 的策略，给我列举不上线的可能原因。

答：

1. Look-ahead bias（无意中用未来数据）
2. Survivorship bias（只回测现存资产）
3. 在 out-of-sample 反复调参
4. 滑点低估（特别是高频或大资金量）
5. Capacity 不足（实盘冲击成本远超回测）
6. 数据延迟（mark price vs 实际可成交价）
7. 单一市场 / 单一时段（非稳健）
8. 过拟合（在 1000 次试错中 cherry-pick）

Q2：怎么诊断 look-ahead bias？

答：

• 代码审计：所有信号 df.shift(1) 至少一次
• 时间戳检查：信号时间戳 + 数据延迟应 ≤ 决策时间
• 滚动检查：把回测分两半，前半段在后半段不该有"提前知道未来"的优势
• 极端测试：把数据 reverse 一下，结果应该完全不同（否则代码有 bug）
• Lopez de Prado 提议：每个步骤明确 t 时点

Q3：Walk-forward 和 cross-validation 的区别？

答：

• CV：随机分 K 折，每折训练 K-1 折预测剩 1 折。违反时序性，金融数据不能用
• Walk-forward：严格时序，过去训练未来预测
• Combinatorial Purged CV (CPCV)：Lopez de Prado 提议，组合 k 个 train/test 段且 purge 重叠期，比 walk-forward 信息利用率更高
• 加密推荐：基础策略用 walk-forward；机器学习类用 CPCV

Q4：vectorbt vs backtrader 怎么选？

答：

• vectorbt：批量参数扫描快（1000+ 组合秒级），适合因子研究和信号筛选
• backtrader：事件驱动，逻辑灵活（可写复杂订单管理），适合实盘准备
• 实战流程：
  1. vectorbt 筛选 → 找前 5% 最有希望的参数
  2. backtrader 验证 → 加真实订单管理、滑点、仓位风控
  3. paper trading 6 月 → 上线
• 自己写框架的好处：完全掌握每个细节，避免库的隐藏 bug

Q5：DSR > 0.95 是什么意思？为什么是 0.95？

答：

• DSR = 在 N 次回测试错下，"真实 Sharpe > 0" 的后验概率
• 0.95 = 95% 置信度，类似 p-value < 0.05 的统计显著性
• N 越大，门槛 SR* 越高（试错越多越要严苛）
• 例：试 100 次回测，N=100，要求 DSR > 0.95，相当于真实 SR 在 1.5 以上才显著
• 实战意义：发表论文 / 上线策略前必检验

---

明日预告

Day 94: 风险管理 — VaR/ES、最大回撤、Sharpe/Sortino/Calmar、Kelly、压力测试。今天的回测框架是架子，明天填充风控的肉。