把“盯大佬持仓”做成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, paramsparams, timeout30) resp.raise_for_status() data resp.json() return data.get(items, [])采集频率需要注意。普通交易日每天盘后跑一次就够到了定期报告密集披露期再临时提高频率。不要全天24小时高频轮询公开网页并发压力大了容易触发访问频率限制反而得不偿失。2.2 存储层SQLite 就能满足个人盯盘项目的数据量不大SQLite 完全够用不需要上 PostgreSQL。关键是建表时就把去重字段想清楚。我用过的核心表结构参考如下字段类型说明record_idTEXT唯一记录IDannouncement_dateTEXT公告日期stock_codeTEXT股票代码stock_nameTEXT股票名称investor_nameTEXT投资者或机构名称change_typeTEXT增持/减持/新进/退出/不变sharesINTEGER变动股数source_urlTEXT原始链接created_atTEXT入库时间在这个表上对 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( modelqwen-plus, # 换成你实际使用的模型 messages[ {role: system, content: PROMPT}, {role: user, content: announcement_text} ], response_format{type: json_object}, temperature0, ) 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, triggercron, day_of_weekmon-fri, hour17, minute30) scheduler.start()这里的“24小时在线”指的是服务可以常开任务按计划到达而不是高频轮询。把这一点想清楚能少踩很多坑。4. 关注名单、判断标准和输出格式4.1 关注名单怎么维护系统不能什么公告都推。你需要维护一份关注名单建议单独用配置文件或数据库表管理字段包括alias显示名比如“示例基金”category分类比如“公募”“牛散”“营业部席位”keywords匹配关键词enabled是否启用。配置文件可以这样组织{ watchlist: [ { alias: 示例基金, category: 公募, keywords: [示例基金, 示例混合型证券投资基金], enabled: true } ] }匹配时先按完整名称精确匹配匹配不到再用关键词做模糊匹配。比如关注某营业部席位就用营业部全名作为第一关键词再补充几个简称词。不要一上来就用大模型做实体识别成本高、误报多规则匹配在这个场景里更可靠。4.2 一条提醒应该包含哪些信息提醒不是越短越好也不是越长越好。每条提醒至少要包含六个字段股票代码和股票名称变动类型增持、减持、新进、退出或不变变动数量变动发生日期公告日期原始来源链接。最后一项最容易忽略。没有来源链接的提醒过两天你想回查就找不到了等于一条无效信息。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 解析失败的排查顺序我遇到解析失败时会按这个顺序排查先看URL能不能直接访问是登录问题、验证码问题还是需要特定请求头再看HTML或PDF文本抽取的结果表格是不是错位、文本是不是乱码、PDF是不是扫描件再看字段名是否变化比如原本叫“股东名称”新版公告改成了“股东全称”再看输入模型的文本是不是被截断超长公告很可能在截断后丢失关键段落最后再怀疑模型本身检查 temperature 是否为0输出格式有没有变化。前四步能解决大部分问题。真正需要改提示词的场景其实没有想象中那么多。6.2 数据延迟和口径要反复强调上市公司定期报告披露和真实持仓变动之间存在时间差基金季报显示的也是季度末时点的快照。所以这套监控的价值是“发现公开信息”不是“知道今天实时持仓”。以基金季报为例公告是在季末之后的一段时间才发布中间隔着收集数据、审计、复核的流程。就算你再快看到的也是过去某个时点的静态名单。理解这个时间差你就不会因为某条数据跟实时行情对不上而怀疑系统坏了。6.3 合规边界和使用心态最后说一个非常重要的事这个项目只使用公开披露的数据做的是信息整理和提醒不是内幕消息收集也不该被当成荐股系统。不要拿它去诱导任何人跟单也不要编造或传播未经核实的内容。使用心态也要放平。公开披露过持仓变化的大佬并不等于他们做得对。历史上高位加仓、随后大幅回撤的例子并不少。这个项目真正能帮到你的是减少手工查资料的时间让你对公开信息的变化有更完整的感知而不是提供一个“跟着买就能赚钱”的信号。我个人更建议先把单条任务跑稳再考虑批量和接口。数据源和通知渠道都稳定之后这个 AI 搭子才会真正可靠。如果只是学习默认配置完全够用如果要长期使用就把关注名单、日志、输出目录和失败重试提前设计好比之后频繁补丁要省心得多。