PTrade 日内交易：跳空缺口 (Gap) 策略实现与回测

这是一个关于量化交易策略设计的问题。

捕捉“跳空缺口”（Gap）是技术分析中常见的策略。缺口通常意味着市场情绪的剧烈波动。

策略逻辑分析

我们将设计一个**“向上跳空高开追涨”（Gap and Go）的日内或短线策略。为了在回测中真实模拟“在缺口出现时立即买入”，我们需要使用分钟级别**的回测（因为日线回测通常在收盘时成交，无法捕捉开盘的缺口收益）。

策略规则如下：

1. 运行频率：分钟级别（Minute）。
2. 标的：以贵州茅台（600519.SS）为例。
3. 买入条件：
   • 当前时间为 09:31（开盘第一分钟）。
   • 今日开盘价相比昨日收盘价上涨超过 1%（跳空高开）。
   • 今日开盘涨幅小于 9%（防止涨停无法买入或高位接盘风险）。
4. 卖出条件：
   • 尾盘平仓：在当日 14:55 强制平仓（日内交易，规避隔夜风险）。
   • 止损：如果盘中价格跌破今日开盘价的一定比例（如 -2%），提前止损。

PTrade 策略代码实现

以下是完整的 Python 策略代码。请在 PTrade 回测界面选择 “分钟” 频率进行运行。

def initialize(context):
    """
    初始化函数，设置股票池和全局变量
    """
    # 设置要操作的股票，这里以贵州茅台为例
    g.security = '600519.SS'
    # 设置股票池
    set_universe(g.security)
    
    # 设置手续费（股票默认为万三）
    set_commission(commission_ratio=0.0003, min_commission=5.0, type='STOCK')
    # 设置滑点（模拟真实交易中的价格偏差）
    set_slippage(slippage=0.002)
    
    # 策略参数
    g.jump_threshold = 0.01  # 跳空幅度阈值：1%
    g.max_jump = 0.09        # 最大跳空幅度：9% (避免涨停)
    g.stop_loss_pct = 0.02   # 盘中回撤止损幅度
    
    # 记录今日是否已经交易过的标志位
    g.traded_today = False
    # 记录今日开盘价
    g.today_open_price = 0.0

def before_trading_start(context, data):
    """
    盘前处理，每天开盘前运行一次
    """
    # 重置每日交易状态
    g.traded_today = False
    g.today_open_price = 0.0
    log.info("盘前准备完成，等待开盘...")

def handle_data(context, data):
    """
    盘中运行函数，分钟模式下每分钟运行一次
    """
    # 获取当前回测时间
    current_dt = context.blotter.current_dt
    hour = current_dt.hour
    minute = current_dt.minute
    
    # -----------------------------------------------------------------
    # 1. 开盘检测逻辑 (09:31)
    # -----------------------------------------------------------------
    if hour == 9 and minute == 31:
        # 获取昨日收盘价
        # count=1, frequency='1d', include=False 表示获取不包含今天的前一天数据
        history = get_history(1, '1d', 'close', g.security, fq='pre', include=False)
        
        if len(history) > 0:
            last_close = history['close'].values[-1]
            
            # 获取今日开盘价 (09:31这一分钟的open即视为今日开盘价)
            current_open = data[g.security]['open']
            g.today_open_price = current_open
            
            # 计算跳空幅度
            jump_ratio = (current_open - last_close) / last_close
            
            log.info("昨日收盘: %.2f, 今日开盘: %.2f, 跳空幅度: %.2f%%" % (last_close, current_open, jump_ratio * 100))
            
            # 判断是否满足买入条件：
            # 1. 跳空幅度大于设定阈值
            # 2. 跳空幅度小于最大阈值（防止追高涨停板）
            if g.jump_threshold < jump_ratio < g.max_jump:
                # 全仓买入
                cash = context.portfolio.cash
                if cash > 0:
                    order_value(g.security, cash)
                    g.traded_today = True
                    log.info("触发跳空高开买入信号，买入 %s" % g.security)
    
    # -----------------------------------------------------------------
    # 2. 盘中风控逻辑 (持仓期间)
    # -----------------------------------------------------------------
    # 如果今日已买入，且当前有持仓
    if g.traded_today and get_position(g.security).amount > 0:
        current_price = data[g.security]['close']
        
        # 止损逻辑：如果价格低于开盘价的一定比例
        if current_price < g.today_open_price * (1 - g.stop_loss_pct):
            order_target(g.security, 0)
            log.info("触发盘中风控止损，当前价: %.2f, 开盘价: %.2f" % (current_price, g.today_open_price))
            g.traded_today = False # 标记为已处理，避免重复发单

    # -----------------------------------------------------------------
    # 3. 尾盘平仓逻辑 (14:55)
    # -----------------------------------------------------------------
    if hour == 14 and minute == 55:
        position = get_position(g.security)
        if position.amount > 0:
            log.info("尾盘时间到，执行强制平仓")
            order_target(g.security, 0)

