不知道有多少人跟我一样，每天要盯二十几只票，开盘前翻一遍行情软件，盘中时不时切过去瞄一眼，收盘后再手动记一下涨跌幅。一天下来，光在这些重复动作上就浪费了大把时间，而且经常是 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, headers=headers, timeout=5) # 注意：这个接口返回的是 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(limit=50): 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:0+t:6,m:0+t:80,m:1+t:2,m:1+t: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, params=params, headers=headers, timeout=5) 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关键参数解释一下：fltt=2表示价格字段按浮点返回，如果不传或传 1，价格会被乘以 1000 变成整数，初学者很容易在这里栽跟头。fid=f3表示按涨跌幅排序，po=1表示降序，所以上面这个请求实际是在拿"全市场涨幅榜前 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.encoding，requests 库自动推断为ISO-8859-1。这就找到根因了：接口返回的实际是 GBK 编码，但响应头里没声明，requests 只能按默认方式处理，中文必然乱码。修复方案很简单，就是请求后显式指定编码：resp = requests.get(url, headers=headers, timeout=5) 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, retries=3): 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(tzinfo=ZoneInfo("Asia/Shanghai"))6. 数据校验、持续性优化和复盘复用流水线稳定运行之后，我开始考虑一个更深的问题：怎么让自己相信这套系统给出的数据是对的。免费接口偶尔返回空值或异常值，必须有自检手段。6.1 双数据源交叉校验我用的最简单有效的校验手段，就是前面提到过的两源交叉比对。腾讯和东财是两家独立的数据源，正常行情下同一只标的的现价应该非常接近。我在判定模块前加了一个校验节点：如果同一只票在两个数据源里的价格偏差超过 0.5%，就认为存在异常，这一轮判定暂时跳过该标的，并输出一条告警日志。这里有个细节：偏差阈值不能设置得太小，因为两个接口的更新频率不完全一致，在极端波动时，同一秒内两家数据差出几厘很正常。0.5% 这个值是观察了一周数据后取的，整体误报率非常低。6.2 从盯盘到复盘：快照数据的复用价值前面提到的"记录"环节，后来证明是整个项目里最值得的一笔投资。每次采集的数据快照，我追加行写入一张 SQLite 表，字段就是统一数据模型的九个字段加一个采集批次号。跑了一个多月后，这张表已经有不少行数据。这些历史快照能做到的事远超盯盘本身：收盘后可以画自定义分时走势，可以统计自己关注池里各标的的波动分布，可以回测简单的阈值策略。比如我会用 SQL 直接统计"上午 10 点到 10 点半之间涨幅超过 3% 的标的，午后回落概率有多大"，这种问题放在以前，完全没有数据支撑，现在随口就能查。6.3 我这个项目的稳定运行体会整套流水线从搭建到现在，让我印象最深的不是某个技术细节，而是它改变了我盯盘的方式。以前是人被动看盘，现在是机器主动把有效信息推到面前。每天收盘后，我只看一眼当天的快照统计和触发的预警记录，就能掌握全局。如果你也想复制这套方案，我给三个建议：第一，代码量不要控制得太少，把解析、判定、通知拆成独立模块，后面维护会感谢自己；第二，接口字段和返回格式一定要在代码里写上注释，数据源隔几个月调整格式是常事，有注释至少能快速定位；第三，任何预警都先跑两天模拟盘再上线，用历史数据回放验证规则，不要上线第一天就被假信号淹没。盯盘这件事，工具只是手段，把精力从盯盘里解放出来做更深入的分析，才是最终目的。
不知道有多少人跟我一样，每天要盯二十几只票，开盘前翻一遍行情软件，盘中时不时切过去瞄一眼，收盘后再手动记一下涨跌幅。一天下来，光在这些重复动作上就浪费了大把时间，而且经常是 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, headers=headers, timeout=5) # 注意：这个接口返回的是 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(limit=50): 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:0+t:6,m:0+t:80,m:1+t:2,m:1+t: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, params=params, headers=headers, timeout=5) 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关键参数解释一下：fltt=2表示价格字段按浮点返回，如果不传或传 1，价格会被乘以 1000 变成整数，初学者很容易在这里栽跟头。fid=f3表示按涨跌幅排序，po=1表示降序，所以上面这个请求实际是在拿"全市场涨幅榜前 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.encoding，requests 库自动推断为ISO-8859-1。这就找到根因了：接口返回的实际是 GBK 编码，但响应头里没声明，requests 只能按默认方式处理，中文必然乱码。修复方案很简单，就是请求后显式指定编码：resp = requests.get(url, headers=headers, timeout=5) 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, retries=3): 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(tzinfo=ZoneInfo("Asia/Shanghai"))6. 数据校验、持续性优化和复盘复用流水线稳定运行之后，我开始考虑一个更深的问题：怎么让自己相信这套系统给出的数据是对的。免费接口偶尔返回空值或异常值，必须有自检手段。6.1 双数据源交叉校验我用的最简单有效的校验手段，就是前面提到过的两源交叉比对。腾讯和东财是两家独立的数据源，正常行情下同一只标的的现价应该非常接近。我在判定模块前加了一个校验节点：如果同一只票在两个数据源里的价格偏差超过 0.5%，就认为存在异常，这一轮判定暂时跳过该标的，并输出一条告警日志。这里有个细节：偏差阈值不能设置得太小，因为两个接口的更新频率不完全一致，在极端波动时，同一秒内两家数据差出几厘很正常。0.5% 这个值是观察了一周数据后取的，整体误报率非常低。6.2 从盯盘到复盘：快照数据的复用价值前面提到的"记录"环节，后来证明是整个项目里最值得的一笔投资。每次采集的数据快照，我追加行写入一张 SQLite 表，字段就是统一数据模型的九个字段加一个采集批次号。跑了一个多月后，这张表已经有不少行数据。这些历史快照能做到的事远超盯盘本身：收盘后可以画自定义分时走势，可以统计自己关注池里各标的的波动分布，可以回测简单的阈值策略。比如我会用 SQL 直接统计"上午 10 点到 10 点半之间涨幅超过 3% 的标的，午后回落概率有多大"，这种问题放在以前，完全没有数据支撑，现在随口就能查。6.3 我这个项目的稳定运行体会整套流水线从搭建到现在，让我印象最深的不是某个技术细节，而是它改变了我盯盘的方式。以前是人被动看盘，现在是机器主动把有效信息推到面前。每天收盘后，我只看一眼当天的快照统计和触发的预警记录，就能掌握全局。如果你也想复制这套方案，我给三个建议：第一，代码量不要控制得太少，把解析、判定、通知拆成独立模块，后面维护会感谢自己；第二，接口字段和返回格式一定要在代码里写上注释，数据源隔几个月调整格式是常事，有注释至少能快速定位；第三，任何预警都先跑两天模拟盘再上线，用历史数据回放验证规则，不要上线第一天就被假信号淹没。盯盘这件事，工具只是手段，把精力从盯盘里解放出来做更深入的分析，才是最终目的。