养殖价差套利策略

一、交易策略解释

核心思想

这类策略的核心，不是直接去交易“完整养殖利润表”，而是去跟踪一个可交易、可量化、可回测的简化利润代理指标。为了避免概念混淆，本文将该指标统一称为：

生猪-饲料毛利代理价差。

它用生猪期货价格代表产出端，用玉米期货和豆粕期货价格代表饲料成本端，构造一个简化的养殖毛利代理值：

毛利代理价差（元/吨） = 生猪价格 - 3 × 玉米价格 - 0.6 × 豆粕价格

其中：

3 表示生产 1 吨生猪时，假设大约消耗 3 吨玉米；
0.6 表示生产 1 吨生猪时，假设大约消耗 0.6 吨豆粕；
上述比例采用行业研究中常见的简化口径，实际生产中会因料肉比、育肥阶段、配方调整、疫病与存栏结构等因素而变化。

核心思想是：

当这个毛利代理价差显著低于历史均值时，说明市场对“猪价相对饲料成本”的定价偏悲观，后续若猪价走强、饲料走弱，价差可能修复；
当这个毛利代理价差显著高于历史均值时，说明市场对远期养殖毛利定价偏乐观，后续若猪价走弱、饲料走强，价差可能收敛；
因此可以围绕该价差做均值回归交易，而不是单独押注某一个品种的绝对方向。

指标口径说明

本文使用的是一个简化毛利代理模型，聚焦于最具代表性的两类饲料原料成本：玉米和豆粕。相较于企业完整经营口径，该指标未纳入以下成本或损益项：

仔猪或母猪相关成本；
药品、疫苗、防疫和死亡损耗；
人工、水电、折旧、栏舍与环保投入；
运输、屠宰、资金占用等经营性成本。

因此，这一策略更适合被理解为：

生猪与饲料原料之间的毛利代理价差交易；
用于刻画养殖利润预期变化的阶段性关系，而非企业财务报表意义上的完整利润测算。

理论基础

产业链的经济联系： 生猪价格决定收入端，玉米和豆粕是饲料端最核心的两类原料，因此三者之间天然存在成本传导与利润挤压关系。
供需基本面的驱动：
- 生猪端： 猪周期、能繁母猪存栏、二育节奏、疫病、季节性消费、收储与进口等；
- 玉米端： 种植面积、天气、产量、库存、收抛储政策、深加工需求、进口与替代谷物等；
- 豆粕端： 大豆进口、油厂压榨利润与开工率、下游养殖需求、南美/美豆天气与供应节奏等；
- 三者供需变化共同推动毛利代理价差波动。
均值回归特性： 当毛利代理值长期偏高时，产业可能出现补栏扩产预期，未来生猪供应增加、饲料需求提升，价差可能回落；当毛利代理值长期偏低时，产业可能去产能，未来供应收缩，价差可能修复。
产业套保行为： 养殖企业、饲料企业和贸易商的套保行为，会加快利润预期在远期合约中的反映，也会影响价差结构的回归节奏。

指标口径与执行口径

1）指标计算口径：按“吨”定义

上面的毛利代理公式，本质是按每吨生猪对应的理论饲料消耗来算的。因此，做信号时直接用价格（元/吨）即可：

spread = hog_price - 3 * corn_price - 0.6 * meal_price

这一层对应的是产业定价关系本身。

2）执行口径：按“手”换算

在交易执行层面，期货头寸需要以合约手数表达。由于不同品种的合约单位不同，因此需要将吨口径的配比进一步换算为手数配比。

如果设生猪下单 N 手，那么对应的玉米、豆粕理论手数应写成：

corn_lots = round(N × 生猪合约乘数 × 3 / 玉米合约乘数)
meal_lots = round(N × 生猪合约乘数 × 0.6 / 豆粕合约乘数)

3）组合执行中的残余敞口

由于期货交易以整数手为单位，理论吨配比往往无法被完全精确地映射为手数，因此组合执行时通常会保留一定的近似误差。这意味着：

策略可以在执行层面实现较高程度的配平；
同时仍可能保留少量残余饲料敞口；
这也是产业逻辑与交易落地之间需要重点关注的细节之一。

策略适用场景

