做短线盯盘的朋友都清楚，盘面上最让人焦虑的不是指数涨跌，而是明明看到有大单进场，却不清楚这些钱到底是真金白银的主力，还是量化资金在自营高抛。我花了一个周末，用300行Python代码搭了个实时股票主力监控系统，专门盯A股的资金流向，把超大单、大单的逐笔成交汇总成主力净流入，一旦发现异动，直接把告警推到手机。

这篇文章就把整套做法的完整设计思路、核心代码逻辑、数据清洗和踩到的坑都拆开讲，你没接触过行情接口也能跟着搭。适合想自己写看盘工具的人，也适合手里有数据源但不知道怎么转化成“可执行信号”的研究型玩家。先说清楚：这不是荐股系统，更不是稳赚策略，它只解决一个需求——让你在主力资金异动发生后的几分钟内看到，而不是收盘后看新闻才知道。

1. 主力监控系统的核心思路：先用300行跑通最小闭环

1.1 主力资金到底怎么定义

很多人把“主力资金”想象成一个神秘机构在偷偷买卖，实际行情软件里它就是个统计口径，通常按单笔成交金额把交易拆成几档：小单、中单、大单、超大单。

各家口径有差异，常见分法是：单笔成交金额小于4万元是小单，4万到20万是中单，20万到100万是大单，超过100万是超大单。所谓主力资金，就是大单和超大单的合计。再进一步，净流入等于主动买入金额减去主动卖出金额，反映了这些大单整体在买还是在卖。

一个合格的监控系统不依赖行情软件给你算好的“主力净流入”字段，而是应该尽量自己去算。因为第三方给的字段口径你喜欢不喜欢是一回事，关键是它往往被平滑过，等数值明显变化时，短线窗口已经关了一半。自己对着快照数据算，你才能控制时间窗，才能知道某个数字到底是“最近5分钟的变化”还是“从早上开盘以来的累计值”。

1.2 300行的边界：哪些功能必须做，哪些必须砍

我说300行不是拍脑袋，是为了给自己设一个边界。实时资金监控这个需求，往深了做没有底：分时重构、Tick回放、Level-2逐笔委托、机器学习预测资金拐点，每一项都能再写几千行。但个人盯盘最需要的其实是四件事：

• 获取行情快照，最好覆盖全市场主要股票；
• 把大单、超大单的成交方向聚合成分钟级净流入；
• 设定规则，把异动股票筛出来；
• 推送提醒并记录日志，方便回来看。

所以我在设计时人为砍掉了K线形态、多因子模型、自动交易执行。这些功能一旦混进来，代码量直接爆炸，调试成本也会指数上升。300行的意义在于让你有一个能跑、能改、能优化的底座，而不是一开始就追求什么都做，最后什么都跑不稳。

1.3 整体架构与数据流向

系统的数据流很简单：行情数据源 → 数据清洗 → 计算主力净流入 → 规则过滤 → 告警与存储。

数据源这里用轮询方式，不接实时推送，理由很实际：A股全市场快照接口用轮询，每秒或每几秒拉一次足够发现分钟级异动。实时推送协议要维护长连接、处理心跳和重连，对300行的小项目来说太重了。

清洗环节处理的是单位不一致、停牌空值、涨跌停导致的失真等问题。清洗完的数据进计算模块，按时间窗口滚动汇总。最后规则模块不只看净流入绝对值，还会看涨跌幅、换手率、市值等辅助条件，筛掉大量无效翻动。整个流程跑完，一次循环大概几百毫秒，完全够用。

2. 数据层设计：怎么选行情源、清洗资金字段、存本地

2.1 实时数据源接入的取舍

做实时监控，数据源是最关键的，也是最容易踩坑的。我的经验是：不要一上来就追求Level-2逐笔委托，先用普通快照跑通逻辑。

快照字段通常包含代码、名称、最新价、涨跌幅、成交额、换手率、各档位的买入卖出金额。这些字段足够算出一个近似主力净流入。Level-2数据虽然更准，但费用、接口复杂度、带宽消耗都上了一个量级，个人工具没必要一开始就用。

接入方式建议用统一封装，哪怕只有一个数据源，也要单独写一个函数返回标准化字典。后续换数据源，只改这一个函数，其他模块不动。我最初没做这层封装，结果数据源字段名一改，牵扯到五个地方的代码，花了半小时排查，教训相当深刻。

提示：接全市场快照时，务必确认接口的频率限制和返回时间戳的来源。有的接口返回的是服务器时间，有的返回的是行情时间，两者相差可能几十秒，直接影响你的时间窗判断。

2.2 资金流向字段的坑与清洗逻辑

