JoinQuant 平台如何实现 DPO 区间震荡线交易策略？

这是一个基于 DPO (Detrended Price Oscillator，去趋势价格震荡指标) 的交易策略。

策略原理

DPO 指标通过消除价格的长周期趋势，来识别短周期的超买/超卖状态和周期性转折点。

1. DPO 计算公式：
   $$DPO = \text{Close} - \text{SMA}(\text{Close}, N)[\text{shifted by } (N/2 + 1)]$$
   即：当前收盘价 减去 $(N/2 + 1)$ 天前的 $N$ 日简单移动平均线。
2. 交易逻辑：
   • 买入信号：DPO 上穿 0 轴（金叉）。意味着短期价格表现强于滞后的移动平均线，可能开启新的短周期上涨。
   • 卖出信号：DPO 下穿 0 轴（死叉）。意味着短期价格表现弱于滞后的移动平均线，可能开启新的短周期下跌。
3. 资金管理：全仓买入/卖出（仅作演示，实盘建议配合风控）。

策略代码

# -*- coding: utf-8 -*-
import jqdata
import numpy as np

def initialize(context):
    """
    初始化函数，设定基准、手续费、全局变量等
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # --- 策略参数设置 ---
    # 要操作的标的：平安银行
    g.security = '000001.XSHE'
    
    # DPO 周期参数 N
    g.N = 20
    
    # 移动平均线的位移量 (Displacement)
    # 标准 DPO 公式通常使用 (N / 2 + 1) 作为位移
    g.displacement = int(g.N / 2) + 1
    
    # 每天开盘时运行
    run_daily(market_open, time='every_bar')

def get_dpo(security, n, displacement):
    """
    计算 DPO 指标
    输入: 标的, 周期N, 位移量
    输出: 当前DPO值, 上一期DPO值 (用于判断交叉)
    """
    # 获取历史数据
    # 我们需要足够的数据来计算位移后的 SMA
    # 需要获取的长度 = 周期 N + 位移量 displacement + 额外缓冲(用于计算上一期DPO)
    count = n + displacement + 5
    
    # 获取收盘价数据
    h = attribute_history(security, count, '1d', ['close'], df=False)
    close_prices = h['close']
    
    if len(close_prices) < count:
        return None, None

    # --- 计算当前的 DPO ---
    # 当前收盘价
    current_close = close_prices[-1]
    
    # 计算位移后的 SMA (Simple Moving Average)
    # 我们需要的是 (displacement) 天前的 N 日均线
    # 数据切片范围：从倒数 (N + displacement) 到 倒数 (displacement)
    sma_slice_curr = close_prices[-(n + displacement) : -displacement]
    sma_ref_curr = sma_slice_curr.mean()
    
    dpo_curr = current_close - sma_ref_curr
    
    # --- 计算上一期的 DPO (用于判断金叉死叉) ---
    # 上一期收盘价
    prev_close = close_prices[-2]
    
    # 上一期的位移 SMA
    # 数据切片整体向左移动 1 位
    sma_slice_prev = close_prices[-(n + displacement + 1) : -(displacement + 1)]
    sma_ref_prev = sma_slice_prev.mean()
    
    dpo_prev = prev_close - sma_ref_prev
    
    return dpo_curr, dpo_prev

def market_open(context):
    """
    每日交易逻辑
    """
    security = g.security
    
    # 获取 DPO 数据
    dpo_curr, dpo_prev = get_dpo(security, g.N, g.displacement)
    
    if dpo_curr is None:
        return

    # 获取当前仓位
    curr_position = context.portfolio.positions[security].closeable_amount
    # 获取可用资金
    cash = context.portfolio.available_cash
    
    # --- 交易信号判断 ---
    
    # 1. 买入信号：DPO 上穿 0 轴 (金叉)
    # 上一期 < 0 且 当前 >= 0
    if dpo_prev < 0 and dpo_curr >= 0:
        if cash > 0:
            # 全仓买入
            order_value(security, cash)
            log.info("DPO金叉 (Prev: %.2f, Curr: %.2f) - 买入 %s" % (dpo_prev, dpo_curr, security))
            
    # 2. 卖出信号：DPO 下穿 0 轴 (死叉)
    # 上一期 > 0 且 当前 <= 0
    elif dpo_prev > 0 and dpo_curr <= 0:
        if curr_position > 0:
            # 清仓卖出
            order_target(security, 0)
            log.info("DPO死叉 (Prev: %.2f, Curr: %.2f) - 卖出 %s" % (dpo_prev, dpo_curr, security))

    # 记录 DPO 值以便在回测图中查看
    record(DPO=dpo_curr, Zero=0)

常见问题 (Q&A)

Q1: 为什么 DPO 指标要使用位移的移动平均线？
A: 传统的移动平均线（MA）通常具有滞后性。DPO 试图通过将移动平均线向后移动（即比较当前价格与 $N/2+1$ 天前的均线值），来消除这种滞后带来的长期趋势影响，从而更清晰地展示价格的短周期波动。

Q2: 这个策略适合什么类型的市场？
A: DPO 是一个震荡指标。

• 适合： 震荡市或具有明显波段周期的市场。
• 不适合： 极端的单边趋势行情。在强劲的单边牛市中，DPO 可能会过早发出卖出信号（因为价格虽然还在涨，但涨速减慢可能导致 DPO 下穿 0 轴）；在单边熊市中可能发出错误的抄底信号。

Q3: 如何优化这个策略？
A:

1. 增加趋势过滤器：例如，只有在长期均线（如 MA60）向上时，才执行 DPO 的买入信号。
2. 调整阈值：不单纯使用 0 轴，而是设置超买线（如 +2%）和超卖线（如 -2%）作为反转信号。
3. 止损机制：代码中未包含止损，实盘建议加入固定比例止损或 ATR 止损。

Q4: 代码中的 attribute_history 为什么要取那么多天数据？
A: 计算 DPO 需要用到“位移”后的均线。假设 $N=20$，位移量约为 11 天。要计算今天的 DPO，我们需要比较今天的价格和 11 天前那个时间点结束的 20 日均线。因此，我们需要的数据长度至少是 $20 + 11 = 31$ 天，为了计算上一期 DPO 进行交叉判断，还需要额外的数据，所以代码中取了更多的缓冲量。