如何在聚宽中先计算候选 ETF，再通过多个盘中时间点复检趋势和成交量后执行买入？

在量化交易中，ETF日内择时常面临“假突破”和“日内噪音”的干扰。为了减少假信号，一种行之有效的方案是：在开盘前筛选出候选ETF，并在盘中设置多个关键时间点，对价格趋势和成交量进行复检，只有当多个时点均确认趋势成立且成交量配合时，才执行买入。

在聚宽（JoinQuant）平台中，我们可以利用 run_daily 注册多个定时任务，配合全局变量 g 来实现这一逻辑。以下是具体的策略设计与 Python 代码实现。

1. 核心设计思路

• 开盘前（09:00）：通过 before_market_open 筛选出候选的 ETF 列表（例如成交量活跃、处于多头排列的 ETF），并存入全局变量 g.candidate_etfs。
• 盘中多时点复检（10:00, 11:00, 14:00）：利用 run_daily 在这些时间点触发复检函数。计算候选 ETF 的短期均线（如5分钟或15分钟均线）以及当前成交量是否放大。若满足条件，则在 g.confirmation_count 中累加确认次数。
• 尾盘执行（14:30）：检查确认次数。如果候选 ETF 在之前的多个复检点均通过了趋势和成交量测试（例如确认次数 $\ge 2$），则在尾盘执行买入，以规避日内波动风险。

2. 聚宽策略源码实现

import pandas as pd
import numpy as np
from jqdata import *

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 股票/ETF交易手续费设置：买入万分之三，卖出万分之三，无印花税，最低5元
    set_order_cost(OrderCost(close_tax=0, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='fund')
    
    # 定义全局变量
    g.candidate_etfs = []  # 候选ETF列表
    g.confirmation = {}    # 记录每个候选ETF的盘中确认次数
    
    # 1. 开盘前运行：筛选候选ETF
    run_daily(before_market_open, time='09:00')
    
    # 2. 盘中多时点复检：10:00, 11:00, 14:00 分别进行趋势与成交量复检
    run_daily(market_check_point, time='10:00')
    run_daily(market_check_point, time='11:00')
    run_daily(market_check_point, time='14:00')
    
    # 3. 尾盘执行交易：14:30
    run_daily(trade_execution, time='14:30')

def before_market_open(context):
    log.info("====== 开盘前准备 ======")
    # 设定一个初始的ETF池（例如主流的宽基ETF）
    etf_pool = [
        '510300.XSHG', # 沪深300ETF
        '510500.XSHG', # 中证500ETF
        '159915.XSHE', # 创业板ETF
        '510050.XSHG', # 上证50ETF
        '159949.XSHE'  # 创业板50ETF
    ]
    
    g.candidate_etfs = []
    g.confirmation = {}
    
    # 筛选昨日未停牌且价格在5日均线之上的ETF作为候选
    for etf in etf_pool:
        current_data = get_current_data()
        if not current_data[etf].paused:
            close_data = attribute_history(etf, 5, '1d', ['close'])
            ma5 = close_data['close'].mean()
            last_price = close_data['close'][-1]
            
            if last_price > ma5:
                g.candidate_etfs.append(etf)
                g.confirmation[etf] = 0 # 初始确认次数为0
                
    log.info("今日候选ETF: %s" % g.candidate_etfs)

def market_check_point(context):
    log.info("====== 盘中时点复检: %s ======" % context.current_dt.time())
    if not g.candidate_etfs:
        return
        
    for etf in g.candidate_etfs:
        # 获取最近15根1分钟K线（15分钟趋势）
        bars = get_bars(etf, count=15, unit='1m', fields=['close', 'volume'], include_now=True)
        if len(bars) < 15:
            continue
            
        # 趋势判断：当前价格高于15分钟均价
        avg_price = np.mean([bar['close'] for bar in bars])
        current_price = bars[-1]['close']
        
        # 成交量判断：当前1分钟成交量大于过去15分钟平均成交量的1.5倍（放量）
        avg_volume = np.mean([bar['volume'] for bar in bars[:-1]])
        current_volume = bars[-1]['volume']
        
        if current_price > avg_price and current_volume > 1.5 * avg_volume:
            g.confirmation[etf] += 1
            log.info("%s 在当前时点通过复检，累计确认次数: %d" % (etf, g.confirmation[etf]))
        else:
            log.info("%s 未通过当前时点复检" % etf)

def trade_execution(context):
    log.info("====== 尾盘执行决策 ======")
    cash = context.portfolio.available_cash
    if cash <= 0:
        return
        
    # 卖出不在候选列表或未获得足够确认的持仓
    for stock in list(context.portfolio.positions.keys()):
        if stock not in g.candidate_etfs or g.confirmation.get(stock, 0) < 2:
            order_target(stock, 0)
            log.info("卖出/清仓未达标ETF: %s" % stock)
            
    # 买入获得至少2次及以上盘中确认的ETF
    buy_targets = [etf for etf, count in g.confirmation.items() if count >= 2]
    if not buy_targets:
        log.info("今日无满足多时点确认条件的ETF")
        return
        
    # 平分资金买入
    allocation_cash = cash / len(buy_targets)
    for etf in buy_targets:
        if context.portfolio.positions[etf].total_amount == 0:
            order_value(etf, allocation_cash)
            log.info("多时点确认成功，买入ETF: %s，金额: %.2f" % (etf, allocation_cash))

3. 为什么这种多时点复检能减少假信号？

1. 过滤日内随机波动：单次突破（如仅在早盘10:00突破）极有可能是主力资金试盘或散户跟风造成的假突破。通过在11:00和14:00多次复检，确保了趋势的持续性。
2. 量价协同确认：策略不仅看价格是否在均线之上，还引入了 current_volume > 1.5 * avg_volume 的放量指标，无量空涨的假突破会被直接过滤。
3. 尾盘交易优势：在14:30进行最终决策，此时全天大势基本定型，能有效避开早盘剧烈波动带来的“过山车”行情。