QMT 策略实战：价量趋势 (PVT) 指标交易模型 Python 源码

这是一个基于 PVT (Price Volume Trend，价量趋势指标) 的量化策略。

策略原理

PVT 指标类似于 OBV（能量潮），但它考虑了价格变动的幅度，而不仅仅是方向。

• 公式：$PVT = \text{昨日PVT} + \text{今日成交量} \times \frac{\text{今日收盘价} - \text{昨日收盘价}}{\text{昨日收盘价}}$
• 交易逻辑：
  1. 计算 PVT 线。
  2. 计算 PVT 的 N 日移动平均线 (MA_PVT)。
  3. 买入信号：当 PVT 向上突破 MA_PVT 时，视为多头信号。
  4. 卖出信号：当 PVT 向下跌破 MA_PVT 时，视为空头信号。

策略代码

# -*- coding: gbk -*-
import pandas as pd
import numpy as np

def init(ContextInfo):
    """
    初始化函数
    """
    # 设置账号 (请替换为您的实际资金账号)
    # ContextInfo.set_account('您的资金账号')
    
    # 设置股票池，这里示例使用 沪深300 成分股
    # 实盘或回测时，请确保在界面或此处设置了股票池
    ContextInfo.trade_code_list = ContextInfo.get_stock_list_in_sector('沪深300')
    
    # 策略参数设置
    ContextInfo.pvt_ma_window = 20  # PVT均线周期
    ContextInfo.lookback_days = 100 # 获取历史数据的长度，需大于均线周期
    
    print("PVT策略初始化完成")

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 获取当前K线位置
    index = ContextInfo.barpos
    
    # 获取当前时间
    realtime = ContextInfo.get_bar_timetag(index)
    
    # 仅在最后一根K线或回测模式下运行，避免历史K线重复计算浪费资源（视需求而定）
    # 如果是回测模式，每一根K线都会运行；如果是实盘，只在最新K线运行
    
    # 获取股票池
    stock_list = ContextInfo.trade_code_list
    if not stock_list:
        print("股票池为空，请检查设置")
        return

    # 获取历史行情数据 (使用 get_market_data_ex 接口)
    # 获取字段：收盘价(close) 和 成交量(volume)
    # count 设置为 lookback_days 以确保有足够数据计算 PVT 和 MA
    market_data = ContextInfo.get_market_data_ex(
        ['close', 'volume'], 
        stock_list, 
        period=ContextInfo.period, 
        count=ContextInfo.lookback_days,
        dividend_type='front' # 使用前复权
    )

    for stock in stock_list:
        # 检查该股票是否有数据
        if stock not in market_data:
            continue
            
        df = market_data[stock]
        
        # 数据长度不足以计算指标时跳过
        if len(df) < ContextInfo.pvt_ma_window + 2:
            continue
            
        # --- 计算 PVT 指标 ---
        # 1. 计算价格涨跌幅: (今日收盘 - 昨日收盘) / 昨日收盘
        pct_change = df['close'].pct_change()
        
        # 2. 计算每日 PVT 变动值: 涨跌幅 * 成交量
        pvt_change = pct_change * df['volume']
        
        # 3. 计算累积 PVT (处理 NaN 值，通常第一天为 NaN)
        pvt_line = pvt_change.fillna(0).cumsum()
        
        # 4. 计算 PVT 的移动平均线
        pvt_ma = pvt_line.rolling(window=ContextInfo.pvt_ma_window).mean()
        
        # --- 生成交易信号 ---
        # 获取最新的 PVT 和 MA 值 (-1 为最新，-2 为上一周期)
        curr_pvt = pvt_line.iloc[-1]
        prev_pvt = pvt_line.iloc[-2]
        
        curr_ma = pvt_ma.iloc[-1]
        prev_ma = pvt_ma.iloc[-2]
        
        # 检查是否计算出有效值 (避免 NaN)
        if np.isnan(curr_ma) or np.isnan(prev_ma):
            continue

        # 获取当前持仓
        position = ContextInfo.get_position(stock)
        hold_vol = 0
        if position:
            hold_vol = position.volume

        # 交易逻辑：金叉买入，死叉卖出
        
        # 金叉：上个周期 PVT < MA，当前周期 PVT > MA
        if prev_pvt < prev_ma and curr_pvt > curr_ma:
            if hold_vol == 0:
                # 满仓买入该股 (示例逻辑，实际请做好资金管理)
                # 这里使用 order_target_percent 调整仓位到 10% (假设持仓10只股)
                # 或者使用 order_value 指定金额
                print(f"{stock} PVT金叉，触发买入信号")
                order_target_percent(stock, 0.1, ContextInfo, ContextInfo.accid)

        # 死叉：上个周期 PVT > MA，当前周期 PVT < MA
        elif prev_pvt > prev_ma and curr_pvt < curr_ma:
            if hold_vol > 0:
                # 清仓该股
                print(f"{stock} PVT死叉，触发卖出信号")
                order_target_percent(stock, 0.0, ContextInfo, ContextInfo.accid)

