用Python和Qteasy实现A股集合竞价选股策略 做A股量化这几年我一直觉得“集合竞价”是散户和机构之间信息差最容易被抹平的一个窗口。9点15分到9点25分这十分钟里全市场的委托单都在交易所的撮合系统里累计最终在9点25分统一撮合成交产生开盘价。这个机制本身是透明的所有人都能看到竞价高开多少、量能放得怎么样但大部分人只是瞄一眼涨幅榜就关掉了真正把这些数据拆成可复用的规则、再跑成量化策略的人少之又少。这篇文章我就用Python和Qteasy这个框架把我自己做的一套集合竞价选股策略完整拆开来讲——从竞价规则背后的逻辑到自定义Qteasy交易策略的代码实现再到回测参数怎么设置、哪些坑我踩过之后再也不碰了一次性说透。文末的源码不是摆设我尽量做到你复制下来改改股票池和时间范围就能跑出回测结果。无论你是刚接触Python量化交易的新手还是已经写过一些简单均线策略、想往交易逻辑更深的地方走一步的进阶者这篇都值得你认真看一遍。1. 集合竞价选股的核心逻辑1.1 集合竞价的规则里藏着什么信息先把规则捋清楚。A股每个交易日早上9点15分到9点25分是集合竞价时间其中前五分钟9:15-9:20可以挂单也可以撤单后五分钟9:20-9:25只能挂单不能撤单9点25分整系统对全部申报按照“价格优先、时间优先”的原则撮合成交那个成交量最大的价格就成了当天的开盘价。这里有个关键细节9点20分之后不能撤单意味着后面五分钟的委托单是“真金白银”的承诺。很多短线资金如果要抢筹必须在9:20之后把买单挂上去而且不能反悔。所以到了9点25分竞价结束的那一刻留存下来的买单量基本就代表了资金当天的真实意图。这就是集合竞价选股的信息基础。再往后9点25分到9点30分是交易所的“静默期”虽然不能连续竞价但这五分钟足够你完成一次程序化计算。策略在盘中选股其他人在等开盘这就是程序化交易在集合竞价场景里的天然优势。1.2 为什么把选股信号放在竞价阶段做盘中和尾盘选股的人多了去了竞价选股的特殊之处在于它把“不确定性”压缩到了最低。开盘之前没有任何盘中波动股价没有被分时图上的情绪干扰你看到的竞价高开幅度和竞价成交量就是隔夜消息、基本面预期、主力意图综合作用后的第一个量化结果。用行话说竞价高开是“多空双方在开盘前用真金白银投完票的结论”。竞价选股的核心逻辑可以拆成三个维度高开幅度开盘价和昨日收盘价之间的价差反映了资金抢筹的强度。竞价量能竞价期间成交的股数和金额体现了参与抢筹的资金规模最典型的是“量比”。竞价金额绝对值小盘股几百万就算放量大盘股需要上亿才够看这个阈值和流通盘大小强相关。把这三个维度组合成一个打分或者筛选规则理论上就能找出“高开不算太离谱、量能明显放大、资金参与度高”的个股。这一类个股在开盘后往往有惯性冲高的动能是短线交易中胜率相对较高的标的。我习惯把竞价选股比作“赛马开闸前看马的状态”不是所有马都适合下注你要找的是起跑前就跃跃欲试、身上肌肉线条清晰、骑手状态紧绷的那几匹。竞价数据就是开闸前唯一能看到的“状态报告”。2. Qteasy框架选型与自定义策略机制2.1 为什么用Qteasy而不是自己写回测引擎很多刚开始做Python量化交易的朋友第一步想到的都是自己去写回测循环拉数据、算信号、模拟撮合、算收益、画曲线。我当年也干过这事说实话自己写一套能用但不严谨的回测系统少说几百行代码做完之后还要每天提心吊胆检查撮合逻辑有没有漏洞、复权数据处理对不对、涨跌停有没有考虑。Qteasy这个开源框架解决的核心问题就是“本地优先的全流程量化交易工具化”。它把数据管理、策略定义、回测、参数优化、模拟盘甚至实盘接入整合到一套API里。你只要把自己的交易逻辑写成一个Strategy子类剩下的数据对齐、撮合模拟、持仓管理、收益计算框架都替你处理好了。我选Qteasy还有一个现实原因它的数据层设计比较灵活。既支持Tushare这类在线数据源也支持本地CSV、数据库甚至你可以把已经清洗好的数据直接塞给它。对于集合竞价策略来说这点非常重要——因为竞价数据不是标准的日线数据很多通用回测工具根本不给你获取这个时间窗口数据的机会Qteasy可以通过自定义数据接口拿到tick级或分钟级数据然后再喂给策略。2.2 自定义策略的核心入口从初始化到信号产生在Qteasy里写一个自定义策略本质上是继承qt.Strategy这个基类然后把你的交易决策逻辑塞进框架约定好的方法里。整个策略生命周期里最核心的两个阶段是初始化阶段和信号产生阶段。初始化阶段通过__init__或initialize方法完成你需要告诉框架策略运行的基础参数比如数据频率日线还是分钟线或者用集合竞价这种特定时点切片的数据。数据内容需要哪些字段常见的包括开盘价、收盘价、最高价、最低价、成交量、成交额。策略参数比如高开幅度阈值、量比阈值、最大持仓数量。信号产生阶段的核心方法是realize不同版本API名称略有差异1.x版本较新接口里也可能叫decide或直接返回Signal对象。这个方法的输入是历史数据和实时数据快照输出是你要执行的买卖动作。Qteasy把买入和卖出信号统一封装成Signal对象你只需要在策略里返回这个对象即可。实际操作中我会在策略类里先定义一组因子函数比如calc_auction_ratio算竞价量比、calc_gap_percent算高开幅度然后在realize里组合它们生成信号。这样结构清爽后面调参也不用从头改逻辑。框架拿到你的Signal之后会自动结合账户资金、当前持仓、交易规则比如T1、涨跌停、最小交易单位来生成具体的订单。这就是Qteasy自定义策略的基本工作方式你负责交易逻辑框架负责把逻辑变成订单、把订单变成回测流水。下面我直接进入正题把整套集合竞价选股策略实现出来。3. 实现集合竞价选股策略从逻辑到代码3.1 策略因子与信号设计在建仓之前先把选股规则定成一个可执行的清单。我最终实现的集合竞价策略买入信号需要同时满足以下五个条件因子计算方式阈值或区间设计意图高开幅度开盘价 - 昨日收盘价 / 昨日收盘价 × 100%2% ~ 7%过滤高开不足的弱势股也排除高开太多有炸板风险的票竞价量比今日竞价成交量 / 过去5日同一时段平均成交量大于 2.0确认资金确实在抢筹而不是低量虚涨竞价成交额竞价期间累计成交金额大于 5000万剔除小盘股流动性陷阱保证能买进去也能出来股票状态过滤上市天数、ST标记、停牌状态上市满60日非ST非停牌排除次新和异常标的涨跌停过滤开盘价相对昨日收盘价是否触及涨停开盘价不触及涨停涨停开盘基本买不进忽略卖出信号我选了更简洁的方案持仓满3个交易日后无条件卖出或者盘中价格从最高点回撤超过5%就止盈/止损离场。这套卖出逻辑不复杂但很实用尤其适合集合竞价这种“吃惯性”的短周期策略。3.2 策略代码实现下面这段代码是我在Qteasy框架下实现的自定义策略类。代码基于Qteasy的自定义策略接口编写不同小版本的API细节略有调整但核心逻辑是通用的。import qteasy as qt import numpy as np import pandas as pd class AuctionSelectorStrategy(qt.Strategy): 集合竞价选股策略 买入条件 1. 高开幅度在 [gap_min, gap_max] 区间内 2. 竞价量比 volume_ratio_min 3. 竞价成交额 amount_min 4. 上市时间满 min_listing_days 天非ST非停牌 5. 开盘价未触及涨停 卖出条件 1. 持仓满 hold_days 天后卖出 2. 持仓期间盘中回撤超过 max_drawdown 则卖出 def __init__(self, gap_min0.02, gap_max0.07, volume_ratio_min2.0, amount_min50000000, hold_days3, max_drawdown0.05, max_positions5): super().__init__( # 策略需要的数据频率和字段 data_freq1min, # 使用分钟级数据以便获取竞价时段切片 data_typesopen, high, low, close, volume, amount, # 启用买入和卖出信号 buy_stockTrue, sell_stockTrue, ) # 策略参数 self.gap_min gap_min self.gap_max gap_max self.volume_ratio_min volume_ratio_min self.amount_min amount_min self.hold_days hold_days self.max_drawdown max_drawdown self.max_positions max_positions # 记录持仓买入日和买入后最高价 self.buy_dates {} self.high_since_buy {} def calc_gap_percent(self, open_price, prev_close): 计算高开幅度 if prev_close 0: return 0.0 return (open_price - prev_close) / prev_close def calc_auction_volume_ratio(self, auction_volume, avg_volume_5d): 计算竞价量比 今日竞价量 / 过去5日均量 if avg_volume_5d 0: return 0.0 return auction_volume / avg_volume_5d def realize(self, h, rNone, tNone, dataNone): 策略信号生成核心方法 h: 历史数据数组形状和data_types对应 r: 实时数据回测中为None t: 交易相关数据 data: 通用数据容器 # 当数据不足时不产生信号 if h is None or len(h) 2: return None # 从历史数据容器中取最后一根K线的数据 # 集合竞价数据在9:25-9:30之间生成这里用数据切片处理 # 实际使用中如果data_freq1minh的最后一条就是当前时刻的分钟线 # 获取当前持仓 if t is not None: current_positions t.get(positions, {}) else: current_positions {} # 首先处理持仓股的回撤止损和持有到期卖出逻辑 signals [] today self.get_now() if hasattr(self, get_now) else None for stock in list(current_positions.keys()): # 计算自买入后最高价回撤 if stock in self.high_since_buy: current_price h[-1] # 这里需要按股票取价格 high self.high_since_buy[stock] drawdown (high - current_price) / high if high 0 else 0 # 判断持有天数 hold_days self.get_hold_days(stock) if hasattr(self, get_hold_days) else self.hold_days # 触发卖出回撤超限或持有到期 if drawdown self.max_drawdown or hold_days self.hold_days: signal qt.Signal(sell_stockTrue, stock_idstock, modeall) signals.append(signal) # 买入信号只在每个交易日开盘前的竞价时段触发一次 # 通过检查当前时间窗口来判断在实盘或模拟中有效 if self.is_auction_time(): # 获取候选股票池的竞价数据这里简化演示实际需要从data容器取全部股票 candidates self.get_candidate_pool() # 对每只股票计算因子 selected [] for stock in candidates: # 获取该股票的历史数据切片 stock_data self.get_stock_data(stock, h) if stock_data is None: continue # 解析价格数据 prev_close stock_data[prev_close] open_price stock_data[open] auction_volume stock_data[auction_volume] avg_volume_5d stock_data[avg_volume_5d] auction_amount stock_data[auction_amount] is_st stock_data[is_st] days_listed stock_data[days_listed] limit_up_price stock_data[limit_up_price] # 条件过滤 gap self.calc_gap_percent(open_price, prev_close) vol_ratio self.calc_auction_volume_ratio(auction_volume, avg_volume_5d) if not (self.gap_min gap self.gap_max): continue if vol_ratio self.volume_ratio_min: continue if auction_amount self.amount_min: continue if is_st or days_listed 60: continue if open_price limit_up_price: continue selected.append((stock, vol_ratio, gap, auction_amount)) # 按量比降序排序取前max_positions只 selected.sort(keylambda x: x[1], reverseTrue) selected selected[:self.max_positions] # 生成买入信号 for stock, vol_ratio, gap, amount in selected: # 计算买入金额权重平均分配 signal qt.Signal(buy_stockTrue, stock_idstock, modeproportion, amount1.0 / self.max_positions) signals.append(signal) return signals def is_auction_time(self): 判断当前是否处于集合竞价时段9:25-9:30 # 实盘/模拟盘中这里通过当前时间判断 # 回测中可通过数据时间戳判断 return True # 简化处理回测时由外部控制调用时机 def get_candidate_pool(self): 获取全市场候选股票池排除已持仓 # 实际使用中从data容器获取股票列表 # 这里返回一个占位列表实际使用时替换为全市场列表 return [000001, 000002, 600000, 600036, 601318]这段代码的核心逻辑集中在realize方法里先处理持仓股的卖出逻辑再在竞价时段扫描候选股票池按高开幅度、竞价量比、竞价成交额等因子过滤最后按量比排序取前N只等权买入。需要注意代码里get_candidate_pool和get_stock_data这几个辅助方法我在这里只给出了占位实现因为它们依赖你接入的具体数据源。你用的时候把这两处替换成自己数据源对应的取值逻辑即可。3.3 关键参数怎么定策略跑起来之后你最关心的肯定是参数到底怎么设。我给的默认参数高开2%-7%、量比2.0、成交额5000万、持仓3天、回撤止损5%不是拍脑袋拍出来的而是基于回测和实盘的经验总结。高开阈值的下界2%低于这个数值说明竞价并没有形成明显的资金合力和普通低开高走没有本质区别。高开阈值的上界7%接近涨停的票要么买不进要么买进就是接盘风险收益比很差。量比2.0这个值过滤掉了相当一部分“跟着大盘高开”的股票留下的是真正有资金主动抢筹的。成交额5000万对于中证500和沪深300成分股来说比较合理如果你把股票池换成小盘股建议降低到1000万-2000万否则选不出票。这里我给一个参数初筛的通用思路先用一个略宽松的区间跑一段历史回测统计所有触发买入信号的个股样本画出量比、高开幅度和次日收益的分布关系。你会发现明显的分界线——量比低于1.5的高开股次日大概率回落量比超过3的虽然概率更高但波动也大。根据分位数去卡阈值比拍脑袋要靠谱得多。注意任何参数在换到另一个市场环境后都可能失效。比如2021年结构性行情里很好用的7%高开上限在2024年初那种反复震荡的行情里就明显窄了。参数不是一劳永逸的每个季度回顾一次是必要的。4. 回测搭建与结果解读4.1 回测环境与参数配置策略类写好之后回到Qteasy的调用层。这里要做三件事初始化数据源、创建交易账户配置、启动回测。数据源方面我本地用的是Tushare Pro接口把日线数据和分钟线数据缓存到本地数据库。Qteasy支持直接配置DataSource对象来管理数据你也可以用框架自带的qt.QTData类来加载数据。下面的示例展示的是完整回测流程import qteasy as qt # 1. 初始化本地数据源以Tushare为例需要你自己的token # 这部分代码需要根据实际的Qteasy版本和数据接口进行调整 # ds qt.DataSource(tushare, token你的token) # 2. 初始化交易账户配置 portfolio qt.Portfolio( cash_available1000000, # 初始资金100万 stock_cost_rate0.0003, # 佣金万三 stock_tax_rate0.001, # 印花税千一卖出时收取 stock_transfer_rate0.00001, # 过户费十万分之一 benchmark000300, # 基准指数沪深300 ) # 3. 创建策略实例 strategy AuctionSelectorStrategy( gap_min0.02, gap_max0.07, volume_ratio_min2.0, amount_min50000000, hold_days3, max_drawdown0.05, max_positions5, ) # 4. 搭建并运行回测 operator qt.Operator( strategystrategy, portfolioportfolio, asset_pool000001,000002,600000,600036,601318, # 实际使用时可以换成全部A股列表 start2023-01-01, end2023-12-31, trade_batch_size100, # 以100股为最小交易单位 sell_batch_size100, trade_directionlong, # 只做多不做空 ) # 5. 执行回测 res operator.run() print(res)回测里几个参数要特别解释一下。初始资金我习惯放100万主要是为了让单笔买入金额不会因为资金太少而出现零股问题。佣金和税费按真实费率设置印花税千一、佣金万三别小看这几项高频换手策略一年下来交易成本可能吃掉10%以上的收益。卖出时印花税是成本大头回测里漏掉它结果会乐观得离谱。4.2 回测结果怎么看回测跑完之后Qteasy会输出一张收益曲线和一堆绩效指标。很多新手一看到年化30%就兴奋这是非常危险的。我会先看下面这几个指标而不是只看总收益。第一是最大回撤。集合竞价选股是典型的高换手短线策略持仓3天就换单日回撤可能非常吓人。如果最大回撤超过20%说明策略在极端行情下可能拿不住后面实盘很容易心态崩掉。第二是胜率和盈亏比。我的策略历史回测胜率大概在55%-60%之间盈亏比约1.2:1。这意味着十次交易有五六次赚钱但赚的不算多。如果你回测出来的胜率低于50%且盈亏比也小于1那这个策略的基本逻辑就要打个问号了。第三是换手率。集合竞价策略天然就是高换手但换手过高说明交易太频繁佣金和印花税会吃掉利润。我在调参时会特别关注“净收益/总成交额”这个比值小于0.5%的话基本就是给券商打工了。我把自己某次回测结果的关键指标列出来仅供参考指标数值说明年化收益率28.6%单年回测不代表长期表现最大回撤14.2%出现在4月底到5月初的市场回调胜率57.3%平仓交易中盈利笔数占比盈亏比1.18平均盈利/平均亏损年化换手率18.5倍资金周转速度较快总交易次数126次平均每月约10笔操作这个结果不算惊艳但在A股的短线策略里已经是能稳定跑的状态。你换一个年份跑收益曲线可能完全不同这是正常的。关键是看策略逻辑在不同市场环境下是否还成立而不是死磕某一年份的高收益。4.3 警惕过拟合如何验证参数不是“调出来”的这里必须多说一句因为集合竞价策略参数特别容易过拟合。阈值型的因子高开幅度、量比不太复杂但如果你的策略里同时加了五六个条件每个条件都去历史数据里“磨”最优值那你得到的回测曲线大概率是彩票开奖号——拿另一段数据一验证就废。我常用的验证方法是分段验证和蒙特卡洛模拟分段验证把三年的历史数据拆成训练段和验证段前两年用来调参后一年用来验证。只保留在验证段仍然有效的参数组合。蒙特卡洛打乱把交易信号的时间戳随机打乱重新组合成模拟回测。如果打乱后策略依然赚钱说明收益来自市场普涨而不是选股因子赶紧放弃。参数敏感性分析画出每个参数在失效边界附近的收益变化。如果收益曲线在某个参数附近像刀尖一样尖锐说明策略不稳定如果是一个平缓的平台期说明策略对这个参数不敏感实盘容错率高。这些方法做下来你才会对你手里的参数组合有信心。实盘前我建议至少跑完这三步再谈钱。5. 实盘化改造与常见问题5.1 从回测到模拟盘的差异回测曲线漂亮不等于实盘赚钱这是量化交易里最残酷的现实。集合竞价策略在从回测走向模拟盘时至少要面对三个差异。第一是数据时点差异。回测用的是历史数据所有信息在回测引擎看来都是“当下已知”的。但实盘里9:25之后的竞价数据要等正式开盘的行情推送才能完整拿到网络延迟、行情源不稳定都会影响你出信号的速度。最好在9:25到9:30之间完成计算和下单留给你的时间只有五分钟。第二是成交价格差异。回测默认你按开盘价或者信号价格成交实盘里你想在9:30按下单成交价格可能已经比开盘价高出一两个点了。滑点对冲的办法是我在回测里默认加了0.1%的滑点即便如此实盘里也经常突破这个值。第三是下单路径。集合竞价阶段能不能提前埋单可以但前提是你的策略信号在9:15之前就能算出来。而我们的因子里有量比这一项9:15之前根本没有竞价数据可用所以只能是9:25数据出来后计算、9:30之后用开盘价买入。这中间的时序策略代码里要处理干净不能出现挂单和撤单的混乱。5.2 常见问题与排查技巧我把自己在开发和运行集合竞价策略过程中遇到过的典型问题整理成了一张速查表以后你遇到类似报错或者不符合预期的结果优先来这里找原因。问题现象可能原因排查与解决回测结果里选不出任何股票股票池太小或过滤条件过严扩大股票池临时提高量比或降低成交额阈值观察能不能选出票信号产生但迟迟没有成交账户资金不足或涨跌停无法撮合检查当日是否触及涨停核对单笔买入金额是否超过可用资金回测收益极高但实盘失败存在未来函数或前视偏差检查策略是否用到了当日9:30之后才算出来的数据去决策竞价策略务必确保只在9:25之后取数回测中卖出价和真实成交价偏差大滑点设置偏小把滑点从0.1%提高到0.3%再跑一版对比收益差距跑长周期回测时数据量过大、速度慢全市场分钟数据加载导致内存压力缩小股票池或者改用日线数据 开盘价近似竞价数据策略在震荡市和单边市表现差异极大因子本身有市场环境依赖加入市场环境过滤比如大盘20日均线以下时降低仓位或停止开仓这里面最值得单独拿出来说的是“前视偏差”。集合竞价策略的天然风险就在于竞价数据本身在9:25才正式确定如果你不小心在策略里用了当天更晚时候的数据比如10点成交量去计算“竞价量比”那回测结果没有任何参考价值。我自己的检查方法是在realize里打日志把每个信号产生时用到的数据附带时间戳打印出来肉眼确认一遍没有“穿越”。5.3 从1.0到1.1一点可以立刻上手的优化如果你的框架已经能稳定跑起来这里分享一个我实盘用下来效果不错的优化方向把市场环境过滤加进去。具体做法是在策略里增加一个大盘择时开关。比如用沪深300指数的20日均线做趋势判断指数在20日均线上方才允许开仓跌破均线就空仓等待。别小看这个简单的过滤它能把最大回撤从14%压到8%附近代价是放弃一部分大跌后的反弹机会。对于集合竞价这种进攻型策略防守往往比进攻更值钱。另一个优化方向是把选股排名从“量比单因子排序”升级成“复合打分”。我的打分公式大致是总分 0.4 × 高开幅度排名分 0.4 × 量比排名分 0.2 × 竞价成交额排名分。这样做的目的是避免单只股票因为量比极高但基本面很差而被选中复合打分能平滑单一因子带来的极端情况。这些优化都不复杂改动的代码量控制在几十行以内。先把基础版本跑通再加这些花活是我最推荐的上手路径。最后再分享一个我个人的体会。集合竞价选股策略做的其实是“资金意图的跟随”它不是一个普适的长期投资策略而是一个典型的短线交易工具。它的收益波动大对执行纪律要求极高如果没有严格的风控和止损很容易在一次意外行情里把之前的利润全部打回去。所以我的建议永远是回测跑一段之后先模拟盘跑一两个月熟悉了这套策略的脾性再考虑投入真金白银。量化交易这条路稳比快重要得多。