【Python实时盯盘与预警 #09】涨了3%就报警太吵？4个条件任意组合的异动雷达

你照着前面的文章搭了个涨跌幅告警——涨 3% 就响。结果开盘半小时，手机响了 10 次。打开一看：有的票成交额才 200 万，有的换手率 0.1%，纯碎是低流通盘的小票随便跳两下就 3% 了。

问题出在：单一条件太粗暴。真正的异动，从来不是一个指标说了算的——涨了 3% 且成交额过 5000 万且换手率超 3% 且量比大于 1.5，这才是"有资金真在干活"的异动。

但问题来了：换手率、量比这些字段，批量接口 stock_ssjy_more 不返回。100 只股票逐只调 stock_ssjy 拿完整字段？那 100 次额度没了，免费版一天只能跑 5 轮。

这篇教你两级过滤法：Level-1 用批量接口快速筛掉 90% 不达标的票（5 次 API），Level-2 只对命中的几只调单股接口精筛（约 10 次 API）。100 只股票扫一遍，总共 15 次额度。

本文你将得到什么

一个多条件组合的异动雷达脚本：4 个指标任意 AND 组合，改配置即生效
两级过滤策略：批量接口粗筛 + 单股接口精筛，省 86% 额度
通用条件判断引擎：> >= < <= == 五种运算符，字段名写错自动跳过
Level-1 / Level-2 规则拆分设计思路：哪些条件放批量层、哪些放精筛层

为什么需要两级过滤

先看两个接口的字段差异：

字段 stock_ssjy_more（批量，≤20只/次） stock_ssjy（单股，1只/次）
涨跌幅 zf ✅ ✅
成交额 cje ✅ ✅
振幅 tr ✅ ✅
市盈率 pe ✅ ✅
换手率 hs ❌ ✅
量比 lb ❌ ✅
涨速 zs ❌ ✅
流通市值 lt ❌ ✅

批量接口快但字段少，单股接口全但贵。两级过滤的核心思路：

Level-1：批量拉 100 只（5 次 API）
→ 用 zf/cje 等"批量字段"过滤
→ 假设命中 10 只

Level-2：只对这 10 只调单股接口（10 次 API）
→ 用 hs/lb/zs 等"单股字段"过滤
→ 最终命中 3 只

总计 15 次 API → 免费版可跑 33 轮/日

对比"全量逐只调单股接口"：100 次 × 1 = 100 次，只能跑 5 轮。两级过滤省了 86% 额度。

代码：异动雷达

import sys
import os
import time

证书加载（只读不打印）

sys.path.insert(0, os.path.join(os.path.dirname(file), "..", "..", "_common"))
from mairui_helper import load_licence
from mairui import Client

===== 配置区 =====

SAMPLE_SIZE = 100 # 扫描股票数
BATCH_SIZE = 20 # 批量接口上限

Level-1 条件（基于 stock_ssjy_more 批量字段，省额度）

字段: zf=涨跌幅% cje=成交额(元) tr=振幅% pe=市盈率

LEVEL1_RULES = [
{"field": "zf", "op": ">", "value": 3.0, "desc": "涨跌幅 > 3%"},
{"field": "cje", "op": ">", "value": 50_000_000, "desc": "成交额 > 5000万"},
]

Level-2 条件（基于 stock_ssjy 单股字段，需逐只调用）

字段: hs=换手率% lb=量比 zs=涨速%

LEVEL2_RULES = [
{"field": "hs", "op": ">", "value": 3.0, "desc": "换手率 > 3%"},
{"field": "lb", "op": ">", "value": 1.5, "desc": "量比 > 1.5"},
]

def check_condition(value, op, target):
"""通用条件判断。"""
if value is None:
return False
try:
v = float(value)
except (TypeError, ValueError):
return False
if op == ">":
return v > target
elif op == ">=":
return v >= target
elif op == "<":
return v < target
elif op == "<=":
return v <= target
elif op == "==":
return abs(v - target) < 1e-6
return False

def filter_rules(data, rules):
"""对一条数据应用所有规则（AND 逻辑）。"""
hits = []
all_pass = True
for rule in rules:
val = data.get(rule["field"])
ok = check_condition(val, rule["op"], rule["value"])
hits.append({"rule": rule["desc"], "field": rule["field"],
"actual": val, "pass": ok})
if not ok:
all_pass = False
return all_pass, hits

def fetch_batch(mc, codes):
"""分批拉取实时行情，返回 {code: data}。"""
result = {}
for i in range(0, len(codes), BATCH_SIZE):
batch = codes[i:i + BATCH_SIZE]
items = mc.stock_ssjy_more(",".join(batch))
for item in items:
result[item["dm"]] = item
return result

