AI自动化监控大佬持仓：从数据采集到定时提醒的完整方案

把“盯大佬持仓”做成AI自动化项目，真正值得研究的不是某个模型有多聪明，而是怎么把散落在定期报告、龙虎榜、大宗交易和上市公司公告里的公开数据，稳定地抓下来、洗干净、归纳成几条能看懂的消息。这个项目的最终形态，就是让AI每天定时把公开信息整理成提醒推给你：谁在增持、谁在减持、谁是新进前十大股东、谁的席位最近上了龙虎榜。

它不是什么神秘的预测工具，更像一个把公开信息查了个底朝天的情报整理系统。适合谁看？适合已经有一定数据基础，想用代码把公开信息整理成信息流的个人投资者或开发者。我实测下来的判断是：这套方案不需要很强算力，一台普通的Linux小服务器或者常开的电脑就能跑，核心难点不在模型选型，而在数据源稳定性、字段口径和提醒去重。下面按实际落地顺序拆一遍。

1. 先搞清楚这个需求到底在盯什么数据

1.1 公开数据的信息口径

先立一个预期：几乎所有公开的“大佬持仓”，都不是实时数据。公募基金的季度报告，只披露季末时点的前十大重仓股；上市公司定期报告，披露报告期末的前十大股东名单；龙虎榜是满足特定条件后盘后公示的当日营业部买卖明细；大宗交易是盘后公示的大额成交记录。

这些数据有一个共同特征：都有明确的时间点，都是事后可查的公开记录。所以这个系统能告诉你的，是“在某个可验证的时间点，某个主体出现过一笔可以查证的持仓变动”，而不是“某大佬今天买了什么”。把这个预期先立住，后面写代码才不会走偏。

整理一下我实际使用的数据源类型：

1.2 手动盯和程序盯的差别

手动盯的痛点是每天要打开好几个网站、逐个搜索关键词、再翻上期数据对比，才知道是增持还是减持。短期做一次还好，连续做一个月人就会开始漏。漏掉一条公告，你根本不知道漏的是什么，这种不确定性比信息本身更消耗精力。

程序化盯盘的优势是“不遗漏”，不是“能预测”。只要数据源没有断，任务每天按时跑，每条新增公告都会进入数据库，即使当天不适合推送，后面也能补查。这才是这个项目最核心的价值。

1.3 AI 在这个项目里的真实角色

有人以为这个项目就是把所有公告丢给大模型，然后等它输出惊天结论。实际上大模型在这里只承担两件事：信息抽取和文案归纳。

公告的形态很杂，有网页、PDF、表格、扫描件转文本，字段名不统一。正则能处理一部分，但公告格式一变，正则就要重写。大模型可以把一段不规则的公告文本变成结构化的JSON，这是它最实用的地方。但要注意，模型擅长抽取，不擅长计算。涉及比例变化、市值变化时，不要让模型算，而是让它在JSON里给原始数值，计算放在代码里做。

2. 整体架构：从采集到推送的四条链路

这个项目本质上就是一个 AI Agent 应用：定时触发、抓取公开信息、调用模型抽取关键字段、最后推送结果。我建议拆成四层：采集层、存储层、分析层、通知层。每一层职责单一，方便单独调试。如果一开始就把所有事情写进一个主循环，任何一个网站改版都会让你重新排查整段逻辑。

2.1 采集层：定时获取增量数据

采集层负责每天定时去公开渠道拿增量。常见做法是使用 Python 的 requests、httpx 拉取页面或接口，再用 BeautifulSoup 或 pyquery 解析 HTML。

从公开渠道来看，我实际会抓取的增量数据有四类：

• 交易所盘后公示的龙虎榜明细；
• 交易所盘后公示的大宗交易记录；
• 上市公司发布的股东减持、增持、权益变动公告；
• 基金定期报告中披露的前十大重仓股名单。

每一类的页面结构和更新频率都不一样，建议拆成四个采集函数，而不是一个万能爬虫。这样任何一类数据源出了问题，都只需要修对应的那部分。

