基于距离的配对交易策略
一、交易策略解释
核心思想
基于距离的配对交易策略（Distance-Based Pairs Trading）的核心在于识别价格行为相似的资产对，并利用价格差异的短期偏离进行交易。该策略假设高度相似的资产价格差异存在均值回归的趋势，当价差异常偏离时提供了交易机会。与传统的协整配对交易不同，距离法更关注价格序列本身的相似性，而非严格的长期均衡关系。
当两个资产价格走势相似时，它们之间的价差通常会在一个相对稳定的范围内波动。当价差显著偏离这个稳定范围时，策略会做出相应的交易决策，预期价差将回归到其历史平均水平。
理论基础
-
相对价值理论：同类资产应有相似的价格行为，价格差异的极端偏离往往代表市场的临时非理性或信息不对称。
-
市场效率假说的缓和版本：市场可能存在短期无效性，但长期会回归有效状态。
-
均值回归现象：许多金融时间序列在短期内可能偏离长期均值，但最终会回归到其长期均值水平。
-
学术研究：Gatev等人(2006)的研究表明，距离法在多个市场中都能产生显著的超额收益，特别是在流动性高的市场中。
策略适用场景
-
同行业或相关产业链的资产：如同一行业的股票、期货合约，或上下游关系紧密的商品期货。
-
相似基本面影响因素的资产：受相同宏观经济因素或供需结构影响的资产。
-
高流动性市场：交易量充足、买卖价差小的市场，以确保有效执行交易。
-
波动性较高但相关性稳定的资产：价格波动给出更多交易信号，而相关性稳定保证策略逻辑有效。
-
短期市场波动中：特别适合在市场出现短期情绪波动或流动性事件导致的价格偏离时产生交易机会。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
策略的核心实现步骤包括：
- 数据收集与处理：
- 获取两个合约的日线收盘价数据
- 维持固定长度的历史窗口（本例为30个交易日）
- 对价格序列进行Z-Score标准化处理
- 标准化价差计算：
- 计算标准化后的价格差异：spread = norm1 - norm2
- 计算价差的均值和标准差
- 确定当前价差相对于均值的偏离程度
- 交易信号生成：
- 当价差低于均值减去K倍标准差时，做多价差（做多第一个合约，做空第二个合约）
- 当价差高于均值加上K倍标准差时，做空价差（做空第一个合约，做多第二个合约）
- 当价差回归至均值附近（偏离小于0.5倍标准差）时平仓
- 仓位管理：
- 根据两个合约的价格比例确定交易比率，保证交易价值平衡
- 使用固定手数进行交易，确保风险可控
交易逻辑
开仓条件：
-
价差显著低于均值时做多价差：当两个标的的标准化价差大幅低于历史均值（超过2个标准差）时，我们判断第一个合约相对被低估、第二个合约相对被高估，因此买入第一个合约同时卖出第二个合约。
-
价差显著高于均值时做空价差：当价差大幅高于历史均值时，判断第一个合约相对被高估、第二个合约相对被低估，因此卖出第一个合约同时买入第二个合约。
-
交易比例动态调整：为了保证两个合约的交易价值大致平衡，根据当前价格比例动态计算第二个合约的交易手数。
平仓条件：
-
价差回归到均值：当价差回归到接近历史均值的区域（偏离小于0.5个标准差）时，认为价格关系回归正常，获利了结。
-
时间止损：如果持仓超过10个交易日仍未达到目标，强制平仓，避免长期占用资金。
-
比例止损：当组合亏损超过5%时触发止损，控制单笔交易风险。
策略的核心逻辑是通过统计方法识别价格关系的异常偏离，在偏离发生时建立相应的对冲头寸，然后等待价格关系回归正常。整个过程不依赖于对市场方向的判断，而是利用两个相关资产间的相对定价偏差获利，属于典型的市场中性策略。
回测
回测初始设置
- 测试周期： 2023 年 2 月 1 日 - 2023 年 4 月 27 日
- 交易品种： SHFE.rb2305/SHFE.hc2305
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'Chaos'
from tqsdk import TqApi, TqAuth, TargetPosTask, TqBacktest, BacktestFinished
from datetime import date, datetime
import numpy as np
# === 全局参数 ===
SYMBOL1 = "SHFE.rb2305"
SYMBOL2 = "SHFE.hc2305"
WINDOW = 30
K_THRESHOLD = 2.0
CLOSE_THRESHOLD = 0.5
MAX_HOLD_DAYS = 10
STOP_LOSS_PCT = 0.05
POSITION_LOTS1 = 200 # 合约1固定手数
POSITION_RATIO = 1.0 # 合约2与合约1的数量比例
# === 全局变量 ===
price_data1, price_data2 = [], []
position_long = False
position_short = False
entry_price1 = 0
entry_price2 = 0
position_time = None
trade_ratio = 1
entry_spread = 0
trade_count = 0
win_count = 0
total_profit = 0
# === API初始化 ===
api = TqApi(backtest=TqBacktest(start_dt=date(2023, 2, 1),end_dt=date(2023, 4, 27)),
auth=TqAuth("快期账号", "快期密码")
)
quote1 = api.get_quote(SYMBOL1)
quote2 = api.get_quote(SYMBOL2)
klines1 = api.get_kline_serial(SYMBOL1, 24*60*60)
klines2 = api.get_kline_serial(SYMBOL2, 24*60*60)
target_pos1 = TargetPosTask(api, SYMBOL1)
target_pos2 = TargetPosTask(api, SYMBOL2)
print(f"策略开始运行，交易品种: {SYMBOL1} 和 {SYMBOL2}")
try:
while True:
api.wait_update()
if api.is_changing(klines1.iloc[-1], "datetime") or api.is_changing(klines2.iloc[-1], "datetime"):
price_data1.append(klines1.iloc[-1]["close"])
price_data2.append(klines2.iloc[-1]["close"])
if len(price_data1) <= WINDOW:
continue
if len(price_data1) > WINDOW:
price_data1 = price_data1[-WINDOW:]
price_data2 = price_data2[-WINDOW:]
data1 = np.array(price_data1)
data2 = np.array(price_data2)
norm1 = (data1 - np.mean(data1)) / np.std(data1)
norm2 = (data2 - np.mean(data2)) / np.std(data2)
spread = norm1 - norm2
mean_spread = np.mean(spread)
std_spread = np.std(spread)
current_spread = spread[-1]
price_ratio = quote2.last_price / quote1.last_price
trade_ratio = round(price_ratio * POSITION_RATIO, 2)
position_lots2 = int(POSITION_LOTS1 * trade_ratio)
current_time = datetime.fromtimestamp(klines1.iloc[-1]["datetime"] / 1e9)
# === 平仓逻辑 ===
if position_long or position_short:
days_held = (current_time - position_time).days
if position_long:
current_profit = (quote1.last_price - entry_price1) * POSITION_LOTS1 - (quote2.last_price - entry_price2) * position_lots2
else:
current_profit = (entry_price1 - quote1.last_price) * POSITION_LOTS1 - (entry_price2 - quote2.last_price) * position_lots2
profit_pct = current_profit / (entry_price1 * POSITION_LOTS1)
close_by_mean = abs(current_spread - mean_spread) < CLOSE_THRESHOLD * std_spread
close_by_time = days_held >= MAX_HOLD_DAYS
close_by_stop = profit_pct <= -STOP_LOSS_PCT
if close_by_mean or close_by_time or close_by_stop:
target_pos1.set_target_volume(0)
target_pos2.set_target_volume(0)
trade_count += 1
if profit_pct > 0:
win_count += 1
total_profit += current_profit
reason = "均值回归" if close_by_mean else "时间限制" if close_by_time else "止损"
print(f"平仓 - {reason}, 盈亏: {profit_pct:.2%}, 持仓天数: {days_held}")
position_long = False
position_short = False
# === 开仓逻辑 ===
else:
if current_spread < mean_spread - K_THRESHOLD * std_spread:
target_pos1.set_target_volume(POSITION_LOTS1)
target_pos2.set_target_volume(-position_lots2)
position_long = True
position_time = current_time
entry_price1 = quote1.last_price
entry_price2 = quote2.last_price
entry_spread = current_spread
print(f"开仓 - 多价差, 合约1: {POSITION_LOTS1}手, 合约2: {-position_lots2}手, 比例: {trade_ratio}")
elif current_spread > mean_spread + K_THRESHOLD * std_spread:
target_pos1.set_target_volume(-POSITION_LOTS1)
target_pos2.set_target_volume(position_lots2)
position_short = True
position_time = current_time
entry_price1 = quote1.last_price
entry_price2 = quote2.last_price
entry_spread = current_spread
print(f"开仓 - 空价差, 合约1: {-POSITION_LOTS1}手, 合约2: {position_lots2}手, 比例: {trade_ratio}")
# 每日统计
if api.is_changing(klines1.iloc[-1], "datetime"):
account = api.get_account()
print(f"日期: {current_time.date()}, 账户权益: {account.balance:.2f}, 可用资金: {account.available:.2f}")
if trade_count > 0:
print(f"交易统计 - 总交易: {trade_count}, 胜率: {win_count/trade_count:.2%}, 总盈亏: {total_profit:.2f}")
except BacktestFinished as e:
print("回测结束")
api.close()