DPO 指标计算与 QMT 策略实现

DPO（Detrended Price Oscillator，去趋势价格摆动指标）是一种试图通过消除价格趋势来识别短期周期的指标。它通过将当前价格与滞后的移动平均线进行比较，从而剔除长期的价格趋势，帮助交易者识别超买超卖水平以及周期性的转折点。

以下是 DPO 的计算原理、策略逻辑以及在 QMT 平台上的完整代码实现。

1. DPO 指标计算原理

DPO 的核心思想是将移动平均线向后平移（滞后），然后用收盘价减去这个滞后的移动平均线。

公式：
$$DPO = Close - SMA(N)[N/2 + 1 \text{ periods ago}]$$

步骤：

1. 确定周期 $N$：通常设置为 20。
2. 计算简单移动平均线 (SMA)：计算过去 $N$ 天的收盘价平均值。
3. 确定位移量：位移量通常为 $\frac{N}{2} + 1$。例如 $N=20$，位移量为 $11$。
4. 计算 DPO：当前收盘价减去（位移量）天前的 SMA 数值。

2. 策略逻辑

本示例策略采用经典的 0 轴穿越（零轴交叉） 策略：

• 买入信号（金叉）：当 DPO 指标由下向上穿越 0 轴时，表明短期价格相对于滞后均线开始走强，视为买入信号。
• 卖出信号（死叉）：当 DPO 指标由上向下穿越 0 轴时，表明短期价格相对于滞后均线开始走弱，视为卖出信号。

3. QMT 策略代码实现

以下代码包含完整的策略结构，可以直接在 QMT 的 Python 策略编辑器中运行。

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

def init(ContextInfo):
    """
    策略初始化函数
    """
    # 设置策略参数
    ContextInfo.N = 20  # DPO的计算周期
    ContextInfo.displacement = int(ContextInfo.N / 2 + 1) # 位移量 (20/2 + 1 = 11)
    
    # 设置股票池（示例：平安银行）
    ContextInfo.stock_code = '000001.SZ'
    ContextInfo.set_universe([ContextInfo.stock_code])
    
    # 设置资金账号（请替换为您自己的资金账号）
    ContextInfo.account_id = 'YOUR_ACCOUNT_ID' 
    ContextInfo.set_account(ContextInfo.account_id)
    
    # 设置回测参数（仅回测有效）
    ContextInfo.set_commission(0.0003) # 手续费
    ContextInfo.set_slippage(1, 0.002) # 滑点