# 采集层示意代码：以某公开公告页为例 import requests from bs4 import BeautifulSoup def fetch_daily_records(date_str: str): url = "https://example.com/api/list" # 以实际公开接口为准 params = {"date": date_str, "page": 1, "limit": 200} resp = requests.get(url, params=params, timeout=30) resp.raise_for_status() data = resp.json() return data.get("items", [])

采集频率需要注意。普通交易日，每天盘后跑一次就够；到了定期报告密集披露期，再临时提高频率。不要全天24小时高频轮询，公开网页并发压力大了容易触发访问频率限制，反而得不偿失。

2.2 存储层：SQLite 就能满足

个人盯盘项目的数据量不大，SQLite 完全够用，不需要上 PostgreSQL。关键是建表时就把去重字段想清楚。

我用过的核心表结构参考如下：

在这个表上对 record_id 建唯一索引，天然去重。重复公告再次入库时会直接失败，不会产生重复提醒。

2.3 分析层：模型抽取，代码计算

分析层是整个项目里最容易被神话的部分，也是最容易翻车的部分。我的做法是分两步。

第一步，用规则或模板把公告里的关键段落截出来。这一步可以显著降低模型输入长度，也能减少无关信息干扰。比如只截“股东持股变动”相关的段落，而不是把整份年报丢给模型。

第二步，用大模型做结构化抽取，输出固定字段的 JSON。这里最关键的是：模型只负责抽取字段和写摘要，所有涉及持股比例变化、市值波动的计算，一律放在代码里用数值完成。

让大模型直接算“较上期变动比例”看起来省事，一旦它把数字看错、把方向说反，你很难在结果里发现。用代码算，至少每次结果可复现、可核对。

2.4 通知层：推送之前先想去重

通知渠道很多：企业微信群机器人、钉钉群机器人、邮件、Server酱，以及各种自建 webhook。选哪个取决于你平时用哪个，不影响整体架构。

真正影响体验的是去重。每条消息推送之前，先按“公告ID_股票代码_投资者_变动类型_日期”生成一个唯一指纹，数据库里查不到才推送。否则同一个公告被重新解析两次，你会收到两条一模一样的提醒，一次两次还能忍，连续三天就会想关掉整个项目。

注意：通知渠道可以换，去重逻辑必须一开始就做。后面再补去重，意味着你已经重复推送了一段时间，还得回头清理消息记录。

3. 环境准备与最小跑通流程

3.1 运行环境和依赖

我建议使用 Python 3.10 以上版本。依赖主要有这些：

pip install requests beautifulsoup4 pandas openai apscheduler

如果你解析 PDF 公告，再装一个 pdfplumber；如果只用网页公告，可以不装。这里要说明一下，openai SDK 只是示例，现在很多大模型平台都提供兼容接口，换 base_url 和模型名就能用。

关于硬件，可以按一个简单的标准判断：如果你只是盯十几个人或机构的持仓变化，4核CPU、8G内存的小服务器完全带得动。模型推理如果放到云端API，本地只需要负责采集、调度和推送。如果你想全部本地跑，就需要一台至少8G显存的机器来跑量化小模型。纯CPU跑7B模型不是不行，但解析一条公告可能要等很久，体验会差很多。

低配环境能跑通单条任务，不代表适合跑定时批量任务。批量解析多份PDF时，内存占用会明显上升，建议先在任务里限制每天处理条数，跑一周看趋势再加量。

3.2 先跑通一条公告解析

不要一上来就全量抓历史数据。第一次测试，我建议只取一条公告，走完整条链路。