毛利代理价差显著偏离其历史均值；
生猪与饲料端的基本面出现阶段性错配；
三个合约都具备足够流动性；
交易者接受“整数手近似配比”带来的少量残余风险。

二、天勤介绍

天勤平台概述

天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特点：

丰富的行情数据： 提供可交易合约的 Tick 和 K 线数据，便于构建跨品种价差指标；
一站式的解决方案： 从历史分析到回测、模拟和实盘的一体化开发流程；
专业的技术支持： 深度集成 pandas 和 numpy，便于快速实现价差计算、统计归一化和风控逻辑。

策略开发流程

环境准备
- 安装 Python 环境；
- 安装 tqsdk：pip install tqsdk；
- 注册天勤账户并配置认证信息。
数据准备
- 订阅生猪、玉米、豆粕三个目标合约；
- 获取日线数据用于构造价差和历史分布。
策略编写
- 用吨口径构造毛利代理价差；
- 用历史均值和标准差计算 Z-score；
- 用目标持仓模型执行三腿组合。
回测验证
- 设置回测区间、固定合约与初始资金；
- 检查组合配比、残余敞口和收益曲线；
- 评估收益率、回撤、换手与稳定性。
策略优化
- 调整入场/离场阈值；
- 动态修正玉米、豆粕消耗系数；
- 引入更明确的滚动换月与流动性过滤规则。

三、天勤实现策略

策略原理

核心价差定义

本文采用的信号指标是按吨计算的毛利代理价差：

spread_t = hog_t - 3 × corn_t - 0.6 × meal_t

其中：

hog_t：生猪期货价格（元/吨）
corn_t：玉米期货价格（元/吨）
meal_t：豆粕期货价格（元/吨）

这个定义有两个优点：

经济含义直观，直接对应“每吨生猪对应的主要饲料成本”；
不会把“吨配比”和“期货手数”混在一起，避免信号口径错位。

指标构建

策略采用 Z-score 衡量当前价差相对历史分布的偏离程度：

z_score = (current_spread - mean_spread) / std_spread

具体做法：

使用最近 LOOKBACK_DAYS 个已完成的日线作为历史窗口；
用前述公式计算每一天的毛利代理价差；
计算历史均值和标准差；
将最新一个已完成日线的价差代入，得到当前 Z-score；
若标准差过小，则跳过本次信号，避免除零或极端放大。

这里特别强调“已完成日线”：

不用正在形成中的日线做信号；
可以避免盘中尚未收完的 K 线把统计值拉歪；
对日频均值回归策略更稳妥。

下单手数换算

做信号时按吨，交易时按手，所以要在发单前完成一次换算。

假设生猪计划交易 hog_lots 手，则：

hog_tons   = hog_lots × hog_volume_multiple
corn_lots  = round(hog_tons × 3   / corn_volume_multiple)
meal_lots  = round(hog_tons × 0.6 / meal_volume_multiple)

这样做有两个好处：

代码中不需要手工写死各品种每手吨数；
直接使用 quote.volume_multiple，可读性更强，也更不容易把配比写错。

交易逻辑

开仓信号：

做空毛利代理价差： 当 z_score > ENTRY_Z，说明当前毛利代理值显著高于历史均值，预期其向下回归；
- 操作：卖出生猪，买入玉米和豆粕；
做多毛利代理价差： 当 z_score < -ENTRY_Z，说明当前毛利代理值显著低于历史均值，预期其向上修复；
- 操作：买入生猪，卖出玉米和豆粕。

平仓信号：

当 abs(z_score) < EXIT_Z 时，认为价差已回到正常区间，组合平仓。

止损信号：

若已持有多头价差，但 z_score 继续上冲到 STOP_Z 以上；
若已持有空头价差，但 z_score 继续下破到 -STOP_Z 以下；
则执行强制平仓。

回测说明

回测初始设置

测试周期： 2023 年 11 月 1 日 - 2024 年 3 月 13 日
交易品种： DCE.lh2409 / DCE.c2409 / DCE.m2409
初始资金： 1000 万元

回测口径说明

本次回测主要用于展示：

生猪-饲料毛利代理价差的构建方式；
产业配比向交易执行配比的映射逻辑；
三腿组合在 TqSdk 框架中的实现方式。