每个行情平台给的字段名都不一样，常见的有buy_large_amount、sell_large_amount、main_net_inflow这类。但你要警惕三个问题。

第一是单位陷阱。有的接口返回元，有的返回万元，有的干脆返回字符串“123.4万”。不统一单位，你计算阈值时会得到离谱的数。我的做法是全部转成元，确保数值可比。

第二是累计值还是增量值。有些字段叫“今日主力净流入”，它就是凌晨以来的累计值。如果你每5分钟比较一次，要用当前累计值 - 5分钟前累计值得到窗口变化。如果字段本身已经做成了增量明细，那就直接按时间窗聚合。这两者一旦搞混，系统会不停误报。

第三是停牌和涨跌停的失真。停牌股没有最新成交，价格字段可能还是昨天的；涨跌停板上的大单可能是排队买入或卖出，主动成交方向和资金真实意图并不完全一致。我处理时会对停牌股票直接跳过，对涨跌停个股单独打标签，不在常规告警里展示。

2.3 轻量存储：为什么先用本地数据库

告警记录不需要上云，本地嵌入式数据库足够了。它没有独立的服务进程，写进代码里就能用，不用额外部署，重启后数据还在，很适合单人工具。

我的表结构很简单：

存储逻辑必须在告警推送成功后再执行，这样即使推送服务临时抽风，原始记录还在，不会丢数据。我还建了一张运行日志表，记录每次轮询的耗时、行情时间和服务端时间，排查延迟问题时特别好用。

3. 实时监控与告警：主力异动怎么自动通知

3.1 净流入计算与时间窗设计

计算方式建议分两层：第一层计算每个采样瞬间的主买主卖差额，第二层做滚动窗口聚合。

单次采样的主力净流入可以用下面的思路：

单次净流入 = 大单主动买入金额 + 超大单主动买入金额 - 大单主动卖出金额 - 超大单主动卖出金额

采集频率我设为每10秒一次，但计算告警用的是5分钟窗口，也就是最近30个采样点求和。为什么用5分钟？因为它比1分钟更能过滤单笔大单脉冲，又比15分钟更灵敏，适合短线盯盘。1分钟窗口容易被一两笔大单骗进告警，15分钟窗口往往会错过最佳跟踪起点。

时间是做滚动窗口最麻烦的点。行情数据的到达时间不均匀，不能简单数30条就算完，而是要考虑每条数据的时间戳，只把“当前时间减去5分钟内”的样本纳入计算。如果某个股票突然停牌或数据源断流，窗口内样本不足15个，我会暂时不触发告警，避免用半残缺数据做决策。

3.2 告警规则：减少无意义通知的实用策略

很多新手只看“主力净流入最多”就报警，结果一天收到几百条，最后懒得看。我实测下来，规则一定要组合过滤。

我的核心规则是四层过滤：

• 5分钟主力净流入大于当日全市场数据的80分位数，而不是固定值。这个动态阈值能自动适应市场冷热；
• 当前涨跌幅处于-1%到5%之间。涨幅太高可能是明显拉升尾段，追的风险大；跌幅过大可能是资金出逃，不是普通异动；
• 换手率大于2%，过滤掉没人气的僵尸票；
• 排除上市不满60个交易日的次新股和风险警示票。

动态阈值我多做了一步：每次全市场扫描完成，把全市场股票的5分钟净流入排序，取出分位数作为本次“候选阈值”。这样大盘沉闷的时候，不会有大量股票头顶着同一个固定阈值来欺骗你；大盘火热的时候，也不会因为阈值太低导致信息爆炸。这个优化大概只加了20行代码，但告警质量提升非常明显。

3.3 推送通道怎么选

推送通道我建议优先选Webhook机器人，不要自己维护App后端。主流的办公聊天工具和企业微信机器人都支持自定义Webhook，只需要一个HTTP POST，把JSON消息发过去，手机端就能收到带标题和正文的卡片。

如果你只是本地调试，可以先写成控制台输出，再逐步加HTTP推送。推荐顺序是：

1. 控制台彩色打印，方便代码调试；
2. 写文本日志，方便回溯；
3. 接Webhook推送，方便实盘盯盘；
4. 最后再考虑邮件通知或短信通知，因为贵的通道对个人工具没必要。

推送内容不要只发一个股票代码，要把触发依据一起带上。我通常发一条：代码、名称、最新价、涨跌幅、5分钟主力净流入、换手率、触发规则编号、行情时间。这样你在手机上看一眼，就能判断这个信号值不值得切进行情软件。

