第二十九节：量化工具——Python实现基差交易策略回测框架

做基差交易，光有理论可不够。

你得有工具，能快速验证你的想法。我见过太多人，脑子里有个策略觉得稳赚，结果一跑回测，亏得底裤都不剩。所以今天，咱们就动手搭一个基差交易策略的回测框架。

说白了，就是用Python把「库存-价差」的逻辑，变成可执行的代码。

29.1 回测框架的核心模块

一个完整的回测框架，至少需要四个部分：

• 数据模块：负责拉取期货价格、现货价格、库存数据
• 信号模块：根据库存和价差的关系，生成买卖信号
• 执行模块：模拟下单、成交、滑点
• 绩效模块：计算收益率、夏普比率、最大回撤

我个人习惯，先把数据模块搞定。数据不干净，后面全是白搭。

29.2 数据准备与预处理

先看数据长什么样。我一般用pandas的DataFrame来存。

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# 模拟数据：日期、期货价格、现货价格、库存
dates = pd.date_range('2023-01-01', periods=500, freq='D')
df = pd.DataFrame({
    'date': dates,
    'futures': np.random.randn(500).cumsum() + 100,
    'spot': np.random.randn(500).cumsum() + 98,
    'inventory': np.random.randn(500).cumsum() + 200
})

# 计算基差
df['basis'] = df['futures'] - df['spot']

# 计算库存的Z-score（标准化）
df['inv_z'] = (df['inventory'] - df['inventory'].rolling(60).mean()) / df['inventory'].rolling(60).std()

# 计算基差的Z-score
df['basis_z'] = (df['basis'] - df['basis'].rolling(60).mean()) / df['basis'].rolling(60).std()

print(df.head())

嗯，这里要注意。滚动窗口选多少？我试过20天、60天、120天。对于大多数商品，60天比较稳。太短了噪音大，太长了反应慢。

29.3 信号生成逻辑

信号怎么来？核心就一句话：库存极端 + 基差极端 = 交易机会。

我常用的规则是这样的：

• 当库存Z-score > 1.5 且 基差Z-score < -1.0，做多基差（买现货，卖期货）
• 当库存Z-score < -1.5 且 基差Z-score > 1.0，做空基差（卖现货，买期货）
• 其他情况，持仓不动或平仓

你想想看，库存高说明供应过剩，基差弱说明期货比现货贵得不多，这时候做多基差，赌的就是库存回归。

# 生成信号
df['signal'] = 0

# 做多基差信号
long_condition = (df['inv_z'] > 1.5) & (df['basis_z'] < -1.0)
df.loc[long_condition, 'signal'] = 1

# 做空基差信号
short_condition = (df['inv_z'] < -1.5) & (df['basis_z'] > 1.0)
df.loc[short_condition, 'signal'] = -1

# 信号不能频繁变动，我习惯做一次持仓判断
df['position'] = df['signal'].replace(0, method='ffill').fillna(0)

print(df[df['signal'] != 0].head())

29.4 回测执行与绩效计算

执行模块，说白了就是算每天的盈亏。基差交易，我们赚的是基差回归的钱。

# 计算每日基差变化
df['basis_change'] = df['basis'].diff()

# 策略收益 = 持仓方向 * 基差变化
df['strategy_return'] = df['position'].shift(1) * df['basis_change']

# 累计收益
df['cum_return'] = (1 + df['strategy_return']).cumprod()

# 绩效指标
total_return = df['cum_return'].iloc[-1] - 1
annual_return = total_return / (len(df) / 252)
sharpe = np.sqrt(252) * df['strategy_return'].mean() / df['strategy_return'].std()
max_drawdown = (df['cum_return'] / df['cum_return'].cummax() - 1).min()

print(f"总收益率: {total_return:.2%}")
print(f"年化收益率: {annual_return:.2%}")
print(f"夏普比率: {sharpe:.2f}")
print(f"最大回撤: {max_drawdown:.2%}")

夏普比率低于1，我基本不会实盘。低于0.5，直接放弃。

29.5 可视化与框架图

光看数字不够直观。我习惯把策略表现画出来。

fig, axes = plt.subplots(3, 1, figsize=(12, 10))

# 基差与库存
axes[0].plot(df['date'], df['basis'], label='Basis', color='blue')
axes[0].plot(df['date'], df['inventory']/10, label='Inventory/10', color='green')
axes[0].legend()
axes[0].set_title('Basis vs Inventory')

# 信号
axes[1].plot(df['date'], df['position'], label='Position', color='red', drawstyle='steps-post')
axes[1].legend()
axes[1].set_title('Trading Signal')

# 累计收益
axes[2].plot(df['date'], df['cum_return'], label='Cumulative Return', color='purple')
axes[2].legend()
axes[2].set_title('Strategy Performance')

plt.tight_layout()
plt.show()

下面这张SVG图，是我自己总结的基差回测框架流程。你看一眼就明白了。

29.6 完整框架封装

最后，我把上面这些代码封装成一个类。以后换品种，改参数就行。

class BasisTradingBacktest:
    def __init__(self, data, window=60, inv_threshold=1.5, basis_threshold=1.0):
        self.data = data.copy()
        self.window = window
        self.inv_threshold = inv_threshold
        self.basis_threshold = basis_threshold
        
    def prepare_data(self):
        self.data['basis'] = self.data['futures'] - self.data['spot']
        self.data['inv_z'] = (self.data['inventory'] - 
                              self.data['inventory'].rolling(self.window).mean()) / \
                              self.data['inventory'].rolling(self.window).std()
        self.data['basis_z'] = (self.data['basis'] - 
                                self.data['basis'].rolling(self.window).mean()) / \
                                self.data['basis'].rolling(self.window).std()
        
    def generate_signals(self):
        self.data['signal'] = 0
        long_cond = (self.data['inv_z'] > self.inv_threshold) & \
                    (self.data['basis_z'] < -self.basis_threshold)
        short_cond = (self.data['inv_z'] < -self.inv_threshold) & \
                     (self.data['basis_z'] > self.basis_threshold)
        self.data.loc[long_cond, 'signal'] = 1
        self.data.loc[short_cond, 'signal'] = -1
        self.data['position'] = self.data['signal'].replace(0, method='ffill').fillna(0)
        
    def run(self):
        self.prepare_data()
        self.generate_signals()
        self.data['basis_change'] = self.data['basis'].diff()
        self.data['strategy_return'] = self.data['position'].shift(1) * self.data['basis_change']
        self.data['cum_return'] = (1 + self.data['strategy_return']).cumprod()
        return self.data

# 使用示例
bt = BasisTradingBacktest(df, window=60, inv_threshold=1.5, basis_threshold=1.0)
result = bt.run()
print(result[['date', 'basis', 'position', 'cum_return']].tail())

好了，框架搭完了。你拿自己的数据跑一跑，调调参数，很快就能找到感觉。记住，回测只是起点，不是终点。