你用前面几篇的脚本搭好了异动雷达，盘中每 5 分钟扫一轮，告警弹出来看一眼——然后呢？
**然后就没有然后了。**弹窗关掉、终端关掉，今天扫了几十轮、命中了哪些票、哪只票反复出现、哪个时段异动最多——全忘了。
盘后想做复盘？“今天好像有只票涨得挺猛的，叫什么来着？”——想不起来。
问题的根源：你的盯盘数据是即时的，没有落盘。异动监控的价值不止"盘中发现"，还有"盘后回放"——哪些票持续异动、哪个时段集中爆发、今天的异动跟昨天比是什么趋势。这些都需要历史数据。
这篇教你用 Python 标准库sqlite3（零额外依赖），把每轮扫描的异动结果自动入库，盘后一条 SQL 查出全部。
本文你将得到什么
- 一个完整的异动落库脚本：多轮扫描 + 自动写 SQLite + 4 种盘后查询
- SQLite 表结构设计：索引、字段、查询优化
- 4 个实用盘后查询：今日全部异动、个股触发频率、时段分布、每轮汇总
- 从"即时告警"到"历史回放"的工程思路：怎么把分散的弹窗变成可回溯的数据库
为什么选 SQLite
| 方案 | 优势 | 劣势 |
|---|---|---|
| CSV 文件 | 最简单，记事本可看 | 不能查询、不能统计、数据量大后难管理 |
| SQLite | 零依赖（Python 内置）、支持 SQL 查询、单文件易备份 | 并发写入有锁（盯盘场景单进程无影响） |
| MySQL/PostgreSQL | 高并发、分布式 | 需装服务、运维成本高（个人盯盘杀鸡用牛刀） |
盯盘场景是单进程写入、低频查询——SQLite 完美匹配。一个.db文件就是整个数据库，拷走就是备份，用sqlite3命令行随时打开查看。
表结构设计
CREATETABLEanomaly_log(idINTEGERPRIMARYKEYAUTOINCREMENT,scan_timeTEXTNOTNULL,-- 扫描时间（精确到秒）scan_roundINTEGERNOTNULL,-- 第几轮扫描codeTEXTNOTNULL,-- 股票代码nameTEXT,-- 股票名称priceREAL,-- 现价zfREAL,-- 涨跌幅%cjeREAL,-- 成交额（元）trREAL,-- 振幅%peREAL-- 市盈率);CREATEINDEXidx_scan_timeONanomaly_log(scan_time);CREATEINDEXidx_codeONanomaly_log(code);设计要点：
scan_time用 TEXT 存 “2026-08-06 09:58:07” 格式——SQLite 的LIKE '2026-08-06%'可走索引查当日数据scan_round记录是第几轮扫描，方便按轮次统计- 两个索引：按时间查（盘后复盘）、按代码查（看某只票今天触发了几次）
- 不设唯一约束——同一只票在同一轮只入库一次，但在不同轮可以多次出现（这正是"持续异动"的证据）
代码：异动落库 + 盘后查询
importsysimportosimporttimeimportsqlite3fromdatetimeimportdatetime,date# 证书加载（只读不打印）sys.path.insert(0,os.path.join(os.path.dirname(__file__),"..","..","_common"))frommairui_helperimportload_licencefrommairuiimportClient# ===== 配置区 =====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 操作 =====definit_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()returnconndefinsert_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))forainalerts]conn.executemany(""" INSERT INTO anomaly_log (scan_time, scan_round, code, name, price, zf, cje, tr, pe) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """,rows)conn.commit()returnlen(rows)# ===== 条件过滤 =====defcheck_condition(value,op,target):ifvalueisNone:returnFalsetry:v=float(value)except(TypeError,ValueError):returnFalseops={">":lambdaa,b:a>b,">=":lambdaa,b:a>=b,"<":lambdaa,b:a<b,"<=":lambdaa,b:a<=b,"==":lambdaa,b:abs(a-b)<1e-6}returnops.get(op,lambdaa,b:False)(v,target)deffilter_alerts(data,rules):forruleinrules:ifnotcheck_condition(data.get(rule["field"]),rule["op"],rule["value"]):returnFalsereturnTrue# ===== 盘后查询 =====defquery_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}%",))returncur.fetchall()defquery_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}%",))returncur.fetchall()defquery_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}%",))returncur.fetchall()defquery_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}%",))returncur.fetchall()# ===== 主流程 =====defmain():lic=load_licence()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]forsinstocks]code_to_name={s["dm"].split(".")[0]:s["mc"]forsinstocks}total_api=0# 多轮扫描forrndinrange(1,SCAN_ROUNDS+1):t0=time.time()snap={}foriinrange(0,len(codes),BATCH_SIZE):batch=codes[i:i+BATCH_SIZE]items=mc.stock_ssjy_more(",".join(batch))foriteminitems:snap[item["dm"]]=item total_api+=(len(codes)+BATCH_SIZE-1)//BATCH_SIZE alerts=[]forcodeincodes:data=snap.get(code)ifdataandfilter_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()-t0print(f"[第{rnd}轮]{datetime.now().strftime('%H:%M:%S')}| "f"扫描{len(codes)}只 | 命中{len(alerts)}只 | 入库{inserted}条 | "f"API 1次 |{elapsed:.1f}s")ifrnd<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 条：")forrinrows[:10]:ts,code,name,price,zf,cje,tr=rprint(f"{ts}{code}{name:<8}{price:>8.2f}{zf:>+6.2f}%{cje/10000:>10.0f}万")print()print("【2】个股触发次数 Top 10（多次命中 = 持续异动）")forrinquery_stock_frequency(conn):code,name,freq,max_zf,avg_cje=rprint(f"{code}{name:<8}触发{freq}次 最大涨跌幅{max_zf:>+6.2f}% "f"平均成交额{avg_cje/10000:>10.0f}万")print()print("【3】异动时间段分布")forrinquery_hourly_dist(conn):hour,count,unique=rprint(f"{hour}:00 异动{count}次 涉及{unique}只个股")print()print("【4】每轮扫描汇总")forrinquery_round_summary(conn):rnd,first_t,last_t,cnt,unique=rprint(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，三种方式任选，让你离开电脑也能第一时间收到异动提醒。
⚠️免责声明：本文为技术分享，所有计算仅展示 Python 编程方法与 API 使用技巧，不构成任何投资建议。股市有风险，投资需谨慎。
📌 关注「Python实时盯盘与预警」合集（共 16 篇），从自选股报价到全市场监控、从免费版到企业证书方案——一站式盘中盯盘不求人。
