22、曲面套利回测框架：数据获取（期权链数据）、信号生成、执行模拟、绩效评估

做曲面套利回测，说白了就是要把你的策略想法放到历史数据里跑一遍。

我见过太多人，策略逻辑写得天花乱坠，一到回测就露馅。为什么？因为框架没搭好。数据拿错了、信号算偏了、执行滑点没考虑——任何一个环节出问题，结果都是废的。

今天我就把整个回测框架拆成四个模块来讲：数据获取、信号生成、执行模拟、绩效评估。每个模块我都会结合自己踩过的坑来说。

22.1 数据获取：期权链数据的那些坑

数据是回测的命根子。期权数据比股票数据复杂得多，我刚开始做的时候，光数据清洗就折腾了两周。

你需要的数据包括：

• 期权合约数据：行权价、到期日、合约类型（Call/Put）、交易代码
• 行情数据：开盘价、收盘价、最高价、最低价、成交量、持仓量
• 标的资产数据：标的价格、分红、拆股等事件
• 无风险利率：通常用国债收益率或SHIBOR
• 隐含波动率曲面：这是核心，需要根据期权价格反推

我个人习惯用 pandas-datareader 或者 yfinance 拉取美股期权数据。但国内期权数据比较麻烦，我一般用万得或者聚宽的API。

下面是一个简单的数据获取代码示例：

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def fetch_option_chain(ticker, expiry_date, source='yahoo'):
    """
    获取期权链数据
    ticker: 标的代码
    expiry_date: 到期日
    """
    # 这里用伪代码表示，实际需要根据数据源调整
    # 从数据源获取所有行权价的Call和Put
    calls = get_calls(ticker, expiry_date)
    puts = get_puts(ticker, expiry_date)
    
    # 合并数据
    chain = pd.concat([calls, puts], axis=0)
    
    # 计算隐含波动率
    chain['implied_vol'] = chain.apply(
        lambda row: calculate_iv(row), axis=1
    )
    
    return chain

def calculate_iv(row):
    """使用牛顿法反推隐含波动率"""
    # 这里省略具体实现
    pass

22.2 信号生成：从曲面到交易信号

数据拿到手了，接下来就是生成交易信号。曲面套利的核心逻辑其实不复杂——找到曲面上的异常点。

我常用的信号生成流程：

1. 构建波动率曲面：用SVI或者样条插值，把离散的隐含波动率点连成曲面
2. 计算理论波动率：根据曲面模型，算出每个合约的"合理"波动率
3. 计算偏差：实际隐含波动率 - 理论波动率
4. 设定阈值：偏差超过2个标准差，就认为是套利机会

你想想看，如果某个合约的隐含波动率比曲面模型算出来的高了3个点，那大概率是市场定价错了。这时候就可以考虑做空该合约的波动率，同时用其他合约对冲。

代码实现大致是这样：

def generate_signals(surface_data, threshold=2.0):
    """
    生成曲面套利信号
    surface_data: 包含所有合约的波动率数据
    threshold: 标准差阈值
    """
    # 构建曲面模型
    model = fit_surface_model(surface_data)
    
    # 计算每个合约的理论波动率
    surface_data['theoretical_vol'] = model.predict(surface_data)
    
    # 计算偏差
    surface_data['deviation'] = (
        surface_data['implied_vol'] - surface_data['theoretical_vol']
    )
    
    # 标准化偏差
    mean_dev = surface_data['deviation'].mean()
    std_dev = surface_data['deviation'].std()
    surface_data['z_score'] = (
        surface_data['deviation'] - mean_dev
    ) / std_dev
    
    # 生成信号
    surface_data['signal'] = 0
    surface_data.loc[surface_data['z_score'] > threshold, 'signal'] = -1  # 做空
    surface_data.loc[surface_data['z_score'] < -threshold, 'signal'] = 1  # 做多
    
    return surface_data[surface_data['signal'] != 0]

22.3 执行模拟：别让理想照不进现实

信号有了，怎么执行？很多人的回测在这里就翻车了。

