市场存在不确定性，投资需谨慎行事，过往的回测结果并不预示未来的实际表现。本文提供的策略示例仅供学习和交流使用，在实际操作前，务必对其进行全面理解和严格测试。以下示例代码基于天勤量化测试平台编写，仅供参考，不对任何交易结果承担责任。

一、交易策略解析

核心理念

压榨价差（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库，采用单线程异步架构以保障高性能。

策略开发步骤

环境搭建

1. 安装Python环境（建议使用Python 3.6及以上版本）。
2. 安装tqsdk库：通过命令行执行 `pip install tqsdk`。
3. 注册天勤账号并获取访问密钥。

数据准备

订阅最近和远期合约的市场信息，获取历史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()