3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在早盘 09:20 - 09:25 的集合竞价阶段,主力资金的意图往往体现得最为明显。通过 PTrade 量化平台,投资者可以在 09:25 集合竞价定型后、09:30 正式开盘前的窗口期,快速抓取全市场或目标股票池的竞价数据,筛选出强势高开或即将涨停的标的并完成下单。
get_trend_dataget_trend_data 是 PTrade 专用于获取集中竞价期间代码数据的 API,能够精准抓取集合竞价结束时的价格、成交量及成交额等关键指标。
get_trend_data(date=None, stocks=None, market=None)
'600570.SS' 或 ['600570.SS', '000001.SZ']。['XSHG', 'XSHE']),与 stocks 参数互斥。返回数据格式为嵌套字典 {stock_code: {fields...}},主要字段包括:
hq_px: 集合竞价撮合成交价wavg_px: 集合竞价加权价格business_amount: 集合竞价匹配成交量business_balance: 集合竞价匹配成交额time_stamp: 时间戳initialize):使用 run_daily 绑定定时任务,设定在每日 09:25 执行竞价选股逻辑。before_trading_start):排除 ST、停牌及退市股票,构建待监控的目标股票池。get_trend_data 获取竞价撮合价 hq_px 和竞价成交额 business_balance。get_snapshot 获取昨日收盘价 preclose_px 与今日涨停价 up_px。order 或 order_value 函数,以限价单(涨停价抢筹)或市价单及时报单。# -*- coding: utf-8 -*-
import time
def initialize(context):
# 设置基准
set_benchmark('000300.SS')
# 设置股票池(以沪深300成分股为例)
g.security_list = get_index_stocks('000300.XBHS')
set_universe(g.security_list)
# 定时任务:每日 09:25 执行集合竞价抢筹策略
run_daily(context, auction_strategy, time='9:25')
def before_trading_start(context, data):
# 过滤 ST、停牌及退市股票
g.target_stocks = filter_stock_by_status(g.security_list, filter_type=["ST", "HALT", "DELISTING"])
def auction_strategy(context):
log.info("=== 开始执行 09:25 集合竞价选股扫单策略 ===")
# 获取竞价阶段数据
trend_data = get_trend_data(stocks=g.target_stocks)
# 获取实时快照(包含昨收价、涨停价)
snapshots = get_snapshot(g.target_stocks)
selected_stocks = []
for stock in g.target_stocks:
if stock not in trend_data or stock not in snapshots:
continue
auction_price = trend_data[stock].get('hq_px', 0) # 竞价价格
auction_balance = trend_data[stock].get('business_balance', 0) # 竞价成交额
pre_close = snapshots[stock].get('preclose_px', 0)
up_limit = snapshots[stock].get('up_px', 0)
if pre_close <= 0 or auction_price <= 0:
continue
# 计算竞价涨幅
auction_pct = (auction_price - pre_close) / pre_close * 100
# 筛选条件:1. 竞价高开 3%~8% 2. 竞价成交额 > 500万
if 3.0 <= auction_pct <= 8.0 and auction_balance >= 5000000:
selected_stocks.append((stock, auction_price, up_limit))
log.info(f"选中标的: {stock}, 竞价价: {auction_price}, 竞价涨幅: {auction_pct:.2f}%, 竞价成交额: {auction_balance/10000:.2f}万")
# 下单逻辑:对精选标的进行抢筹买入
cash_per_stock = context.portfolio.cash / max(len(selected_stocks), 1)
for stock, curr_price, up_limit in selected_stocks:
if cash_per_stock > 10000:
# 采用涨停价挂单,确保开盘优先成交(实际成交价为开盘撮合价)
order_id = order_value(stock, cash_per_stock, limit_price=up_limit)
log.info(f"集合竞价买单已发出: {stock}, 目标金额: {cash_per_stock:.2f}, 委托限价(涨停价): {up_limit}")
def handle_data(context, data):
pass
09:25 之后撮合价格已经确定且买卖单不可撤销,是量化策略捕捉开盘信息的最佳时间。limit_price 时需确保精度符合柜台要求,避免报单被拒。limit_price 进行限价委托,可以在开盘集合竞价撮合时享受最高优先级的撮合顺序,而实际成交价依然是 09:25 确定的撮合开盘价。