做短线盯盘的朋友都清楚盘面上最让人焦虑的不是指数涨跌而是明明看到有大单进场却不清楚这些钱到底是真金白银的主力还是量化资金在自营高抛。我花了一个周末用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 轻量存储为什么先用本地数据库告警记录不需要上云本地嵌入式数据库足够了。它没有独立的服务进程写进代码里就能用不用额外部署重启后数据还在很适合单人工具。我的表结构很简单字段说明code股票代码name股票名称price触发时刻最新价change_pct涨跌幅net_inflow_5m5分钟主力净流入turnover换手率alert_rule触发规则编号create_time告警时间存储逻辑必须在告警推送成功后再执行这样即使推送服务临时抽风原始记录还在不会丢数据。我还建了一张运行日志表记录每次轮询的耗时、行情时间和服务端时间排查延迟问题时特别好用。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推送。推荐顺序是控制台彩色打印方便代码调试写文本日志方便回溯接Webhook推送方便实盘盯盘最后再考虑邮件通知或短信通知因为贵的通道对个人工具没必要。推送内容不要只发一个股票代码要把触发依据一起带上。我通常发一条代码、名称、最新价、涨跌幅、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, datadata, headers{Content-Type: application/json} ) with urllib.request.urlopen(req, timeout5) as resp: return resp.status这个函数对任何Webhook都通用只改JSON结构就行。timeout5一定加上否则推送服务卡住会把主循环一起拖死。4. 实战落地300行代码的模块拆分与核心代码4.1 项目文件结构我强烈建议哪怕只有300行也要拆成几个文件。全塞在一个文件里改起来绝对想骂人。我的拆分方式文件职责行数config.py全局配置、阈值参数20source.py行情数据源接入与清洗70monitor.py窗口计算与主力净流入聚合90alert.py规则判断与Webhook推送60store.py本地存储与日志40main.py主循环与控制逻辑20这样每个文件函数都不复杂单文件职责一清晰出问题能快速定位。三个文件之间只通过统一的数据结构交互不要互相混用。统一数据结构长这样{ 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_seconds300): 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_count5100 [10:00:41] cycle done, 0.76s, stock_count5100 [10:01:02] ALERT code600XXX name示例股份 net3210万 pct3.15% [10:01:03] webhook send success, status200第一轮跑完如果发现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行版本跑通至少积累两周告警日志再去增加指标。你一定会发现资金流字段背后的口径问题远比你预想的要复杂也远比多写一百行花里胡哨的筛选有趣。
