---

你照着前面的文章搭了个涨跌幅告警——涨 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 = "LICENCE-66D8-9F96-0C7F0FBCD073"  # 演示证书，演示证书（读者复制即用，无需替换）
    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

统计: 样本 100 只 | Level-1命中 10 | 最终告警 3 | API调用 15次 (L1=5+L2=10) | 耗时 0.5s
额度: 15/500 (3.0%) | 可跑 33 轮/日

关键数据解读：

1. Level-1 过滤掉 90%：100 只中只有 10 只同时满足"涨跌幅>3% + 成交额>5000万"。这一步只花了 5 次 API
2. Level-2 再砍掉 70%：10 只中只有 3 只同时满足"换手率>3% + 量比>1.5"。这一步花了 10 次 API
3. 最终 3 只异动股：兴业银锡（量比 7.16 最高）、招金黄金（换手率 5.77% 最高）、新金路（涨跌幅 +3.97%）
4. 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。

---

常见坑

1. Level-1 条件太松：如果 Level-1 只设 zf > 0.5%，可能 80 只都命中，Level-2 要调 80 次单股接口——两级过滤白做了。Level-1 的目标是把命中数砍到 10-20 只以内
2. Level-2 条件太严：4 个条件全设 > 且阈值太高，可能一只都不命中。建议先用宽松阈值跑一轮看看分布，再逐步收紧
3. 字段名写错不报错：check_condition 遇到 None 或非数字值会返回 False，不会报错。如果某条规则永远不命中，检查字段名是否拼对（参考上文字段表）
4. 成交额 cje 单位是元：不是万元。5000 万 = 50_000_000
5. 量比 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——每轮扫描结果自动入库，一天跑下来积累几百条记录，盘后可以回放、统计、做策略复盘。一天异动全留痕，不漏不丢。

---

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

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