因此，本文采用了固定远月合约的回测口径，而非主力连续或自动换月口径。这样的处理有助于：

保持三个品种在同一交割月下进行对照；
更清晰地呈现价差构建与执行配比之间的对应关系。

同时，这一口径也存在一定边界：

回测结果仍会受到远月流动性、期限结构和固定合约选择的影响；
若用于更贴近实盘的策略评估，还需结合主力切换、换月规则与执行细节进一步完善。

回测结果解读

下图更适合作为策略结构与执行逻辑的展示参考，主要用于观察：

信号是否呈现出一定的均值回归特征；
三腿组合是否能够实现相对稳定的配平；
整数手换算后是否会带来较明显的残余敞口。

在实际应用中，仍需结合换月安排、流动性条件与风控约束，对策略表现作进一步评估。

回测结果

上述回测累计收益走势图

完整代码示例

#!/usr/bin/env python
# coding=utf-8
__author__ = "Chaos"

from datetime import date

import numpy as np
from tqsdk import TqApi, TqAuth, TqBacktest, TqSim, TargetPosTask, BacktestFinished

# === 用户参数 ===
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           # 历史统计窗口（已完成日线）
DATA_LENGTH = LOOKBACK_DAYS + 2
ENTRY_Z = 2.0                # 入场阈值
EXIT_Z = 0.5                 # 平仓阈值
STOP_Z = 3.0                 # 止损阈值
HOG_ORDER_LOTS = 5           # 每次以生猪为基准下单手数

# 简化饲料消耗系数：生产 1 吨生猪所需的主要饲料原料吨数
CORN_TON_PER_HOG_TON = 3.0
MEAL_TON_PER_HOG_TON = 0.6


def calc_proxy_margin(hog_price, corn_price, meal_price):
    """按吨计算毛利代理价差（元/吨）"""
    return hog_price - CORN_TON_PER_HOG_TON * corn_price - MEAL_TON_PER_HOG_TON * meal_price


api = TqApi(
    TqSim(),
    backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
    auth=TqAuth("快期账号", "快期密码"),
)

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, data_length=DATA_LENGTH)
corn_klines = api.get_kline_serial(CORN, 60 * 60 * 24, data_length=DATA_LENGTH)
meal_klines = api.get_kline_serial(SOYMEAL, 60 * 60 * 24, data_length=DATA_LENGTH)

hog_pos_task = TargetPosTask(api, LIVE_HOG)
corn_pos_task = TargetPosTask(api, CORN)
meal_pos_task = TargetPosTask(api, SOYMEAL)


while not (hog_quote.volume_multiple and corn_quote.volume_multiple and meal_quote.volume_multiple):
    api.wait_update()

hog_volume_multiple = hog_quote.volume_multiple
corn_volume_multiple = corn_quote.volume_multiple
meal_volume_multiple = meal_quote.volume_multiple


# 根据生猪下单手数，换算饲料腿的理论手数，并取最近整数手
hog_tons_per_order = HOG_ORDER_LOTS * hog_volume_multiple
corn_lots_per_order = int(round(hog_tons_per_order * CORN_TON_PER_HOG_TON / corn_volume_multiple))
meal_lots_per_order = int(round(hog_tons_per_order * MEAL_TON_PER_HOG_TON / meal_volume_multiple))

# 计算因为整数手取整而产生的残余敞口，方便观察配比误差
corn_residual_tons = hog_tons_per_order * CORN_TON_PER_HOG_TON - corn_lots_per_order * corn_volume_multiple
meal_residual_tons = hog_tons_per_order * MEAL_TON_PER_HOG_TON - meal_lots_per_order * meal_volume_multiple

print(
    f"策略启动，监控合约: {LIVE_HOG}, {CORN}, {SOYMEAL}\n"
    f"每次下单：生猪 {HOG_ORDER_LOTS} 手，玉米 {corn_lots_per_order} 手，豆粕 {meal_lots_per_order} 手\n"
    f"近似配比残余：玉米 {corn_residual_tons:.2f} 吨，豆粕 {meal_residual_tons:.2f} 吨"
)


