Python实时盯盘与预警
【Python实时盯盘与预警 #04】炸板预警，刚封涨停又开板的票别错过（按炸板次数+修复状态排序）
系列：Python实时盯盘与预警 · 第 04 篇 · 适合初学 · 7 分钟读完即可跑通
痛点开场
03 把"涨停池"拉出来后，我们立刻会关心它的反面——今天有没有"封了又开"的票？哪些在剧烈分歧？哪些正在修复、哪些已经沉沦？。交易终端的"炸板股"模块只给一个名单，没有"修复力度"维度——你不知道一只"已回封"的票和一只"远离 13%"的票差多少倍。
本文用一份证书（任一品牌免费版就够），把炸板股池按"炸板次数 + 现价距涨停价"两轴排序，盘中 60 秒一次轮询——"刚开板"的票"立刻在屏幕弹出"，比你刷交易终端快 5~10 分钟。
本文你将得到什么
- 一段 90 行的 Python，单文件可跑，零外部依赖。
- 一张"炸板次数 → 距涨停% → 状态"复合排序的炸板快照表（30+ 行 1 屏，盘中 60 秒刷一次）。
- 4 张子表：炸板次数分布 / 状态分布（已回封/修复中/远离/沉沦）/ 行业分布。
- 一份"盘中持续盯盘 30 秒接入"骨架（结合 #01 30 只自选 + #03 涨停池 + #04 炸板 = 完整盘中雷达）。
环境准备（5 分钟）
复用 #01~#03：conda 环境 python、mairui 1.0.0、证书只读不打印。
完整可运行脚本
自验环境：
python/ Python 3.9 / mairui 1.0.0；默认查 2026-07-31（最近一个完整交易日 107 条）已 PASS 跑通；盘中跑今天（TARGET_DATE改今天）建议等 10:00 之后再启动。
"""【Python实时盯盘与预警 #04】炸板预警,刚封涨停又开板的票别错过"""
from __future__ import annotations
import os
import sys
from datetime import datetime
# 证书加载器(只读不打印，从环境变量读取)
from mairui import Client # noqa: E402
# 默认看最近一个完整交易日(07-31 是上一个完整交易日,107 条样本最具代表性)
# 盘中:把 TARGET_DATE 改成今天即可;开盘前 30 分钟(09:30 之前)池子可能为空
TARGET_DATE = os.environ.get("TARGET_DATE") or "2026-07-31"
def fetch_broken(api: Client, date: str) -> list[dict]:
"""拉炸板股池。返回原始 list(键为拼音首字母:dm/mc/p/ztp/zbc/fbt/...)."""
raw = api.broken_board_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 status_label(p, ztp):
"""根据「现价距涨停价」给一个状态标签(快速判断修复力度)。"""
if p is None or ztp is None or ztp <= 0:
return "数据缺"
diff_pct = (ztp - p) / ztp * 100.0
if diff_pct <= 0.5:
return "已回封"
if diff_pct <= 3.0:
return "修复中"
if diff_pct <= 6.0:
return "远离"
return "沉沦"
def render(rows):
if not rows:
return f"[{TARGET_DATE}] 当日无炸板股(全市场封板都很稳,或接口还没拉到数据)。"
# 排序:炸板次数降序 → 现价距涨停价升序(越接近=越强修复) → 封板时间升序
def key(r):
dist_pct = ((r.get("ztp") or 0) - (r.get("p") or 0)) / (r.get("ztp") or 1) * 100
return (
-(r.get("zbc") or 0),
dist_pct,
fmt_time(r.get("fbt")) or "99:99:99",
)
rows.sort(key=key)
lines = [
f"【炸板预警】日期 {TARGET_DATE} 共 {len(rows)} 只 (封了又开/多次开板的高分歧票)",
f"{'排名':>3} {'代码':<7} {'名称':<8} {'炸板':>4} {'统计':<6} {'首次封板':<9} {'现价':>7} {'涨停价':>7} {'距涨停%':>8} {'状态':<6} {'行业':<6}",
]
for i, r in enumerate(rows, 1):
p = r.get("p") or 0
ztp = r.get("ztp") or 0
dist_pct = (ztp - p) / ztp * 100 if ztp else 0
lines.append(
f"{i:>3} {str(r.get('dm') or ''):<7} {str(r.get('mc') or ''):<8} "
f"{r.get('zbc') or 0:>4} {str(r.get('tj') or '-'):<6} "
f"{fmt_time(r.get('fbt')):<9} {p:>7.2f} {ztp:>7.2f} "
f"{dist_pct:>7.2f}% {status_label(p, ztp):<6} {str(r.get('hy') or '-'):<6}"
)
return "\n".join(lines)
def render_stats(rows):
"""关键统计:炸板梯队 / 状态分布 / 行业分布"""
out = []
# 炸板次数分布
zbc_counter = {}
for r in rows:
n = r.get("zbc") or 0
zbc_counter[n] = zbc_counter.get(n, 0) + 1
out.append("【炸板次数分布】")
for n in sorted(zbc_counter.keys(), reverse=True):
out.append(f" 炸板 {n} 次: {zbc_counter[n]} 只")
# 状态分布
status_counter = {}
for r in rows:
s = status_label(r.get("p"), r.get("ztp"))
status_counter[s] = status_counter.get(s, 0) + 1
out.append("")
out.append("【状态分布】(现价距涨停价的修复力度)")
for s, cnt in sorted(status_counter.items(), key=lambda x: -x[1]):
out.append(f" {s}: {cnt} 只")
# 行业分布
hy_counter = {}
for r in rows:
hy = r.get("hy") or "未知"
hy_counter[hy] = hy_counter.get(hy, 0) + 1
out.append("")
out.append("【行业分布】")
for hy, cnt in sorted(hy_counter.items(), key=lambda x: -x[1])[:15]:
out.append(f" {hy}: {cnt} 只")
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_broken(api, TARGET_DATE)
print(render(rows))
print()
print(render_stats(rows))
return 0
if __name__ == "__main__":
sys.exit(main())
真实运行输出（截取 2026-08-03 09:56 跑 TARGET_DATE=2026-07-31）
[info] licence present: True (value hidden, len=36)
[info] target_date=2026-07-31 current=2026-08-03 09:56:17
【炸板预警】日期 2026-07-31 共 107 只 (封了又开/多次开板的高分歧票)
排名 代码 名称 炸板 统计 首次封板 现价 涨停价 距涨停% 状态 行业
1 002787 华源控股 22 0/0 09:30:03 19.47 19.73 1.32% 修复中 包装印刷
2 600183 生益科技 17 0/0 09:30:02 105.64 110.09 4.04% 远离 元件
3 688661 和林微纳 15 0/0 09:41:09 81.28 82.14 1.05% 修复中 半导体
4 002484 江海股份 15 0/0 09:39:51 58.71 61.05 3.83% 远离 元件
5 603306 华懋科技 15 0/0 09:47:09 57.88 60.49 4.31% 远离 汽车零部
6 603061 金海通 14 0/0 09:48:52 263.39 268.38 1.86% 修复中 半导体
7 002655 共达电声 14 0/0 09:56:48 22.85 23.42 2.43% 修复中 消费电子
8 002281 光迅科技 14 0/0 09:49:09 161.65 167.56 3.53% 远离 通信设备
9 002916 深南电路 13 0/0 09:40:27 308.00 312.95 1.58% 修复中 元件
10 002353 杰瑞股份 13 0/0 10:06:24 134.70 138.30 2.60% 修复中 专用设备
…
71 300795 米奥会展 1 0/0 10:59:54 16.44 16.44 0.00% 已回封 专业服务
…
107 688766 普冉股份 1 0/0 09:25:01 354.90 423.74 16.25% 沉沦 半导体
【炸板次数分布】
炸板 22 次: 1 只
炸板 17 次: 1 只
炸板 15 次: 3 只
炸板 14 次: 3 只
炸板 13 次: 2 只
炸板 10 次: 3 只
炸板 9 次: 3 只
炸板 7 次: 4 只
炸板 6 次: 3 只
炸板 5 次: 4 只
炸板 4 次: 10 只
炸板 3 次: 13 只
炸板 2 次: 20 只
炸板 1 次: 37 只
【状态分布】(现价距涨停价的修复力度)
修复中: 41 只
远离: 38 只
沉沦: 26 只
已回封: 2 只
【行业分布】
元件: 15 只
半导体: 8 只
汽车零部: 7 只
软件开发: 5 只
自动化设: 5 只
塑料: 5 只
专用设备: 4 只
光学光电: 4 只
消费电子: 3 只
通信设备: 3 只
广告营销: 3 只
计算机设: 3 只
通用设备: 3 只
金属新材: 3 只
包装印刷: 2 只
09:56 跑今天（
TARGET_DATE=2026-08-03）会返回空数组——09:30 开盘才 26 分钟，炸板池要 10:00 之后才陆续有数据；本脚本默认回看 07-31（最近一个完整交易日 107 条）让"完整样本"立刻可看。盘中把TARGET_DATE改今天即可。
代码要点拆解
- 排序三轴：zbc → 距涨停% → 首次封板时间——炸板次数是"分歧强度"（22 次的华源控股是"市场抛弃型"），距涨停价是"修复力度"（修复中的票是"还有救"），时间轴是"先发"（越早封板的越强）。换排序轴：把 key 第一项改成
r.get("hs") or 0就能直接按"换手率"找异动。 - 状态标签用现价距涨停价自动归类：≤0.5% = 已回封 / ≤3% = 修复中 / ≤6% = 远离 / >6% = 沉沦。这是个经验阈值——实战中可以按自己的风险偏好调宽窄。
zbc字段是核心：和 #03 涨停池的"连板数"不同，炸板池的 zbc 永远 >= 1（"没炸板"就不会进炸板池）。zbc >= 5 的是"剧烈分歧"票、zbc >= 10 的是"市场用脚投票"票，盘中盯 zbc >= 3 的就足够。- 字段是拼音首字母键（
dm/mc/p/ztp/zbc/fbt/tj/hy），与 #03 完全一致——读者照搬代码不踩坑。 - 盘中 09:30 之前查会返回 0——这是数据生成时机问题（开盘才有"封板/炸板"），不是脚本问题。脚本里
if not rows那段空数据保护就是为这一刻写的。
接入盘中实时（30 秒接入）
import time
def loop():
while True:
with Client(licence="<你的证书>") as api: # 演示证书，请替换成你自己的正式证书
rows = fetch_broken(api, datetime.now().strftime("%Y-%m-%d"))
os.system("cls" if os.name == "nt" else "clear")
print(render(rows))
print()
print(render_stats(rows))
time.sleep(60) # 炸板池 60 秒够,不用像 #01 自选股那么高频
想"刚开板就桌面通知"：把
status_label(p, ztp) == "沉沦"替换成自己的告警条件，再接 #02 的win10toast / plyer三级降级即可。
常见坑
- 盘中 09:30 之前查 = 空数组：开盘前的池子还没生成。脚本里"当日无炸板股"的兜底文案就是这个场景。
tj字段含义多版本：老版本是"连板数/封板次数"，新版本是"已封次数/已开次数"——本脚本不依赖tj做判定，核心只用zbc和ztp/p（这两个字段所有版本都稳）。- 盘后跑当天数据是"盘后快照"：那时大部分炸板票已经远离涨停价甚至沉沦，不会"修复中"了——要看盘中变化就盘中跑。
- 不同接口版本字段名可能略变：如果你拿到的字段名是
zdf而不是zf、或fbtime而不是fbt，把本脚本里的字段名替换一下即可，逻辑不变。
小结 + 下篇预告
107 只炸板 + 行业 Top3（元件/半导体/汽车零部）= "科技板块当天整体大分歧"——这正是 #05 五档盘口读语言的实战背景：板块分歧剧烈时，五档盘口的大单动向就是"谁在接力、谁在出货"的关键证据。下一篇 #05 我们用 stock_real_five 把"封板瞬间的五档委托"拉出来，读"封单 vs 砸单"。
合规与引导
数据/接口演示仅作技术示意，不构成投资建议；本系列代码仅用于自验与教学，请勿用于非法牟利。炸板池属于"按日切片"接口，盘后 1 次/日 + 盘中 60 秒重跑，免费版每日 500 次额度足够盘中持续盯盘；套餐与价格参见 https://www.mairuiapi.com 。