做日内盯盘的人大多经历过这种场景开盘半小时内全市场五千多只股票谁在拉升、谁快封板了只能在几个行情软件之间来回切换。热点板块一启动手速根本跟不上等你在软件里翻到那只票它可能已经封死涨停了。我自己也折腾过不少数据源最后还是回到MiniQMT——它的实时行情接口拿全市场快照非常方便速度和本地交易终端一致Python直接就能处理。这篇文章就来完整复盘一下怎么用MiniQMT把沪深A股的实时涨跌幅全部拉出来再按交易规则把涨停股筛出来。适合已经装好MiniQMT、想用Python做盘中监控的朋友也适合刚接触量化交易、想找一个低门槛实时行情方案的新手。1. 为什么要自己拉全市场涨跌幅而不是靠行情软件很多人第一反应是同花顺、通达信不是自带涨跌幅排名和涨停板池子吗为什么要费劲自己写这个问题我一开始也问过自己。实际用了两周之后我确定了自己写这套东西的价值不在“看排名”而在“可控”。1.1 行情软件解决不了的三件事第一个痛点是刷新频率和交互方式不可控。行情软件确实能排序但它不会在你需要的时候自动把数据导出来也不会在你自定义的触发条件满足时给你推送。你看到的永远是软件想让你看到的视图而不是你代码里定义的变量。第二个痛点是数据无法直接进策略。如果你后面想做涨停打板、龙头战法、情绪周期统计最终都需要把行情数据接入Python去做分析和回测。用手在软件里看再手动记录这个流程既慢又容易漏。第三个痛点是自定义规则几乎为零。比如说“昨天涨停、今天盘中涨幅超过3%”的股票或者“当前封单金额大于流通市值1%”的票这类叠加条件在通达信里写公式也能做但门槛不低而且数据导出还是麻烦。用MiniQMT拉数据这些规则就是几行字典过滤的事。1.2 为什么最终选了MiniQMT我当时对比过几条技术路线挨个说下我的实际感受。一是爬虫抓网页行情接口。免费、不用开户但延迟不稳定盘中高峰期经常拿不到全市场数据而且页面结构一改你的代码就废。做历史数据回测还能忍做实盘监控就算了。二是券商提供的官方数据接口。稳定性和行情源没问题但大部分都绑定特定软件和特定语言文档也写得比较简略想灵活操作不容易。三是MiniQMT。它的核心优势在于行情数据从本地交易终端走等于你本地装了一个行情接收端Python通过接口直接读数据速度很快。同时它支持批量获取全市场快照一次调用就能拿到几千只股票的实时价格、成交量、盘口等字段这对全市场扫描场景来说非常关键。说句大白话MiniQMT是你自己机器上的行情终端数据到你家门口了怎么写、怎么算、怎么筛都是你说了算这个自由度是网页接口和行情软件给不了的。1.3 整体实现链路拆解这个项目的完整链路其实不复杂核心就四步拿到全市场股票代码列表、批量拉实时快照、计算涨跌幅、按涨停规则筛选。这里面真正要动脑子的不是拉数据本身而是“涨停判断”的规则。A股的涨跌幅限制不是所有股票都是10%主板、创业板、科创板、北交所、ST股都不一样而且涨停价是按昨收价乘比例后四舍五入到分计算的。这些细节如果不处理筛出来的涨停池一定是错的。另外还要处理新股和停牌股这两类股票没有昨收价涨跌幅无从谈起需要过滤掉。整个实现思路一句话概括先保证数据源干净可靠再用交易规则去定义“涨停”最后用代码把规则翻译成过滤器。2. 环境准备MiniQMT权限、软件安装与连接测试这个项目的环境门槛不高但有几个坑特别容易卡人。我先按我的操作顺序走一遍你照着弄就行。2.1 MiniQMT权限与软件获取首先要明确一个概念QMT和MiniQMT是两个形态的同一个东西。QMT是完整的行情交易终端带界面、带策略编辑器MiniQMT则是拿掉了界面的精简版本主要通过Python调用。实际操作中你大概率是先安装了QMT或者MiniQMT终端然后用Python去连接这个本地终端进程。注意并不是随便开个股票账户就能用MiniQMT。你需要联系开户券商申请量化交易或QMT/MiniQMT接口的使用权限。每家券商的申请门槛和政策不一样有的要求资金量门槛有的没有。我个人建议直接打客服电话问“能不能开通MiniQMT或者QMT权限”网上的说法不一定准因为政策经常调整。权限开通后券商会给你软件下载地址。不同券商给的安装包可能略有差异界面叫法也可能不一样但核心的Python接口包是同一个也就是你装好终端后在安装目录的某个子目录里能找到xtquant这个Python包。2.2 Python环境与xtquant包MiniQMT对Python版本的要求不苛刻我用的3.9和3.10都跑过。拿到xtquant包之后把它安装到Python环境里有两种方式。第一种是把终端安装目录下的xtquant目录整体复制到你的Python的site-packages里。这种方式的优点是版本一定和你的终端匹配不会出现接口版本不兼容的问题。缺点是升级终端后你可能得重新复制一次。第二种是通过pip安装。如果你所在的券商版本支持可以直接pip install xtquant但我这里的经验是**优先用终端自带的xtquant包不要乱用pip源里的版本。**曾经有朋友为了省事直接用pip装了最新版结果连本地终端时一直握手失败查了半天才发现是接口版本比终端新两边协议对不上。排错成本远高于手动复制那一下。2.3 连接测试与初始化装好包之后第一步先测试能不能连上本地终端。按照我自己的习惯写任何正式代码之前先跑一个最小连通性测试。from xtquant import xtdata # 建立连接 xtdata.connect() # 随便取一只股票的最新行情验证数据通路 tick xtdata.get_full_tick([600000.SH]) print(tick)运行这个脚本之前记得先把MiniQMT终端启动并登录。终端没有登录的情况下connect()大概率会卡住或者返回空数据这是最容易踩的一个坑。我第一次做连通性测试的时候终端没开Python进程挂在那儿一分钟没反应我还以为是自己代码写错了。后来才明白MiniQMT的Python接口本质上是本地终端进程的客户端终端是服务端服务端没启动客户端当然连不上。3. 核心代码拉取全市场实时快照、计算涨跌幅并筛选涨停股这一节是正题。我会按“股票池获取 → 批量快照拉取 → 涨跌幅计算 → 涨停判断 → 主流程整合”的顺序把完整代码拆给你看。3.1 获取全市场A股股票代码列表MiniQMT里获取板块成分股用get_stock_list_in_sector。这里要注意的是不同券商版本里板块名称不一定完全一致有的版本用“沪深A股”有的版本可能需要“上证A股”和“深证A股”分开拉。稳妥的做法是几个候选板块名称都拉一次最后取并集去重。from xtquant import xtdata def get_all_a_stock_list(): stock_list [] for sector_name in [沪深A股, 上证A股, 深证A股]: try: codes xtdata.get_stock_list_in_sector(sector_name) if codes: stock_list.extend(codes) except Exception: continue return sorted(set(stock_list)) all_codes get_all_a_stock_list() print(f共获取到 {len(all_codes)} 只股票)这里有个小细节需要说一下get_stock_list_in_sector返回的代码格式是带交易所后缀的比如600000.SH、000001.SZ后面判断股票代码前缀的时候要记得先去掉后缀或者直接用前缀匹配的方式。实测下来全市场正常交易状态的沪深A股大概在5400只左右数量会因为新股上市、退市整理等因素浮动。3.2 批量获取实时快照数据拿到股票列表后直接调用get_full_tick批量拉快照。这个方法一次可以传几千只股票返回的是一个字典key是股票代码value是该股票的tick数据。ticks xtdata.get_full_tick(all_codes) print(f成功获取 {len(ticks)} 只股票的实时数据)使用这个方法有几点要注意。第一它是快照式接口拉一次返回一个当前时刻的数据不会自动持续推送。要做盘中监控需要自己在循环里反复调用或者改用订阅方式。第二全市场几千只股票一次调用耗时大约在几百毫秒到一两秒之间和你的网络环境、本地终端处理速度都有关系。实测下来这个速度对盘口监控完全够用但如果你的循环里还做了大量其他计算要注意节奏。第三tick数据里常见的字段有lastPrice最新价、lastClose昨收价、open今开、high最高、low最低、volume成交量等。不同版本的接口字段名可能有细微差异我建议你在正式写逻辑之前先打印一只股票的tick看看字段结构。# 先打印看看字段结构 sample_code all_codes[0] sample_tick ticks.get(sample_code, {}) print(sample_tick)这一步非常重要别跳过。我见过有人直接按网上教程写tick[lastPrice]结果自己的版本里字段叫别的名程序一直报KeyError还把锅甩给接口。3.3 计算涨跌幅与涨停判断规则现在到了整个项目的核心部分涨跌幅计算和涨停过滤。先说涨跌幅公式很简单涨跌幅 (最新价 - 昨收价) / 昨收价 × 100%这个逻辑没问题但边界条件要处理。比如新股上市首日没有昨收价停牌股没有最新有效价格这些在tick数据里通常表现为lastClose是0或者缺失直接计算会得到错误结果要跳过。然后是涨停判断。这里我详细展开说下规则因为它是这个项目里最容易写错的部分。A股不同板块的涨停幅度不同我整理了一张规则表供参考股票类型代码特征涨停幅度沪深主板含原中小板60开头、00开头10%创业板30开头20%科创板68开头20%北交所8开头、4开头、92开头30%ST股风险警示股无法从代码识别5%主板ST股的涨停幅度是5%创业板和科创板ST股的涨停幅度也是20%。ST的识别需要股票名称不能只看代码。涨停价的计算规则是昨收价 × (1 涨停幅度)按交易规则四舍五入到分0.01元。举个例子一只主板股票昨收10.35元涨停幅度10%涨停价 10.35 × 1.10 11.385四舍五入到分是11.39元。也就是说这只股票今天最高能到11.39元涨幅达到10.05%因为11.39/10.35 - 1 ≈ 0.1005。如果你只是简单判断涨幅是否大于等于10%这只票就会被漏掉。所以必须计算精确的涨停价不能直接用涨跌幅阈值。判断涨停的方式是最新价 涨停价 - 一个极小容差。为什么要减容差因为浮点数计算和四舍五入可能带来细微误差。比如涨停价算出来是11.64实际最新价是11.64但因为浮点数精度问题比较时可能出现11.64 11.64的情况。我惯用的容差是0.001或者0.01都不影响实际结果只是保险起见。下面是完整的涨停判断函数和主流程代码。def get_limit_up_price(code, last_close, stock_name): 根据股票代码和昨收价计算涨停价。 code: 股票代码格式如 600000.SH last_close: 昨收价 stock_name: 股票名称用于ST识别可为空 pure_code code.split(.)[0] # 板块涨停幅度判断 if pure_code.startswith(30) or pure_code.startswith(68): # 创业板、科创板 limit_ratio 0.20 elif pure_code.startswith(8) or pure_code.startswith(4) or pure_code.startswith(92): # 北交所 limit_ratio 0.30 else: # 沪深主板 limit_ratio 0.10 # ST股判断需要股票名称 if stock_name and ST in stock_name.upper(): limit_ratio 0.05 # 按四舍五入到分计算涨停价 limit_up_price round(last_close * (1 limit_ratio), 2) return limit_up_price然后是把所有逻辑串起来的主流程。这里我为了把逻辑说清楚把ST识别先跳过名称参数传空字符串。后面在第4节单独讲ST股票怎么处理。from xtquant import xtdata import time from datetime import datetime def get_all_a_stock_list(): stock_list [] for sector_name in [沪深A股, 上证A股, 深证A股]: try: codes xtdata.get_stock_list_in_sector(sector_name) if codes: stock_list.extend(codes) except Exception: continue return sorted(set(stock_list)) def monitor_once(all_codes): 拉取全市场实时行情计算涨跌幅筛选涨停股。 ticks xtdata.get_full_tick(all_codes) results [] for code in all_codes: tick ticks.get(code) if not tick: continue last_price tick.get(lastPrice, 0) last_close tick.get(lastClose, 0) # 过滤没有昨收价的股票新股上市首日、无数据 if last_close 0: continue # 计算涨跌幅 change_pct (last_price - last_close) / last_close * 100 # 计算涨停价并判断是否涨停 limit_up_price get_limit_up_price(code, last_close) is_limit_up last_price limit_up_price - 0.001 results.append({ code: code, last_close: last_close, last_price: last_price, change_pct: change_pct, limit_up_price: limit_up_price, is_limit_up: is_limit_up, }) # 按涨跌幅降序排列 results.sort(keylambda x: x[change_pct], reverseTrue) # 筛出涨停股 limit_up_list [r for r in results if r[is_limit_up]] print(f[{datetime.now().strftime(%H:%M:%S)}] 全市场 {len(results)} 只 f涨停 {len(limit_up_list)} 只) for r in limit_up_list[:30]: print(f{r[code]} 涨幅 {r[change_pct]:.2f}% | f现价 {r[last_price]:.2f} | 涨停价 {r[limit_up_price]:.2f}) return results, limit_up_list if __name__ __main__: # 1. 连接 xtdata.connect() # 2. 获取股票池 codes get_all_a_stock_list() print(f股票池数量: {len(codes)}) # 3. 先跑一次 monitor_once(codes)运行之后你会在控制台看到所有股票按涨跌幅从高到低排列涨停股被单独筛出来并且标注了现价和精确涨停价。这套逻辑已经可以作为盘中监控的雏形用起来了。3.4 做成持续监控的循环上面的代码只跑一次就退出了。要做盘中监控还需要外层加一个循环。我实际用的循环逻辑很简单if __name__ __main__: xtdata.connect() codes get_all_a_stock_list() while True: try: monitor_once(codes) except Exception as e: print(f监控异常: {e}) time.sleep(3)这里的时间间隔我建议至少3秒。原因有二一是全市场几千只股票的快照每次拉取本身要一秒左右太频繁没有意义二是如果无限快速循环调用本地终端和网络压力都会增大遇到数据量大时容易把接口拉崩溃。3秒到5秒的刷新频率对绝大多数盘中监控场景都够用。另一个经验是这个循环要包try-except。盘中偶尔会出现单次快照拉取失败的情况比如本地终端暂时卡顿、网络抖动如果不捕获异常监控程序会直接挂掉你可能直到复盘时才发现中断了十几个小时。加个try-except让它自动跳过本轮、继续下一轮稳定很多。4. 涨停筛选的边界情况和避坑记录涨停筛选看着简单实际跑起来之后各种边界情况特别多。我把自己踩过的坑和对应解决方案都整理出来了这部分建议仔细看。4.1 ST股识别与涨停价的特殊情况前面说过ST股的涨停幅度是5%但ST无法从股票代码判断。怎么处理有两种方式。方式一是从tick数据里取股票名称。部分版本的MiniQMT tick数据会带股票名称字段如果有直接判断名称里是否含有“ST”即可。方式二是在开盘前先拉一次全市场股票名称做缓存。MiniQMT提供了获取基础信息的接口你可以批量查询股票名称然后在运行监控脚本时把名称传进去。我实际维护了一个简单的本地映射字典每天开盘前更新一次监控时直接查字典省去盘中重复请求。# 伪代码示例不同版本接口名称可能有差异以实际为准 def build_st_name_map(all_codes): name_map {} for code in all_codes: try: # 获取股票基础信息字段名请以实际打印为准 info xtdata.get_instrument_detail(code) name_map[code] info.get(InstrumentName, ) except Exception: name_map[code] return name_map这里要提示一点不要每次筛选涨停时都现查股票名称几千只股票逐个查接口压力很大场面会很尴尬。一次缓存后面都用内存里的字典。4.2 新股、停牌股、退市整理股怎么过滤新股上市首日没有昨收价在代码里通过last_close 0直接过滤掉了。但这里有个衍生问题科创板、创业板新股上市前5个交易日不设涨跌幅限制它们的代码也是30、68开头。如果不做额外处理创业板第一天上市的新股会被误判成“涨停”——因为现价大概率远超按昨收价算出的涨停价。要解决这个问题最直接的方式就是把上市日期距离今天不超过5个交易日的股票排除。MiniQMT的接口可以查上市日期但这个逻辑做起来会复杂不少。对绝大多数监控场景来说新股数量本身就少我建议先把它们标记出来单独跟踪不要混进正常涨停池等上市满5个交易日后再纳入统一规则。停牌股的判断更简单停牌股票当天没有有效成交tick数据里lastPrice要么是0要么和lastClose完全一致不动。我实测发现不同券商的版本表现不完全一样有的是0有的是停牌前价格。如果你发现某只票的价格连着一天都没动过那就是停牌可以在代码里加个匹配置比如连续N次快照现价和昨收相同就排除。退市整理期的股票也需要注意这类股票虽然还在交易但存在退市风险涨停规则也和正常股票有差异。我个人的做法是维护一份退市风险股名单在筛选结果里单独标注或者直接过滤避免它们混进涨停池干扰分析。4.3 一字板、炸板、封单量的进阶判断基础的涨停筛选只能告诉你“这只票当前价格到了涨停价”但如果想更精细地判断“是否真正封住涨停”还需要看盘口数据。真正封死涨停的股票典型特征是买一档有巨额封单卖一档挂单量很少甚至没有。炸板则是指股价曾经涨停过但随后又打开不再维持在涨停价。如果你只看最新价判断炸板股票会被错误识别为涨停。要处理这部分场景需要用到tick数据里的委买委卖字段。MiniQMT的tick数据中通常包含bidPrice买价、bidVolumes买量、askPrice卖价、askVolumes卖量等盘口字段。判断“封住涨停”的经验逻辑是最新价到达涨停价同时卖一档价格大于等于涨停价或卖一量为0买一档仍然有巨量挂单。更严格的判断还可以看封单金额即买一档价格乘以买一档数量。封单金额越大说明涨停越“牢靠”。我当时额外加了一个规则封单金额小于流通市值的0.5%时标记为“弱封单”提示需要留意炸板风险。4.4 复权和除权除息导致的昨收价偏差这个坑比较隐蔽。MiniQMT实时行情接口返回的lastClose通常是当天的昨收价也就是交易所公布的上一交易日收盘价理论上不需要做复权处理。但如果你在计算涨跌幅时混用了其他数据源的历史行情比如拿不复权的价格去对账就可能出现对不上的情况。我遇到过的情况是某只股票前一天除权除息除权后昨收价和软件里显示的前日收盘价对不上。这不是数据错了而是除权除息后交易所自动调整了昨收价。所以记住一点实时行情场景下直接用接口返回的lastClose不要去自己计算复权价更不要把第三方数据的昨收价填进来。5. 常见问题与排查办法这部分是我在实际运行中碰到的问题汇总以及对应的排查顺序。整理成表格方便你对照。问题现象可能原因排查方法connect()一直卡住或超时MiniQMT终端没启动或没登录先打开终端挂机登录再运行Python脚本get_full_tick返回空字典行情数据还没初始化完成或尚未连接成功先跑连通性测试等终端行情稳定后再拉数据股票池数量只有几百只板块名称不对只拉到了一个子板块用沪深A股、上证A股、深证A股并集涨停筛出大量涨幅超过20%的票把创业板当主板判断了或没过滤新股检查30、68开头代码的涨停幅度是否为20%某些ST股没有出现在涨停池ST识别未生效涨停价按10%算成13%之类确认名称缓存是否存在ST涨停幅度是否设为5%监控程序跑一段时间后卡死循环未加异常捕获本地终端死锁外层包try-excepttime.sleep设为3秒以上涨跌幅计算出现负数或正数异常大最新价字段取错或tick数据结构不对打印一只tick确认lastPrice、lastClose字段名再补一个排查技巧当你发现输出结果和自己预期不一致时先不要看筛选逻辑先把原始tick打印出来对着看。很多问题不是逻辑写错了而是接口字段名、字段含义在券商版本之间有差异。数据接口的字段结构就是你这套系统的协议花十分钟把协议摸清楚后面能少加一个礼拜的班。写在最后的个人体会跑这个监控脚本跑了大概两个月我最大的体会是实时行情接口本身不复杂但把“涨停股筛选”从能用做到好用重头戏全在规则细节上。ST识别、新股过滤、涨停价四舍五入、盘口封单判断每一个看起来都是小事漏掉任何一个你的涨停池就会混进奇怪的东西。我第一版写出来的时候因为没处理ST池子里同时出现了涨停的ST股和正常股涨跌幅排序看起来非常诡异排查半天才反应过来。如果让我给后来者一个建议那就是第一版先求稳只用“现价到达涨停价”这一个条件做基础筛选跑通全流程。等数据通了、结果稳定了再逐步叠加封单量、炸板检测、连板统计这些进阶维度。另外说一点真实感受做这套东西不会让你在盘中成为什么股神但它能让你把注意力从枯燥的刷新排名里解放出来给后续的策略和复盘留出更多时间。后面我大概率会再写一篇关于封单强度排序和连板梯队统计的内容如果这条路上你也遇到了有意思的坑欢迎一起交流。