量价趋势策略（VPT）
一、交易策略解释
核心思想
量价趋势策略（Volume Price Trend，简称VPT）的核心思想是通过成交量和价格变动的协同关系来判断市场趋势的有效性和持续性。该策略 基于这样一个基本原则：真实的、可持续的价格趋势应当由相应的成交量变化来支撑。
具体来说：
- 价格上涨伴随成交量增加，表明买方力量强劲，上涨趋势更可靠
- 价格下跌伴随成交量增加，表明卖方占据优势，下跌趋势更可靠
- 价格变动但成交量不足，则趋势可能不可持续，或即将出现反转
该策略通过VPT指标量化这种关系，指标值随着价格与成交量的协同变化而累积增长或降低，从而反映市场内在力量的变化。
理论基础
VPT策略的理论基础来源于市场技术分析的基本原理和市场微观结构理论：
-
道氏理论：道氏理论强调市场趋势需要成交量的确认，"成交量应该随趋势增长"的原则被广泛应用于技术分析。
-
供需关系：金融市场本质上反映了供需关系，成交量增加时价格上涨表明需求增加；成交量增加时价格下跌则表明供应增加。
-
市场参与度 成交量代表市场参与度，高成交量表明更多投资者认同当前价格走向，使趋势更可靠。
-
资金流向理论：VPT实际上是在追踪资金流向，通过价格和成交量的乘积累计来量化资金流入或流出的强度。
研究表明，价格变动与成交量的协同关系在大部分成熟市场中都存在明显的统计显著性，尤其在商品期货和股票市场中表现尤为明显。
策略适用场景
-
流动性充足的市场：如主要期货合约、大盘股等交易活跃的市场，成交量信号更可靠。
-
明显趋势市场：在有明显上升或下降趋势的市场中，VPT策略能更好地捕捉趋势的有效性。
-
中长期交易：该策略更适合日线级别以上的中长期交易，因为短期内成交量波动可能过于剧烈。
-
转折点识别：在市场可能形成顶部或底部时，VPT的背离信号尤为有价值。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
VPT 指标的基本计算公式：
VPT = 前一期VPT + 当期成交量 × (当期收盘价 - 前期收盘价) / 前期收盘价
或简化为：
VPT = 前一期VPT + 当期成交量 × 价格变化百分比
初始VPT值通常设为首个交易日的成交量。该指标通过累加每个周期内价格百分比变化与成交量的乘积，形成一条反映量价协同关系的曲线。
在本策略中，我们进一步计算了VPT的移动平均线（VPT_MA）作为参考基准，用于生成交易信号：
VPT_MA = VPT的N周期简单移动平均线
我们还引入了成交量放大的概念，通过比较当前成交量与过去N个周期平均成交量的比值，判断成交量是否显著增加：
1 成交量放大倍数 = 当前成交量 / N周期平均成交量
交易逻辑
开仓信号：
-
做多信号： 同时满足以下条件
- VPT值大于其移动平均线（VPT > VPT_MA）
- 当前成交量大于前N周期平均成交量的1.5倍
- 当前收盘价高于前一周期收盘价
-
做空信号： 同时满足以下条件
- VPT值小于其移动平均线（VPT < VPT_MA）
- 当前成交量大于前N周期平均成交量的1.5倍
- 当前收盘价低于前一周期收盘价
平仓信号：
- 平多信号： VPT下穿其移动平均线且成交量放大
- 平空信号： VPT上穿其移动平均线且成交量放大
回测
回测初始设置
- 测试周期： 2023 年 1 月 15 日至 2023 年 5 月 15 日
- 交易品种： SHFE.au2306
- 周期选择： 日线级别
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from datetime import date
import numpy as np
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
# ===== 全局参数设置 =====
SYMBOL = "SHFE.au2306" # 交易合约
POSITION_SIZE = 30 # 持仓手数
START_DATE = date(2023, 1, 15) # 回测开始日期
END_DATE = date(2023, 5, 15) # 回测结束日期
# VPT策略参数
VPT_MA_PERIOD = 14 # VPT均线周期
VOLUME_THRESHOLD = 1.5 # 成交量放大倍数阈值
print(f"开始回测 {SYMBOL} 的量价趋势(VPT)策略...")
print(f"参数: VPT均线周期={VPT_MA_PERIOD}, 成交量阈值={VOLUME_THRESHOLD}")
api = None
try:
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)
# 初始化交易状态
position = 0 # 当前持仓
entry_price = 0 # 入场价格
vpt_values = [] # 存储VPT值
while True:
api.wait_update()
if api.is_changing(klines):
# 确保有足够的数据
if len(klines) < VPT_MA_PERIOD + 1:
continue
# 计算VPT指标
close = klines.close.values
volume = klines.volume.values
# 计算最新的VPT值
if len(vpt_values) == 0:
vpt_values.append(volume[-1])
else:
price_change_pct = (close[-1] - close[-2]) / close[-2]
new_vpt = vpt_values[-1] + volume[-1] * price_change_pct
vpt_values.append(new_vpt)
# 保持VPT列表长度与K线数据同步
if len(vpt_values) > len(klines):
vpt_values.pop(0)
# 计算VPT均线
if len(vpt_values) >= VPT_MA_PERIOD:
vpt_ma = np.mean(vpt_values[-VPT_MA_PERIOD:])
# 获取当前价格和成交量数据
current_price = float(close[-1])
current_volume = float(volume[-1])
avg_volume = np.mean(volume[-VPT_MA_PERIOD:-1])
# 判断成交量是否放大
volume_increased = current_volume > avg_volume * VOLUME_THRESHOLD
# 交易信号判断
vpt_trend_up = vpt_values[-1] > vpt_ma
vpt_trend_down = vpt_values[-1] < vpt_ma
# 交易逻辑
if position == 0: # 空仓
if vpt_trend_up and volume_increased and close[-1] > close[-2]:
position = POSITION_SIZE
entry_price = current_price
target_pos.set_target_volume(position)
print(f"开多仓: 价格={current_price:.2f}, VPT上穿均线")
elif vpt_trend_down and volume_increased and close[-1] < close[-2]:
position = -POSITION_SIZE
entry_price = current_price
target_pos.set_target_volume(position)
print(f"开空仓: 价格={current_price:.2f}, VPT下穿均线")
elif position > 0: # 持有多头
if vpt_trend_down and volume_increased:
profit_pct = (current_price / entry_price - 1) * 100
target_pos.set_target_volume(0)
print(f"平多仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
position = 0
entry_price = 0
elif position < 0: # 持有空头
if vpt_trend_up and volume_increased:
profit_pct = (entry_price / current_price - 1) * 100
target_pos.set_target_volume(0)
print(f"平空仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
position = 0
entry_price = 0
except BacktestFinished as e:
api.close()
print("回测结束")