第13章 Python实现基差计算

第13章：Python实现基差计算：Pandas数据处理、滚动窗口统计、向量化回测实现

各位同学，欢迎来到实战环节。前面我们聊了很多基差交易的逻辑和策略框架，但说实话，纸上谈兵没意思。今天我们就用Python把基差算出来，把策略跑起来。

我个人习惯是，先搞定数据，再谈策略。基差计算的核心就三件事：数据对齐、滚动统计、回测验证。咱们一个一个来。

13.1 数据准备：Pandas处理期货与现货数据

做基差交易，你至少需要两套数据：现货价格和期货价格。而且时间戳必须对齐。我在项目中遇到过最坑的事，就是现货数据是日频，期货数据是分钟频，直接合并会出大问题。

先看代码：

import pandas as pd
import numpy as np

# 模拟现货数据（日频）
spot_data = {
    'date': pd.date_range('2024-01-01', periods=100, freq='D'),
    'spot_price': np.random.randn(100).cumsum() + 5000
}
spot_df = pd.DataFrame(spot_data).set_index('date')

# 模拟期货数据（日频，主力合约）
futures_data = {
    'date': pd.date_range('2024-01-01', periods=100, freq='D'),
    'futures_price': np.random.randn(100).cumsum() + 5050
}
futures_df = pd.DataFrame(futures_data).set_index('date')

# 合并数据
df = pd.merge(spot_df, futures_df, left_index=True, right_index=True, how='inner')
df.head()

这里我用的是merge，而不是concat。为什么？因为merge能自动对齐索引，保证两个数据集在相同日期上匹配。你想想看，如果某天现货有数据但期货停牌，inner模式会直接剔除，避免脏数据。

我的小技巧： 实际项目中，期货数据经常有换月跳空。我建议用复权处理，或者只取主力合约连续数据。否则基差会突然跳变，策略信号全乱套。

13.2 基差计算：向量化操作

基差 = 期货价格 - 现货价格。就这么简单。但要注意，不要用for循环。Pandas的向量化操作快得多，尤其数据量上百万行时，差距是秒级和分钟级的区别。

# 向量化计算基差
df['basis'] = df['futures_price'] - df['spot_price']

# 计算基差率（相对值）
df['basis_pct'] = df['basis'] / df['spot_price'] * 100

print(df[['spot_price', 'futures_price', 'basis', 'basis_pct']].head())

嗯，这里要注意：基差率用百分比表示，方便不同品种之间对比。比如螺纹钢基差50块，和铜基差500块，绝对值没法比，但基差率就能看出谁更偏离。

  核心要点： 基差计算本身不复杂，复杂的是数据清洗和异常值处理。我曾经因为期货数据里混入了非主力合约的报价，导致基差曲线出现尖峰，回测结果完全失真。

13.3 滚动窗口统计：捕捉基差的均值回归特性

基差交易的核心假设是：基差会围绕某个均值波动。所以我们需要计算滚动均值和滚动标准差，来判断当前基差是否处于极端位置。

# 设置滚动窗口（比如20个交易日）
window = 20

# 滚动均值
df['basis_ma'] = df['basis'].rolling(window=window).mean()

# 滚动标准差
df['basis_std'] = df['basis'].rolling(window=window).std()

# 计算Z-score（标准化基差）
df['z_score'] = (df['basis'] - df['basis_ma']) / df['basis_std']

# 生成交易信号：|z_score| > 2 时开仓
df['signal'] = 0
df.loc[df['z_score'] > 2, 'signal'] = -1  # 基差过大，做空基差
df.loc[df['z_score'] < -2, 'signal'] = 1  # 基差过小，做多基差

print(df[['basis', 'basis_ma', 'z_score', 'signal']].tail(10))

为什么用Z-score？说白了，就是标准化。不同品种的基差波动幅度不一样，Z-score把他们都映射到同一个尺度上，阈值2和-2就是通用的开仓标准。

避坑指南： 我曾经用30天滚动窗口做螺纹钢，结果发现基差均值变化太慢，信号滞后严重。后来改成15天，效果才好。窗口大小没有标准答案，一定要根据品种波动特性去调。

13.4 向量化回测实现：一行代码算收益

回测最怕什么？循环。尤其是逐行判断开平仓，数据量一大就卡死。向量化回测的思路是：用shift和cumprod一次性算出所有持仓的收益。

# 假设每次开仓1手，不考虑手续费和滑点
df['position'] = df['signal'].shift(1)  # 信号延迟一天执行

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

# 策略每日收益 = 持仓方向 * 基差变化
df['strategy_return'] = df['position'] * df['basis_change']

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

# 基准收益（简单持有现货）
df['benchmark_return'] = (1 + df['spot_price'].pct_change()).cumprod()

print(df[['position', 'basis_change', 'strategy_return', 'cumulative_return']].tail())

这里有个细节：signal.shift(1)是为了避免未来函数。你想想看，如果当天收盘时信号出现，你最早也只能在第二天开盘交易。不shift的话，回测结果会虚高。

  回测结果评估： 别只看累计收益。我一般会算夏普比率、最大回撤、胜率这三个指标。尤其是最大回撤，基差交易看似稳健，但遇到极端行情（比如2020年原油），回撤可能超乎想象。

13.5 知识体系总览

下面这张图，是我自己梳理的基差计算与回测的完整流程。你可以把它当作一个检查清单，做项目时对照着来。

13.6 完整代码整合

最后，我把上面所有步骤整合成一个函数。这样你以后做其他品种时，直接调用就行。

def basis_backtest(spot_df, futures_df, window=20, z_threshold=2):
    """
    基差交易回测函数
    :param spot_df: 现货DataFrame，索引为日期，列包含'spot_price'
    :param futures_df: 期货DataFrame，索引为日期，列包含'futures_price'
    :param window: 滚动窗口大小
    :param z_threshold: Z-score阈值
    :return: 包含信号和收益的DataFrame
    """
    # 合并数据
    df = pd.merge(spot_df, futures_df, left_index=True, right_index=True, how='inner')
    
    # 计算基差
    df['basis'] = df['futures_price'] - df['spot_price']
    df['basis_pct'] = df['basis'] / df['spot_price'] * 100
    
    # 滚动统计
    df['basis_ma'] = df['basis'].rolling(window=window).mean()
    df['basis_std'] = df['basis'].rolling(window=window).std()
    df['z_score'] = (df['basis'] - df['basis_ma']) / df['basis_std']
    
    # 信号生成
    df['signal'] = 0
    df.loc[df['z_score'] > z_threshold, 'signal'] = -1
    df.loc[df['z_score'] < -z_threshold, 'signal'] = 1
    
    # 回测
    df['position'] = df['signal'].shift(1)
    df['basis_change'] = df['basis'].diff()
    df['strategy_return'] = df['position'] * df['basis_change']
    df['cumulative_return'] = (1 + df['strategy_return']).cumprod()
    
    return df

# 使用示例
result = basis_backtest(spot_df, futures_df, window=20, z_threshold=2)
print(result[['basis', 'z_score', 'signal', 'cumulative_return']].tail())

我的建议： 刚开始做回测时，先不加手续费，看看策略本身有没有盈利逻辑。如果连毛收益都是负的，那加手续费只会亏更多。等策略逻辑验证通过后，再按千分之一到万分之五的费率加上去，看看还能不能活。

好了，这一章的内容就到这里。基差计算和回测框架已经搭好，下一章我们会深入讨论多品种组合的基差交易策略，以及如何避免过拟合。记住，回测只是起点，实盘才是真正的考验。

📚 课程目录