你用前面几篇的脚本搭好了异动雷达盘中每 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_SIZE20# 每轮扫描股票数BATCH_SIZE20# 批量接口上限SCAN_ROUNDS3# 扫描轮数demo 模拟实盘设 30SCAN_INTERVAL5# 每轮间隔秒数demo 短间隔实盘 300-600DB_PATHos.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):初始化数据库创建异动记录表。connsqlite3.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):批量插入异动记录。tsdatetime.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:vfloat(value)except(TypeError,ValueError):returnFalseops{:lambdaa,b:ab,:lambdaa,b:ab,:lambdaa,b:ab,:lambdaa,b:ab,: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):今日全部异动记录。todaydate.today().strftime(%Y-%m-%d)curconn.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。todaydate.today().strftime(%Y-%m-%d)curconn.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):异动按小时分布。todaydate.today().strftime(%Y-%m-%d)curconn.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):每轮扫描命中数。todaydate.today().strftime(%Y-%m-%d)curconn.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():licload_licence()mcClient(lic)conninit_db(DB_PATH)print(f数据库:{DB_PATH})print(f扫描配置:{SCAN_ROUNDS}轮 x{SAMPLE_SIZE}只, 间隔{SCAN_INTERVAL}s)print(*70)# 获取股票列表all_stocksmc.stock_list()stocksall_stocks[:SAMPLE_SIZE]codes[s[dm].split(.)[0]forsinstocks]code_to_name{s[dm].split(.)[0]:s[mc]forsinstocks}total_api0# 多轮扫描forrndinrange(1,SCAN_ROUNDS1):t0time.time()snap{}foriinrange(0,len(codes),BATCH_SIZE):batchcodes[i:iBATCH_SIZE]itemsmc.stock_ssjy_more(,.join(batch))foriteminitems:snap[item[dm]]item total_api(len(codes)BATCH_SIZE-1)//BATCH_SIZE alerts[]forcodeincodes:datasnap.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)),})insertedinsert_alerts(conn,rnd,alerts)elapsedtime.time()-t0print(f[第{rnd}轮]{datetime.now().strftime(%H:%M:%S)}| f扫描{len(codes)}只 | 命中{len(alerts)}只 | 入库{inserted}条 | fAPI 1次 |{elapsed:.1f}s)ifrndSCAN_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】今日全部异动记录)rowsquery_today_all(conn)print(f共{len(rows)}条前 10 条)forrinrows[:10]:ts,code,name,price,zf,cje,trrprint(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_cjerprint(f{code}{name:8}触发{freq}次 最大涨跌幅{max_zf:6.2f}% f平均成交额{avg_cje/10000:10.0f}万)print()print(【3】异动时间段分布)forrinquery_hourly_dist(conn):hour,count,uniquerprint(f{hour}:00 异动{count}次 涉及{unique}只个股)print()print(【4】每轮扫描汇总)forrinquery_round_summary(conn):rnd,first_t,last_t,cnt,uniquerprint(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 次 APIdemo 间隔 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 篇从自选股报价到全市场监控、从免费版到企业证书方案——一站式盘中盯盘不求人。