裂解价差套利策略

一、交易策略解释

核心思想

裂解价差套利的核心思想，是敏锐地捕捉并利用原油期货与其主要炼化产品（如汽油、柴油或取暖油）期货之间出现的非正常价差关系来进行套利。 这种策略的根基在于，尽管原油是炼化产品的直接原料，它们的价格在大多数时候会保持一种相对稳定的、反映炼油成本和合理利润的比例关系，但市场短期内的各种因素冲击，如季节性需求波动、炼厂意外停产、原油供应中断、地缘政治风险或是投机力量的推动，都可能导致这一价差暂时偏离其历史平均水平或经济学上的合理区间。 当这种偏离达到一定程度，使得价差显得过宽（炼油利润异常丰厚）或过窄（炼油利润异常微薄甚至亏损）时，套利机会便可能显现。

此时，洞察到这种失衡的套利者会果断介入，同时在期货市场上建立方向相反的头寸：如果他们判断价差过大，意味着成品油相对于原油价格过高，他们便会“卖出”裂解价差，即卖出成品油期货（如汽油和柴油），同时买入原油期货，赌的是未来成品油价格会相对原油回落，价差收窄；反之，如果判断价差过小，意味着成品油相对于原油价格过低，他们则会“买入”裂解价差，即买入成品油期货，同时卖出原油期货，预期未来成品油价格会相对原油上涨，价差扩大。 这种操作的精髓在于，套利者并不去预测单一商品价格的绝对涨跌方向，而是专注于原油与炼化产品之间相对价值的变动。他们通过在金融市场上构建一个模拟炼油厂运营的头寸组合——例如，买入原油期货并卖出汽油和柴油期货，就如同在纸面上“锁定”了一个炼油过程的潜在利润——目标是在价差从当前的“非正常”状态回归到历史均值或更为“正常”的水平时，通过平掉这些相互对冲的头寸来获取利润。

理论基础

• 商品间的长期均衡关系： 原油是炼化产品的上游，它们之间的价格在长期内存在着经济上的联系。炼油成本、运输成本、仓储成本以及供需关系决定了一个合理的裂解价差区间。

• 市场短期失衡与价差偏离： 由于各种短期因素（如季节性需求变化、突发性供给中断、投机行为等），原油和炼化产品的价格可能会出现暂时的失衡，导致裂解价差偏离其长期均衡水平。

• 套利机制与价差回归： 当裂解价差过大时（意味着炼油利润过高），套利者会买入原油期货，卖出炼化产品期货，这会增加对原油的需求，抑制其价格上涨；同时增加炼化产品的供给，抑制其价格下跌，从而缩小裂解价差。反之，当裂解价差过小时，套利者会卖出原油期货，买入炼化产品期货，推动价差扩大。这种套利行为有助于将价差拉回其正常的波动区间。

策略适用场景

• 历史价差波动稳定且存在均值回归特性： 如果历史数据显示特定的原油和炼化产品组合的裂解价差在一定范围内波动，并且有明显的均值回归趋势，那么进行套利的成功率会更高。

• 季节性需求变化显著： 如 Investopedia 文章提到的，夏季汽油需求旺盛，冬季取暖油需求增加。这些季节性因素可能导致裂解价差在特定时间段内出现规律性的扩大或缩小，为套利者提供机会。

• 炼油产能或库存出现意外变化： 炼油厂的计划外停产、检修延期，或者原油及炼化产品库存的异常波动，都可能导致短期的供需失衡，使得裂解价差偏离正常水平。

• 地缘政治或自然灾害影响原油或炼化产品供给： 这些突发事件可能导致原油和炼化产品价格出现不同幅度的波动，从而创造套利机会。

• 期货合约流动性良好： 由于套利需要同时交易多个期货合约，因此相关的原油和炼化产品期货合约必须具有充足的流动性，以保证能够以合理的成本快速建仓和平仓。

二、天勤介绍

天勤平台概述

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

• 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
• 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
• 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。

策略开发流程

• 环境准备
  • 安装Python环境（推荐Python 3.6或以上版本）
  • 安装tqsdk包：pip install tqsdk
  • 注册天勤账户获取访问密钥
• 数据准备
  • 订阅近月与远月合约行情
  • 获取历史K线或者Tick数据（用于分析与行情推进）