import json import urllib.request def post_webhook(url, content): data = json.dumps({ "msgtype": "text", "text": {"content": content} }).encode("utf-8") req = urllib.request.Request( url, data=data, headers={"Content-Type": "application/json"} ) with urllib.request.urlopen(req, timeout=5) as resp: return resp.status

这个函数对任何Webhook都通用，只改JSON结构就行。timeout=5一定加上，否则推送服务卡住会把主循环一起拖死。

4. 实战落地：300行代码的模块拆分与核心代码

4.1 项目文件结构

我强烈建议哪怕只有300行，也要拆成几个文件。全塞在一个文件里，改起来绝对想骂人。我的拆分方式：

这样每个文件函数都不复杂，单文件职责一清晰，出问题能快速定位。三个文件之间只通过统一的数据结构交互，不要互相混用。

统一数据结构长这样：

{ "code": "600XXX", "name": "示例股份", "price": 12.34, "change_pct": 2.15, "amount": 34560000, "turnover": 3.2, "buy_large": 1234567.0, "sell_large": 890123.0, "main_net_inflow": 344444.0, "ts": 1699999999 }

所有字段统一用浮点或字符串，单位统一成元，时间统一用Unix时间戳。这套结构贯穿全系统，清洗和计算都围绕它展开。

4.2 核心代码逐段拆解

先看主力净流入聚合的核心逻辑。我用一个字典做滚动窗口缓冲，每个股票维护一个最近5分钟的采样列表。

class MainFundBuffer: def __init__(self, window_seconds=300): self.window_seconds = window_seconds self.buffer = {} def __delitem__(self, code): ... def add_snapshot(self, snap): code = snap["code"] ts = snap["ts"] if code not in self.buffer: self.buffer[code] = [] self.buffer[code].append(snap) cutoff = ts - self.window_seconds self.buffer[code] = [ x for x in self.buffer[code] if x["ts"] > cutoff ] def window_stats(self, code): snaps = self.buffer.get(code, []) if len(snaps) < 15: return None total_buy = sum(x["buy_large"] + x["buy_super"] for x in snaps) total_sell = sum(x["sell_large"] + x["sell_super"] for x in snaps) net = total_buy - total_sell return {"code": code, "net_inflow": net, "sample_count": len(snaps)}

注意一个细节，cutoff = ts - self.window_seconds，我使用的是每个采样点自带的时间戳，不是系统当前时间。这能避免数据源延迟导致的“窗口错位”，同时如果某股票最新快照时间明显滞后，它自然会被清理出窗口。

再来看主循环。主循环每10秒跑一次，先拉全市场快照，清洗后喂给缓冲，然后计算分位数和触发候选。

def main_loop(): config = load_config() buffer = MainFundBuffer(config["window_seconds"]) while True: start = time.time() raw_quotes = fetch_all_quotes() quotes = clean_quotes(raw_quotes) for q in quotes: buffer.add_snapshot(q) candidates = compute_candidates(buffer, quotes) for cand in candidates: alert_message = build_alert(cand, config) if alert_message: post_webhook(config["webhook_url"], alert_message) save_alert(cand) elapsed = time.time() - start log("cycle", elapsed) sleep(max(0.5, config["interval"] - elapsed))

这里我给轮询间隔加了自适应：interval设为10秒，但循环体耗时会占掉一部分，所以真正等待的时间是interval - elapsed。这个细节在行情波动大、全市场股票多的时候特别重要，否则你会发现看似10秒循环，实际每圈跑了13秒，时间窗被越拉越长。

计算候选和动态阈值这部分，核心逻辑是先算所有股票5分钟净流入，再取80分位，然后对超过阈值的股票做规则过滤。

def compute_candidates(buffer, quotes): stats = [] for q in quotes: st = buffer.window_stats(q["code"]) if st: stats.append(st) if not stats: return [] net_values = [s["net_inflow"] for s in stats] threshold = percentile(net_values, 80) return [ s for s in stats if s["net_inflow"] >= threshold and pass_additional_filters(s, quotes_index) ]

pass_additional_filters里就是前面说的涨跌幅、换手率和上市日期过滤。因为逻辑本身不复杂，我建议直接在一个文件里写成普通函数即可，不要过度设计成策略模式，300行的目标需要克制。

4.3 运行效果与调试

启动后日志大概是这个画风：

[10:00:31] cycle done, 0.82s, stock_count=5100 [10:00:41] cycle done, 0.76s, stock_count=5100 [10:01:02] ALERT code=600XXX name=示例股份 net=3210万 pct=3.15% [10:01:03] webhook send success, status=200

