PTrade 日内交易策略：如何基于分钟成交量 (volume) 寻找交易机会？

这是一个基于PTrade平台的日内分钟级别交易策略。

策略逻辑说明

该策略的核心思想是**“量价配合”**。单纯的成交量放大可能意味着抛压（下跌放量）或买盘涌入（上涨放量）。因此，本策略结合了价格走势：

1. 计算基准成交量：获取过去 N 分钟（例如20分钟）的平均成交量。
2. 判断成交量突变：如果当前分钟的成交量超过基准成交量的 M 倍（例如2倍），视为成交量放大。
3. 价格过滤：仅在成交量放大且价格上涨（收盘价 > 开盘价）时买入，防止接飞刀。
4. 日内风控：
   • 设置止盈止损比例。
   • 尾盘清仓：在收盘前（如14:55）强制卖出，以规避隔夜风险（注意：A股为T+1制度，当日买入的股票当日无法卖出，除非您有底仓进行T+0操作。本代码示例默认包含尾盘清仓逻辑，适用于有底仓或回测T+0环境）。

策略代码

import numpy as np

def initialize(context):
    """
    初始化函数，设置策略参数和股票池
    """
    # 1. 设置要操作的股票，这里以恒生电子为例
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 2. 策略参数设置
    g.vol_window = 20       # 计算平均成交量的窗口（过去20分钟）
    g.vol_multiplier = 2.0  # 成交量放大的倍数阈值（当前量 > 平均量 * 2.0）
    g.trade_value = 50000   # 每次交易的金额
    
    # 3. 止盈止损设置
    g.stop_loss_pct = 0.02  # 止损 2%
    g.take_profit_pct = 0.05 # 止盈 5%
    
    # 4. 记录买入成本，用于止盈止损
    g.cost_price = 0
    
    # 5. 设置手续费（可选，回测时建议设置）
    set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")

def handle_data(context, data):
    """
    分钟级别的主逻辑函数，每分钟运行一次
    """
    # 获取当前时间
    current_dt = context.blotter.current_dt
    current_time = current_dt.strftime('%H:%M')
    
    security = g.security
    
    # -----------------------------------------------------------------
    # 1. 尾盘清仓逻辑 (14:50之后不再买入，14:55强制卖出)
    # 注意：A股T+1，当日买入无法当日卖出。此逻辑在回测T+0或有底仓时有效。
    # -----------------------------------------------------------------
    if current_time >= '14:50':
        if current_time >= '14:55':
            position = get_position(security)
            # 如果有可卖持仓，全部卖出
            if position.enable_amount > 0:
                log.info("尾盘清仓: 卖出 %s" % security)
                order_target(security, 0)
                g.cost_price = 0
        return # 尾盘时间不执行买入逻辑

    # -----------------------------------------------------------------
    # 2. 获取历史数据
    # 获取过去 g.vol_window + 1 根分钟K线数据
    # 包含：成交量(volume), 收盘价(close), 开盘价(open)
    # -----------------------------------------------------------------
    hist = get_history(g.vol_window + 1, '1m', ['volume', 'close', 'open'], security, fq='pre', include=True)
    
    # 如果数据不足（例如刚开盘），则跳过
    if len(hist) < g.vol_window + 1:
        return

    # -----------------------------------------------------------------
    # 3. 计算指标
    # -----------------------------------------------------------------
    # 提取成交量数组
    volumes = hist['volume'].values
    
    # 当前分钟的成交量（数组最后一个）
    current_vol = volumes[-1]
    
    # 过去N分钟的平均成交量（不包含当前分钟，取前N个）
    avg_vol = volumes[:-1].mean()
    
    # 获取当前分钟的价格信息
    current_close = hist['close'].values[-1]
    current_open = hist['open'].values[-1]
    
    # 获取当前持仓
    position = get_position(security)
    
    # -----------------------------------------------------------------
    # 4. 止盈止损逻辑 (如果有持仓)
    # -----------------------------------------------------------------
    if position.amount > 0 and g.cost_price > 0:
        # 计算收益率
        pnl_pct = (current_close - g.cost_price) / g.cost_price
        
        # 止损
        if pnl_pct < -g.stop_loss_pct:
            if position.enable_amount > 0:
                log.info("触发止损: 当前价格 %.2f, 成本 %.2f, 卖出" % (current_close, g.cost_price))
                order_target(security, 0)
                g.cost_price = 0
                return
        
        # 止盈
        elif pnl_pct > g.take_profit_pct:
            if position.enable_amount > 0:
                log.info("触发止盈: 当前价格 %.2f, 成本 %.2f, 卖出" % (current_close, g.cost_price))
                order_target(security, 0)
                g.cost_price = 0
                return

    # -----------------------------------------------------------------
    # 5. 买入信号判断
    # 条件A: 成交量放大 (当前量 > 平均量 * 倍数)
    # 条件B: 价格上涨 (收盘价 > 开盘价)，即阳线，防止下跌放量接盘
    # 条件C: 当前无持仓 (简化逻辑，避免重复加仓)
    # -----------------------------------------------------------------
    if position.amount == 0:
        if current_vol > avg_vol * g.vol_multiplier and current_close > current_open:
            # 记录日志
            log.info("交易信号触发: 标的 %s" % security)
            log.info("当前成交量: %.2f, 平均成交量: %.2f, 放大倍数: %.2f" % (current_vol, avg_vol, current_vol/avg_vol))
            
            # 执行买入
            order_value(security, g.trade_value)
            
            # 更新持仓成本
            g.cost_price = current_close

