聚宽平台如何编写首板半路策略？（附集合竞价买入与尾盘卖出Python源码）

在聚宽（JoinQuant）平台上编写“首板半路/追涨策略”需要精确控制时间节点与执行逻辑。该策略的核心逻辑包括：昨日首板筛选、今日集合竞价过滤、盘中动态止盈以及尾盘强制清仓。

以下是完整的策略设计思路与Python代码实现：

一、 核心逻辑设计

1. 昨日首板识别：
   • 在 before_trading_start 中，获取全市场股票昨日及前日的收盘价与涨停价。
   • 筛选出昨日涨停（昨日收盘价 = 昨日涨停价）且前日未涨停（前日收盘价 < 前日涨停价）的股票，避免选入连板股。
2. 集合竞价筛选（09:25 - 09:30）：
   • 使用聚宽特有的 get_call_auction 接口获取今日 09:25 的集合竞价数据。
   • 计算高开幅度：(今日开盘价 - 昨收价) / 昨收价。筛选高开幅度在 $2% \sim 5%$ 之间、且集合竞价成交量异常放大的标的。
3. 盘中动态止盈：
   • 盘中实时监控持仓，若个股涨幅达到止盈点（如相比买入价上涨 $6%$）或触及涨停，可选择分批或全部止盈。
4. 尾盘强制清仓（14:50）：
   • 通过 run_daily(tail_sell, time='14:50') 注册定时任务，在 14:50 准时将所有持仓一键清空，确保不留套牢盘过夜。

二、 聚宽策略源码实现

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)
    
    # 佣金印花税设置：买入万三，卖出万三加千一印花税
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 策略参数设置
    g.max_buy_count = 3          # 每日最多买入股票数
    g.min_gap = 0.02             # 最小高开幅度 (2%)
    g.max_gap = 0.05             # 最大高开幅度 (5%)
    g.take_profit_ratio = 0.06   # 盘中动态止盈幅度 (6%)
    
    # 定时运行任务
    # 1. 09:30 开盘立即执行集合竞价买入逻辑
    run_daily(market_open_buy, time='09:30')
    # 2. 14:50 尾盘强制清仓
    run_daily(tail_sell_all, time='14:50')

def before_trading_start(context):
    # 获取当前日期
    today = context.current_dt.date()
    # 获取全市场股票
    all_stocks = list(get_all_securities(['stock'], date=today).index)
    
    # 筛选昨日首板股票
    g.target_list = []
    for stock in all_stocks:
        # 获取过去2天的日K线数据（包含收盘价、涨停价）
        df = attribute_history(stock, 2, '1d', ['close', 'high_limit', 'paused'])
        if len(df) < 2 or df['paused'].iloc[-1] == 1:
            continue
        
        # df.iloc[-1] 为昨日，df.iloc[-2] 为前日
        yesterday_limit = df['close'].iloc[-1] >= df['high_limit'].iloc[-1]
        before_yest_limit = df['close'].iloc[-2] >= df['high_limit'].iloc[-2]
        
        # 昨日涨停且前日未涨停 = 首板
        if yesterday_limit and not before_yest_limit:
            g.target_list.append(stock)
            
    log.info(f"今日候选昨日首板股票数量: {len(g.target_list)}")

def market_open_buy(context):
    if not g.target_list:
        return
        
    today_str = context.current_dt.strftime('%Y-%m-%d')
    
    # 获取候选股今日09:25的集合竞价数据
    try:
        auction_df = get_call_auction(g.target_list, start_date=today_str, end_date=today_str, fields=['time', 'current', 'volume'])
    except Exception as e:
        log.error(f"获取集合竞价数据失败: {e}")
        return
        
    if auction_df.empty:
        return
        
    buy_candidates = []
    for stock in g.target_list:
        stock_auction = auction_df[auction_df['code'] == stock]
        if stock_auction.empty:
            continue
            
        # 获取昨日收盘价
        hist = attribute_history(stock, 1, '1d', ['close'])
        last_close = hist['close'].iloc[0]
        
        # 集合竞价开盘价
        open_price = stock_auction['current'].iloc[0]
        volume = stock_auction['volume'].iloc[0]
        
        if open_price <= 0 or last_close <= 0:
            continue
            
        # 计算高开幅度
        gap_ratio = (open_price - last_close) / last_close
        
        # 筛选高开在 2% ~ 5% 之间，且集合竞价有成交量（成交量 > 1000手）的股票
        if g.min_gap <= gap_ratio <= g.max_gap and volume > 100000:
            buy_candidates.append((stock, gap_ratio))
            
    # 按高开幅度从大到小排序，优先买入强势高开股
    buy_candidates.sort(key=lambda x: x[1], reverse=True)
    buy_list = [x[0] for x in buy_candidates[:g.max_buy_count]]
    
    # 执行买入
    if buy_list:
        cash_per_stock = context.portfolio.available_cash / len(buy_list)
        for stock in buy_list:
            order_value(stock, cash_per_stock)
            log.info(f"集合竞价符合条件，买入: {stock}")

def handle_data(context, data):
    # 盘中实时监控：动态止盈
    for stock in list(context.portfolio.positions.keys()):
        position = context.portfolio.positions[stock]
        if position.closeable_amount <= 0:
            continue
            
        current_price = data[stock].close
        avg_cost = position.avg_cost
        
        # 计算当前收益率
        profit_ratio = (current_price - avg_cost) / avg_cost
        
        # 达到止盈点，盘中清仓
        if profit_ratio >= g.take_profit_ratio:
            order_target(stock, 0)
            log.info(f"{stock} 达到止盈点 {g.take_profit_ratio*100}%, 盘中卖出清仓！")

def tail_sell_all(context):
    # 14:50 尾盘无条件清仓所有持仓
    for stock in list(context.portfolio.positions.keys()):
        order_target(stock, 0)
        log.info(f"尾盘14:50强制清仓卖出: {stock}")

三、 避坑指南与注意事项

1. get_call_auction 权限与时间：该接口在回测中支持历史数据，但在模拟盘中，当日的集合竞价数据最晚于 09:28 分返回。因此，将买入逻辑放在 09:30 开盘时运行是最安全、最稳定的做法。
2. 真实价格模式（use_real_price）：首板策略对价格极其敏感，必须在 initialize 中开启 set_option('use_real_price', True)，否则前复权价格会导致买入股数计算错误。
3. 停牌与退市处理：在筛选昨日首板时，务必通过 paused 字段过滤掉停牌股，避免下单失败占用资金。