涡旋指标期货量化策略
一、交易策略解释
核心思想
涡旋指标（Vortex Indicator，VI）量化交易策略的核心思想是利用价格运动中的"涡旋"模式来识别和跟踪市场趋势。 该策略基于价格运动在形成趋势时展现的特定方向性运动模式，通过比较正向涡旋指标(+VI)和负向涡旋指标(-VI)的相对强度来确定趋势的方向和力度。
当+VI线上穿-VI线时，表明上升动量增强，可能形成上升趋势，产生做多信号；当-VI线上穿+VI线时，表明下降动量增强，可能形成下降趋势，产生做空信号。 该策略的本质是抓住市场趋势的起始阶段并伴随趋势发展获利，同时通过涡旋指标值的大小判断趋势强度，协助进行动态仓位管理。
理论基础
涡旋指标由Etienne Botes和Douglas Siepman于2010年首次引入交易社区，其理论基础源于自然界中涡旋流动的现象（如水流绕过障碍物时形成的涡流模式）。 该指标理论受到J. Welles Wilder方向运动概念的影响，认为连续价格柱之间的关系可以提供关于市场方向的有价值见解。学术研究表明，价格运动在形成趋势时确实表现出类似涡旋的特性：
-
价格波动的方向性集聚：当市场形成趋势时，价格波动会在某一方向上集聚，表现为连续的高点比前一个高点更高，低点比前一个低点更高（上升趋势）；或连续的低点比前一个低点更低，高点比前一个高点更低（下降趋势）。
-
多周期相关性：趋势形成通常跨越多个时间周期，形成自相似的价格模式，这与自然界中的涡旋结构类似。
-
动量累积效应：市场趋势往往伴随着动量的积累过程，这可以通过涡旋指标中+VI和-VI的相对变化捕捉到。
策略适用场景
-
明显趋势市场：策略在明显趋势形成时表现最佳，特别是在大宗商品、外汇和期货等波动性较大的市场中。
-
中长期交易：虽然涡旋指标可用于不同时间框架，但通常在日线、4小时线等中长期时间框架上表现更佳，可以过滤掉短期噪音。
-
高波动性品种：指标在波动较大的期货品种中能更好地捕捉趋势起点，如能源期货、金属期货和农产品期货等。
-
宏观经济转折期：在宏观经济出现明显转折，导致商品价格趋势发生改变时，涡旋指标能较早捕捉到这种变化。
-
季节性波动市场：对于具有季节性波动特征的商品期货，涡旋指标能够有效识别季节性价格趋势的开始。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
涡旋指标（VI）的计算涉及以下几个步骤：
1. 计算真实范围（TR）： TR = max(当前高点 - 当前低点, |当前高点 - 前一收盘价|, |当前低点 - 前一收盘价|)
2. 计算正向涡旋运动（+VM）： +VM = |当前高点 - 前一低点|
3. 计算负向涡旋运动（-VM）： -VM = |当前低点 - 前一高点|
4. 计算N周期内TR、+VM和-VM的总和： 通常N=14，但可根据不同市场条件优化
5. 计算正向涡旋指标（+VI）： +VI = (N周期+VM总和) / (N周期TR总和)
6. 计算负向涡旋指标（-VI）： -VI = (N周期-VM总和) / (N周期TR总和)
交易逻辑
-
入场信号：
-
当+VI上穿-VI时，产生做多信号
-
当-VI上穿+VI时，产生做空信号
-
为减少假信号，加入条件：只有当穿越线的VI值大于1时，信号才有效
-
-
出场信号：
-
当持有多头仓位时，如果-VI上穿+VI，平仓离场
-
当持有空头仓位时，如果+VI上穿-VI，平仓离场
-
-
止损设置：
-
基于ATR（平均真实范围）设置动态止损
-
多头止损位 = 入场价 - 2 * ATR
-
空头止损位 = 入场价 + 2 * ATR
-
回测
回测初始设置
- 测试周期： 2022 年 11 月 1 日至 2023 年 4 月 19 日
- 交易品种： CFFEX.IC2306
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from datetime import date
import numpy as np
import pandas as pd
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
from tqsdk.ta import ATR
# ===== 全局参数设置 =====
SYMBOL = "CFFEX.IC2306" # 螺纹钢期货合约
POSITION_SIZE = 30 # 固定交易手数
START_DATE = date(2022, 11, 1) # 回测开始日期
END_DATE = date(2023, 4, 19) # 回测结束日期
# 涡旋指标参数
VI_PERIOD = 14 # 涡旋指标周期
ATR_PERIOD = 14 # ATR指标周期
ATR_MULTIPLIER = 2.0 # 止损倍数
VI_THRESHOLD = 1.0 # VI值阈值，筛选强度较大的信号
# ===== 全局变量 =====
current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓
entry_price = 0 # 开仓价格
stop_loss_price = 0 # 止损价格
# ===== 涡旋指标计算函数 =====
def calculate_vortex(df, period=14):
"""计算涡旋指标"""
# 计算真实范围(TR)
df['tr'] = np.maximum(
np.maximum(
df['high'] - df['low'],
np.abs(df['high'] - df['close'].shift(1))
),
np.abs(df['low'] - df['close'].shift(1))
)
# 计算正向涡旋运动(+VM)
df['plus_vm'] = np.abs(df['high'] - df['low'].shift(1))
# 计算负向涡旋运动(-VM)
df['minus_vm'] = np.abs(df['low'] - df['high'].shift(1))
# 计算N周期内的总和
df['tr_sum'] = df['tr'].rolling(window=period).sum()
df['plus_vm_sum'] = df['plus_vm'].rolling(window=period).sum()
df['minus_vm_sum'] = df['minus_vm'].rolling(window=period).sum()
# 计算涡旋指标
df['plus_vi'] = df['plus_vm_sum'] / df['tr_sum']
df['minus_vi'] = df['minus_vm_sum'] / df['tr_sum']
return df
# ===== 策略开始 =====
print("开始运行涡旋指标(VI)期货策略...")
# 创建API实例
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账号", "快期密码"))
# 订阅合约的日K线数据
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24) # 日线数据
# 创建目标持仓任务
target_pos = TargetPosTask(api, SYMBOL)
try:
while True:
# 等待更新
api.wait_update()
# 如果K线有更新
if api.is_changing(klines.iloc[-1], "datetime"):
# 确保有足够的数据计算指标
if len(klines) < max(VI_PERIOD, ATR_PERIOD) + 5:
continue
# 计算涡旋指标
df = pd.DataFrame(klines)
df = calculate_vortex(df, VI_PERIOD)
# 计算ATR
atr_data = ATR(klines, ATR_PERIOD)
current_atr = float(atr_data.atr.iloc[-1])
# 获取最新和前一个周期的数据
current_price = float(klines.close.iloc[-1])
current_plus_vi = float(df.plus_vi.iloc[-1])
current_minus_vi = float(df.minus_vi.iloc[-1])
prev_plus_vi = float(df.plus_vi.iloc[-2])
prev_minus_vi = float(df.minus_vi.iloc[-2])
# 获取当前日期
current_datetime = pd.to_datetime(klines.datetime.iloc[-1], unit='ns')
date_str = current_datetime.strftime('%Y-%m-%d')
# 输出调试信息
print(f"日期: {date_str}, 价格: {current_price}, +VI: {current_plus_vi:.4f}, -VI: {current_minus_vi:.4f}, ATR: {current_atr:.2f}")
# ===== 交易逻辑 =====
# 空仓状态 - 寻找开仓机会
if current_direction == 0:
# 多头信号: +VI上穿-VI且+VI > 阈值
if prev_plus_vi <= prev_minus_vi and current_plus_vi > current_minus_vi and current_plus_vi > VI_THRESHOLD:
# 设置入场价格
entry_price = current_price
# 设置止损价格
stop_loss_price = entry_price - ATR_MULTIPLIER * current_atr
# 设置持仓方向和目标持仓
current_direction = 1
target_pos.set_target_volume(POSITION_SIZE)
print(f"多头开仓: 价格={entry_price}, 手数={POSITION_SIZE}, 止损价={stop_loss_price:.2f}")
# 空头信号: -VI上穿+VI且-VI > 阈值
elif prev_minus_vi <= prev_plus_vi and current_minus_vi > current_plus_vi and current_minus_vi > VI_THRESHOLD:
# 设置入场价格
entry_price = current_price
# 设置止损价格
stop_loss_price = entry_price + ATR_MULTIPLIER * current_atr
# 设置持仓方向和目标持仓
current_direction = -1
target_pos.set_target_volume(-POSITION_SIZE)
print(f"空头开仓: 价格={entry_price}, 手数={POSITION_SIZE}, 止损价={stop_loss_price:.2f}")
# 多头持仓 - 检查平仓条件
elif current_direction == 1:
# 条件1: 止损触发
if current_price <= stop_loss_price:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止损平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
# 条件2: 信号反转 (-VI上穿+VI)
elif prev_minus_vi <= prev_plus_vi and current_minus_vi > current_plus_vi:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头信号平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
# 空头持仓 - 检查平仓条件
elif current_direction == -1:
# 条件1: 止损触发
if current_price >= stop_loss_price:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止损平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
# 条件2: 信号反转 (+VI上穿-VI)
elif prev_plus_vi <= prev_minus_vi and current_plus_vi > current_minus_vi:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头信号平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%")
except BacktestFinished as e:
print(f"策略运行异常: {e}")
api.close()