量价趋势策略（VPT）

一、交易策略解释

核心思想

量价趋势策略（Volume Price Trend，简称VPT）的核心思想是通过成交量和价格变动的协同关系来判断市场趋势的有效性和持续性。该策略基于这样一个基本原则：真实的、可持续的价格趋势应当由相应的成交量变化来支撑。

具体来说：

价格上涨伴随成交量增加，表明买方力量强劲，上涨趋势更可靠
价格下跌伴随成交量增加，表明卖方占据优势，下跌趋势更可靠
价格变动但成交量不足，则趋势可能不可持续，或即将出现反转

该策略通过VPT指标量化这种关系，指标值随着价格与成交量的协同变化而累积增长或降低，从而反映市场内在力量的变化。

理论基础

VPT策略的理论基础来源于市场技术分析的基本原理和市场微观结构理论：

道氏理论：道氏理论强调市场趋势需要成交量的确认，"成交量应该随趋势增长"的原则被广泛应用于技术分析。
供需关系：金融市场本质上反映了供需关系，成交量增加时价格上涨表明需求增加；成交量增加时价格下跌则表明供应增加。
市场参与度成交量代表市场参与度，高成交量表明更多投资者认同当前价格走向，使趋势更可靠。
资金流向理论：VPT实际上是在追踪资金流向，通过价格和成交量的乘积累计来量化资金流入或流出的强度。

研究表明，价格变动与成交量的协同关系在大部分成熟市场中都存在明显的统计显著性，尤其在商品期货和股票市场中表现尤为明显。

策略适用场景

流动性充足的市场：如主要期货合约、大盘股等交易活跃的市场，成交量信号更可靠。
明显趋势市场：在有明显上升或下降趋势的市场中，VPT策略能更好地捕捉趋势的有效性。
中长期交易：该策略更适合日线级别以上的中长期交易，因为短期内成交量波动可能过于剧烈。

转折点识别：在市场可能形成顶部或底部时，VPT的背离信号尤为有价值。

策略原理

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
周期选择：日线级别

回测结果

量价趋势策略（VPT） - 图片 1

上述回测累计收益走势图

量价趋势策略（VPT） - 图片 2

代码

#!/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("回测结束")

量价趋势策略（VPT）

投资思想 · 其他文章 (273)