Python 实时监控 A 股行情并自动筛选强势股REST WebSocket 两种方案盘中实时监控全市场行情自动筛选涨停、放量上涨、突破均线的股票 – 这是很多量化交易者的刚需。本文用 Python 实现两种方案REST 轮询方案简单易懂和 WebSocket 推送方案低延迟附完整可运行代码。一、盘中行情监控能做什么实时行情监控是从回测走向实盘的关键一步。常见用途涨停板监控第一时间发现涨停或接近涨停的股票放量突破筛选找出成交量异常放大且价格突破关键位的标的自选股盯盘监控持仓或关注的股票触发条件时提醒日内交易信号结合分钟 K 线生成实时买卖信号二、环境准备安装依赖pipinstalltickflow[all]--upgrade实时行情功能需要注册获取 API key访问 tickflow.org登录后在控制台生成 API key配置环境变量推荐或直接传入# Linux / MacexportTICKFLOW_API_KEYyour-api-key# WindowssetTICKFLOW_API_KEYyour-api-key三、方案一REST API 轮询入门推荐REST 方案最简单通过定时请求获取全市场行情快照适合对延迟要求不高的场景。1. 获取全市场实时行情fromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)# 获取全部 A 股实时行情quotestf.quotes.get(universes[CN_Equity_A],as_dataframeTrue)print(f共获取{len(quotes)}只 A 股行情)print(quotes[[symbol,last_price,volume,amount]].head())一次调用即可拿到 5000 只 A 股的实时行情包含最新价、开盘价、最高价、最低价、昨收、成交量、成交额等字段。2. 涨跌幅排行quotes[change_pct](quotes[last_price]-quotes[prev_close])/quotes[prev_close]*100# 涨幅榜 Top 10print(涨幅榜 Top 10:)top_gainersquotes.nlargest(10,change_pct)[[symbol,last_price,change_pct]]print(top_gainers.to_string(indexFalse))# 跌幅榜 Top 10print(\n跌幅榜 Top 10:)top_losersquotes.nsmallest(10,change_pct)[[symbol,last_price,change_pct]]print(top_losers.to_string(indexFalse))3. 涨停股监控# A 股主板涨跌停限制为 10%创业板和科创板为 20%limit_upquotes[quotes[change_pct]9.9]print(f\n涨停或接近涨停:{len(limit_up)}只)print(limit_up[[symbol,last_price,change_pct,amount]].to_string(indexFalse))4. 放量上涨股筛选# 成交额 5 亿 且 涨幅 3%strongquotes[(quotes[amount]5e8)(quotes[change_pct]3)]strong_sortedstrong.sort_values(change_pct,ascendingFalse)print(f\n放量上涨成交额5亿 且 涨幅3%:{len(strong_sorted)}只)print(strong_sorted[[symbol,last_price,change_pct,amount]].head(20).to_string(indexFalse))5. 定时轮询监控将上述逻辑封装为一个定时轮询程序importtimeimportdatetimefromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)defscan_market():扫描全市场返回符合条件的强势股quotestf.quotes.get(universes[CN_Equity_A],as_dataframeTrue)quotes[change_pct]((quotes[last_price]-quotes[prev_close])/quotes[prev_close]*100)# 涨停股limit_upquotes[quotes[change_pct]9.9]# 放量上涨strongquotes[(quotes[amount]5e8)(quotes[change_pct]3)]# 大跌预警跌幅超过 7%big_dropquotes[quotes[change_pct]-7]return{total:len(quotes),limit_up:limit_up,strong:strong,big_drop:big_drop,avg_change:quotes[change_pct].mean(),}defrun_monitor(interval30):每隔 interval 秒扫描一次print(开始监控全市场行情...)whileTrue:nowdatetime.datetime.now()hournow.hour# 只在交易时间运行9:25 - 15:05ifhour9orhour16:print(f[{now:%H:%M:%S}] 非交易时间等待中...)time.sleep(60)continueresultscan_market()print(f\n[{now:%H:%M:%S}] 市场概况:)print(f 全市场平均涨幅:{result[avg_change]:.2f}%)print(f 涨停:{len(result[limit_up])}只)print(f 放量上涨:{len(result[strong])}只)print(f 大跌预警:{len(result[big_drop])}只)iflen(result[limit_up])0:print( --- 涨停股 ---)for_,rowinresult[limit_up].head(5).iterrows():print(f{row[symbol]}:{row[last_price]}({row[change_pct]:.2f}%))time.sleep(interval)run_monitor(interval30)四、方案二WebSocket 实时推送低延迟如果需要更低延迟推荐使用 WebSocket 方案。服务端主动推送行情变动无需轮询。1. 基本用法importdatetimefromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)streamtf.streamstream.on_quotesdefon_quotes(quotes):forqinquotes:tsdatetime.datetime.fromtimestamp(q[timestamp]/1000)extq.get(ext,{})nameext.get(name,)change_pctext.get(change_pct)change_strf{change_pct:.2%}ifchange_pctisnotNoneelseN/Aprint(f[{ts:%H:%M:%S}]{q[symbol]}{name}最新:{q[last_price]}涨跌:{change_str})stream.on_errordefon_error(msg):print(f错误:{msg})# 订阅感兴趣的标的stream.subscribe(quotes,[600519.SH,000001.SZ,600000.SH])stream.connect()# 阻塞运行输出示例[09:31:05] 600519.SH 贵州茅台 最新:1580.00 涨跌:0.82% [09:31:05] 000001.SZ 平安银行 最新:11.12 涨跌:0.54% [09:31:06] 600000.SH 浦发银行 最新:9.89 涨跌:-0.10%2. 带筛选条件的实时监控importdatetimefromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)streamtf.stream alert_threshold5.0# 涨幅超过 5% 时提醒stream.on_quotesdefon_quotes(quotes):forqinquotes:extq.get(ext,{})change_pctext.get(change_pct)ifchange_pctisNone:continuechange_pct_valchange_pct*100ifabs(change_pct_val)alert_threshold:nameext.get(name,)tsdatetime.datetime.fromtimestamp(q[timestamp]/1000)direction大涨ifchange_pct_val0else大跌print(f[{ts:%H:%M:%S}]{direction}{q[symbol]}{name}f最新:{q[last_price]}涨跌:{change_pct_val:.2f}%)symbols[600519.SH,000858.SZ,601318.SH,000001.SZ,600036.SH]stream.subscribe(quotes,symbols)stream.connect()3. 非阻塞模式 动态订阅importtimefromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)streamtf.stream watched{}stream.on_quotesdefon_quotes(quotes):forqinquotes:watched[q[symbol]]q# 后台运行 WebSocketstream.subscribe(quotes,[600519.SH,000001.SZ])stream.connect(blockFalse)# 主线程可以做其他事time.sleep(5)# 动态追加订阅stream.subscribe(quotes,[600036.SH,601318.SH])# 定时打印最新行情for_inrange(10):time.sleep(3)forsymbol,qinwatched.items():extq.get(ext,{})print(f{symbol}:{q[last_price]}({ext.get(name,)}))print(---)stream.close()4. 异步 WebSocketimportasyncioimportdatetimefromtickflowimportAsyncTickFlowasyncdefmain():asyncwithAsyncTickFlow(api_keyyour-api-key)astf:streamtf.streamstream.on_quotesdefon_quotes(quotes):forqinquotes:tsdatetime.datetime.fromtimestamp(q[timestamp]/1000)extq.get(ext,{})nameext.get(name,)print(f[{ts:%H:%M:%S}]{q[symbol]}{name}:{q[last_price]})awaitstream.subscribe(quotes,[600519.SH,000001.SZ])awaitstream.connect()asyncio.run(main())五、进阶结合日内 K 线做实时选股不仅可以看行情快照还可以结合日内分钟 K 线做更精细的分析日内 VWAP 偏离度筛选fromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)symbols[600519.SH,000858.SZ,601318.SH,000001.SZ,600036.SH]dfstf.klines.intraday_batch(symbols,period5m,as_dataframeTrue,show_progressTrue)forsymbol,dfindfs.items():iflen(df)0:continuedf[vwap](df[amount]/df[volume]).round(2)latestdf.iloc[-1]vwap_dev(latest[close]-latest[vwap])/latest[vwap]*100status高于VWAPifvwap_dev0else低于VWAPprint(f{symbol}: 最新价{latest[close]}, VWAP{latest[vwap]},{status}{abs(vwap_dev):.2f}%)分钟级均线突破监控fromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)# 获取日内 5 分钟 K 线dftf.klines.intraday(600519.SH,period5m,as_dataframeTrue)df[ma10]df[close].rolling(10).mean()df[ma30]df[close].rolling(30).mean()# 检查最新是否金叉iflen(df)30:latestdf.iloc[-1]prevdf.iloc[-2]iflatest[ma10]latest[ma30]andprev[ma10]prev[ma30]:print(f600519.SH 日内 5 分钟级别 MA10 上穿 MA30关注)else:print(f600519.SH 日内暂无均线交叉信号)六、ETF 行情监控除了个股ETF 也是很好的监控对象fromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)# 获取全部 ETF 行情etf_quotestf.quotes.get(universes[CN_ETF],as_dataframeTrue)etf_quotes[change_pct]((etf_quotes[last_price]-etf_quotes[prev_close])/etf_quotes[prev_close]*100)# 成交额最大的 ETF通常是宽基 ETFtop_volumeetf_quotes.nlargest(20,amount)print(成交额 Top 20 ETF:)print(top_volume[[symbol,last_price,change_pct,amount]].to_string(indexFalse))# 涨幅最大的 ETF发现热门板块top_changeetf_quotes.nlargest(10,change_pct)print(\n涨幅 Top 10 ETF:)print(top_change[[symbol,last_price,change_pct]].to_string(indexFalse))通过 ETF 涨幅排行可以快速发现当天最热门的板块方向。七、五档盘口数据如果需要更细粒度的分析TickFlow 还提供五档买卖盘口需 Pro/Expert 套餐fromtickflowimportTickFlow tfTickFlow(api_keyyour-api-key)depthtf.depth.get(600519.SH)print(f标的:{depth[symbol]})foriinrange(5):bid_pricedepth[bid_prices][i]bid_voldepth[bid_volumes][i]ask_pricedepth[ask_prices][i]ask_voldepth[ask_volumes][i]print(f 卖{5-i}:{ask_price}x{ask_vol}| 买{i1}:{bid_price}x{bid_vol})八、两种方案对比维度REST 轮询WebSocket 推送延迟取决于轮询间隔秒级毫秒级实现难度简单稍复杂适用场景选股扫描、定时监控盯盘、日内交易资源消耗每次请求有网络开销长连接效率更高全市场支持一次拉取全部行情按标的订阅建议全市场扫描选股- 用 REST一次拿全部数据做筛选自选股盯盘- 用 WebSocket延迟更低两者结合- REST 做定时全市场扫描WebSocket 订阅筛选出的标的做实时监控九、总结本文介绍了两种 Python 实时监控 A 股行情的方案REST 轮询一行代码拿到全市场 5000 只 A 股行情适合定时扫描选股WebSocket 推送毫秒级延迟适合盯盘和日内交易核心功能包括涨跌幅排行涨停股监控放量上涨筛选ETF 板块监控日内 VWAP 偏离分析分钟级均线突破信号TickFlow 提供了统一的 API无论是 REST 还是 WebSocketSDK 都做了良好的封装几行代码即可实现复杂的监控逻辑。相关链接官网https://tickflow.org文档https://docs.tickflow.orgGithubhttps://github.com/tickflow-org/tickflow实盘交易的第一步不是下单而是看到实时的市场。