我刚开始做回测的时候，假设信号一出来就能以当前价成交。结果实盘发现，滑点能把利润吃掉一大半。尤其是期权市场，流动性差的合约，买卖价差大得吓人。

执行模拟要考虑的因素：

• 滑点：我一般按买卖价差的50%来估算
• 流动性限制：成交量太小的合约，直接过滤掉
• 交易成本：手续费、印花税、保证金占用成本
• 冲击成本：大单交易对市场的影响

执行模拟的代码框架：

def simulate_execution(signals, market_data, slippage=0.0005):
    """
    模拟交易执行
    signals: 交易信号
    market_data: 市场行情数据
    slippage: 滑点比例
    """
    trades = []
    
    for idx, signal in signals.iterrows():
        # 获取当前市场报价
        bid = market_data.loc[idx, 'bid']
        ask = market_data.loc[idx, 'ask']
        
        # 根据信号方向确定成交价
        if signal['signal'] == 1:  # 买入
            exec_price = ask * (1 + slippage)
        else:  # 卖出
            exec_price = bid * (1 - slippage)
        
        # 检查流动性
        volume = market_data.loc[idx, 'volume']
        if volume < signal['position_size'] * 10:
            continue  # 流动性不足，跳过
        
        trades.append({
            'timestamp': idx,
            'contract': signal['contract'],
            'side': 'buy' if signal['signal'] == 1 else 'sell',
            'price': exec_price,
            'size': signal['position_size']
        })
    
    return pd.DataFrame(trades)

22.4 绩效评估：别只看收益率

回测跑完了，怎么评价策略好不好？很多人只看年化收益率和最大回撤，这远远不够。

我常用的绩效指标：

指标	计算公式	说明
年化收益率	(总收益率+1)^(252/交易天数)-1	衡量盈利能力
夏普比率	(策略收益率-无风险利率)/波动率	风险调整后收益
最大回撤	max(峰值-谷值)/峰值	衡量下行风险
胜率	盈利交易次数/总交易次数	交易质量
盈亏比	平均盈利/平均亏损	风险回报比
Calmar比率	年化收益率/最大回撤	回撤调整后收益

我个人最看重的是夏普比率和Calmar比率。夏普比率低于1的策略，我基本不会考虑。Calmar比率能告诉你，为了获得这些收益，你承受了多大的回撤风险。

绩效评估的代码实现：

def evaluate_performance(trades, equity_curve, risk_free_rate=0.03):
    """
    绩效评估
    trades: 交易记录
    equity_curve: 权益曲线
    risk_free_rate: 无风险利率
    """
    # 计算收益率
    returns = equity_curve.pct_change().dropna()
    
    # 年化收益率
    total_return = equity_curve.iloc[-1] / equity_curve.iloc[0] - 1
    trading_days = len(returns)
    annual_return = (1 + total_return) ** (252 / trading_days) - 1
    
    # 夏普比率
    excess_returns = returns - risk_free_rate / 252
    sharpe = np.sqrt(252) * excess_returns.mean() / returns.std()
    
    # 最大回撤
    cumulative = equity_curve / equity_curve.iloc[0]
    running_max = cumulative.expanding().max()
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = drawdown.min()
    
    # 胜率
    winning_trades = trades[trades['pnl'] > 0]
    win_rate = len(winning_trades) / len(trades)
    
    # 盈亏比
    avg_win = winning_trades['pnl'].mean()
    avg_loss = trades[trades['pnl'] <= 0]['pnl'].mean()
    profit_loss_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else np.inf
    
    return {
        'annual_return': annual_return,
        'sharpe_ratio': sharpe,
        'max_drawdown': max_drawdown,
        'win_rate': win_rate,
        'profit_loss_ratio': profit_loss_ratio,
        'calmar_ratio': annual_return / abs(max_drawdown) if max_drawdown != 0 else np.inf
    }

22.5 整体框架流程图

说了这么多，我把整个框架画成一张图，方便你理解各个模块之间的关系。

这张图把整个框架串起来了。你从数据获取开始，一步步走到绩效评估，然后根据评估结果反过来优化数据清洗逻辑和信号参数。嗯，这就是一个完整的回测闭环。