def main():
lic = load_licence()
mc = Client(lic)

1. 获取股票列表

all_stocks = mc.stock_list()
stocks = all_stocks[:SAMPLE_SIZE]
code_to_name = {}
codes = []
for s in stocks:
raw_code = s["dm"].split(".")[0]
codes.append(raw_code)
code_to_name[raw_code] = s["mc"]

print("=" * 70)
print("异动雷达 — 多条件组合扫描")
print("=" * 70)
print(f"扫描样本: {len(codes)} 只")
print(f"\nLevel-1 条件（批量过滤，省额度）😊
for r in LEVEL1_RULES:
print(f" - {r['desc']}")
print(f"\nLevel-2 条件（逐只精筛）😊
for r in LEVEL2_RULES:
print(f" - {r['desc']}")
print()

2. Level-1: 批量拉取 + 过滤

t0 = time.time()
snap = fetch_batch(mc, codes)
api_calls_l1 = (len(codes) + BATCH_SIZE - 1) // BATCH_SIZE
print(f"[Level-1] 批量拉取 {len(snap)} 只 ({api_calls_l1} 次API)")

l1_hits = []
for code in codes:
data = snap.get(code)
if not data:
continue
ok, _ = filter_rules(data, LEVEL1_RULES)
if ok:
l1_hits.append(code)

print(f"[Level-1] 命中 {len(l1_hits)} 只: {', '.join(l1_hits[:10])}")
print()

3. Level-2: 逐只精筛

api_calls_l2 = 0
final_alerts = []
for code in l1_hits:
detail = mc.stock_ssjy(code)
api_calls_l2 += 1
if not detail:
continue
ok, hits = filter_rules(detail, LEVEL2_RULES)
if ok:
final_alerts.append({
"code": code,
"name": code_to_name.get(code, ""),
"p": float(detail.get("p", 0)),
"zf": float(detail.get("zf", 0)),
"cje": float(detail.get("cje", 0)),
"hs": float(detail.get("hs", 0)),
"lb": float(detail.get("lb", 0)),
"zs": float(detail.get("zs", 0)),
"tr": float(detail.get("tr", 0)),
})

total_api = api_calls_l1 + api_calls_l2
elapsed = time.time() - t0

4. 输出告警

print(f"[Level-2] 逐只精筛 {len(l1_hits)} 只 ({api_calls_l2} 次API)")
print(f"[最终告警] {len(final_alerts)} 只命中全部条件\n")

if final_alerts:
print(f"{'代码':<8} {'名称':<8} {'现价':>8} {'涨跌幅%':>8} "
f"{'成交额(万)':>12} {'换手率%':>8} {'量比':>6} {'涨速%':>8} {'振幅%':>8}")
print("-" * 90)
for a in final_alerts:
cje_wan = a["cje"] / 10000
print(f"{a['code']:<8} {a['name']:<8} {a['p']:>8.2f} {a['zf']:>+8.2f} "
f"{cje_wan:>12.0f} {a['hs']:>8.2f} {a['lb']:>6.2f} "
f"{a['zs']:>+8.3f} {a['tr']:>8.2f}")

print()
print(f"统计: 样本 {len(codes)} 只 | Level-1命中 {len(l1_hits)} | "
f"最终告警 {len(final_alerts)} | "
f"API调用 {total_api}次 (L1={api_calls_l1}+L2={api_calls_l2}) | "
f"耗时 {elapsed:.1f}s")
print(f"额度: {total_api}/500 ({total_api/500*100:.1f}%) | "
f"可跑 {500//total_api} 轮/日")

if name == "main":
main()

实测结果（2026-08-06 09:55，真实运行）

======================================================================
异动雷达 — 多条件组合扫描
======================================================================
扫描样本: 100 只

Level-1 条件（批量过滤，省额度）:
- 涨跌幅 > 3%
- 成交额 > 5000万

Level-2 条件（逐只精筛）:
- 换手率 > 3%
- 量比 > 1.5

[Level-1] 批量拉取 100 只 (5 次API)
[Level-1] 命中 10 只: 000021, 000034, 000035, 000048, 000065, 000069, 000426, 000428, 000506, 000510

[Level-2] 逐只精筛 10 只 (10 次API)
[最终告警] 3 只命中全部条件

代码名称现价涨跌幅% 成交额(万) 换手率% 量比涨速% 振幅%
------------------------------------------------------------------------------------------
000426 兴业银锡 38.47 +4.63 210132 3.11 7.16 -0.340 0.00
000506 招金黄金 16.44 +3.46 88458 5.77 4.79 -0.180 0.00
000510 新金路 16.35 +3.97 53935 5.44 6.23 -0.300 0.00

关键数据解读：

Level-1 过滤掉 90%：100 只中只有 10 只同时满足"涨跌幅>3% + 成交额>5000万"。这一步只花了 5 次 API
Level-2 再砍掉 70%：10 只中只有 3 只同时满足"换手率>3% + 量比>1.5"。这一步花了 10 次 API
最终 3 只异动股：兴业银锡（量比 7.16 最高）、招金黄金（换手率 5.77% 最高）、新金路（涨跌幅 +3.97%）
15 次额度 = 3%：免费版一天能跑 33 轮，每 15 分钟一轮能覆盖整个交易日

怎么自定义你的条件

改配置区的 LEVEL1_RULES 和 LEVEL2_RULES 就行。规则结构：

{"field": "字段名", "op": "运算符", "value": 阈值, "desc": "描述"}

运算符支持：> >= < <= ==

常用条件模板

激进型：抓异动龙头

LEVEL1_RULES = [
{"field": "zf", "op": ">", "value": 5.0},
{"field": "cje", "op": ">", "value": 100_000_000}, # 1亿
]
LEVEL2_RULES = [
{"field": "lb", "op": ">", "value": 2.0},
{"field": "zs", "op": ">", "value": 0.3},
]

保守型：找温和放量

LEVEL1_RULES = [
{"field": "zf", "op": ">", "value": 1.0},
{"field": "cje", "op": ">", "value": 30_000_000}, # 3000万
{"field": "tr", "op": ">", "value": 2.0}, # 振幅 > 2%
]
LEVEL2_RULES = [
{"field": "hs", "op": ">", "value": 1.0},
{"field": "lb", "op": ">", "value": 1.0},
]

跌幅异动：找急跌的票

LEVEL1_RULES = [
{"field": "zf", "op": "<", "value": -3.0}, # 跌超 3%
{"field": "cje", "op": ">", "value": 50_000_000},
]
LEVEL2_RULES = [
{"field": "lb", "op": ">", "value": 1.5}, # 放量下跌
]

条件分层的原则

放 Level-1（批量层）放 Level-2（精筛层）
涨跌幅 zf、成交额 cje、振幅 tr、市盈率 pe、市净率 pb_ratio 换手率 hs、量比 lb、涨速 zs、流通市值 lt、总市值 sz
批量接口有的字段只有单股接口才有的字段

原则：能把 90% 的票筛掉的条件放 Level-1，精确定位的条件放 Level-2。

常见坑

Level-1 条件太松：如果 Level-1 只设 zf > 0.5%，可能 80 只都命中，Level-2 要调 80 次单股接口——两级过滤白做了。Level-1 的目标是把命中数砍到 10-20 只以内
Level-2 条件太严：4 个条件全设 > 且阈值太高，可能一只都不命中。建议先用宽松阈值跑一轮看看分布，再逐步收紧
字段名写错不报错：check_condition 遇到 None 或非数字值会返回 False，不会报错。如果某条规则永远不命中，检查字段名是否拼对（参考上文字段表）
成交额 cje 单位是元：不是万元。5000 万 = 50_000_000
量比 lb 在开盘前 5 分钟不稳定：9:30-9:35 量比偏高是正常现象（分母小），建议 9:35 后再跑

小结

异动雷达的核心不是"多条件"——任何人都会写 if zf > 3 and cje > 5000万。核心是两级过滤的额度策略：用批量接口的 5 次调用砍掉 90% 的噪音，再只对剩下的几只调单股接口拿完整字段。

100 只股票、4 个条件、15 次额度。这就是"免费版也能做专业级异动监控"的关键。

加上前八篇的实时报价（#01）、阈值告警（#02）、涨停池（#03）、炸板预警（#04）、盘口分析（#05）、批量接口（#06）、涨速排行（#07）、成交额异动（#08），你的盯盘工具链已经很完整了。

下一篇我们把异动雷达的输出落进 SQLite——每轮扫描结果自动入库，一天跑下来积累几百条记录，盘后可以回放、统计、做策略复盘。一天异动全留痕，不漏不丢。

本文所用行情数据来自麦蕊智数 API（mairuiapi.com）。文中代码均已本地验证并可复现。证书免费领取地址： https://www.mairuiapi.com 。

⚠️ 免责声明：本文为技术分享，所有计算仅展示 Python 编程方法与 API 使用技巧，不构成任何投资建议。股市有风险，投资需谨慎。

📌 关注「Python实时盯盘与预警」合集（共 16 篇），从自选股报价到全市场监控、从免费版到企业证书方案——一站式盘中盯盘不求人。

资讯详情

【Python实时盯盘与预警 #09】涨了3%就报警太吵？4个条件任意组合的异动雷达