代码关键点解析

1. get_history 的使用：

   • get_history(1, '1d', 'close', ..., include=False)：这是获取昨日收盘价的关键。include=False 确保我们取到的是昨天的数据，而不是今天尚未走完的数据。
   • 注意：PTrade API 中 get_history 必须直接调用，不需要导入任何库。

2. 时间控制：

   • 使用 context.blotter.current_dt 获取当前回测时间。
   • if hour == 9 and minute == 31: 确保策略只在开盘第一分钟判断缺口，这是捕捉缺口最准确的时间点。

3. 数据获取：

   • data[g.security]['open']：在分钟回测模式下，这代表当前这一分钟（09:31）的开盘价，近似等于当天的开盘价。

4. 风险管理：

   • 涨停保护：g.max_jump 设置为 9%，防止在已经涨停或接近涨停时买入，导致无法成交或风险过大。
   • 日内平仓：策略设定为日内交易（T+0逻辑，虽然A股是T+1，但如果是底仓滚动或ETF可实现，此处为演示策略逻辑，假设是T+0或次日卖出逻辑的变体）。如果是纯A股无底仓策略，你需要将卖出逻辑修改为“持有到第二天”。

如何修改为持仓过夜（T+1）？

如果你希望买入后持有到第二天再卖出（符合A股T+1规则），请修改 handle_data 中的卖出逻辑：

1. 删除 14:55 的强制平仓代码。
2. 在 09:31 的逻辑块最开始，添加卖出昨日持仓的代码：

    # 在 09:31 检查是否有昨天的持仓，如果有，先卖出
    if hour == 9 and minute == 31:
        pos = get_position(g.security)
        # 如果有可用持仓（A股enable_amount代表昨仓），则卖出
        if pos.enable_amount > 0:
            order_target(g.security, 0)
            log.info("卖出昨日持仓")
        
        # ... 之后再执行获取历史数据和判断今日缺口的逻辑 ...

Q&A

Q: 为什么不在日线级别（Daily）回测这个策略？
A: PTrade 的日线级别回测，handle_data 是在每天收盘时（通常是 15:00）运行的。如果在日线级别写“如果今天开盘跳空就买入”，实际上代码执行买入动作时已经是收盘价了，这会造成巨大的“未来函数”偏差（你以为买在开盘，实际买在收盘）。捕捉缺口必须使用分钟级别回测。

Q: 如何监控全市场股票的缺口？
A: 在 initialize 中使用 get_index_stocks('000300.SS') 获取沪深300成分股列表，然后在 handle_data 中使用 for 循环遍历这 300 只股票，分别计算它们的跳空幅度。注意要注意 API 的调用频率限制和计算效率。

Q: get_history 函数报错说找不到？
A: 请确保没有在代码顶部写 import ptrade 或类似语句。PTrade 的 API 是内置注入的，直接使用函数名即可。另外，请检查 count 参数是否合理，建议设置为 100 以确保指标计算（虽然本策略只用了 1 天，但习惯上取多一点数据更稳健）。