# 辅助函数：获取持仓对象
def get_position(self, stock_code):
    # 这是一个绑定到 ContextInfo 的辅助方法，方便获取持仓
    positions = get_trade_detail_data(self.accid, 'stock', 'position')
    for pos in positions:
        if pos.m_strInstrumentID + '.' + pos.m_strExchangeID == stock_code:
            return pos
    return None

# 将辅助函数绑定到 ContextInfo，方便调用
ContextInfo.get_position = get_position.__get__(ContextInfo)

代码关键点解析

1. 数据获取 (get_market_data_ex):

   • 我们使用了 get_market_data_ex 接口，这是 QMT 推荐的高效数据接口。
   • 获取了 close (收盘价) 和 volume (成交量)。
   • count 设置为 100，是为了确保有足够的数据来计算累积的 PVT 值以及随后的 20 日均线。

2. PVT 计算逻辑:

   • 利用 Pandas 的 pct_change() 快速计算收益率。
   • 利用 cumsum() 计算累积值。
   • 注意：在量化策略中，PVT 的绝对数值取决于计算的起始点（即获取数据的第1天），但这不影响策略逻辑，因为我们比较的是 PVT 线与其自身的移动平均线的相对位置（交叉关系）。

3. 交易信号:

   • 金叉：prev_pvt < prev_ma 且 curr_pvt > curr_ma。
   • 死叉：prev_pvt > prev_ma 且 curr_pvt < curr_ma。

4. 下单接口:

   • 使用了 order_target_percent。这是一个非常方便的函数，它会自动计算需要买入或卖出的数量，将该股票的持仓调整到账户总资产的指定比例（代码中示例为 10%）。
   • 卖出时设置为 0.0 即代表清仓。

使用说明

1. 新建策略：在 QMT 策略编辑器中新建一个 Python 策略。
2. 粘贴代码：将上述代码完整复制进去。
3. 设置账号：在 init 函数中，虽然代码里注释掉了 set_account，但在实盘运行时，请确保在界面右侧参数栏或代码中正确设置了资金账号。
4. 补充数据：运行回测前，请务必在“数据管理”中下载所需品种的日线数据，否则 get_market_data_ex 可能返回空值。
5. 运行模式：此策略适用于日线级别（period='1d'）。

Q&A

Q: 为什么 PVT 指标比单纯的 OBV 指标更有效？
A: OBV 只是简单地将上涨日的成交量全部加，下跌日全部减。而 PVT 考虑了涨跌幅的权重。例如，股价微涨 0.1% 但成交量巨大，OBV 会大幅增加，可能产生误导；而 PVT 因为乘以了极小的涨跌幅，增加的数值会很小，更真实地反映了资金对价格推动的有效性。

Q: 如何调整策略的灵敏度？
A: 修改 init 函数中的 ContextInfo.pvt_ma_window 参数。

• 调小该值（如 10）：策略更灵敏，交易更频繁，但假信号可能增多。
• 调大该值（如 60）：策略更稳健，主要捕捉大趋势，但可能滞后。

Q: get_market_data_ex 返回的数据结构是什么？
A: 当请求多个股票时，它返回一个字典 {stock_code: dataframe}。每个 DataFrame 的索引是时间字符串，列是请求的字段（如 close, volume）。