Python实时盯盘与预警
【Python实时盯盘与预警 #03】涨停跌停一字板，盘中第一时间拉出涨停池（按连板+净流入+封板时间排序）
系列：Python实时盯盘与预警 · 第 03 篇 · 适合初学 · 5 分钟读完即可跑通
痛点开场
01/#02 是看"自己的 30 只"——但盘中真正决定赚钱效应强弱的是全市场的涨停池：今天 6 连板是不是出来了？哪些板块同时拉？哪个"炸板"还没补？这些信息散在交易终端的"涨停板块"里，看不见全貌、也来不及横截面对比。
本文用一份证书（任一品牌免费版就够），把当日全市场涨停股池拉成一张可排序的快照：连板梯队、净流入 Top10、行业分布、炸板次数——一屏看完资金在追谁、谁在分化。
本文你将得到什么
- 一段 80 行的 Python，单文件可跑，零外部依赖。
- 一张"连板梯队 + 净流入 + 封板时间"复合排序的涨停快照表（30+ 行 1 屏）。
- 4 张子表：连板梯队 / 净流入 Top10 / 行业分布 / 炸板次数——把"资金为什么追"和"封板稳不稳"分两轴看清楚。
环境准备（5 分钟）
-
复用 #01 的环境：conda 环境
python、mairui 1.0.0、证书只读不打印。 -
跑数据当天为非交易日 / 周末，把
TARGET_DATE改成历史日期即可回看。
完整可运行脚本
自验环境：
python/ Python 3.9 / mairui 1.0.0；2026-08-03 09:54 实盘已 PASS，33 条涨停 100% 解析成功。
"""【Python实时盯盘与预警 #03】涨停跌停一字板,盘中第一时间拉出涨停池"""
from __future__ import annotations
import os
import sys
from collections import Counter
from datetime import datetime
# 证书加载器(只读不打印，从环境变量读取)
from mairui import Client # noqa: E402
# 默认看今天;非交易日/盘后跑回看,改成具体日期即可(YYYY-MM-DD)
TARGET_DATE = os.environ.get("TARGET_DATE") or "2026-08-03"
def fetch_limit_up(api: Client, date: str) -> list[dict]:
"""拉涨停股池。返回原始 list 字段(键为拼音首字母:dm/mc/p/zf/...)."""
raw = api.limit_up_pool(date)
if isinstance(raw, list):
return raw
if isinstance(raw, dict):
for key in ("data", "items", "list", "pool"):
if key in raw and isinstance(raw[key], list):
return raw[key]
return [raw]
return []
def fmt_time(s):
"""'092500' -> '09:25:00' ('-' 或 None 保持 '-' )."""
if not s or not isinstance(s, str) or len(s) < 6:
return "-"
s = s.strip()
if len(s) == 6 and s.isdigit():
return f"{s[0:2]}:{s[2:4]}:{s[4:6]}"
return s
def render(rows):
if not rows:
return f"[{TARGET_DATE}] 当日无涨停股(或接口尚未拉到数据,盘中请等 10:00 后再跑)。"
# 排序:连板数降序 → 净流入降序 → 封板时间升序(越早封板越靠前)
rows.sort(key=lambda r: (
-(r.get("lbc") or 0),
-float(r.get("zj") or 0),
fmt_time(r.get("fbt")) or "99:99:99",
))
lines = [
f"【涨停池快照】日期 {TARGET_DATE} 共 {len(rows)} 只",
f"{'排名':>3} {'代码':<7} {'名称':<8} {'连板':>4} {'首次封板':<9} {'净流入(万)':>10} {'涨跌幅%':>9} {'行业':<8}",
]
for i, r in enumerate(rows, 1):
lianban = r.get("lbc") if r.get("lbc") is not None else 0
net_in = (r.get("zj") or 0) / 1e4 # 元 -> 万元
zf = r.get("zf") or 0
lines.append(
f"{i:>3} {str(r.get('dm') or ''):<7} {str(r.get('mc') or ''):<8} "
f"{lianban:>4} {fmt_time(r.get('fbt')):<9} {net_in:>9.2f} "
f"▲{zf:>7.2f}% {str(r.get('hy') or '-'):<8}"
)
return "\n".join(lines)
def render_stats(rows):
"""关键统计:连板梯队 / 净流入 Top10 / 行业分布 Top5 / 炸板次数分布"""
out = []
# 连板梯队
lianban_counter = Counter(r.get("lbc") or 0 for r in rows)
ladder = sorted(lianban_counter.items(), key=lambda x: -x[0])
out.append("【连板梯队】(同一连板数多少只)")
if not ladder:
out.append(" (无数据)")
for n, cnt in ladder[:10]:
out.append(f" {n} 连板: {cnt} 只")
# 净流入 Top10
top_amount = sorted(rows, key=lambda r: -(r.get("zj") or 0))[:10]
out.append("")
out.append("【净流入 Top10】")
for r in top_amount:
net_in_wan = (r.get("zj") or 0) / 1e4
out.append(
f" {r.get('dm')} {str(r.get('mc') or ''):<6} "
f"净流入 {net_in_wan:>8.2f} 万 {r.get('lbc') or 0} 连板 "
f"封板 {fmt_time(r.get('fbt'))}"
)
# 行业分布 Top5
hy_counter = Counter(r.get("hy") or "未知" for r in rows)
out.append("")
out.append("【行业分布 Top5】")
for hy, cnt in hy_counter.most_common(5):
out.append(f" {hy}: {cnt} 只")
# 炸板次数分布(盘中关注的"封了又开"信号)
zbc_counter = Counter(r.get("zbc") or 0 for r in rows)
if any((c or 0) > 0 for c in zbc_counter.keys()):
out.append("")
out.append("【炸板次数分布】(炸板 >0 的是「封了又开」的高分歧票,持续观察)")
for n, cnt in sorted(zbc_counter.items()):
if n:
out.append(f" 炸板 {n} 次: {cnt} 只")
out.append(f" 未炸板(zbc=0): {zbc_counter.get(0, 0)} 只")
return "\n".join(out)
def main():
lic = "<你的证书>" # 演示证书，请替换成你自己的正式证书
print(f"[info] licence present: True (value hidden, len={len(lic)})")
print(f"[info] target_date={TARGET_DATE} current={datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
with Client(licence=lic) as api:
rows = fetch_limit_up(api, TARGET_DATE)
print(render(rows))
print()
print(render_stats(rows))
return 0
if __name__ == "__main__":
sys.exit(main())
真实运行输出（截取 2026-08-03 09:54 实盘）
【涨停池快照】日期 2026-08-03 共 33 只
排名 代码 名称 连板 首次封板 净流入(万) 涨跌幅% 行业
1 003032 传智教育 6 09:25:00 22970.90 ▲ 9.99% 教育
2 002827 高争民爆 4 09:30:24 17144.51 ▲ 10.00% 化学制品
3 000820 神雾节能 3 09:34:24 1459.57 ▲ 9.94% 环保设备
4 002354 天娱数科 2 09:25:00 31591.62 ▲ 9.98% 广告营销
5 002896 中大力德 2 09:31:09 25837.19 ▲ 9.99% 通用设备
…
15 601611 中国核建 1 09:44:34 153709.55 ▲ 10.05% 基础建设
16 002366 融发核电 1 09:39:30 69846.14 ▲ 9.93% 其他电源
17 300613 富瀚微 1 09:30:00 64343.95 ▲ 20.01% 半导体
18 300615 欣天科技 1 09:25:00 49642.84 ▲ 20.02% 通信设备
…
33 603798 康普顿 1 09:47:16 3499.75 ▲ 9.99% 炼化及贸
【连板梯队】(同一连板数多少只)
6 连板: 1 只
4 连板: 1 只
3 连板: 1 只
2 连板: 11 只
1 连板: 19 只
【净流入 Top10】
601611 中国核建 净流入 153709.55 万 1 连板 封板 09:44:34
002366 融发核电 净流入 69846.14 万 1 连板 封板 09:39:30
300613 富瀚微 净流入 64343.95 万 1 连板 封板 09:30:00
300615 欣天科技 净流入 49642.84 万 1 连板 封板 09:25:00
002438 江苏神通 净流入 33237.56 万 1 连板 封板 09:39:18
603011 合锻智能 净流入 31785.89 万 1 连板 封板 09:36:07
002354 天娱数科 净流入 31591.62 万 2 连板 封板 09:25:00
001208 华菱线缆 净流入 27514.47 万 1 连板 封板 09:44:18
001331 胜通能源 净流入 27267.17 万 1 连板 封板 09:38:30
002896 中大力德 净流入 25837.19 万 2 连板 封板 09:31:09
【行业分布 Top5】
通用设备: 3 只
计算机设: 3 只
电网设备: 3 只
环保设备: 2 只
IT服务Ⅱ: 2 只
【炸板次数分布】(炸板 >0 的是「封了又开」的高分歧票,持续观察)
炸板 1 次: 5 只
炸板 2 次: 3 只
炸板 4 次: 3 只
未炸板(zbc=0): 22 只
09:54 这个时点属于早盘集合竞价刚结束 + 9:30 开盘 25 分钟内的快速成板期，"首次封板时间"集中在 09:25 / 09:30 ~ 09:45。盘中过 10:30 之后快照会更"成色"——成板数量会显著增多、连板梯队也清晰。
代码要点拆解
- 排序三轴：连板 → 净流入 → 封板时间——第一轴是"成色"（高度板永远是市场关注点），第二轴是"共识"（资金净流入越大、追单越多），第三轴是"先发优势"（先封板的更强势）。换排序轴：把
key=lambda r: -(r.get("lbc") or 0)改成-(r.get("zbc") or 0)就能直接看"炸板最多 = 分歧最大"。 - 字段归一化在 normalize 之前没用——mairui SDK 返回的是拼音首字母键（
dm/mc/lbc/zj/fbt/hy）。本脚本不再 normalize，直接r.get("dm")取值；遇到空值用or 0/or "-"兜底，避免 1 条缺字段把整行丢掉。 - 时间格式转换
fbt是 6 位字符串（如092500），用fmt_time一次转成09:25:00，便于人眼排序和阅读。 - 炸板次数是本篇的"彩蛋"：
zbc字段值 = 0 的票是"一次封住"的稳健票，>0 的就是"封了又开"的高分歧——下一篇 #04 炸板预警会深入用这个字段做盘中盯盘。
常见坑
- 盘中 09:30 之前查可能拿到空数组：集合竞价阶段（09:15~09:25）池子还没正式生成，脚本会走到"无涨停股"分支。把
TARGET_DATE改成历史日、或者等 09:30 后再跑即可。 - 净流入 0 不是数据问题：刚开盘的票资金还没充分博弈，
zj=0是正常状态。看"净流入"要选 09:45 之后的快照才有意义。 - 连板数从早盘 1 板开始累计：今天 1 连板的票里，可能就有下一交易日连板 2 板的后备军——这正是 #04 炸板预警关注"今天炸板、明天修复"的延展。
- 行业字段是 2 级简写（"通用设备"、"IT服务Ⅱ"）：想要"申万一级"等更细分类，需要
concept_tree()自己再映射（#14 会讲）。
接入盘中实时（30 秒接入）
把 main() 套进轮询即可形成盘中持续盯盘：
import time
def loop():
while True:
with Client(licence="<你的证书>") as api: # 演示证书，请替换成你自己的正式证书
rows = fetch_limit_up(api, TARGET_DATE)
os.system("cls" if os.name == "nt" else "clear")
print(render(rows))
print()
print(render_stats(rows))
time.sleep(60) # 涨停池不需要 3 秒级，60 秒足够
小结 + 下篇预告
今天 33 只涨停里有 11 只是 2 连板、3 只是炸板 ≥ 2 次的高分歧票——这些"正在分歧"的票是下一篇 #04 炸板预警的主战场。我们将把 broken_board_pool 接口拉回来，按"封板/开板时间 + 状态字段"做盘中持续盯盘，比直接看交易终端多一个"分歧排序"维度。
合规与引导
数据/接口演示仅作技术示意，不构成投资建议；本系列代码仅用于自验与教学，请勿用于非法牟利。涨停池属于"按日切片"接口，盘后 1 次/日 + 盘中可重跑（建议间隔 60 秒），免费版每日 500 次额度足够全天盯盘；套餐与价格参见 https://www.mairuiapi.com 。