养殖价差套利策略
一、交易策略解释
核心思想
养殖套利的核心在于监控和交易“饲养利润价差”。这个价差代表了养殖企业（此处特指生猪养殖）购买饲料（主要为玉米和豆粕）并将其转化为最终产品（生猪）后所能获得的理论毛利润。
- 饲养利润价差：饲养利润 = 生猪价值 - 饲料成本价值
核心思想是：
-
这个饲养利润价差会受到生猪、玉米、豆粕各自供需基本面以及养殖行业整体状况的影响，但通常会围绕一个由生产成本、合理利润和市场预期构成的“均衡水平”波动。
-
做多饲养利润 (Long the Feed Spread / "Buying the Margin"): 当交易者认为当前市场计算出的饲养利润过低，未来有望扩大时，他们会买入生猪期货，同时卖出玉米期货和豆粕期货。这在金融市场上模拟了养殖户锁定未来更高饲养利润的行为（或对冲当前低利润风险，预期未来饲料成本下降或猪价上涨）。
-
做空饲养利润 (Short the Feed Spread / "Selling the Margin"): 当交易者认为当前市场计算出的饲养利润过高，未来可能收窄时，他们会卖出生猪期货，同时买入玉米期货和豆粕期货。这模拟了养殖户锁定当前较高饲养利润的行为。
养殖套利策略预期当此价差显著偏离其历史均值或行业盈亏平衡点时，会向其“正常”水平回归。
理论基础
-
产业链的经济联系: 生猪养殖业中，饲料成本通常占总成本的60%-70%。玉米提供主要的能量，豆粕提供主要的蛋白质。因此，生猪价格与玉米、豆粕价格之间存在紧密的经济联系。
-
供需基本面的驱动:
-
生猪端: 猪周期（母猪存栏量、能繁母猪数量）、疫病情况（如非洲猪瘟）、季节性消费需求、国家收储政策、进出口等。
-
玉米端: 种植面积、天气、产量、库存、国家收储抛储政策、工业需求（如深加工、燃料乙醇）、进出口等。
-
豆粕端: 大豆进口量、压榨利润（影响大豆加工厂开工率）、国内养殖规模（特别是禽类和水产，与猪争夺豆粕需求）、天气（影响大豆主产国产量）等。
-
这些因素的变化导致三者价格波动，进而影响饲养利润价差。
-
均值回归特性 (Mean Reversion): 历史数据显示，饲养利润价差虽然波动剧烈，但往往具有均值回归的特性。当利润过高时，可能刺激养殖户补栏扩产，未来生猪供应增加将压低猪价，同时增加对饲料的需求可能推高饲料价格，从而压缩利润。当利润过低甚至亏损时，养殖户可能去产能，未来生猪供应减少将推升猪价，同时减少对饲料的需求可能压低饲料价格，从而修复利润。
-
养殖户的套期保值行为: 养殖企业和饲料企业会利用期货市场进行套期保值。养殖户可能在远期饲养利润可观时卖出生猪期货、买入玉米和豆粕期货（锁定利润）；或在预期饲料价格上涨时提前买入饲料期货。这些行为本身也会影响价差的动态。
策略适用场景
-
饲养利润价差显著偏离: 当计算出的饲养利润（经过适当的比例和标准化处理后）显著高于或低于其历史均值或关键的盈亏平衡点时。
-
预期均值回归: 交易者基于对市场周期或短期供需失衡的判断，预期价差将向更可持续的水平回归。
-
基本面分析支持: 结合对生猪、玉米、豆粕各自基本面的分析，如果基本面也支持价差向预期方向回归，则策略的可靠性更高。
-
流动性充足: 策略涉及三个品种的期货合约（国内主要是大连商品交易所的生猪、玉米、豆粕期货），需要保证各合约均有良好的流动性。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
核心价差定义与计算(饲养利润价差):
该策略的核心是追踪一个根据预设价值权重构建的“饲养利润价差”。这个价差通过以下方式计算得出：
-
首先，为生猪、玉米和豆粕期货分别设定一个价值贡献的比例因子。
-
然后，将各期货品种的最新收盘价乘以其对应的合约乘数（代表每手合约的价值规模），再乘以各自的价值贡献比例因子，得到它们在价差组合中的调整后价值。
-
饲养利润价差被定义为：生猪期货的调整后价值，减去玉米期货和豆粕期货调整后价值的总和。 这个计算出的价差，反映了按照预设价值比例关系模拟的“原料成本”与“成品价值”之间的差异。
指标构建:
策略采用 标准化得分（Z-score） 来衡量当前计算出的饲养利润价差与其近期历史水平的偏离程度。
- 历史价差序列: 收集特定生猪、玉米、豆粕期货合约在过去一段固定交易日（例如30天）内的每日收盘价。基于这些价格和上述价差计算方法，得到一个每日“饲养利润价差”的历史序列。
- 计算历史均值: 对这个历史价差序列计算其算术平均值。
- 计算历史标准差: 对同一个历史价差序列计算其标准差。
- 计算标准化得分: 用最新的每日收盘价计算出当前的“饲养利润价差”，然后通过以下公式得到其标准化得分： 标准化得分 = (当前饲养利润价差 - 历史均值) / 历史标准差 （若历史标准差为零，则此得分无意义或需特殊处理）。
交易逻辑
开仓信号 (入场):
-
预期价差缩小（做空饲养利润价差）:
- 信号: 当计算出的“标准化得分”显著高于一个预设的正阈值（例如，大于2个标准差）。这表明当前的饲养利润价差远高于其历史平均水平，策略预期其会向均值回落。
- 操作: 卖出指定数量的生猪期货，同时买入相应比例数量的玉米期货和豆粕期货。
-
预期价差扩大（做多饲养利润价差）:
- 信号: 当计算出的“标准化得分”显著低于一个预设的负阈值（例如，小于负2个标准差）。这表明当前的饲养利润价差远低于其历史平均水平，策略预期其会向均值回升。
- 操作: 买入指定数量的生猪期货，同时卖出相应比例数量的玉米期货和豆粕期货。
平仓信号 (止盈离场):
- 信号: 当“标准化得分”的绝对值回落到一个较小的预设阈值以内（例如，小于0.5个标准差）。这表示饲养利润价差已经回归到接近其历史平均的水平。
- 操作: 将所有持有的期货头寸全部平掉（即目标持仓设为零）。
特定条件下的平仓信号 (类似止损或极端行情退出): 当策略当前已持有头寸时，若出现以下情况也会触发平仓：
- 信号:
- 若最初是因价差过低而做多（生猪多头，饲料空头），之后价差不仅未按预期回升，反而大幅反向运动至显著高于历史均值（例如，标准化得分超过一个更高的正阈值，如3个标准差）。
- 若最初是因价差过高而做空（生猪空头，饲料多头），之后价差不仅未按预期回落，反而大幅反向运动至显著低于历史均值（例如，标准化得分低于一个更低的负阈值，如负3个标准差）。
- 操作: 与止盈操作类似，立即平掉所有持有的期货头寸。
回测
回测初始设置
- 测试周期： 2023 年 11 月 1 日 - 2024 年 3 月 13 日
- 交易品种： DCE.c2409/DCE.m2409/DCE.lh2409
- 初始资金： 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
# === 用户参数 ===
# 合约参数
LIVE_HOG = "DCE.lh2409" # 生猪期货合约
CORN = "DCE.c2409" # 玉米期货合约
SOYMEAL = "DCE.m2409" # 豆粕期货合约
START_DATE = date(2023, 11, 1) # 回测开始日期
END_DATE = date(2024, 3, 13) # 回测结束日期
# 套利参数
LOOKBACK_DAYS = 30 # 计算历史价差的回溯天数
STD_THRESHOLD = 2.0 # 标准差阈值，超过此阈值视为套利机会
ORDER_VOLUME = 100 # 生猪的下单手数
CLOSE_THRESHOLD = 0.5 # 平仓阈值(标准差)
# 饲养利润价差比例 - 生产1吨生猪约需要3吨玉米和0.6吨豆粕
# 可根据实际养殖转化比调整
HOG_RATIO = 1
CORN_RATIO = 3
SOYMEAL_RATIO = 0.6
# === 初始化API ===
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账号", "快期密码"))
# 获取合约行情和K线
hog_quote = api.get_quote(LIVE_HOG)
corn_quote = api.get_quote(CORN)
meal_quote = api.get_quote(SOYMEAL)
hog_klines = api.get_kline_serial(LIVE_HOG, 60*60*24, LOOKBACK_DAYS)
corn_klines = api.get_kline_serial(CORN, 60*60*24, LOOKBACK_DAYS)
meal_klines = api.get_kline_serial(SOYMEAL, 60*60*24, LOOKBACK_DAYS)
# 创建目标持仓任务
hog_pos = TargetPosTask(api, LIVE_HOG)
corn_pos = TargetPosTask(api, CORN)
meal_pos = TargetPosTask(api, SOYMEAL)
# 获取合约乘数
hog_volume_multiple = hog_quote.volume_multiple
corn_volume_multiple = corn_quote.volume_multiple
meal_volume_multiple = meal_quote.volume_multiple
# 初始化状态变量
position_time = 0 # 建仓时间
in_position = False # 是否有持仓
mean_spread = 0 # 历史价差均值
std_spread = 0 # 历史价差标准差
print(f"策略启动，监控合约: {LIVE_HOG}, {CORN}, {SOYMEAL}")
# === 主循环 ===
try:
# 初始计算历史统计值
spreads = []
for i in range(len(hog_klines) - 1):
hog_price = hog_klines.close.iloc[i] * hog_volume_multiple * HOG_RATIO
corn_price = corn_klines.close.iloc[i] * corn_volume_multiple * CORN_RATIO
meal_price = meal_klines.close.iloc[i] * meal_volume_multiple * SOYMEAL_RATIO
# 饲养利润 = 生猪价值 - 饲料成本价值
spread = hog_price - (corn_price + meal_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(hog_klines) or api.is_changing(corn_klines) or api.is_changing(meal_klines):
# 重新计算历史价差统计
spreads = []
for i in range(len(hog_klines) - 1):
hog_price = hog_klines.close.iloc[i] * hog_volume_multiple * HOG_RATIO
corn_price = corn_klines.close.iloc[i] * corn_volume_multiple * CORN_RATIO
meal_price = meal_klines.close.iloc[i] * meal_volume_multiple * SOYMEAL_RATIO
spread = hog_price - (corn_price + meal_price)
spreads.append(spread)
mean_spread = np.mean(spreads)
std_spread = np.std(spreads)
# 计算当前饲养利润价差
hog_price = hog_klines.close.iloc[-1] * hog_volume_multiple * HOG_RATIO
corn_price = corn_klines.close.iloc[-1] * corn_volume_multiple * CORN_RATIO
meal_price = meal_klines.close.iloc[-1] * meal_volume_multiple * SOYMEAL_RATIO
current_spread = hog_price - (corn_price + meal_price)
# 计算z-score (标准化的价差)
z_score = (current_spread - mean_spread) / std_spread
print(f"当前饲养利润: {current_spread:.2f}, Z-score: {z_score:.2f}")
# 获取当前持仓
hog_position = api.get_position(LIVE_HOG)
corn_position = api.get_position(CORN)
meal_position = api.get_position(SOYMEAL)
current_hog_pos = hog_position.pos_long - hog_position.pos_short
current_corn_pos = corn_position.pos_long - corn_position.pos_short
current_meal_pos = meal_position.pos_long - meal_position.pos_short
# 计算实际下单手数（依据比例）
corn_volume = int(ORDER_VOLUME * CORN_RATIO / HOG_RATIO)
meal_volume = int(ORDER_VOLUME * SOYMEAL_RATIO / HOG_RATIO)
# === 交易信号判断 ===
if not in_position: # 如果没有持仓
if z_score > STD_THRESHOLD: # 饲养利润显著高于均值
# 做空饲养利润：卖出生猪，买入玉米和豆粕
print(f"做空饲养利润：卖出生猪{ORDER_VOLUME}手，买入玉米{corn_volume}手和豆粕{meal_volume}手")
hog_pos.set_target_volume(-ORDER_VOLUME)
corn_pos.set_target_volume(corn_volume)
meal_pos.set_target_volume(meal_volume)
position_time = time.time()
in_position = True
elif z_score < -STD_THRESHOLD: # 饲养利润显著低于均值
# 做多饲养利润：买入生猪，卖出玉米和豆粕
print(f"做多饲养利润：买入生猪{ORDER_VOLUME}手，卖出玉米{corn_volume}手和豆粕{meal_volume}手")
hog_pos.set_target_volume(ORDER_VOLUME)
corn_pos.set_target_volume(-corn_volume)
meal_pos.set_target_volume(-meal_volume)
position_time = time.time()
in_position = True
else: # 如果已有持仓
# 检查是否应当平仓
if abs(z_score) < CLOSE_THRESHOLD: # 饲养利润恢复正常
print("饲养利润回归正常水平，平仓所有头寸")
hog_pos.set_target_volume(0)
corn_pos.set_target_volume(0)
meal_pos.set_target_volume(0)
in_position = False
# 止损逻辑
if (z_score > STD_THRESHOLD * 1.5 and current_hog_pos > 0) or \
(z_score < -STD_THRESHOLD * 1.5 and current_hog_pos < 0):
print("止损：饲养利润向不利方向进一步偏离")
hog_pos.set_target_volume(0)
corn_pos.set_target_volume(0)
meal_pos.set_target_volume(0)
in_position = False
except BacktestFinished as e:
print("回测结束")
api.close()