几个月前我开始折腾“A股投资助手”这个项目,起因其实很朴素:每天打开行情软件看自选股,再翻一遍券商研报,两边来回切,实在太耗精力。行业研报散落在不同平台,实时行情又是另一套系统,做投资研究变成了纯体力活。于是我用Python写了一条完整链路——行业研报爬虫负责自动抓取和解析研究报告,行情模块负责实时拉取A股行情,最后用智能对话分析把两者串联起来,慢慢做成一个小型的股票行情分析系统和投资数据知识库。这篇文章不打算讲高大上的架构,只讲我实际调研、写代码、跑数据时踩过的路,以及我认为最值得复用的那部分经验。适合有Python基础、想用技术解决信息整合问题的朋友参考。
先说明一个原则:这个项目定位是个人学习和研究工具,所有数据源都使用公开渠道,严格遵守网站访问规则,输出内容也只作为信息整理参考,不构成任何投资建议。
1. 整体架构与技术选型:为什么把三件事拆成三层来做
1.1 项目拆解:这个工具到底要解决什么问题
我归纳下来,每天做投研信息整理,核心就三件事:第一,把散落在各处的研报集中到一个地方,按行业、公司、发布时间归档;第二,快速拿到当前行情、历史K线、涨跌幅这些基础数据,不用每次打开不同软件手动看;第三,能把研报里几千字的内容和实时行情结合起来,形成可回答问题的知识库。说得直接一点,我需要一个“研报爬虫 + 行情数据源 + 问答系统”的组合。
所以项目拆成三个相对独立的模块:数据采集层负责抓研报列表、下载PDF、提取文本;数据服务层负责行情拉取、指标计算、缓存;智能分析层负责把研报文本向量化,做成检索问答。三个模块之间通过数据库和消息解耦,任何一个挂了都不会拖垮其他部分。这种分层方式最大的好处是调试方便,爬虫被限流了,行情模块照常工作,问答系统还能基于已有知识库回答问题。
1.2 技术选型对比:为什么是Python
整个项目我用的是Python,理由很现实,生态成熟。研报爬虫需要requests、BeautifulSoup、pdfplumber,行情拉取有现成的akshare和tushare,知识库和问答有LangChain生态,这些库可以无缝拼在一起。如果换其他语言,每个环节都几乎要从头造轮子,在单人开发场景下非常不划算。
存储层我做了分层。研报元数据和行情日线数据放MySQL,因为要支持比较灵活的SQL查询,比如“统计某行业最近一月研报数量”;实时行情快照放Redis,因为高频更新且不需要持久化;研报文本切片放向量库Chroma,用来做语义检索。文件本身直接落磁盘,按日期和报告ID组织目录。这个组合实测下来最省心:MySQL稳定、Redis快、Chroma部署简单,没有引入重型中间件。
| 数据类别 | 存储方案 | 理由 |
|---|---|---|
| 研报元数据 | MySQL | 支持复杂查询,字段固定 |
| 研报PDF文件 | 本地文件系统 | 方便人工查阅和备份 |
| 研报文本切片 | Chroma向量库 | 轻量,适合做RAG检索 |
| 实时行情快照 | Redis | 高频读写,天然带过期机制 |
| 每日收盘数据 | MySQL | 后续做趋势统计和回测 |
1.3 数据流设计:从原始数据到可回答问题
整个数据流是一条单向管道:爬虫定时任务抓取研报列表,解析PDF后写入MySQL,同时把文本切片写入向量库;行情任务在交易时段定时拉取快照,缓存到Redis,收盘后把日线数据写库;问答系统接收用户问题后,先判断意图,再决定是从Redis取行情、从MySQL查指标、还是从向量库检索研报片段,最后组装上下文交给大模型生成答案。
这样设计之后,每条数据都有明确的“流向”,排查问题非常快。比如研报没入库,我先看爬虫日志,再看解析日志,不用一头扎进代码里猜。所有环节都打上了简单日志,这个习惯帮了我大忙。
2. 行业研报爬虫实操:公开数据源、反爬与PDF解析全流程
2.1 研报数据源怎么选:先列几个公开渠道
爬虫的第一步不是写代码,是先找数据源。我优先用公开、稳定的渠道,巨潮资讯网是官方信息披露平台,研报和公告覆盖比较全;券商研报聚合页面也提供公开列表;新浪财经的研报中心同样能拿到机构研报。这些渠道不需要登录,数据字段也比较规整,适合做自动化。
抓取字段我固定为:标题、机构名称、行业分类、评级、发布日期、PDF链接。这几个字段已经能支撑大部分查询需求。实际开发中,我看到很多朋友一上来就想把研报正文全文抓下来,其实没必要,先拿到结构和元数据,再按需下载PDF正文,性价比高得多。官方渠道往往有反爬策略,但只抓公开列表页并控制频率,保持基本礼貌,一般不会遇到太激进的风控。
2.2 列表抓取与增量更新:别每次都全量跑
研报列表的接口返回通常是JSON格式,里面包含标题、日期和PDF地址。我写了一个定时任务,每天拉一次近三天的研报列表,用“发布时间 + 标题”作为唯一键做去重,已经存在的记录直接跳过,实现增量更新。这个方法比全量抓取高效得多,也减轻了对方服务器的压力。
import requests import time import random session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" }) def fetch_report_list(date_str): url = "https://example-financial-public-api/report/list" params = {"date": date_str, "page": 1, "page_size": 50} for retry in range(3): try: resp = session.get(url, params=params, timeout=10) resp.raise_for_status() return resp.json()["data"]["list"] except Exception as e: print(f"第{retry + 1}次请求失败: {e}") time.sleep(2 ** retry) return []这段代码没什么特别的,但有一个细节值得注意:重试用指数退避而不是固定间隔,第一次失败等2秒,第二次4秒,第三次8秒。这样既不会对服务器造成压力,也能避开临时抖动。请求之间再加随机延时,比如time.sleep(random.uniform(0.5, 1.5)),避免形成明显的定时访问特征。
2.3 PDF下载与文本解析:从非结构化数据里提炼要点
研报列表拿到后,PDF下载这步其实比较简单,但解析是真正的坑。研报PDF质量参差不齐,有的是文本型PDF可以直接提取,有的是扫描图片需要OCR。我优先用pdfplumber处理文本型PDF,它提取表格和段落的准确率比一般的库要高一些。
import pdfplumber def extract_pdf_text(pdf_path): text_parts = [] with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: content = page.extract_text() if content: text_parts.append(content) return "\n".join(text_parts)提取完之后,我会做一次简单的清洗:把多余空行去掉,把“本报告由XXX公司提供”这类页脚信息过滤掉,然后按标题层级切分。研报通常有“一、行业概况”“二、公司分析”“三、盈利预测”这类结构,我利用正则定位这些标题,把正文切成多个带小标题的段落片段,每个片段控制在800字左右。这一步直接决定后面知识库的检索效果,切得太粗检索容易命中无关内容,切得太细又会丢失上下文。
2.4 反爬避坑要点:保持克制比技术更重要
说到反爬,我的态度是:不要老想着绕过风控,而是通过规范请求行为来降低被限制的概率。具体来说,第一,严格控制请求频率,单线程跑,一小时最多抓几百条数据;第二,带上正常的User-Agent和Referer,不要伪装成搜索引擎;第三,遇到验证码或者连续几次返回异常,立即停下任务,等半小时再继续。我唯一一次被临时限制,就是因为写了个多线程脚本同时抓列表和详情,半天就被识别了。后来改成单线程加延时,连续跑了两个多月没出过问题。
提示:调用任何公开数据接口前,先看看对方网站是否声明了使用条款,尽量选择允许程序访问的数据,并在本地做好数据备份,避免接口不可用导致整套系统停摆。
3. 实时行情接入:免费数据源、缓存与指标计算
3.1 免费行情数据源怎么选:akshare、tushare还是直接调接口
实时行情这块,我先后试过三条路:直接用akshare封装的行情接口、申请tushare的积分接口、还有直接请求新浪和腾讯的公开HTTP接口。三者的差异很明显:
| 数据源 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| akshare | 接口丰富,上手快,A股覆盖全 | 依赖上游接口,字段偶尔变化 | 个人项目首选 |
| tushare | 数据质量高,提供积分制 | 部分接口需要积分,门槛高 | 需要专业数据的场景 |
| 新浪/腾讯公开接口 | 完全免费,实时性高 | 字段不固定,连接不稳定 | 轻量展示场景 |
最终我选akshare作为主力行情源,因为它的接口设计对个人开发者很友好,stock_zh_a_spot_em()能一次性拿到全市场快照,省去自己拼接口的麻烦。实时性在分钟级别完全够用,但要注意它本质是对上游公开接口的封装,所以代码里必须做兼容处理,不能假设字段永远不变。
3.2 行情快照、K线与指标计算:先把数据算对
我的做法是,交易时段每5分钟拉一次全市场快照,然后在内存里过滤出自选股票列表,计算涨跌幅、量比、换手率等指标。日线级别用stock_zh_a_hist拿历史K线,同时计算MA5和MA20,用于趋势判断。
import akshare as ak # 获取全市场实时快照 spot_df = ak.stock_zh_a_spot_em() # 自选股列表 watchlist = ["600519", "000858", "300750"] target = spot_df[spot_df["代码"].isin(watchlist)].copy() # 计算基础指标 target["涨跌幅"] = target["涨跌幅"].astype(float) target["量比"] = target["量比"].astype(float) print(target[["代码", "名称", "最新价", "涨跌幅", "量比"]].to_string())这里有一个我踩过的坑:akshare返回的字段在除权除息日前后会有差别,尤其是复权因子变化会导致历史K线看起来“断崖”。解决方法是统一用前复权数据做技术分析,同时保留不复权原始价格,避免在计算收益率时出现偏差。交易日判断也不能想当然,我引入了交易日历接口,节假日自动跳过行情任务,否则周末跑任务只会拿到一堆空数据。
3.3 轮询频率、缓存与推送:不要每秒钟都去打接口
实时行情最忌讳高频率轮询公共接口。我一开始图省事,写了个每2秒拉一次全市场快照的脚本,结果没跑多久接口就开始超时。后来调整为:分钟级快照用轮询,秒级变化用WebSocket连接单一数据源,前端展示直接订阅Redis里的最新值。
具体做法是,把全市场快照写入Redis,设置过期时间60秒;自选股票的明细数据单独存一份,每5秒更新一次;每日收盘后把当日快照写入MySQL,作为历史数据留存。这样的层级设计既保证了实时性,又不会把公共接口打爆。前端页面通过WebSocket订阅Redis中的变化,实测刷新延迟在1秒以内,体验非常流畅。
4. 投资数据知识库:从研报文本到智能问答的完整链路
4.1 知识库怎么建:切块、向量化和标签缺一不可
研报正文不是拿来就能问答的,直接扔给大模型既费token又容易答非所问。我的做法是先把研报文本切片,每个切片附带元数据标签:报告标题、机构、发布日期、涉及行业、评级。然后通过embedding模型把切片转成向量,存入向量库,同时保留原始文本和标签字段,方便后面做过滤。
embedding我用的是bge-m3这类开源中文模型,在本地跑,这样不用每次调用外部API产生费用。向量库选Chroma,因为部署简单,几百MB的研报数据完全能撑住。这里的关键在于切块策略:我按研报的小标题切分,而不是固定按字数硬切。比如一个研报的小标题是“行业景气度分析”,那这个切片天然就是一个完整语义单元,检索命中后直接能把整段分析拿给模型,比硬切800字的效果好很多。
4.2 检索增强问答的实现思路:先用意图路由再召回
问答模块不能所有问题都走同一条链路,我做了简单的意图路由。用户问“贵州茅台现在多少钱”,直接查Redis里的实时行情,根本不需要动大模型;用户问“白酒行业最近有什么研报”,走MySQL查询元数据;用户问“新能源行业景气度如何”,这属于研报内容理解,走向量检索加LLM生成。这三类意图分开处理,既省成本又准确。
研报内容的问答流程也很直接:把用户问题转成向量,在Chroma里做相似度检索,取前5个最相关切片,拼成上下文,再带上问题去请求大模型。为了让答案可验证,我会在提示词里明确要求模型引用切片来源,并在返回结果时把对应的研报标题和日期贴出来。实测下来,这种方案的安全性比直接让模型凭空回答高得多。
4.3 提示词模板与成本控制:少花钱也能答得靠谱
提示词的核心原则是限定边界。我给问答系统写的提示词模板大致是这样:
你是一个A股投研信息助手。请仅基于以下研报片段回答用户问题。 如果片段中没有足够信息,请直接回答“资料库中没有相关信息”,不要自行推测。 每个回答末尾标注信息来源的研报标题和发布日期。 研报片段: {context} 用户问题: {question}这个模板的约束效果很明显,模型不再“自由发挥”,遇到不知道的内容会老实承认。token成本方面,我实测一次完整问答大约消耗800到1200个token,每天几十次查询的量级完全可控。再加上常见问题缓存,比如“XX公司主营是什么”这类固定问题,第一次回答后直接存Redis,下次直接命中,费用能再降一半。
5. 实战中的坑与排查记录:一张速查表和三次现场复盘
5.1 高频问题速查表:先收藏再实操
项目跑起来之后,遇到的问题五花八门,我整理成了一张速查表,给后面自己排查问题省了大量时间。
| 问题现象 | 根本原因 | 解决方式 |
|---|---|---|
| 研报PDF提取出来是空文本 | 扫描版PDF,无文本层 | 换OCR方案,或者下载时过滤文本型PDF |
| 实时行情接口偶尔返回503 | 请求频率过高触发限流 | 降频、加指数退避重试、错峰运行 |
| 研报列表字段突然错位 | 上游接口字段顺序变动 | 用字段名取值不要用序号,加字段类型校验 |
| Redis里行情数据过期 | 任务调度与交易日不匹配 | 接入交易日历,非交易日不调度 |
| 问答答案把旧研报当最新信息 | 检索时没有限制时间范围 | 检索前按日期过滤,只接受近一年数据 |
| PDF文件重复下载占满磁盘 | 缺少文件级去重 | 用报告MD5做唯一索引,重复的直接跳过 |
| 前端推送卡顿 | WebSocket未做心跳重连 | 加心跳机制和断线自动重连 |
这张表的价值在于,每一个问题后面都带着真实的排查路径。比如“字段错位”那次,我的解析代码是用序号取字段的,上游在中间插了一列,所有数据全部错位,花了半天才定位。后来我统一改成按字段名取值,加上类型断言,再也没出过类似问题。
5.2 三次现场复盘:从限流到模型幻觉
第一次被限流,发生在某天下午开盘后。我同时开了三个脚本,一个抓研报,一个拉行情,还有一个在批量下载PDF,很快公共接口就开始大量超时。排查后发现问题不在单次请求,而是并发总量太高。从那以后我给自己定了一条铁律:任何外部接口的并发数不超过2,抓取任务之间错开半小时执行。
第二次是研报字段变更。某个公开接口原本返回日期字段是字符串,某天突然变成了时间戳,导致入库全是1970年。这个问题提醒我,爬虫代码必须时刻对上游保持敬畏,任何解析逻辑都要加一层防御,字段类型不对就告警而不是直接写库。
第三次是模型幻觉。我向问答系统问“当前光伏行业估值水平”,结果系统引用了三个月前的一份研报,给出的结论明显滞后。问题出在向量检索没有做时间过滤。后来我在检索阶段加了硬性约束:默认只召回最近180天内的切片,用户可以通过参数扩大范围。这个改动让答案的时效性有了质的提升。
5.3 这个项目还能往哪些方向扩展
做到这一步,工具的框架已经完整,后面值得扩展的方向也很多。公告监控是下一步的明显需求,每天自动抓取公告并推送核心股东变化;财报数据整合可以把三大报表的关键指标量化成表格;机构持仓变动跟踪能结合研报评级形成更完整的信号。我的计划是在现有数据管道上增加一些定时分析任务,比如每天收盘后自动生成风格简短的行业热点摘要,推送到Web页面或消息通知。
提示:项目扩展时优先复用一个数据管道,不要为每个新功能单独造一套数据库。研报、行情、问答三层已经能支撑大部分新增需求,加功能主要是在分析层做文章。
最后说两句实在话。我个人实际操作下来最大的感受是:数据的稳定性远比模型的聪明程度重要,一个能稳定跑半年的爬虫管道,比一个偶尔惊艳的提示词更值得投入时间。知识库的边界决定了问答系统的可靠程度,只要检索到的上下文是准确的,大模型回答就不会离谱。这套A股投资助手目前已经稳定运行了几个月,每天自动更新研报、推送行情快照、支持自然语言查询,我反而开始有更多时间去看报告本身,而不是花在找数据和复制粘贴上。如果你也在做类似的信息整合工具,建议先跑通最小闭环,再慢慢加功能。记住,它始终只是一个辅助研究的信息工具,别把它当成自动赚钱机器。