• 策略编写
  • 设计信号生成逻辑（基于价差、均值和标准差）
  • 编写交易执行模块（开仓、平仓逻辑）
  • 实现风险控制措施（止损、资金管理）
• 回测验证
  • 设置回测时间区间和初始资金
  • 运行策略获取回测结果
  • 分析绩效指标（胜率、收益率、夏普率等）
• 策略优化
  • 调整参数（标准差倍数、窗口大小等）
  • 添加过滤条件（成交量、波动率等）
  • 完善风险控制机制

三、天勤实现策略

策略原理

这个裂解价差套利策略基于原油与其成品油(汽油和柴油)之间的价格关系偏离正常范围时进行交易。具体计算方法如下：

1. 裂解价差计算：

• 使用公式：裂解价差 = (汽油价值 + 柴油价值) - 原油价值

• 考虑了合约乘数和设定的比例关系(2:1:1)，使价差反映实际市场价值关系

2. 统计指标构建：

• 计算过去30天的历史裂解价差数据

• 计算历史价差的均值(mean_spread)和标准差(std_spread)

• 将当前价差标准化为Z-score: (current_spread - mean_spread) / std_spread

• Z-score表示当前价差偏离历史均值的程度，是核心指标

3. 价差均值回归假设：

• 策略假设裂解价差长期会回归到历史均值

• 当价差显著偏离均值时(高于或低于1.5个标准差)，认为这是暂时性失衡，存在套利机会

交易逻辑

开仓信号

• 卖出裂解价差 (当Z-score > 1.5)：
  • 表示当前裂解价差过高，预期将向均值回归下降
  • 操作：买入原油，同时卖出汽油和柴油
  • 逻辑：如果成品油价格相对于原油过高，买入原料卖出产品以获利
• 买入裂解价差 (当Z-score < -1.5)：
  • 表示当前裂解价差过低，预期将向均值回归上升
  • 操作：卖出原油，同时买入汽油和柴油
  • 逻辑：如果成品油价格相对于原油过低，卖出原料买入产品以获利

平仓信号：

• 价差回归 (当|Z-score| < 0.3)：
  • 表示价差已经回归到接近历史均值的水平
  • 操作：平掉所有持仓
  • 逻辑：套利机会已消失，获利了结

回测

回测初始设置

• 测试周期： 2024 年 1 月 26 日 - 2024 年 3 月 1 日
• 交易品种： INE.sc2405/SHFE.fu2405/INE.nr2405
• 初始资金： 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

# === 用户参数 ===
# 合约参数
CRUDE_OIL = "INE.sc2405"  # 原油期货合约
GASOLINE = "SHFE.fu2405"   # 燃料油期货合约
DIESEL = "INE.nr2405"      # 柴油期货合约
START_DATE = date(2024, 1, 26)  # 回测开始日期
END_DATE = date(2024, 3, 1)  # 回测结束日期

# 套利参数
LOOKBACK_DAYS = 30         # 计算历史价差的回溯天数
STD_THRESHOLD = 1.5        # 标准差阈值，超过此阈值视为套利机会
ORDER_VOLUME = 30           # 下单手数
CLOSE_THRESHOLD = 0.3      # 平仓阈值(标准差)

# 裂解价差比例
CRUDE_RATIO = 2
GASOLINE_RATIO = 1
DIESEL_RATIO = 1

# === 初始化API ===
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
            auth=TqAuth("快期账号", "快期密码"))

# 获取合约行情和K线
crude_quote = api.get_quote(CRUDE_OIL)
gasoline_quote = api.get_quote(GASOLINE)
diesel_quote = api.get_quote(DIESEL)

crude_klines = api.get_kline_serial(CRUDE_OIL, 60*60*24, LOOKBACK_DAYS)
gasoline_klines = api.get_kline_serial(GASOLINE, 60*60*24, LOOKBACK_DAYS)
diesel_klines = api.get_kline_serial(DIESEL, 60*60*24, LOOKBACK_DAYS)

# 创建目标持仓任务
crude_pos = TargetPosTask(api, CRUDE_OIL)
gasoline_pos = TargetPosTask(api, GASOLINE)
diesel_pos = TargetPosTask(api, DIESEL)

