如何在 PTrade 平台用 Python 实现经典的 Dual Thrust 日内交易策略？

这是一个基于 Dual Thrust 策略的 PTrade 量化交易策略实现。

策略原理

Dual Thrust 是一种经典的区间突破策略，由 Michael Chalek 开发。它的核心逻辑是根据前 N 天的历史数据计算一个波动区间（Range），并在当天的开盘价基础上，加上或减去这个区间的特定比例，形成上轨（Buy Line）和下轨（Sell Line）。

1. 计算区间 (Range)：

   • $HH$ = 过去 $N$ 日的最高价的最高值
   • $HC$ = 过去 $N$ 日的收盘价的最高值
   • $LC$ = 过去 $N$ 日的收盘价的最低值
   • $LL$ = 过去 $N$ 日的最低价的最低值
   • $Range = \max(HH - LC, HC - LL)$

2. 计算轨道：

   • 上轨 (Buy Line) = 当日开盘价 + $K_1 \times Range$
   • 下轨 (Sell Line) = 当日开盘价 - $K_2 \times Range$

3. 交易信号：

   • 当价格突破上轨时，做多（买入）。
   • 当价格跌破下轨时，平仓或做空（股票策略通常只平仓）。

PTrade 策略代码

以下代码适用于 PTrade 回测及实盘环境（股票单向做多逻辑）。

# Dual Thrust 策略实现
# 适用于 PTrade 平台

def initialize(context):
    """
    初始化函数，设置策略参数和股票池
    """
    # 1. 设置要操作的标的，这里以恒生电子为例，也可以换成ETF如 '510300.SS'
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 2. 策略参数设置
    g.N = 5       # 回溯天数，用于计算 Range
    g.k1 = 0.2    # 上轨系数 (Buy trigger)
    g.k2 = 0.2    # 下轨系数 (Sell trigger)
    
    # 全局变量初始化
    g.range = 0.0
    g.buy_line = 0.0
    g.sell_line = 0.0

def before_trading_start(context, data):
    """
    盘前处理函数，每日开盘前运行一次
    用于计算前 N 天的 Range
    """
    # 获取过去 N 天的历史数据（不包含今天），前复权
    # 字段需要：最高价(high), 最低价(low), 收盘价(close)
    hist_data = get_history(g.N, '1d', ['high', 'low', 'close'], g.security, fq='pre', include=False)
    
    # 如果数据不足 N 天，则不计算，直接返回
    if len(hist_data) < g.N:
        g.range = 0.0
        log.info("历史数据不足，跳过今日计算")
        return

    # 计算 HH, HC, LC, LL
    hh = hist_data['high'].max()
    hc = hist_data['close'].max()
    lc = hist_data['close'].min()
    ll = hist_data['low'].min()
    
    # 计算 Range = max(HH - LC, HC - LL)
    r1 = hh - lc
    r2 = hc - ll
    g.range = max(r1, r2)
    
    log.info("今日 Range 计算完成: %s" % g.range)

def handle_data(context, data):
    """
    盘中处理函数，分钟回测模式下每分钟运行一次
    """
    # 如果 Range 未计算成功（例如上市不久），则不交易
    if g.range == 0.0:
        return

    security = g.security
    
    # 获取当天的开盘价
    # 注意：data[security]['open'] 在分钟级别回测中，取的是当天的日开盘价
    current_open = data[security]['open']
    
    # 获取当前最新价格
    current_price = data[security]['price']
    
    # 如果开盘价无效（如停牌），则跳过
    if current_open is None or current_open == 0:
        return

    # 计算上下轨
    # 上轨 = 开盘价 + K1 * Range
    g.buy_line = current_open + g.k1 * g.range
    # 下轨 = 开盘价 - K2 * Range
    g.sell_line = current_open - g.k2 * g.range
    
    # 获取当前持仓和资金
    position = get_position(security).amount
    cash = context.portfolio.cash
    
    # --- 交易逻辑 ---
    
    # 1. 买入逻辑：价格突破上轨 且 当前无持仓
    if current_price > g.buy_line and position == 0:
        # 全仓买入
        order_value(security, cash)
        log.info("突破上轨，买入。当前价: %s, 上轨: %s" % (current_price, g.buy_line))
        
    # 2. 卖出逻辑：价格跌破下轨 且 当前有持仓
    elif current_price < g.sell_line and position > 0:
        # 清仓卖出
        order_target(security, 0)
        log.info("跌破下轨，卖出。当前价: %s, 下轨: %s" % (current_price, g.sell_line))

代码要点解析

1. 参数设置 (initialize):

   • g.N: 决定了策略对历史波动的敏感度。N 越小，Range 变化越快。
   • g.k1 和 g.k2: 决定了突破的难易程度。K 值越小，越容易触发交易（但也可能带来更多假突破）。通常 K1 和 K2 可以设为不同值，例如在牛市中可以减小 K1 以更容易买入。

2. 数据获取 (before_trading_start):

   • 使用 get_history 获取过去 N 天的日线数据。
   • include=False 确保只获取昨天及之前的数据，不包含今天，避免未来函数。
   • 计算逻辑严格遵循 Dual Thrust 公式。

3. 交易执行 (handle_data):

   • 策略依赖于当天的 open (开盘价) 来确定当天的轨道。
   • 在分钟级别回测中，data[security]['open'] 获取的是当天的开盘价，这是固定不变的。
   • order_value 用于按金额下单，order_target 用于调整持仓到目标数量（这里设为0即清仓）。

常见问题 (Q&A)

Q1: 这个策略适合什么类型的标的？
A: Dual Thrust 属于趋势跟踪策略，适合波动率较大、趋势性较强的标的，如股指 ETF（沪深300、中证500）或某些活跃的科技股。在震荡市中可能会频繁止损。

Q2: 如何在期货市场使用此策略？
A: 期货市场支持做空。你需要修改 handle_data 中的逻辑：

• 价格 < 下轨时：如果有多单则平多，并开空单（sell_open）。
• 价格 > 上轨时：如果有空单则平空，并开多单（buy_open）。

Q3: 为什么使用 get_history 而不是 attribute_history？
A: PTrade API 标准文档中使用 get_history 来获取历史行情数据。它返回的是 DataFrame 格式（在 Python 3 环境下），方便进行 max() 和 min() 的计算。

Q4: 回测频率应该选什么？
A: 建议选择 分钟级别 回测。因为 Dual Thrust 是日内判断价格是否突破轨道的策略，日线级别回测无法捕捉盘中的突破动作。