Z-Score Mean Reversion for a Coke and Coking Coal Futures Spread
Summary
This code describes a mean-reversion strategy for the spread between Dalian Commodity Exchange coke and coking coal futures. It calculates a weighted value spread using contract prices, contract multipliers, and a specified leg ratio, then estimates the spread's mean and standard deviation from recent daily bars. A standardized score measures the current spread's distance from its historical mean. The strategy enters opposing positions in the two contracts when the score crosses either side of a threshold, exits when it returns close to the mean, and closes positions if the spread moves further against the trade.
The script specifies a backtest period, lookback window, entry and exit thresholds, and leg sizes, and uses target-position tasks to submit orders. It does not provide backtest results or establish that the spread is stationary. The chosen ratio, rolling statistics, and contract selection may affect behavior, while transaction costs, slippage, margin, and roll handling are not addressed. The displayed code also needs careful review before practical use, including checking that both legs stay aligned and that position state matches actual fills.
Key ideas
- The strategy trades the weighted price spread between coke and coking coal futures using opposing positions in the two legs.
- It standardizes the spread against its recent mean and standard deviation to generate entry signals.
- Positions are closed when the score approaches the mean or crosses a further adverse threshold.
- The script defines contract ratios, order sizes, and a historical backtest window but reports no performance evidence.
- Practical evaluation would need to examine spread stability, execution costs, margin, and synchronization of both legs.
Tags
Full text
# jm-j-spread.py
```py
#!/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
# === 用户参数 ===
# 合约参数
J = "DCE.j2409" # 焦炭期货合约
JM = "DCE.jm2409" # 焦煤期货合约
START_DATE = date(2023, 11, 1) # 回测开始日期
END_DATE = date(2024, 4, 30) # 回测结束日期
# 套利参数
LOOKBACK_DAYS = 30 # 计算历史价差的回溯天数
STD_THRESHOLD = 2.0 # 标准差阈值，超过此阈值视为套利机会
ORDER_VOLUME = 50 # 焦炭的下单手数
CLOSE_THRESHOLD = 0.5 # 平仓阈值(标准差)
# 配比参数（可根据实际工艺调整）
J_RATIO = 10 # 10手焦炭
JM_RATIO = 22 # 22手焦煤（约1.32配比）
# === 初始化API ===
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账户", "快期密码"))
# 获取合约行情和K线
j_quote = api.get_quote(J)
jm_quote = api.get_quote(JM)
j_klines = api.get_kline_serial(J, 60 * 60 * 24, LOOKBACK_DAYS)
jm_klines = api.get_kline_serial(JM, 60 * 60 * 24, LOOKBACK_DAYS)
# 创建目标持仓任务
j_pos = TargetPosTask(api, J)
jm_pos = TargetPosTask(api, JM)
# 获取合约乘数
j_volume_multiple = j_quote.volume_multiple
jm_volume_multiple = jm_quote.volume_multiple
# 初始化状态变量
position_time = 0 # 建仓时间
in_position = False # 是否有持仓
mean_spread = 0 # 历史价差均值
std_spread = 0 # 历史价差标准差
print(f"策略启动，监控合约: {J}, {JM}")
# === 主循环 ===
try:
# 初始计算历史统计值
spreads = []
for i in range(len(j_klines) - 1):
j_value = j_klines.close.iloc[i] * j_volume_multiple * J_RATIO
jm_value = jm_klines.close.iloc[i] * jm_volume_multiple * JM_RATIO
spread = j_value - jm_value
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(j_klines) or api.is_changing(jm_klines):
# 重新计算历史价差统计
spreads = []
for i in range(len(j_klines) - 1):
j_value = j_klines.close.iloc[i] * j_volume_multiple * J_RATIO
jm_value = jm_klines.close.iloc[i] * jm_volume_multiple * JM_RATIO
spread = j_value - jm_value
spreads.append(spread)
mean_spread = np.mean(spreads)
std_spread = np.std(spreads)
# 计算当前炼焦利润价差
j_value = j_klines.close.iloc[-1] * j_volume_multiple * J_RATIO
jm_value = jm_klines.close.iloc[-1] * jm_volume_multiple * JM_RATIO
current_spread = j_value - jm_value
# 计算z-score (标准化的价差)
z_score = (current_spread - mean_spread) / std_spread
print(f"当前炼焦利润: {current_spread:.2f}, Z-score: {z_score:.2f}")
# 获取当前持仓
j_position = api.get_position(J)
jm_position = api.get_position(JM)
current_j_pos = j_position.pos_long - j_position.pos_short
current_jm_pos = jm_position.pos_long - jm_position.pos_short
# === 交易信号判断 ===
if not in_position:
if z_score > STD_THRESHOLD:
# 做空炼焦利润：卖出焦炭，买入焦煤
print(f"做空炼焦利润：卖出焦炭{ORDER_VOLUME}手，买入焦煤{int(ORDER_VOLUME * JM_RATIO / J_RATIO)}手")
j_pos.set_target_volume(-ORDER_VOLUME)
jm_pos.set_target_volume(int(ORDER_VOLUME * JM_RATIO / J_RATIO))
position_time = time.time()
in_position = True
elif z_score < -STD_THRESHOLD:
# 做多炼焦利润：买入焦炭，卖出焦煤
print(f"做多炼焦利润：买入焦炭{ORDER_VOLUME}手，卖出焦煤{int(ORDER_VOLUME * JM_RATIO / J_RATIO)}手")
j_pos.set_target_volume(ORDER_VOLUME)
jm_pos.set_target_volume(-int(ORDER_VOLUME * JM_RATIO / J_RATIO))
position_time = time.time()
in_position = True
else: # 如果已有持仓
# 检查是否应当平仓
if abs(z_score) < CLOSE_THRESHOLD: # 利润回归正常
print("利润回归正常，平仓所有头寸")
j_pos.set_target_volume(0)
jm_pos.set_target_volume(0)
in_position = False
# 止损逻辑
if (z_score > STD_THRESHOLD * 1.5 and current_j_pos < 0) or \
(z_score < -STD_THRESHOLD * 1.5 and current_j_pos > 0):
print("止损：利润向不利方向进一步偏离")
j_pos.set_target_volume(0)
jm_pos.set_target_volume(0)
in_position = False
except BacktestFinished as e:
print("回测结束")
api.close()
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.