# 获取合约乘数
crude_volume_multiple = crude_quote.volume_multiple
gasoline_volume_multiple = gasoline_quote.volume_multiple
diesel_volume_multiple = diesel_quote.volume_multiple

# 初始化状态变量
position_time = 0  # 建仓时间
in_position = False  # 是否有持仓
mean_spread = 0  # 历史价差均值
std_spread = 0  # 历史价差标准差

print(f"策略启动，监控合约: {CRUDE_OIL}, {GASOLINE}, {DIESEL}")

# === 主循环 ===
try:
    # 初始计算历史统计值
    spreads = []
    for i in range(len(crude_klines) - 1):
        crude_price = crude_klines.close.iloc[i] * crude_volume_multiple * CRUDE_RATIO
        gasoline_price = gasoline_klines.close.iloc[i] * gasoline_volume_multiple * GASOLINE_RATIO
        diesel_price = diesel_klines.close.iloc[i] * diesel_volume_multiple * DIESEL_RATIO
        
        spread = (gasoline_price + diesel_price) - crude_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(crude_klines) or api.is_changing(gasoline_klines) or api.is_changing(diesel_klines):
            # 重新计算历史价差统计
            spreads = []
            for i in range(len(crude_klines) - 1):
                crude_price = crude_klines.close.iloc[i] * crude_volume_multiple * CRUDE_RATIO
                gasoline_price = gasoline_klines.close.iloc[i] * gasoline_volume_multiple * GASOLINE_RATIO
                diesel_price = diesel_klines.close.iloc[i] * diesel_volume_multiple * DIESEL_RATIO
                
                spread = (gasoline_price + diesel_price) - crude_price
                spreads.append(spread)
            
            mean_spread = np.mean(spreads)
            std_spread = np.std(spreads)
            
            # 计算当前裂解价差
            crude_price = crude_klines.close.iloc[-1] * crude_volume_multiple * CRUDE_RATIO
            gasoline_price = gasoline_klines.close.iloc[-1] * gasoline_volume_multiple * GASOLINE_RATIO
            diesel_price = diesel_klines.close.iloc[-1] * diesel_volume_multiple * DIESEL_RATIO
            
            current_spread = (gasoline_price + diesel_price) - crude_price
            
            # 计算z-score (标准化的价差)
            z_score = (current_spread - mean_spread) / std_spread
            
            print(f"当前裂解价差: {current_spread:.2f}, Z-score: {z_score:.2f}")
            
            # 获取当前持仓
            crude_position = api.get_position(CRUDE_OIL)
            gasoline_position = api.get_position(GASOLINE)
            diesel_position = api.get_position(DIESEL)
            
            current_crude_pos = crude_position.pos_long - crude_position.pos_short
            current_gasoline_pos = gasoline_position.pos_long - gasoline_position.pos_short
            current_diesel_pos = diesel_position.pos_long - diesel_position.pos_short
            
            # === 交易信号判断 ===
            if not in_position:  # 如果没有持仓
                if z_score > STD_THRESHOLD:  # 价差显著高于均值
                    # 卖出裂解价差：买入原油，卖出汽油和柴油
                    print("卖出裂解价差：买入原油，卖出汽油和柴油")
                    crude_pos.set_target_volume(ORDER_VOLUME)
                    gasoline_pos.set_target_volume(-ORDER_VOLUME)
                    diesel_pos.set_target_volume(-ORDER_VOLUME)
                    position_time = time.time()
                    in_position = True
                    
                elif z_score < -STD_THRESHOLD:  # 价差显著低于均值
                    # 买入裂解价差：卖出原油，买入汽油和柴油
                    print("买入裂解价差：卖出原油，买入汽油和柴油")
                    crude_pos.set_target_volume(-ORDER_VOLUME)
                    gasoline_pos.set_target_volume(ORDER_VOLUME)
                    diesel_pos.set_target_volume(ORDER_VOLUME)
                    position_time = time.time()
                    in_position = True
            
            else:  # 如果已有持仓
                # 检查是否应当平仓
                if abs(z_score) < CLOSE_THRESHOLD:  # 价差恢复正常
                    print("价差恢复正常，平仓所有头寸")
                    crude_pos.set_target_volume(0)
                    gasoline_pos.set_target_volume(0)
                    diesel_pos.set_target_volume(0)
                    in_position = False

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