压榨价差套利策略
一、交易策略解释
核心思想
压榨价差（Crush Spread）代表了油籽加工行业的毛利润。具体到大豆，就是大豆压榨企业购买大豆（成本），将其压榨成豆粕（主要用作动物饲料）和豆油（食用油或生物柴油原料）（收入）后所能获得的价值差。
-
正向压榨 (Long Crush Spread / "Buying the Crush"): 当交易者认为豆粕和豆油相对于大豆的价格未来会上涨（即压榨利润会扩大），他们会买入大豆期货，同时卖出豆粕期货和豆油期货。这实质上是在金融市场上模拟了压榨商锁定未来压榨利润的行为。
-
反向压榨 (Short Crush Spread / "Selling the Crush" / Reverse Crush): 当交易者认为豆粕和豆油相对于大豆的价格未来会下跌（即压榨利润会收缩），他们会卖出大豆期货，同时买入豆粕期货和豆油期货。
压榨价差套利的核心思想是，这个价差在大多数情况下会围绕一个由加工成本、供需关系和合理利润构成的“公允价值”波动。 当市场因素导致价差显著偏离这个公允价值或其历史均值时，套利者预期价差会向其“正常”水平回归，从而通过构建上述的正向或反向压榨头寸来获利。 这不是一个无风险套利，而更像是一种基于统计和基本面分析的相对价值交易。
理论基础
-
产业链的经济联系: 大豆、豆粕和豆油在产业链上紧密相连。大豆是原料，豆粕和豆油是产成品。它们的相对价格受到各自供需基本面以及压榨行业产能、开工率、压榨利润等因素的共同影响。这种经济联系是价差存在的根本原因。
-
供需失衡的短期效应:
- 大豆端: 天气（影响产量）、种植面积、进出口政策、库存等。
- 豆粕端: 畜牧养殖规模（猪、禽）、饲料配方变化、替代品（如菜粕、DDGS）价格等。
- 豆油端: 食用消费、餐饮业景气度、生物柴油政策、替代品（如棕榈油、菜籽油）价格等。
- 这些因素的短期变化可能导致三者价格波动幅度不一致，从而使压榨价差偏离正常范围。
-
均值回归特性 (Mean Reversion): 历史数据通常显示，压榨价差虽然会波动，但倾向于在一个特定区间内运行，或者围绕其长期移动平均线波动。当价差由于短期冲击而走向极端（过高或过低）时，市场力量（如压榨商调整开工率、贸易商调整采购节奏）往往会促使其向历史均值或更可持续的水平回归。这是统计套利策略的主要理论依据。
-
市场参与者的行为:
- 压榨企业 (Processors): 是压榨价差的天然参与者。他们利用期货市场进行套期保值，锁定远期压榨利润。当远期压榨利润可观时，他们会买入大豆期货并卖出豆粕、豆油期货。
- 投机者/基金: 基于对价差未来走势的判断进行投机交易。他们的参与增加了市场流动性，同时也可能放大或加速价差的波动和回归。
策略适用场景
-
价差显著偏离: 当压榨价差（经过适当计算和标准化后）显著高于或低于其历史均值或某个统计区间（如+/-2个标准差）时。
-
预期均值回归: 交易者有理由相信当前的价差偏离是暂时的，未来会向更“正常”的水平回归。
-
基本面支持: 结合对大豆、豆粕、豆油各自基本面的分析，如果基本面也支持价差向预期方向回归，则策略的可靠性更高。例如，如果压榨价差极低，同时预期未来豆粕需求将回暖，则做多压榨价差（买大豆，卖粕卖油——这里应为卖大豆，买粕买油，或者说买入压榨利润，即买粕油卖大豆，如果价差代表利润的话，价差低是买入利润）的信号会更强。 修正：如果价差代表（粕+油）-豆的利润，价差低意味着利润低，此时预期利润扩大，应买入价差，即买粕、买油、卖豆。
-
市场波动性: 在市场波动性较大时，价差偏离的机会可能更多，但风险也相应增加。
-
流动性充足: 策略涉及三个品种的期货合约，需要保证各合约均有良好的流动性，以减少冲击成本和确保顺利成交。国内大商所的黄大豆1号（a）、豆粕（m）、豆油（y）是常用的标的。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
核心价差定义与计算:
该策略的核心是追踪一组按固定手数比例配置的大豆、豆粕和豆油期货合约所共同构成的压榨价差。在这个组合中，三种期货的合约数量遵循一个预设的整数比例（例如，10份大豆对应8份豆粕和2份豆油）。
这个压榨价差的计算方式是：用组合中豆粕和豆油部分的总市场价值，减去大豆部分的总市场价值。计算各项市场价值时，会用到对应期货品种的每日收盘价以及其合约规定的每手代表的价值（合约乘数）。这个计算结果代表了，如果按照预设的手数比例在期货市场上进行模拟的压榨操作，理论上可以实现的价差额。
指标构建 (统计套利基础):
策略采用标准化得分（Z-score）来衡量当前计算出的压榨价差与其历史水平的偏离程度。
-
历史价差序列: 收集所选大豆、豆粕、豆油合约过去一段时间（例如30个交易日）的每日收盘价。基于这些价格，计算出每日的压榨价差，形成一个历史序列。
-
计算历史均值: 对这个历史压榨价差序列计算其算术平均值。
-
计算历史标准差: 对同一个历史压榨价差序列计算其标准差。
-
计算标准化得分: 用最新的每日收盘价计算出当前的压榨价差，然后通过以下公式得到其标准化得分： 标准化得分 = (当前压榨价差 - 历史均值) / 历史标准差 （若历史标准差为零，则此得分无意义或需特殊处理）。
交易逻辑
开仓信号
- 预期压榨价差缩小（做空压榨价差）:
- 信号: 当计算出的“标准化得分”显著高于一个预设的正阈值（例如，大于2）。这表明当前的压榨价差远高于历史平均水平，策略预期其会向历史均值回落。
- 操作: 买入指定数量的大豆期货，同时卖出相应比例数量的豆粕期货和豆油期货。
- 预期压榨价差扩大（做多压榨价差）:
- 信号: 当计算出的“标准化得分”显著低于一个预设的负阈值（例如，小于-2）。这表明当前的压榨价差远低于历史平均水平，策略预期其会向历史均值回升。
- 操作: 卖出指定数量的大豆期货，同时买入相应比例数量的豆粕期货和豆油期货。
平仓信号：
- 信号: 当“标准化得分”的绝对值回落到一个较小的预设阈值以内（例如，小于0.5）。这表示压榨价差已经回归到接近其历史平均的水平。
- 操作: 将所有持有的期货头寸全部平掉（即目标持仓设为零）。
止损信号：
- 信号:
- 若持有的是“预期压榨价差缩小”的头寸（即大豆多头，粕油空头），而“标准化得分”继续大幅上升，超过一个更高的正阈值（例如，大于3）。
- 若持有的是“预期压榨价差扩大”的头寸（即大豆空头，粕油多头），而“标准化得分”继续大幅下降，超过一个更低的负阈值（例如，小于-3）。
- 这两种情况均表示压榨价差向不利方向进一步显著偏离。
- 操作: 与止盈操作类似，立即平掉所有持有的期货头寸。
回测
回测初始设置
- 测试周期： 2023 年 11 月 1 日 - 2024 年 4 月 30 日
- 交易品种： DCE.a2409/DCE.m2409/DCE.y2409
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# coding=utf-8
__author__ = "Chaos"
from datetime import date
from tqsdk import TqApi, TqAuth, TargetPosTask, TqBacktest, BacktestFinished
import numpy as np
import time
# === 用户参数 ===
# 合约参数
SOYBEAN = "DCE.a2409" # 大豆期货合约
SOYMEAL = "DCE.m2409" # 豆粕期货合约
SOYOIL = "DCE.y2409" # 豆油期货合约
START_DATE = date(2023, 11, 1) # 回测开始日期
END_DATE = date(2024, 4, 30) # 回测结束日期
# 套利参数
LOOKBACK_DAYS = 30 # 计算历史价差的回溯天数
STD_THRESHOLD = 2.0 # 标准差阈值，超过此阈值视为套利机会
ORDER_VOLUME = 500 # 大豆的下单手数
CLOSE_THRESHOLD = 0.5 # 平仓阈值(标准差)
# 压榨价差比例 - 1吨大豆压榨可得约0.785吨豆粕和0.18吨豆油
# 为了简化，使用10:8:2的整数比例
BEAN_RATIO = 10
MEAL_RATIO = 8
OIL_RATIO = 2
# === 初始化API ===
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账号", "快期密码"))
# 获取合约行情和K线
bean_quote = api.get_quote(SOYBEAN)
meal_quote = api.get_quote(SOYMEAL)
oil_quote = api.get_quote(SOYOIL)
bean_klines = api.get_kline_serial(SOYBEAN, 60*60*24, LOOKBACK_DAYS)
meal_klines = api.get_kline_serial(SOYMEAL, 60*60*24, LOOKBACK_DAYS)
oil_klines = api.get_kline_serial(SOYOIL, 60*60*24, LOOKBACK_DAYS)
# 创建目标持仓任务
bean_pos = TargetPosTask(api, SOYBEAN)
meal_pos = TargetPosTask(api, SOYMEAL)
oil_pos = TargetPosTask(api, SOYOIL)
# 获取合约乘数
bean_volume_multiple = bean_quote.volume_multiple
meal_volume_multiple = meal_quote.volume_multiple
oil_volume_multiple = oil_quote.volume_multiple
# 初始化状态变量
position_time = 0 # 建仓时间
in_position = False # 是否有持仓
mean_spread = 0 # 历史价差均值
std_spread = 0 # 历史价差标准差
print(f"策略启动，监控合约: {SOYBEAN}, {SOYMEAL}, {SOYOIL}")
# === 主循环 ===
try:
# 初始计算历史统计值
spreads = []
for i in range(len(bean_klines) - 1):
bean_price = bean_klines.close.iloc[i] * bean_volume_multiple * BEAN_RATIO
meal_price = meal_klines.close.iloc[i] * meal_volume_multiple * MEAL_RATIO
oil_price = oil_klines.close.iloc[i] * oil_volume_multiple * OIL_RATIO
# 压榨价差 = (豆粕价值 + 豆油价值) - 大豆价值
spread = (meal_price + oil_price) - bean_price
spreads.append(spread)
mean_spread = np.mean(spreads)
std_spread = np.std(spreads)
print(f"历史压榨价差均值: {mean_spread:.2f}, 标准差: {std_spread:.2f}")
# 主循环
while True:
api.wait_update()
# 当K线数据有变化时进行计算
if api.is_changing(bean_klines) or api.is_changing(meal_klines) or api.is_changing(oil_klines):
# 重新计算历史价差统计
spreads = []
for i in range(len(bean_klines) - 1):
bean_price = bean_klines.close.iloc[i] * bean_volume_multiple * BEAN_RATIO
meal_price = meal_klines.close.iloc[i] * meal_volume_multiple * MEAL_RATIO
oil_price = oil_klines.close.iloc[i] * oil_volume_multiple * OIL_RATIO
spread = (meal_price + oil_price) - bean_price
spreads.append(spread)
mean_spread = np.mean(spreads)
std_spread = np.std(spreads)
# 计算当前压榨价差
bean_price = bean_klines.close.iloc[-1] * bean_volume_multiple * BEAN_RATIO
meal_price = meal_klines.close.iloc[-1] * meal_volume_multiple * MEAL_RATIO
oil_price = oil_klines.close.iloc[-1] * oil_volume_multiple * OIL_RATIO
current_spread = (meal_price + oil_price) - bean_price
# 计算z-score (标准化的价差)
z_score = (current_spread - mean_spread) / std_spread
print(f"当前压榨价差: {current_spread:.2f}, Z-score: {z_score:.2f}")
# 获取当前持仓
bean_position = api.get_position(SOYBEAN)
meal_position = api.get_position(SOYMEAL)
oil_position = api.get_position(SOYOIL)
current_bean_pos = bean_position.pos_long - bean_position.pos_short
current_meal_pos = meal_position.pos_long - meal_position.pos_short
current_oil_pos = oil_position.pos_long - oil_position.pos_short
# 计算实际下单手数（依据比例）
meal_volume = int(ORDER_VOLUME * MEAL_RATIO / BEAN_RATIO)
oil_volume = int(ORDER_VOLUME * OIL_RATIO / BEAN_RATIO)
# === 交易信号判断 ===
if not in_position: # 如果没有持仓
if z_score > STD_THRESHOLD: # 价差显著高于均值，压榨利润偏高
# 卖出压榨价差：买入大豆，卖出豆粕和豆油
print(f"卖出压榨价差：买入大豆{ORDER_VOLUME}手，卖出豆粕{meal_volume}手和豆油{oil_volume}手")
bean_pos.set_target_volume(ORDER_VOLUME)
meal_pos.set_target_volume(-meal_volume)
oil_pos.set_target_volume(-oil_volume)
position_time = time.time()
in_position = True
elif z_score < -STD_THRESHOLD: # 价差显著低于均值，压榨利润偏低
# 买入压榨价差：卖出大豆，买入豆粕和豆油
print(f"买入压榨价差：卖出大豆{ORDER_VOLUME}手，买入豆粕{meal_volume}手和豆油{oil_volume}手")
bean_pos.set_target_volume(-ORDER_VOLUME)
meal_pos.set_target_volume(meal_volume)
oil_pos.set_target_volume(oil_volume)
position_time = time.time()
in_position = True
else: # 如果已有持仓
# 检查是否应当平仓
if abs(z_score) < CLOSE_THRESHOLD: # 价差恢复正常
print("价差恢复正常，平仓所有头寸")
bean_pos.set_target_volume(0)
meal_pos.set_target_volume(0)
oil_pos.set_target_volume(0)
in_position = False
# 也可以添加止损逻辑
if (z_score > STD_THRESHOLD * 1.5 and current_bean_pos > 0) or \
(z_score < -STD_THRESHOLD * 1.5 and current_bean_pos < 0):
print("止损：价差向不利方向进一步偏离")
bean_pos.set_target_volume(0)
meal_pos.set_target_volume(0)
oil_pos.set_target_volume(0)
in_position = False
except BacktestFinished as e:
print("回测结束")
api.close()