第一轮跑完如果发现stock_count比预期少，优先检查数据清洗时是不是把正常股票误删了。我在调试时就犯过这个错，把所有“委卖量为0”的股票都过滤掉了，结果全市场只剩下不到一半。

调试技巧是加一个“影子模式”，也就是先不推送给手机，只把候选告警写入日志。跑几天积累足够样本后，再回看哪些告警是有效异动、哪些是噪声，然后调阈值。直接上实盘推送，大概率第一天就被噪音淹没。

5. 常见问题与排查实录

5.1 数据源延迟与断线

实时接口最常见的坑是“行情时间”和“本地时间”不一致。我排查过一个告警，程序显示9点31分触发，但行情软件里那只股票9点29分就开始拉了。后来打日志才发现，数据源返回的时间戳比实际行情慢了近2分钟，窗口计算被整体平移，导致告警完全失真。

建议在每轮循环固定打印一次服务器时间、本地时间和最新行情时间，三者如果有偏差，要能一眼看出来。断线重连也要处理，我曾经让程序裸奔了一个小时，结果数据源断线后程序直接异常退出，什么记录都没留下。最稳的方案是在主循环外层加个异常捕获，断线后指数退避重试，比如每次失败等待5秒、10秒、20秒……直到恢复。

注意：轮询间隔不要低于3秒。全市场快照接口普遍有频率限制，太激进会触发封禁。我自己实际用的10秒，对5分钟窗口完全够，而且不容易被数据源拉黑。

5.2 频繁误报与漏报

误报的最典型原因是把累计净流入当成窗口净流入。很多接口的“资金流”字段是当日累计值，不做差值直接丢进窗口函数，数字会随着时间越滚越大，每天下午很难不触发告警。

漏报则往往是因为固定阈值设太高。行情冷清时，市场前20名的净流入可能只有1000万；行情火热时，第100名都超过5000万。所以我后来改成80分位数动态阈值，基本解决了这个问题。

还有一个小经验：不要在全市场实时数据上去做“涨跌幅小于X%”的日线级过滤，应该先拿到涨速。我增加了一个“过去5分钟涨速超过0.3%”的条件，专门捕捉资金正在快速推动价格的场景，比单看日涨幅准确得多。

5.3 长时间运行的稳定性

个人脚本挂在电脑上长期跑，最大的敌人是内存和日志。

内存问题出在我的缓冲区：全市场5100只股票，每10秒一个采样，5分钟就有30个采样点，3000万只股票的字典会让内存稳步上涨。解决方案是每小时清理一次快照中超过10分钟的过期数据，以及删除当天已经没有成交的股票。这个清理逻辑不算复杂，加在缓冲区写入函数里即可。

日志建议每天一个文件，文件名带日期，并且只保留最近7天。Python内置的logging模块可以配置轮转，避免单文件无限膨胀。

还有一件事很玄但很现实：别让程序休眠。有些桌面环境下，程序挂机久了会被系统休眠，轮询停止。我当时用电脑挂机过夜做回测，第二天发现进程还在，但日志时间停在了凌晨3点。后来我加了看门狗逻辑，超过2分钟没有新行情事件就主动重建数据源连接，并把异常状态通过告警推给自己。

5.4 风险与合规提醒

必须说清楚的一点是：这个系统只能用于技术学习和研究，不构成任何投资建议。股票市场的涨跌受宏观、行业、情绪、政策等多重因素影响，资金流只是众多观测维度之一。你不能看到5分钟主力净流入为正就认为股票一定会涨，更不能用它来指导真金白银的短线交易。

另外，行情数据接口通常有数据版权条款。个人学习使用问题不大，但如果要部署成对外服务，或者拿来做商业化产品，一定要确认数据授权边界，别让你辛苦搭好的系统因为版权问题翻车。

最后分享一个小技巧

整个系统做完后，我最后悔的一版改动是急着加了好几个新指标：MACD、RSI、分时均线乖离，全部塞进告警规则里，代码量一下子从300行涨到450行，而且实盘盯了几天，发现大部分新指标并没有提升信号质量，反而因为条件太苛刻漏掉了不少有效异动。

后来我强迫自己回到“单变量清晰、双变量验证”的思路：先用主力净流入作为核心信号，只加涨速和换手率做过滤，把系统压回350行，告警准确率反而感觉更稳了。这给我一个很大的触动：监控系统不是指标越多越好，而是要让每一个数字都能说清楚它为什么出现在这里。

如果你也想做类似工具，我的建议是先把这个300行版本跑通，至少积累两周告警日志，再去增加指标。你一定会发现，资金流字段背后的口径问题，远比你预想的要复杂，也远比多写一百行花里胡哨的筛选有趣。