# 最小示例：单条公告解析 from openai import OpenAI client = OpenAI(base_url="你的模型接口", api_key="你的密钥") PROMPT = """ 根据公告内容提取持仓变动信息，输出 JSON： { "stock_code": "股票代码", "stock_name": "股票名称", "investor_name": "投资者名称", "change_type": "增持/减持/新进/退出/不变", "shares": 1234567, "record_date": "变动发生日期", "announcement_date": "公告日期" } 只依据原文提取，不要推测，无法确定时填空字符串。 """ def extract_holdings(announcement_text: str) -> str: resp = client.chat.completions.create( model="qwen-plus", # 换成你实际使用的模型 messages=[ {"role": "system", "content": PROMPT}, {"role": "user", "content": announcement_text} ], response_format={"type": "json_object"}, temperature=0, ) return resp.choices[0].message.content

跑通之后，立刻对着原文人工核对一遍返回的JSON，重点看两处：股票代码和股票名称有没有对应错位；增持和减持的方向有没有搞反。这两处一旦出错，后面的提醒全部失真。

3.3 放进定时调度

单条链路没问题后，再加入调度。我用的是 APScheduler，直接在进程里跑，不依赖外部 cron 也能完成最基本的定时任务。

# 调度示意：工作日17:30执行一次 from apscheduler.schedulers.blocking import BlockingScheduler def run_daily_job(): items = fetch_daily_records("today") for item in items: try: process_one(item) except Exception as exc: log_error(item, exc) scheduler = BlockingScheduler() scheduler.add_job(run_daily_job, trigger="cron", day_of_week="mon-fri", hour=17, minute=30) scheduler.start()

这里的“24小时在线”，指的是服务可以常开，任务按计划到达，而不是高频轮询。把这一点想清楚，能少踩很多坑。

4. 关注名单、判断标准和输出格式

4.1 关注名单怎么维护

系统不能什么公告都推。你需要维护一份关注名单，建议单独用配置文件或数据库表管理，字段包括：

• alias：显示名，比如“示例基金”；
• category：分类，比如“公募”“牛散”“营业部席位”；
• keywords：匹配关键词；
• enabled：是否启用。

配置文件可以这样组织：

{ "watchlist": [ { "alias": "示例基金", "category": "公募", "keywords": ["示例基金", "示例混合型证券投资基金"], "enabled": true } ] }

匹配时先按完整名称精确匹配，匹配不到再用关键词做模糊匹配。比如关注某营业部席位，就用营业部全名作为第一关键词，再补充几个简称词。不要一上来就用大模型做实体识别，成本高、误报多，规则匹配在这个场景里更可靠。

4.2 一条提醒应该包含哪些信息

提醒不是越短越好，也不是越长越好。每条提醒至少要包含六个字段：

1. 股票代码和股票名称；
2. 变动类型：增持、减持、新进、退出或不变；
3. 变动数量；
4. 变动发生日期；
5. 公告日期；
6. 原始来源链接。

最后一项最容易忽略。没有来源链接的提醒，过两天你想回查就找不到了，等于一条无效信息。

4.3 结构化输出的判断标准

我习惯让提醒最终长这样：

{ "alert_id": "20250120_600000_示例基金_增持", "stock": "600000", "stock_name": "示例股份", "investor": "示例基金", "change_type": "增持", "shares": 1234567, "record_date": "2025-01-15", "announcement_date": "2025-01-20", "source_url": "https://example.com/announcement/xxx" }

判断标准是：四个核心字段（股票、投资者、变动类型、日期）缺一个，宁可不发。信息不全的提醒，推送出去只会制造焦虑。另外，每天跑完可以看一眼“新增归档数和实际推送数”的差值，如果差值一直很大，说明大部分公告没有命中关注名单，这时候要回去检查名单关键词，而不是继续调模型。

5. 批量运行时的稳定性设计

5.1 去重、重试和失败记录

批量任务跑了几天之后，你会遇到三类问题：重复公告、解析失败、网络异常。

重复公告靠唯一索引和指纹去重解决。解析失败不能静默跳过，要落到单独的 error_log 表，至少记录公告URL、失败原因和时间。网络异常则要设置超时时间和重试次数，重试间隔建议用指数退避，比如第一次等30秒，第二次等60秒，第三次等120秒。

