第三节:价差计算与信号生成
好,咱们直接进入正题。价差计算和信号生成,是整个套利系统的核心引擎。说白了,你前面数据清洗得再漂亮,这一步要是搞砸了,后面全是白搭。我当年刚入行时,就吃过这个亏——数据算得飞快,信号却频频误报,后来才发现是价差计算里埋了个坑。
3.1 实时价差计算
价差,就是两个相关品种的价格差。但这里有个关键点:价差不是简单的减法。
举个例子,螺纹钢和热卷,一个在期货市场,一个在现货市场。它们的价格单位、最小变动价位都可能不同。你直接拿螺纹钢价格减去热卷价格?那算出来的东西毫无意义。
我个人习惯的做法是:
- 统一价格尺度:先把两个品种的价格都转换成同一单位。比如都换算成每吨人民币。
- 考虑合约乘数:期货有合约乘数,比如一手螺纹钢是10吨。价差计算时,要按每吨价格来算,别按每手算。
- 时间对齐:两个品种的报价时间必须精确到同一毫秒。差一秒,价差可能就变味了。
来看一段我常用的实时价差计算代码:
import numpy as np
import pandas as pd
def calculate_spread(price_a, price_b, method='simple'):
"""
实时价差计算
:param price_a: 品种A的价格序列
:param price_b: 品种B的价格序列
:param method: 'simple' 简单价差, 'log' 对数价差
:return: 价差序列
"""
if method == 'simple':
spread = price_a - price_b
elif method == 'log':
# 对数价差,适合价格波动较大的品种
spread = np.log(price_a) - np.log(price_b)
else:
raise ValueError("不支持的价差计算方法")
return spread
# 示例:螺纹钢和热卷的实时价差
rebar_price = 3800.0 # 螺纹钢当前价格
hot_coil_price = 3950.0 # 热卷当前价格
spread = calculate_spread(rebar_price, hot_coil_price, method='simple')
print(f"当前价差: {spread:.2f}")
嗯,这里要注意:对数价差在价格波动剧烈时更稳定。我曾在原油和燃料油的套利中用过,效果不错。但如果你做的是股指期货,简单价差就够用了。
3.2 Z-score标准化
价差算出来了,但怎么判断它是不是「异常」?直接看绝对值?不行。不同品种的价差波动范围天差地别。螺纹钢价差可能就几十块,而股指期货价差可能几百点。
这时候就需要Z-score标准化。它把价差转换成标准正态分布下的数值,告诉你当前价差偏离均值多少个标准差。
公式很简单:
Z = (当前价差 - 均值) / 标准差
但这里有个坑:均值和标准差用多长的窗口?
我建议:
- 短窗口(20-50期):适合高频套利,反应快,但容易受噪声干扰。
- 长窗口(100-200期):适合中低频套利,信号更稳定,但滞后明显。
我在项目中遇到过一个问题:用20期窗口做螺纹钢套利,信号频繁触发,手续费都亏进去了。后来改成60期窗口,效果好了很多。你想想看,这其实是个平衡问题。
代码实现:
def calculate_zscore(spread, window=60):
"""
计算Z-score
:param spread: 价差序列
:param window: 滚动窗口大小
:return: Z-score序列
"""
rolling_mean = spread.rolling(window=window).mean()
rolling_std = spread.rolling(window=window).std()
zscore = (spread - rolling_mean) / rolling_std
return zscore
# 示例
spread_series = pd.Series([...]) # 假设这是历史价差数据
zscore_series = calculate_zscore(spread_series, window=60)
print(f"当前Z-score: {zscore_series.iloc[-1]:.2f}")
3.3 布林带策略
布林带,说白了就是给价差画了个「通道」。上轨是均值加两倍标准差,下轨是均值减两倍标准差。当价差突破上轨或下轨时,就产生了交易信号。
但布林带不是万能的。我见过很多人直接拿布林带做交易,结果亏得一塌糊涂。为什么?因为布林带假设价格服从正态分布,但金融市场哪有那么乖?
我的经验是:
- 上轨突破:做空价差(卖高买低)
- 下轨突破:做多价差(买低卖高)
- 中轨回归:平仓信号
但别急着下单。我曾经在布林带信号上吃过亏——价差突破上轨后继续飙升,我反向开仓,结果被套牢。后来我加了二次确认机制:价差突破布林带后,必须连续两期都保持在轨道外,才触发信号。
代码实现:
def bollinger_bands(spread, window=60, num_std=2):
"""
布林带计算
:param spread: 价差序列
:param window: 窗口大小
:param num_std: 标准差倍数
:return: 上轨、中轨、下轨
"""
rolling_mean = spread.rolling(window=window).mean()
rolling_std = spread.rolling(window=window).std()
upper_band = rolling_mean + (rolling_std * num_std)
lower_band = rolling_mean - (rolling_std * num_std)
return upper_band, rolling_mean, lower_band
# 生成信号
def generate_signals(spread, upper, lower):
"""
根据布林带生成交易信号
1: 做多价差(买入价差)
-1: 做空价差(卖出价差)
0: 无操作
"""
signals = pd.Series(index=spread.index, data=0)
# 价差跌破下轨,做多
signals[spread < lower] = 1
# 价差突破上轨,做空
signals[spread > upper] = -1
return signals
3.4 阈值触发信号
布林带是一种动态阈值,但有时候我们需要固定阈值。比如,当价差超过50个点时触发交易,低于-30个点时反向交易。
固定阈值的好处是简单直观,但缺点也很明显:市场波动率变化时,固定阈值可能失效。我曾在2015年股灾时用过固定阈值,结果市场波动率飙升,阈值形同虚设。
所以,我现在的做法是混合使用:
- 动态阈值(布林带):作为主要信号源
- 固定阈值:作为风控底线,防止极端行情
举个例子:
def hybrid_signal(zscore, fixed_threshold=2.5):
"""
混合信号生成
:param zscore: 当前Z-score值
:param fixed_threshold: 固定阈值
:return: 信号强度(-1到1之间的连续值)
"""
# 布林带信号(Z-score超过2倍标准差)
bollinger_signal = 0
if zscore > 2.0:
bollinger_signal = -1 # 做空
elif zscore < -2.0:
bollinger_signal = 1 # 做多
# 固定阈值信号(作为风控)
fixed_signal = 0
if zscore > fixed_threshold:
fixed_signal = -1 # 强制做空
elif zscore < -fixed_threshold:
fixed_signal = 1 # 强制做多
# 取两者中更保守的信号
final_signal = min(bollinger_signal, fixed_signal) if zscore > 0 else max(bollinger_signal, fixed_signal)
return final_signal
知识体系总览
下面这张图,是我自己总结的价差计算与信号生成流程。你看一眼,心里就有数了:
这张图把整个流程串起来了。从价格输入,到价差计算,再到标准化和信号生成,每一步都环环相扣。你照着这个流程做,基本不会出大错。
好了,这一节的内容就到这儿。记住:价差计算是基础,Z-score是桥梁,布林带和阈值是武器。三者缺一不可。