Python实时盯盘与预警
【Python实时盯盘与预警 #10】盘中异动看完就忘把盯盘结果落SQLite一天全留痕
你用前面几篇的脚本搭好了异动雷达，盘中每 5 分钟扫一轮，告警弹出来看一眼——然后呢？
然后就没有然后了。弹窗关掉、终端关掉，今天扫了几十轮、命中了哪些票、哪只票反复出现、哪个时段异动最多——全忘了。
盘后想做复盘？"今天好像有只票涨得挺猛的，叫什么来着？"——想不起来。
问题的根源：你的盯盘数据是即时的，没有落盘。 异动监控的价值不止"盘中发现"，还有"盘后回放"——哪些票持续异动、哪个时段集中爆发、今天的异动跟昨天比是什么趋势。这些都需要历史数据。
这篇教你用 Python 标准库 sqlite3（零额外依赖），把每轮扫描的异动结果自动入库，盘后一条 SQL 查出全部。
本文你将得到什么
- 一个完整的异动落库脚本：多轮扫描 + 自动写 SQLite + 4 种盘后查询
- SQLite 表结构设计：索引、字段、查询优化
- 4 个实用盘后查询：今日全部异动、个股触发频率、时段分布、每轮汇总
- 从"即时告警"到"历史回放"的工程思路：怎么把分散的弹窗变成可回溯的数据库
为什么选 SQLite
| 方案 | 优势 | 劣势 |
|---|---|---|
| CSV 文件 | 最简单，记事本可看 | 不能查询、不能统计、数据量大后难管理 |
| SQLite | 零依赖（Python 内置）、支持 SQL 查询、单文件易备份 | 并发写入有锁（盯盘场景单进程无影响） |
| MySQL/PostgreSQL | 高并发、分布式 | 需装服务、运维成本高（个人盯盘杀鸡用牛刀） |
盯盘场景是单进程写入、低频查询——SQLite 完美匹配。一个 .db 文件就是整个数据库，拷走就是备份，用 sqlite3 命令行随时打开查看。
表结构设计
CREATE TABLE anomaly_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
scan_time TEXT NOT NULL, -- 扫描时间（精确到秒）
scan_round INTEGER NOT NULL, -- 第几轮扫描
code TEXT NOT NULL, -- 股票代码
name TEXT, -- 股票名称
price REAL, -- 现价
zf REAL, -- 涨跌幅%
cje REAL, -- 成交额（元）
tr REAL, -- 振幅%
pe REAL -- 市盈率
);
CREATE INDEX idx_scan_time ON anomaly_log(scan_time);
CREATE INDEX idx_code ON anomaly_log(code);
设计要点：
scan_time用 TEXT 存 "2026-08-06 09:58:07" 格式——SQLite 的LIKE '2026-08-06%'可走索引查当日数据scan_round记录是第几轮扫描，方便按轮次统计- 两个索引：按时间查（盘后复盘）、按代码查（看某只票今天触发了几次）
- 不设唯一约束——同一只票在同一轮只入库一次，但在不同轮可以多次出现（这正是"持续异动"的证据）
代码：异动落库 + 盘后查询
import os
import time
import sqlite3
from datetime import datetime, date
# 证书加载（只读不打印）：环境变量注入，不引用内部 helper
from mairui import Client
# ===== 配置区 =====
SAMPLE_SIZE = 20 # 每轮扫描股票数
BATCH_SIZE = 20 # 批量接口上限
SCAN_ROUNDS = 3 # 扫描轮数（demo 模拟，实盘设 30+）
SCAN_INTERVAL = 5 # 每轮间隔秒数（demo 短间隔，实盘 300-600）
DB_PATH = os.path.join(os.path.dirname(__file__), "anomaly_log.db")
# 异动条件（只用批量字段，避免逐只调单股接口消耗额度）
ALERT_RULES = [
{"field": "zf", "op": ">", "value": 1.0},
{"field": "cje", "op": ">", "value": 30_000_000},
]
# ===== SQLite 操作 =====
def init_db(db_path):
"""初始化数据库，创建异动记录表。"""
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS anomaly_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
scan_time TEXT NOT NULL,
scan_round INTEGER NOT NULL,
code TEXT NOT NULL,
name TEXT,
price REAL,
zf REAL,
cje REAL,
tr REAL,
pe REAL
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_scan_time ON anomaly_log(scan_time)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_code ON anomaly_log(code)")
conn.commit()
return conn
def insert_alerts(conn, scan_round, alerts):
"""批量插入异动记录。"""
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rows = [(ts, scan_round, a["code"], a["name"], a["p"], a["zf"],
a["cje"], a["tr"], a.get("pe", 0)) for a in alerts]
conn.executemany("""
INSERT INTO anomaly_log (scan_time, scan_round, code, name, price, zf, cje, tr, pe)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", rows)
conn.commit()
return len(rows)
# ===== 条件过滤 =====
def check_condition(value, op, target):
if value is None:
return False
try:
v = float(value)
except (TypeError, ValueError):
return False
ops = {">": lambda a, b: a > b, ">=": lambda a, b: a >= b,
"<": lambda a, b: a < b, "<=": lambda a, b: a <= b,
"==": lambda a, b: abs(a - b) < 1e-6}
return ops.get(op, lambda a, b: False)(v, target)
def filter_alerts(data, rules):
for rule in rules:
if not check_condition(data.get(rule["field"]), rule["op"], rule["value"]):
return False
return True
# ===== 盘后查询 =====
def query_today_all(conn):
"""今日全部异动记录。"""
today = date.today().strftime("%Y-%m-%d")
cur = conn.execute("""
SELECT scan_time, code, name, price, zf, cje, tr
FROM anomaly_log WHERE scan_time LIKE ?
ORDER BY scan_time DESC, zf DESC
""", (f"{today}%",))
return cur.fetchall()
def query_stock_frequency(conn):
"""个股触发次数 Top 10。"""
today = date.today().strftime("%Y-%m-%d")
cur = conn.execute("""
SELECT code, name, COUNT(*) as freq,
MAX(zf) as max_zf, AVG(cje) as avg_cje
FROM anomaly_log WHERE scan_time LIKE ?
GROUP BY code, name
ORDER BY freq DESC, max_zf DESC LIMIT 10
""", (f"{today}%",))
return cur.fetchall()
def query_hourly_dist(conn):
"""异动按小时分布。"""
today = date.today().strftime("%Y-%m-%d")
cur = conn.execute("""
SELECT substr(scan_time, 12, 2) as hour,
COUNT(*) as count, COUNT(DISTINCT code) as unique_stocks
FROM anomaly_log WHERE scan_time LIKE ?
GROUP BY hour ORDER BY hour
""", (f"{today}%",))
return cur.fetchall()
def query_round_summary(conn):
"""每轮扫描命中数。"""
today = date.today().strftime("%Y-%m-%d")
cur = conn.execute("""
SELECT scan_round, MIN(scan_time), MAX(scan_time),
COUNT(*), COUNT(DISTINCT code)
FROM anomaly_log WHERE scan_time LIKE ?
GROUP BY scan_round ORDER BY scan_round
""", (f"{today}%",))
return cur.fetchall()
# ===== 主流程 =====
def main():
lic = "<你的证书>" # 演示证书，请替换成你自己的正式证书
mc = Client(lic)
conn = init_db(DB_PATH)
print(f"数据库: {DB_PATH}")
print(f"扫描配置: {SCAN_ROUNDS} 轮 x {SAMPLE_SIZE} 只, 间隔 {SCAN_INTERVAL}s")
print("=" * 70)
# 获取股票列表
all_stocks = mc.stock_list()
stocks = all_stocks[:SAMPLE_SIZE]
codes = [s["dm"].split(".")[0] for s in stocks]
code_to_name = {s["dm"].split(".")[0]: s["mc"] for s in stocks}
total_api = 0
# 多轮扫描
for rnd in range(1, SCAN_ROUNDS + 1):
t0 = time.time()
snap = {}
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:
snap[item["dm"]] = item
total_api += (len(codes) + BATCH_SIZE - 1) // BATCH_SIZE
alerts = []
for code in codes:
data = snap.get(code)
if data and filter_alerts(data, ALERT_RULES):
alerts.append({
"code": code,
"name": code_to_name.get(code, ""),
"p": float(data.get("p", 0)),
"zf": float(data.get("zf", 0)),
"cje": float(data.get("cje", 0)),
"tr": float(data.get("tr", 0)),
"pe": float(data.get("pe", 0)),
})
inserted = insert_alerts(conn, rnd, alerts)
elapsed = time.time() - t0
print(f"[第{rnd}轮] {datetime.now().strftime('%H:%M:%S')} | "
f"扫描{len(codes)}只 | 命中{len(alerts)}只 | 入库{inserted}条 | "
f"API 1次 | {elapsed:.1f}s")
if rnd < SCAN_ROUNDS:
time.sleep(SCAN_INTERVAL)
print("=" * 70)
print(f"扫描完成: {SCAN_ROUNDS}轮 | 总API {total_api}次 | "
f"额度 {total_api}/500 ({total_api/500*100:.1f}%)\n")
# 盘后查询
print("【1】今日全部异动记录")
rows = query_today_all(conn)
print(f"共 {len(rows)} 条，前 10 条：")
for r in rows[:10]:
ts, code, name, price, zf, cje, tr = r
print(f" {ts} {code} {name:<8} {price:>8.2f} {zf:>+6.2f}% {cje/10000:>10.0f}万")
print()
print("【2】个股触发次数 Top 10（多次命中 = 持续异动）")
for r in query_stock_frequency(conn):
code, name, freq, max_zf, avg_cje = r
print(f" {code} {name:<8} 触发{freq}次 最大涨跌幅{max_zf:>+6.2f}% "
f"平均成交额{avg_cje/10000:>10.0f}万")
print()
print("【3】异动时间段分布")
for r in query_hourly_dist(conn):
hour, count, unique = r
print(f" {hour}:00 异动{count}次 涉及{unique}只个股")
print()
print("【4】每轮扫描汇总")
for r in query_round_summary(conn):
rnd, first_t, last_t, cnt, unique = r
print(f" 第{rnd}轮 {first_t} 命中{cnt}条 {unique}只个股")
conn.close()
print(f"\n数据库已保存: {DB_PATH}")
if __name__ == "__main__":
main()
实测结果（2026-08-06 09:58，真实运行）
数据库: anomaly_log.db
扫描配置: 3 轮 x 20 只, 间隔 5s
======================================================================
[第1轮] 09:58:07 | 扫描20只 | 命中8只 | 入库8条 | API 1次 | 0.0s
[第2轮] 09:58:12 | 扫描20只 | 命中8只 | 入库8条 | API 1次 | 0.0s
[第3轮] 09:58:17 | 扫描20只 | 命中8只 | 入库8条 | API 1次 | 0.0s
======================================================================
扫描完成: 3轮 | 总API 3次 | 额度 3/500 (0.6%)
【1】今日全部异动记录
共 24 条，前 10 条：
2026-08-06 09:58:17 000021 深科技 39.62 +5.89% 197105万
2026-08-06 09:58:17 000026 飞亚达 17.24 +3.34% 4321万
2026-08-06 09:58:17 000025 特 力Ａ 15.28 +2.23% 4312万
2026-08-06 09:58:17 000009 中国宝安 7.58 +2.23% 10489万
2026-08-06 09:58:17 000002 万 科Ａ 3.24 +1.83% 7550万
...
【2】个股触发次数 Top 10（多次命中 = 持续异动）
000021 深科技 触发3次 最大涨跌幅+5.89% 平均成交额 196487万
000026 飞亚达 触发3次 最大涨跌幅+3.34% 平均成交额 4318万
000025 特 力Ａ 触发3次 最大涨跌幅+2.23% 平均成交额 4311万
000009 中国宝安 触发3次 最大涨跌幅+2.23% 平均成交额 10477万
...
【3】异动时间段分布
09:00 异动24次 涉及8只个股
【4】每轮扫描汇总
第1轮 2026-08-06 09:58:07 命中8条 8只个股
第2轮 2026-08-06 09:58:12 命中8条 8只个股
第3轮 2026-08-06 09:58:17 命中8条 8只个股
数据库已保存: anomaly_log.db
关键数据解读：
- 3 轮扫描共 24 条记录入库：8 只股票 × 3 轮 = 24 条，全部成功写入 SQLite
- 深科技 3 轮全部命中：涨跌幅 +5.89% 最高、成交额近 20 亿——典型"持续异动"，这种票就是复盘重点
- 3 次 API = 0.6% 额度：实盘跑 30 轮（每 10 分钟一轮，覆盖 5 小时交易时段）也只需 30 次 API
- demo 间隔 5 秒太短：3 轮结果几乎相同；实盘设 300-600 秒间隔，才能看到股票进出异动列表的变化
盘后查询的实战价值
查询 1：今日全部异动——"今天到底发生了什么"
按时间倒序排列，最新异动在最前面。盘后从上往下翻，相当于"回放"了一天的异动时间线。
查询 2：个股触发频率——"哪些票反复出现"
SELECT code, name, COUNT(*) as freq, MAX(zf), AVG(cje)
FROM anomaly_log
WHERE scan_time LIKE '2026-08-06%'
GROUP BY code, name
ORDER BY freq DESC
触发 3 次的票比触发 1 次的票值得多看一眼——3 次说明不是偶尔跳一下，而是持续在异动状态。这是"持续异动"vs"瞬时波动"的关键区分。
查询 3：时段分布——"哪个时段异动最多"
实盘跑一天后，你会看到类似这样的分布：
09:00 异动45次 涉及20只 ← 开盘集中爆发
10:00 异动18次 涉及12只
11:00 异动8次 涉及5只 ← 盘中平淡
13:00 异动12次 涉及8只
14:00 异动35次 涉及18只 ← 尾盘再次活跃
这帮你判断什么时间该盯紧屏幕、什么时间可以去喝咖啡。
查询 4：每轮汇总——"监控是否稳定运行"
如果某轮命中数突然变成 0，不是市场没异动——更可能是你的脚本挂了或 API 断了。这个查询是监控你的监控。
常见坑
- SQLite 并发写入锁：
sqlite3在写入时会锁整个数据库文件。如果同时有其他进程在写，会报database is locked。盯盘场景单进程写入无影响，但如果加了多线程要注意 LIKE查询要走索引：scan_time LIKE '2026-08-06%'可以走idx_scan_time索引；但如果写成date(scan_time) = '2026-08-06'则不走索引（函数导致全表扫描）- db 文件会越来越大：每条记录约 100 字节，一天 30 轮 × 20 条 = 600 条 ≈ 60KB。一个月约 2MB，完全不是问题。但如果存了更多字段或更高频率，定期
VACUUM压缩 - 不要把
anomaly_log.db放到代码仓库里：它是运行时产物，加进.gitignore。文章里的代码默认放在脚本同目录，实盘建议放到专门的数据目录 - 时间格式统一用
strftime：不要依赖 SQLite 的datetime函数处理中文环境的时间字符串，统一用 Python 格式化后存入
小结
从 #01 到 #09，你的盯盘工具链已经能实时发现异动了。但这只是"发现"——发现之后的数据沉淀，才是复盘和策略迭代的基础。
SQLite 落库只需要 Python 标准库，零额外依赖。每轮扫描命中的异动股自动入库，盘后一条 SQL 就能查出：今天哪些票反复异动、哪个时段集中爆发、每轮监控是否正常。
3 次 API、24 条记录、0.6% 额度——这就是"一天异动全留痕"的成本。
下一篇我们做微信模板消息推送告警——异动雷达命中后，不只是终端打印，而是自动推送到手机。Server 酱 / 企业微信 / 钉钉 webhook，三种方式任选，让你离开电脑也能第一时间收到异动提醒。
本文所用行情数据来自麦蕊智数 API（mairuiapi.com）。文中代码均已本地验证并可复现。证书免费领取地址：https://www.mairuiapi.com。
⚠️ 免责声明：本文为技术分享，所有计算仅展示 Python 编程方法与 API 使用技巧，不构成任何投资建议。股市有风险，投资需谨慎。
关注「Python实时盯盘与预警」合集（共 16 篇），从自选股报价到全市场监控、从免费版到企业证书方案——一站式盘中盯盘不求人。