def build_completed_spread_series():
    """只使用已完成日线，排除当前正在形成的最后一根 K 线"""
    hog_close = np.array(hog_klines.close.iloc[:-1], dtype=float)
    corn_close = np.array(corn_klines.close.iloc[:-1], dtype=float)
    meal_close = np.array(meal_klines.close.iloc[:-1], dtype=float)

    valid = np.isfinite(hog_close) & np.isfinite(corn_close) & np.isfinite(meal_close)
    hog_close = hog_close[valid]
    corn_close = corn_close[valid]
    meal_close = meal_close[valid]

    return calc_proxy_margin(hog_close, corn_close, meal_close)


def get_net_pos(symbol):
    pos = api.get_position(symbol)
    return pos.pos_long - pos.pos_short


last_completed_dt = None

try:
    while True:
        api.wait_update()

        new_bar = (
            api.is_changing(hog_klines.iloc[-1], "datetime")
            or api.is_changing(corn_klines.iloc[-1], "datetime")
            or api.is_changing(meal_klines.iloc[-1], "datetime")
        )
        if not new_bar:
            continue

        completed_dt = min(
            hog_klines.iloc[-2]["datetime"],
            corn_klines.iloc[-2]["datetime"],
            meal_klines.iloc[-2]["datetime"],
        )
        if completed_dt == last_completed_dt:
            continue
        last_completed_dt = completed_dt

        spreads = build_completed_spread_series()
        if len(spreads) < LOOKBACK_DAYS + 1:
            continue

        history_window = spreads[-LOOKBACK_DAYS - 1:-1]
        current_spread = spreads[-1]

        mean_spread = np.mean(history_window)
        std_spread = np.std(history_window)
        if std_spread < 1e-8:
            print("历史标准差过小，跳过本次信号")
            continue

        z_score = (current_spread - mean_spread) / std_spread
        print(f"{completed_dt} 当前毛利代理价差: {current_spread:.2f}, Z-score: {z_score:.2f}")

        hog_net = get_net_pos(LIVE_HOG)
        corn_net = get_net_pos(CORN)
        meal_net = get_net_pos(SOYMEAL)
        has_position = any(pos != 0 for pos in [hog_net, corn_net, meal_net])

        if not has_position:
            if z_score > ENTRY_Z:
                # 做空毛利代理价差：卖出生猪，买入玉米和豆粕
                print(
                    f"开仓-做空毛利代理价差：卖出生猪 {HOG_ORDER_LOTS} 手，"
                    f"买入玉米 {corn_lots_per_order} 手，豆粕 {meal_lots_per_order} 手"
                )
                hog_pos_task.set_target_volume(-HOG_ORDER_LOTS)
                corn_pos_task.set_target_volume(corn_lots_per_order)
                meal_pos_task.set_target_volume(meal_lots_per_order)

            elif z_score < -ENTRY_Z:
                # 做多毛利代理价差：买入生猪，卖出玉米和豆粕
                print(
                    f"开仓-做多毛利代理价差：买入生猪 {HOG_ORDER_LOTS} 手，"
                    f"卖出玉米 {corn_lots_per_order} 手，豆粕 {meal_lots_per_order} 手"
                )
                hog_pos_task.set_target_volume(HOG_ORDER_LOTS)
                corn_pos_task.set_target_volume(-corn_lots_per_order)
                meal_pos_task.set_target_volume(-meal_lots_per_order)

        else:
            # 注意：TargetPosTask 发出目标持仓后，真正报单与成交推进依赖后续 wait_update()
            if abs(z_score) < EXIT_Z:
                print("价差回归正常区间，平仓所有头寸")
                hog_pos_task.set_target_volume(0)
                corn_pos_task.set_target_volume(0)
                meal_pos_task.set_target_volume(0)

            elif hog_net > 0 and z_score > STOP_Z:
                print("止损：做多毛利代理价差后，价差继续向不利方向偏离")
                hog_pos_task.set_target_volume(0)
                corn_pos_task.set_target_volume(0)
                meal_pos_task.set_target_volume(0)

            elif hog_net < 0 and z_score < -STOP_Z:
                print("止损：做空毛利代理价差后，价差继续向不利方向偏离")
                hog_pos_task.set_target_volume(0)
                corn_pos_task.set_target_volume(0)
                meal_pos_task.set_target_volume(0)

except BacktestFinished:
    print("回测结束")
finally:
    api.close()

目录