def after_trading_end(context, data):
    """
    盘后处理
    """
    log.info("今日交易结束")

代码关键点解析

1. get_history 的使用:

   • 我们使用了 get_history(g.vol_window + 1, '1m', ...)。
   • 取 N+1 个数据是为了用前 N 个计算平均值，用最后 1 个作为当前值进行比较。
   • include=True 确保包含当前正在形成的或刚刚结束的这根K线数据。

2. 成交量均线 (MA Volume):

   • avg_vol = volumes[:-1].mean()：这行代码计算了除当前分钟外，过去20分钟的平均成交量。这是判断“放量”的基准。

3. 价格过滤 (current_close > current_open):

   • 这是一个非常重要的过滤条件。成交量放大既可能发生在暴涨时，也可能发生在暴跌时。我们只希望在放量上涨（资金流入）时买入。

4. T+1 限制:

   • 代码中包含了一个 if position.enable_amount > 0 的检查。
   • 在A股实盘中，当天买入的股票 enable_amount 为0，因此止盈止损和尾盘清仓逻辑在当天买入后不会生效（除非您持有昨仓）。如果是做T+0策略，前提是您账户里本身就有该股票的底仓。

5. Python 版本兼容性:

   • 代码使用了 %s 和 % 格式化字符串，而不是 f-string，以确保兼容 PTrade 可能使用的旧版本 Python (如 Python 3.5)。

常见问题 (Q&A)

Q: 这个策略可以直接用于实盘吗？
A: 该策略是一个基础的量化模型。在实盘前，建议您：

1. 回测：在PTrade回测模块中验证参数（如20分钟窗口、2倍放量）在历史数据上的表现。
2. 滑点设置：实盘中放量时价格变动快，建议使用 order_target_value 或在下单时增加限价保护。
3. T+1问题：如果您没有底仓，当日买入后若发生亏损无法止损，需注意隔夜风险。

Q: 如何修改成交量放大的判断标准？
A: 修改 initialize 函数中的 g.vol_multiplier。例如，将其改为 3.0 表示需要成交量达到平均值的3倍才交易。

Q: 为什么开盘前几分钟不交易？
A: 代码中 if len(hist) < g.vol_window + 1: return 这行逻辑会确保只有当积累了足够的数据（比如开盘20分钟后）才开始计算均线，避免开盘瞬间数据波动造成的误判。