def get_dpo(close_series, n, displacement):
    """
    计算 DPO 指标
    :param close_series: 收盘价序列 (pandas Series)
    :param n: 周期
    :param displacement: 位移量
    :return: DPO 序列
    """
    # 1. 计算 N 周期简单移动平均线 (SMA)
    sma = close_series.rolling(window=n).mean()
    
    # 2. 将 SMA 向后平移 (shift 正数表示获取过去的数据到当前行)
    # DPO = Close - SMA(shifted)
    # 注意：Pandas的shift(k)是将数据向下移动k行，意味着当前索引i处获取的是i-k处的数据
    shifted_sma = sma.shift(displacement)
    
    # 3. 计算 DPO
    dpo = close_series - shifted_sma
    
    return dpo

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 获取当前正在处理的股票代码
    stock_code = ContextInfo.stock_code
    
    # 确保有足够的数据进行计算 (N + displacement + 安全余量)
    required_len = ContextInfo.N + ContextInfo.displacement + 10
    
    # 获取历史行情数据
    # 注意：get_market_data_ex 返回的是字典 {code: dataframe}
    data_dict = ContextInfo.get_market_data_ex(
        ['close'], 
        [stock_code], 
        period='1d', 
        count=required_len, 
        dividend_type='front' # 前复权
    )
    
    if stock_code not in data_dict:
        return
        
    df = data_dict[stock_code]
    
    # 如果数据长度不足，直接返回
    if len(df) < required_len:
        return

    # 计算 DPO
    dpo_series = get_dpo(df['close'], ContextInfo.N, ContextInfo.displacement)
    
    # 获取最近两个周期的 DPO 值用于判断交叉
    # iloc[-1] 是当前K线（如果是盘中则是最新tick合成的），iloc[-2] 是上一根K线
    current_dpo = dpo_series.iloc[-1]
    prev_dpo = dpo_series.iloc[-2]
    
    # 检查 DPO 是否为 NaN (数据初期计算不出 DPO)
    if np.isnan(current_dpo) or np.isnan(prev_dpo):
        return

    # 获取当前持仓
    positions = ContextInfo.get_trade_detail_data(ContextInfo.account_id, 'stock', 'position')
    current_holding = 0
    for pos in positions:
        if pos.m_strInstrumentID + '.' + pos.m_strExchangeID == stock_code:
            current_holding = pos.m_nVolume
            break

    # --- 交易逻辑 ---
    
    # 1. 买入信号：DPO 上穿 0 轴 (前值 < 0, 现值 > 0)
    if prev_dpo < 0 and current_dpo > 0:
        if current_holding == 0:
            # 全仓买入 (示例逻辑，实际请做好资金管理)
            # 获取可用资金
            account_info = ContextInfo.get_trade_detail_data(ContextInfo.account_id, 'stock', 'account')
            available_cash = account_info[0].m_dAvailable if account_info else 0
            
            if available_cash > 0:
                # 计算可买股数 (向下取整到100股)
                current_price = df['close'].iloc[-1]
                buy_vol = int(available_cash / current_price / 100) * 100
                
                if buy_vol > 0:
                    print(f"时间: {df.index[-1]}, 信号: DPO金叉 (上穿0轴), 动作: 买入 {stock_code}")
                    passorder(23, 1101, ContextInfo.account_id, stock_code, 5, -1, buy_vol, ContextInfo)

    # 2. 卖出信号：DPO 下穿 0 轴 (前值 > 0, 现值 < 0)
    elif prev_dpo > 0 and current_dpo < 0:
        if current_holding > 0:
            print(f"时间: {df.index[-1]}, 信号: DPO死叉 (下穿0轴), 动作: 卖出 {stock_code}")
            passorder(24, 1101, ContextInfo.account_id, stock_code, 5, -1, current_holding, ContextInfo)

    # --- (可选) 绘图用于回测观察 ---
    # 注意：paint 只能画数值，不能画在主图坐标系上，因为 DPO 是摆动指标
    ContextInfo.paint('DPO', current_dpo, -1, 0)
    ContextInfo.paint('Zero', 0, -1, 0)

4. 代码关键点解析

1. 数据获取 (get_market_data_ex)：

   • 我们使用 get_market_data_ex 获取历史收盘价。
   • count 参数必须足够大。因为 DPO 需要计算 $N$ 周期均线，还需要向后位移 $N/2+1$ 周期，所以获取的数据长度必须大于 $N + N/2 + 1$。

2. Pandas 的 shift 用法：

   • 公式要求减去“过去的”均线值。
   • 在 Pandas 中，sma.shift(11) 表示将 SMA 数据列整体向下移动 11 格。这意味着，当前索引（今天）对应的 shifted_sma 值，实际上是 11 天前计算出来的 SMA 值。这完全符合 DPO 的定义。

3. 交易信号判断：

   • 使用 prev_dpo < 0 和 current_dpo > 0 判断上穿（金叉）。
   • 使用 prev_dpo > 0 和 current_dpo < 0 判断下穿（死叉）。

4. 实盘注意事项：

   • 请务必将 ContextInfo.account_id 替换为您真实的资金账号。
   • passorder 函数中的 opType (23买, 24卖) 和 orderType (1101) 是 QMT 标准下单参数，代表股票买入/卖出和单股单账号普通交易。

Q&A

Q: 为什么 DPO 指标在最新的几根 K 线上可能会有漂移或不准确？
A: DPO 的标准定义是 $Close(t) - SMA(t - (N/2+1))$。这意味着要计算今天的 DPO，我们需要 $N/2+1$ 天前的移动平均线。这是确定的，不会漂移。
但是，有些变种算法试图用未来的数据来居中移动平均线，这会导致“未来函数”。上述代码使用的是标准的、无未来函数的计算方式（引用过去的数据），因此在实盘中是安全的。

Q: 如何优化 DPO 策略？
A: 单纯的 0 轴交叉可能会在震荡市中产生很多假信号。可以考虑：

1. 增加过滤条件：例如结合趋势指标（如 EMA200），只有在长期趋势向上时才执行 DPO 金叉买入。
2. 设置阈值：不以 0 为界限，而是设定超买线（如 +2）和超卖线（如 -2），在极端位置反转时交易。
3. 结合 K 线形态：DPO 发出信号时，确认 K 线是否有吞没、锤子线等形态。

Q: ContextInfo.paint 画出的图在哪里看？
A: 在 QMT 的回测界面或模型运行界面的副图（指标窗口）中可以看到绘制的 DPO 曲线和 0 轴。