我一般会在主循环外面再包一层 try/except，对单条失败只记录不中断，保证一批公告里面有一条解析失败，不影响其他公告继续处理。

识别解析失败的常见信号也很重要。比如模型抽取出来的 stock_name 一直为空，先看输入文本里是不是把股东名称和股票名称放在同一列，很多PDF表格解析时会吞掉表头，导致字段对不上。这种问题改提示词没用，得从文本抽取那一层修。

5.2 不同时期的调度策略

普通交易日，每天 17:00 之后跑一次当天的公告和生产公示。定期报告密集期（4月、8月、10月）可以改为每小时跑一次增量。周末和节假日没有新公告，不跑也行。

这个调度策略背后有两个考虑：一是给数据源和审核留出更新时间，二是避免给自己制造不必要的通知疲劳。真到了公开信息密集披露的几天，系统每小时推一次消息都很正常，平时保持一天一次，反而更容易坚持使用。

这里的“24小时在线”还有一个含义：长时间无人值守跑批时，要保证重启后任务能自动恢复。最简单的方式是给调度任务加一个启动补偿逻辑，任务启动时先补拉最近24小时的增量，再进入正常调度。

5.3 日志是排查的第一现场

每条任务的开始时间、抓取数量、解析成功数、失败数、推送数，都要写进日志。这个日志不一定是复杂的日志框架，Python logging 就够用，但必须保证每次运行能复盘。

有一个经验值得记住：某天解析成功数为0，先别急着怀疑模型，大概率是网站改版、公告格式换了、或者采集接口的字段名变了。日志会告诉你问题发生在哪一层，而不是让你在模型参数里瞎调。

注意：如果连续多次解析失败，优先停掉任务，人工打开一条公告确认页面结构，再决定是改规则还是改提示词。盲目重试只会让错误日志堆得更厚。

6. 常见问题排查与边界提醒

6.1 解析失败的排查顺序

我遇到解析失败时，会按这个顺序排查：

1. 先看URL能不能直接访问：是登录问题、验证码问题，还是需要特定请求头；
2. 再看HTML或PDF文本抽取的结果：表格是不是错位、文本是不是乱码、PDF是不是扫描件；
3. 再看字段名是否变化：比如原本叫“股东名称”，新版公告改成了“股东全称”；
4. 再看输入模型的文本是不是被截断：超长公告很可能在截断后丢失关键段落；
5. 最后再怀疑模型本身：检查 temperature 是否为0，输出格式有没有变化。

前四步能解决大部分问题。真正需要改提示词的场景，其实没有想象中那么多。

6.2 数据延迟和口径要反复强调

上市公司定期报告披露和真实持仓变动之间存在时间差，基金季报显示的也是季度末时点的快照。所以这套监控的价值是“发现公开信息”，不是“知道今天实时持仓”。

以基金季报为例，公告是在季末之后的一段时间才发布，中间隔着收集数据、审计、复核的流程。就算你再快，看到的也是过去某个时点的静态名单。理解这个时间差，你就不会因为某条数据跟实时行情对不上而怀疑系统坏了。

6.3 合规边界和使用心态

最后说一个非常重要的事：这个项目只使用公开披露的数据，做的是信息整理和提醒，不是内幕消息收集，也不该被当成荐股系统。不要拿它去诱导任何人跟单，也不要编造或传播未经核实的内容。

使用心态也要放平。公开披露过持仓变化的大佬，并不等于他们做得对。历史上高位加仓、随后大幅回撤的例子并不少。这个项目真正能帮到你的，是减少手工查资料的时间，让你对公开信息的变化有更完整的感知，而不是提供一个“跟着买就能赚钱”的信号。

我个人更建议先把单条任务跑稳，再考虑批量和接口。数据源和通知渠道都稳定之后，这个 AI 搭子才会真正可靠。如果只是学习，默认配置完全够用；如果要长期使用，就把关注名单、日志、输出目录和失败重试提前设计好，比之后频繁补丁要省心得多。