不知道有多少人跟我一样每天要盯二十几只票开盘前翻一遍行情软件盘中时不时切过去瞄一眼收盘后再手动记一下涨跌幅。一天下来光在这些重复动作上就浪费了大把时间而且经常是 self 刚离开屏幕想看的那只票就突然异动等切回来已经错过了最佳记录节点。后来我给自己定了个目标能不能用零成本的方式搭一条自动盯盘流水线行情软件能看到的免费数据本质都来自各自的网页接口而这些接口很多都是公开可用的。于是就有了这篇文章记录的项目用 WorkBuddy 做任务编排数据源走腾讯公开行情接口和东方财富公开行情接口写几个 Python 脚本把采集、解析、预警、通知串成一条无人值守的流水线。整个过程不花钱买数据、不租服务器一台常开的电脑就够。期间踩了不少数据接口的坑我把其中 5 个最典型的排查过程完整记录下来希望对想自己掌控行情数据节奏的读者有帮助。适合有一点 Python 基础、想摆脱人工盯盘、又不想上付费终端的人。1. 自建盯盘流水线的前提先弄清楚你到底需要盯什么动手之前我先把需求列清楚。这个环节最容易被人跳过但恰恰决定了整个方案的复杂度。很多人一上来就想着把行情软件里的所有功能都替代掉结果越做越重最后放弃。实际上个人盯盘真正需要的往往是极小一部分能力。1.1 行情软件能查到的不代表你拿得到市面上的免费行情软件展示层面很友好红红绿绿、K线、分时图一应俱全。但这里有几个问题一是数据刷新有延迟免费版本普遍是秒级到分钟级不等对于某些需要快速反应的场景不够用二是自定义预警规则往往受限比如我只想看涨幅超过 5% 或跌幅超过 3% 且成交量大于过去 5 日均量两倍的标的很多软件要么需要开会员要么设置条件特别繁琐三是历史数据不便于导出软件里看得见的东西想批量拉出来做后续复盘统计就很麻烦。所以我当时的结论是真实需求不是要一个更好的看盘软件而是要一个能定时替我拉数据、按规则判断、再推送给我的自动化流程。行情软件是把数据呈现给人看而我需要的是把数据喂给机器去判断。1.2 自建方案的能力边界与适用读者自建流水线的能力边界大概是这样没有专业的 Level-2 逐笔成交也没有经过深度清洗的财务因子免费接口给到的日线级别行情和实时快照足够覆盖绝大多数个人盯盘场景。如果你的需求是打板、抢龙头、毫秒级交易那这套方案不适合你那是量化柜台和收费行情源的领域。适合这套方案的读者有两类一类是有明确盯盘规则但不想一直坐在屏幕前的人比如设置了涨跌幅阈值、希望跌破某条均线时收到提醒另一类是想要积累自己的行情数据集为后续做简单策略回测或者复盘统计做准备的人。两类需求都不需要多复杂的设施一台常开电脑加几个脚本就能覆盖。1.3 两个数据源的分工设计选腾讯行情和东方财富接口不是随便拍的。一开始我也考虑过只用单一数据源后来实际测试发现两家各有不可替代的场景。腾讯行情接口最核心的优势是单次请求可以拼多个股票代码一次拿到一组标的的快照数据字段里有现价、涨跌、涨跌幅、最高最低、成交量等适合做自选股池的轮询监控。东财接口这边强在列表式获取它可以按市场、板块、涨幅排序去拉一批股票适合做全市场扫描比如找出当前涨幅前 50 的标的。两者配合起来一个管深度广撒网一个管精度盯重点。2. 整体架构与 WorkBuddy 的任务编排思路有了需求边界下一步就是设计流水线的骨架。这个阶段不用写代码先在纸上把数据流画清楚。2.1 一条盯盘流水线的四个环节我把整个过程拆成四个环节采集、解析、判定、通知。采集指定时触发脚本去请求两家行情接口拿到原始数据解析是把接口返回的字符串或 JSON 转成统一的 Python 对象判定是拿着统一的数据结构去跑预设的条件规则比如某只票涨跌幅超过阈值、成交量放大、价格跌破 N 日均线通知是当判定命中时把结果通过即时通讯工具的机器人推送到手机。再加一个可选的第五环节记录。每次采集的数据快照落一份到本地数据库或 CSV 文件里。这一步当时觉得没用后来做复盘才发现价值巨大后面我会单独说。2.2 WorkBuddy 的角色值班机器人而不是数据仓库WorkBuddy 在这条流水线里的角色通俗点说就是一个值班机器人。它负责三件事按 Cron 表达式定时唤醒任务、按顺序执行编排好的步骤、在步骤失败时记录日志并重试。它不解决数据怎么解析、条件怎么判断的问题那些是 Python 脚本的活。它解决的是谁在 9 点 31 分准时启动脚本脚本挂了要不要重新跑跑了之后结果怎么送给下一个步骤这类工程问题。这个分工非常重要千万别把业务逻辑硬塞进调度层否则后面每次改一个预警阈值都要动整个任务配置维护成本会陡然上升。2.3 数据流走向具体数据流是这样的WorkBuddy 里的定时任务每 2 分钟触发一次采集脚本脚本先请求腾讯接口拿到自选股的实时快照再请求东财接口拿一次全市场涨幅榜前 N 的数据两个数据源各自解析后归一成统一结构写入本地 SQLite 的行情快照表同时交给判定模块跑规则。判定命中后WorkBuddy 调通知节点把标的名称、代码、触发条件、现价等信息通过 Webhook 送到手机端。整个过程从触发到通知完成正常情况下不超过 10 秒。3. 腾讯行情与东财接口的接入实战请求、解析与统一这一节是实操的核心。我会把两个接口的请求写法、返回格式和解析要点都过一遍并给出可直接改改就用的代码。3.1 腾讯行情接口的请求写法与字段拆解腾讯行情接口的特点是返回格式比较复古一种按;分隔的多段结构每一段对应一个股票代码内部字段用~分隔。刚开始看到这种格式会觉得怪但习惯了反而觉得直接没有嵌套 JSON解析成本很低。import requests def fetch_tencent(codes): codes: 形如 [sh600000, sz000001] 的代码列表 url https://qt.gtimg.cn/q ,.join(codes) headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) } resp requests.get(url, headersheaders, timeout5) # 注意这个接口返回的是 GBK 编码不能直接用 resp.text resp.encoding gbk result {} for seg in resp.text.strip().split(;): if not in seg: continue key, value seg.split(, 1) code key.replace(v_, ) fields value.strip().split(~) result[code] { name: fields[1], code: fields[2], price: float(fields[3]), prev_close: float(fields[4]), open: float(fields[5]), volume: int(float(fields[6])), change: float(fields[31]), change_pct: float(fields[32]), high: float(fields[33]), low: float(fields[34]), timestamp: fields[30], } return result这里有个特别要提醒的地方接口字段索引的布局并非一眼能看出来不同版本的公开资料里对字段位置的描述可能略有差异。我建议第一次接入时先把原始返回打印出来对照着数一遍字段索引确认无误后再写成死代码。另外fields[30]是行情时间格式类似20250102150405这是北京时间后面判断数据新鲜度会用到。3.2 东财列表接口批量筛选与涨跌排序东财接口返回的是标准 JSON比腾讯那边人畜无害得多但参数更多适合做列表式筛选。import requests def fetch_eastmoney_top(limit50): url https://push2.eastmoney.com/api/qt/clist/get params { pn: 1, pz: limit, po: 1, np: 1, fltt: 2, invt: 2, fid: f3, fs: m:0t:6,m:0t:80,m:1t:2,m:1t:23, fields: f2,f3,f5,f6,f12,f14,f15,f16,f17,f18, } headers { User-Agent: Mozilla/5.0, Referer: https://quote.eastmoney.com/, } resp requests.get(url, paramsparams, headersheaders, timeout5) data resp.json() items [] for item in data.get(data, {}).get(diff, []): items.append({ code: item.get(f12), name: item.get(f14), price: item.get(f2), change_pct: item.get(f3), volume: item.get(f5), amount: item.get(f6), high: item.get(f15), low: item.get(f16), open: item.get(f17), prev_close: item.get(f18), }) return items关键参数解释一下fltt2表示价格字段按浮点返回如果不传或传 1价格会被乘以 1000 变成整数初学者很容易在这里栽跟头。fidf3表示按涨跌幅排序po1表示降序所以上面这个请求实际是在拿全市场涨幅榜前 50。fs参数是证券过滤条件我这里的取值覆盖的是沪深 A 股和创业板、科创板的主流板块具体每个枚举值代表什么网上有现成整理通常不用全改。headers 里面的Referer很重要这个先记下后面讲坑的时候会展开。3.3 统一数据模型一份数据结构喂给所有下游逻辑两个接口返回风格完全不同所以在解析之后我定义了一个统一的字典结构键名固定为code / name / price / change_pct / high / low / volume / source / timestamp这九个字段。不管是腾讯来的还是东财来的最终都转成这个结构再进入后续环节。这样做的好处是判定模块只认识一种数据结构跟数据源彻底解耦。后续想新增一个同花顺或者别的数据源只需要多写一个解析函数不需要改预警规则也不需要改通知模块。我把这步叫做数据归一化是整条流水线里投入产出比最高的代码。def normalize(raw_item, source): return { code: raw_item[code], name: raw_item[name], price: raw_item[price], change_pct: raw_item.get(change_pct, 0.0), high: raw_item.get(high, 0.0), low: raw_item.get(low, 0.0), volume: raw_item.get(volume, 0), source: source, timestamp: raw_item.get(timestamp, ), }4. WorkBuddy 落地定时触发、预警与通知的实际配置脚本层面跑通之后剩下的是把脚本挂到 WorkBuddy 上变成真正的无人值守流水线。这个环节看起来简单实际上有不少细节值得抠。4.1 定时任务的交易时段表达盯盘任务不能像普通的每分钟任务那样全天跑必须限制在 A 股交易时段内否则非交易时段拉到的数据没有意义还白白增加接口负载。我的定时表达式长这样task: name: a_share_watch cron: */2 9-11,13-14 * * 1-5 timezone: Asia/Shanghai简单解释*/2是每 2 分钟跑一次9-11,13-14是上午 9 点到 11 点、下午 13 点到 14 点之间1-5是周一到周五。这里有个细节A 股下午收盘是 15 点但我的表达式只到 14 点后面再加一个条件判断如果当前时间已经超过 14:55直接跳过采集中间环节只做收盘快照记录。这样安排是为了避开尾盘最后几分钟高频波动减少不必要的数据记录。4.2 预警规则的配置与去重预警规则我放在一个独立的 JSON 文件里没有写死在 Python 代码里。每跑一轮判定模块会重新读取这个文件好处是改阈值不用动任务、不用重启改完下一轮自动生效。{ rules: [ { name: 急涨提醒, field: change_pct, op: , value: 5 }, { name: 急跌提醒, field: change_pct, op: , value: -3 }, { name: 放量提醒, field: volume_ratio, op: , value: 2.5 } ] }光有规则还不够必须解决重复通知的问题。如果某只票连续 30 分钟涨幅都大于 5%每 2 分钟推一条消息手机通知栏就废了。我的做法是维护一个通知去重表记录代码 规则名 最近一次通知时间同一对组合在 30 分钟内只推一次。这个去重逻辑不复杂但极大提升体验。4.3 通知渠道与异常自愈通知我用的通用 Webhook 方式把命中的结果拼成一条文本消息推到手机端的机器人。Webhook 的好处是通用只要目标环境支持就能接。通知内容至少包含标的名称、代码、触发规则、当前价格、涨跌幅、数据时间。- name: send_alert type: webhook url: https://example.invalid/hooks/xxx headers: Content-Type: application/json body: {\msg\: \{{alert_text}}\}另外还有一个容易被忽略的点单次通知失败不能影响后续循环。我习惯在采集脚本里把所有请求包上 try-except通知节点失败时 WorkBuddy 会记录日志并继续下一轮而不是把整个任务卡死。4.4 全链路日志怎么设计日志是排查问题的唯一线索。我在脚本里埋了三类日志采集日志请求了多少个代码、耗时多久、返回多少条、解析日志解析成功几条、字段缺失几条、判定日志命中规则几条、各条命中的标的是什么。WorkBuddy 的任务日志负责记录每一次调度的开始时间、结束时间和退出码。两边配合起来哪个环节出问题一眼就能定位。5. 五个接口坑的完整排查复盘这一部分是我最想分享的。这五个坑每一个都真实消耗过我不少时间而且它们的排查思路是通用的先确认现象再最小化复现最后锁定根因。5.1 坑一腾讯返回的中文全变乱码问题出在解码而不在接口现象很直接第一次请求腾讯接口拿回来的字符串里股票名称全是类似的乱码价格数字正常。我的第一反应是接口坏了但换个浏览器访问同样的地址完全正常说明服务端数据没问题问题出在我的请求解析环节。排查过程先打印resp.headers.get(Content-Type)发现接口响应头里没有显式声明charset。再打印resp.encodingrequests 库自动推断为ISO-8859-1。这就找到根因了接口返回的实际是 GBK 编码但响应头里没声明requests 只能按默认方式处理中文必然乱码。修复方案很简单就是请求后显式指定编码resp requests.get(url, headersheaders, timeout5) resp.encoding gbk这个坑的教训是resp.text的解析结果取决于resp.encoding而它并不总是准确的。遇到非 UTF-8 编码的接口不要依赖自动推断直接用resp.content手动解码最稳妥。5.2 坑二东财接口偶发 403是 Referer 校验在作祟东财接口的 403 现象比较迷惑有时候请求成功有时候失败而且失败集中在连续请求次数较多的场景。最初我以为是频率问题但把频率降到很低之后依旧偶发 403。排查过程我打开浏览器的开发者工具手动访问东财行情页面看它发出的请求头和自己代码里的请求头做对比。差异非常明显浏览器请求里带了Referer: https://quote.eastmoney.com/而我的代码里只有User-Agent。为了验证我在代码里加上这个 Referer连续请求一百次一次 403 都没有再出现。headers { User-Agent: Mozilla/5.0, Referer: https://quote.eastmoney.com/, }这里的原理是服务端会根据请求的来源页面做基础校验没有合法来源的请求会被判定为异常访问。这个坑提醒我接入任何免费接口前先用浏览器抓一次真实请求当模板把主要头字段复制过来比自己瞎猜有效得多。5.3 坑三自己算的涨跌幅和接口对不上是除权除息里的复权细节运行几天后我发现一个诡异的数据不一致某只票在腾讯接口里显示涨跌幅为 3.2%但我用接口返回的现价和昨收价手动一算算出来是 7.8%差距巨大。起初怀疑是接口返回有误后来查了这只票的公告才知道当天是除权除息日交易所对昨收价做了调整。复盘一下这个逻辑所谓昨收价在除权除息日当天会按除权除息参考价重新计算而不是上一交易日的实际收盘价。腾讯接口返回的prev_close字段是调整后的昨收价现价也是基于这个调整后价格计算涨跌幅的所以两者算出来自然一致。而我手动用上一交易日的实际收盘价去套算出来的数值当然对不上。这个坑的最终修复是凡是涉及涨跌幅判断直接使用接口返回的change_pct字段不在本地重新计算。需要自己算的场景也一定要先确认使用的是不是同一基准的昨收价。复权问题对个人盯盘影响最大的是分红送转季每年那几个月份集中爆发。5.4 坑四批量请求触发限流批量拼接和退避重试是解药自选股数量少的时候没感觉后来我把监控池扩大到一百多只票腾讯接口逐个请求发现跑到一半后面的请求开始大面积超时。排查过程分了两步先在代码里统计每个请求的耗时发现耗时呈逐渐上升趋势再用单次批量请求替代多次逐个请求问题立刻消失。腾讯接口本身支持在一个 URL 里拼接多个代码批量请求一次就能拿回所有数据不仅网络往返次数少了触发限流的概率也大幅下降。除此之外我还加了两道保险一个是最简单的退避重试机制某次请求失败后等待 1 秒、2 秒、4 秒递增重试最多 3 次另一个是本地缓存同一个标的在 30 秒内如果已经成功拉取过数据后续重复请求直接走缓存不再访问网络。import time def safe_request(fn, retries3): for i in range(retries): try: return fn() except Exception as exc: wait 2 ** i time.sleep(wait) raise RuntimeError(request failed after retries)经历这个坑之后我才意识到对免费接口最大的尊重就是减少无效请求能批量就批量能缓存就缓存。5.5 坑五数据看起来慢了一分钟其实是时间戳和调度周期的问题流水线跑了一周后我开始怀疑接口是不是有延迟有时候行情软件上显示某只票已经 15.20 元而我这里拉到的还是 15.18 元感觉慢了一分钟。为了定位我在采集脚本里加了数据时间打印结果发现接口返回的时间戳和行情软件显示的当前时间基本一致慢的其实是调度周期。复盘整个链路WorkBuddy 每 2 分钟跑一次任务假设 10:00:00 触发拿到的时间戳是 10:00:01这是实时的当行情软件在 10:01:30 显示新价格时我的流水线还没到下一次触发时间所以展示的仍是 10:00:01 的快照看起来就像是延迟了一分半。实际上只要把数据时间和通知时间区分开展示这个误解就会消失。我把通知文案从当前价格15.18改成了数据时间 10:00:01现价 15.18收到消息的人自然知道这是哪个时间点的快照。另外我已经在脚本里对时间戳做了严格解析统一用Asia/Shanghai时区避免服务器时区设置导致的时间偏差。from zoneinfo import ZoneInfo from datetime import datetime data_time datetime.strptime(timestamp, %Y%m%d%H%M%S) data_time data_time.replace(tzinfoZoneInfo(Asia/Shanghai))6. 数据校验、持续性优化和复盘复用流水线稳定运行之后我开始考虑一个更深的问题怎么让自己相信这套系统给出的数据是对的。免费接口偶尔返回空值或异常值必须有自检手段。6.1 双数据源交叉校验我用的最简单有效的校验手段就是前面提到过的两源交叉比对。腾讯和东财是两家独立的数据源正常行情下同一只标的的现价应该非常接近。我在判定模块前加了一个校验节点如果同一只票在两个数据源里的价格偏差超过 0.5%就认为存在异常这一轮判定暂时跳过该标的并输出一条告警日志。这里有个细节偏差阈值不能设置得太小因为两个接口的更新频率不完全一致在极端波动时同一秒内两家数据差出几厘很正常。0.5% 这个值是观察了一周数据后取的整体误报率非常低。6.2 从盯盘到复盘快照数据的复用价值前面提到的记录环节后来证明是整个项目里最值得的一笔投资。每次采集的数据快照我追加行写入一张 SQLite 表字段就是统一数据模型的九个字段加一个采集批次号。跑了一个多月后这张表已经有不少行数据。这些历史快照能做到的事远超盯盘本身收盘后可以画自定义分时走势可以统计自己关注池里各标的的波动分布可以回测简单的阈值策略。比如我会用 SQL 直接统计上午 10 点到 10 点半之间涨幅超过 3% 的标的午后回落概率有多大这种问题放在以前完全没有数据支撑现在随口就能查。6.3 我这个项目的稳定运行体会整套流水线从搭建到现在让我印象最深的不是某个技术细节而是它改变了我盯盘的方式。以前是人被动看盘现在是机器主动把有效信息推到面前。每天收盘后我只看一眼当天的快照统计和触发的预警记录就能掌握全局。如果你也想复制这套方案我给三个建议第一代码量不要控制得太少把解析、判定、通知拆成独立模块后面维护会感谢自己第二接口字段和返回格式一定要在代码里写上注释数据源隔几个月调整格式是常事有注释至少能快速定位第三任何预警都先跑两天模拟盘再上线用历史数据回放验证规则不要上线第一天就被假信号淹没。盯盘这件事工具只是手段把精力从盯盘里解放出来做更深入的分析才是最终目的。
