做投资研究的朋友都有这种体会A股市场的行业研报一天能更新上百份叠加实时行情、历史复盘和知识整理单靠手工完全忙不过来。市面上的研报平台要么订阅费高得离谱要么只能在线看不能做二次分析。所以我自己动手搭了一套A股研报整合工具用爬虫抓取公开的行业研报和公司研报清洗成结构化数据再对接实时行情最后通过RAG方式让大模型基于这堆资料做智能对话分析。这篇文章不是讲泛泛的理论而是把从选型到落地、从踩坑到修复的完整过程写出来希望对想自己做投研工具的人有点帮助。1. 为什么我要自己动手做A股研报整合工具1.1 个人研究A股信息时的真实痛点先说背景。我平时做股票研究主要依赖公开的券商研报、公司公告和行情数据。理想状态下我打开一个工具能同时看到某只股票最近有哪些机构覆盖、评级是上调还是下调、目标价大概多少以及对应的K线走势。但现实情况是研报分散在多个平台很多网站只展示摘要想找历史研报翻好几页都找不到。关键词检索很弱。我想查“某个行业过去一年被上调评级的个股名单”在常规研报平台基本搜不出来。行情系统和分析系统是分开的看盘软件管行情研报平台管资料中间没有联动。复制粘贴到本地笔记后资料多了就变成一潭死水没法检索也没法追问。于是“做一个自己的助手”这个念头越来越强。标题里那串定义——A股研报整合工具、股票行情分析系统、投资数据知识库、A股智能助手、行业研究报告平台——其实本质上是一条数据流水线先爬取研报建立知识库再叠加实时行情最后用智能对话的方式把信息“读”给人听。1.2 我给自己定的功能边界动手之前先把边界画清楚避免做着做着变成一个大而无当的炒股软件。我的目标功能如下研报采集自动抓取公开渠道的研报标题、发布时间、机构名称、研究对象、评级、摘要等元数据而不是抓取完整付费PDF内容。行情接入支持A股日线行情、最近一段时间的K线数据和基本交易指标。知识库把清洗后的研报摘要存进数据库并同步做向量化供检索使用。智能对话允许用户问“某股最近一个月的研报观点变化”“行业景气度相关的研报有哪些”模型必须基于知识库内容回答并附上来源。同时明确不做不碰自动交易下单不做个股推荐式的输出不采集任何需要付费或破解的内容。这套工具的价值是“提高信息整理效率”不是“给你致富代码”。这个定位很重要它决定了后续所有技术选型。2. 研报爬虫公开数据源的选型与技术边界2.1 哪些源值得爬哪些碰不得爬虫是这个项目里最容易被误解的部分。很多人一听到爬虫就想到各种黑科技但真实做研究型爬虫的人第一考虑不是技术而是“数据来源是否合规”。我选择的源严格限定在公开信息层面上市公司公告与披露信息这类是法定义务披露公开属性最强抓取元数据完全没风险。财经网站公开的研报摘要列表页很多平台会公开展示研报标题、发布时间、机构、评级这些元信息不需要登录就能看到。我抓的就是这一层。交易所官网的公开数据页面。不去碰的需要付费订阅才能看的正文、任何带会员墙的内容。明确在robots.txt里禁止抓取的站点。需要输入账号密码、验证码才能访问的内部页面。为什么这么保守因为投研工具是给自己长期用的数据源的稳定性比“多抓一点”重要得多。你辛辛苦苦把某个付费源的结构扒下来过两天它改版或者反爬升级你的整个系统就瘫了还会惹上麻烦。公开源虽然数据没那么全但胜在稳定而且研报摘要这个颗粒度已经完全够做知识库和对话分析了。2.2 爬虫的实现与字段提取技术栈层面我选择了最稳的组合requests做HTTP请求lxml配合xpath做页面解析pandas做数据整理。有人可能觉得都用2025年了还用requests是不是太土但实际经验是对于静态公开列表页requests加xpath就是最简单、最容易维护的方案没有之一。scrapy对单一数据源来说有点重beautifulsoup的解析速度也不如lxml。核心爬取逻辑大概是这样的import time import random import requests from lxml import html def fetch_report_list(page_url): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 } resp requests.get(page_url, headersheaders, timeout10) resp.encoding utf-8 tree html.fromstring(resp.text) items [] # 这里的xpath需要根据公开页面的实际结构调整 for node in tree.xpath(//div[contains(class,report-item)]): title node.xpath(.//span[classtitle]/text()) stock_code node.xpath(.//span[classcode]/text()) agency node.xpath(.//span[classagency]/text()) rating node.xpath(.//span[classrating]/text()) pub_time node.xpath(.//time/datetime) items.append({ title: title[0] if title else , stock_code: stock_code[0] if stock_code else , agency: agency[0] if agency else , rating: rating[0] if rating else , pub_time: pub_time[0] if pub_time else , }) return items这里有个细节请求频率一定要控制好。我之前一上来就写了多线程并发抓取结果跑了不到两分钟就被对方限流整个IP被临时封了。后来改成单线程加随机延时每次请求之间sleep 2到5秒反而更稳定——因为研报页面不是那种需要极高时效性的数据晚几分钟拿完全没有影响。请求头里也要带真实的User-Agent不要用默认的python-requests很容易被识别。延时加随机化的目的是让访问模式看起来更像自然浏览而不是机器在高速横扫。2.3 数据清洗与质检爬下来只是第一步真正花时间的是清洗。原始页面上经常出现缺失字段、乱码和重复记录。比如股票代码格式不统一有的带前缀sh、sz有的只有6位数字机构名称同样是“中金”有的写成“中金公司”有的写成“CICC China”。不统一的话后续和行情表join就会出大问题。我的清洗流程分三步字段归一化股票代码统一为6位数字机构名称做一次别名映射把常见简称归一到标准名称。去重以“标题股票代码机构发布时间”为联合键去重。同一个研报可能在列表页出现两次或者同一天被抓了两次需要剔除。过滤只保留标题里带“行业研究”“公司研究”“深度报告”“首次覆盖”“评级调整”等典型特征的记录把无关的新闻资讯过滤掉。清洗完的数据长这样字段示例说明stock_code6005196位股票代码stock_name贵州茅台冗余存储方便看agency华泰证券机构简称已归一rating买入评级结论title白酒行业深度报告静待景气度回升研报标题pub_time2025-02-18发布时间summary行业整体估值处于历史低位……公开摘要这一步做扎实之后知识库的质量就有了保障。如果清洗不做后面RAG检索出来的内容会出现大量同名错乱或时间错乱对话答案稍微有点金融素养的人一看就觉得不可信。3. 行情模块实时行情与研报联动分析3.1 开源行情库的选型研报只是静态信息要让它活起来必须叠加实时行情。行情数据我一开始考虑过自己写爬虫去抓后来果断放弃——自己抓交易数据既不规范又容易被反爬搞死。最后选了akshare和tushare这类开源数据接口。两者都提供A股历史行情和实时快照区别在于akshare不需要注册就能拿部分公开数据免费适合快速验证。tushare需要token但数据质量和稳定性更好字段也更规范。我的做法是先用akshare做开发验证确认整个链路没问题后再引入tushare作为正式数据源。主要原因是我需要精确的复权因子akshare部分接口的复权方式我不太信任tushare的pro接口字段更清晰。实际上复权这件事对投研分析非常重要后面我会专门讲踩坑。import tushare as ts import pandas as pd pro ts.pro_api(你的token) # 获取最近一年的日线行情 df pro.daily( ts_code600519.SH, start_date20240101, end_date20241231 ) # 按时间升序排列 df df.sort_values(trade_date)tushare的daily接口返回单位是手成交额单位是千元这些字段语义在文档里写得很清楚但新手特别容易忽略导致算出来的数据差好几个数量级。3.2 数据库设计与增量更新行情和研报都存在本地SQLite里。为什么不用MySQL或PostgreSQL因为这是个人工具单机使用数据量撑死也就几百万行SQLite完全够用而且零部署成本、备份就是一个文件。我设计的核心表有三张-- 股票基础信息表 CREATE TABLE stock_basic ( stock_code TEXT PRIMARY KEY, stock_name TEXT, industry TEXT, list_date TEXT ); -- 日行情表 CREATE TABLE daily_quote ( stock_code TEXT, trade_date TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL, volume INTEGER, amount REAL, adjust_flag TEXT, PRIMARY KEY (stock_code, trade_date) ); -- 研报元数据表 CREATE TABLE report_meta ( id INTEGER PRIMARY KEY AUTOINCREMENT, stock_code TEXT, agency TEXT, rating TEXT, title TEXT, pub_time TEXT, summary TEXT, source_url TEXT, created_at TEXT );注意daily_quote用了联合主键stock_code加trade_date这样重复跑同一交易日的数据时可以直接做INSERT OR REPLACE不会产生脏数据。增量更新的逻辑很简单每次更新前查询表里已有的最大trade_date只拉取从这个日期到今天的部分避免全量重跑浪费时间。第一次初始化全量数据时A股五千多只股票加一年行情大概要跑十几分钟之后每天增量更新只需要几十秒。3.3 研报与行情如何联动数据有了联动就有意思了。我做了一个最简单的联动视图选中某只股票左侧显示最近几条研报信息右侧显示对应时间段的K线并在K线上用竖线标记出研报发布日期。这样一眼就能看出某家机构发布评级上调之后股价是涨是跌。再进一步可以统计某机构过去半年的评级调整胜率——比如“华泰证券发布买入评级后的20个交易日平均收益是多少”。这种统计在研报平台上是做不了的因为数据在别人那里你只能看不能算。但自己做知识库加行情库这就是一个SQL查询的事。我还算过不同行业研报发布密度的热力图用来感受资金关注度变化。强调一下这些分析本质是信息整理不构成任何投资建议。工具只能告诉你“发生了什么、哪些信息相关”顶多给你一个可回溯的研究线索真正的判断和决策还是得靠自己。4. 知识库与RAG智能对话分析4.1 为什么我选RAG而不是微调大模型研报整合工具做到知识库和行情联动之后目前卡在检索环节。我想问“新能源行业近期被上调评级的个股有哪些”SQL能查到但我想问“从中长期视角看哪些机构认为新能源行业已经触底”SQL就很难表达了因为这是语义问题。所以自然就想到引入大模型。大模型接入有两种主流方案微调和RAG。我选了RAG原因有三点研报更新频率高每周都有新数据注入微调模型没法跟着数据实时变成本也高。RAG可以让模型指出“答案来自哪篇文章、发布时间是什么”这个对投研场景非常重要。微调模型做不到可追溯。我的知识库规模不大用嵌入向量即可完成检索完全没必要为这个场景付出训练成本。本质上RAG做的事情很简单把问题转成向量去知识库里找语义相近的文本片段再把找到的文本片段拼进提示词让大模型基于片段回答。4.2 文本切分、向量化与检索知识库里的研报摘要有的长有的短。直接把整篇摘要作为一条记录丢给模型效果会很差因为模型上下文窗口有限而且检索粒度太粗。我按“段落”维度做了切分每段控制在500字左右重叠部分设50字尽量让语义完整的段落不被切断。from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, ] ) texts [] for _, row in report_meta.iterrows(): content f{row[title]}\n{row[summary]} chunks splitter.split_text(content) for chunk in chunks: texts.append({ text: chunk, metadata: { stock_code: row[stock_code], agency: row[agency], pub_time: row[pub_time], } })向量化模型我用了国产文本嵌入模型bge-m3当时考虑的是它在中文金融文本上的表现比OpenAI的text-embedding-ada-002更稳而且可以本地部署不产生token费用。检索部分用的也是开源的向量库chroma个人数据量根本不需要上eschroma足够启动快、API简单。import chromadb from chromadb.utils import embedding_functions client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection( namereport_rag, embedding_functionembedding_functions.SentenceTransformerEmbeddingFunction( model_nameBAAI/bge-m3 ) ) # 问答时检索 results collection.query( query_texts[新能源行业触底反弹], n_results5 )一个很重要的调优点是纯向量检索在股票场景下不够用。如果你问的是“600519”向量检索根本不知道那是什么含义这时候需要先把问题里的股票代码和名称做实体识别转成筛选项再去检索。我的做法是先做一个轻量规则实体识别——把常见股票名称和代码互相映射命中后在查询时加metadata filter只用向量检索辅助排序。这样“贵州茅台近期的研报观点”就能稳定命中正确的那几只股票的内容。4.3 对话回答的工程化细节对话接口我用FastAPI包了一层专门做两件事检索和生成。生成的大模型我选的是支持长上下文的开源模型跑在本地推理这样对话数据不外传做研究更安心。提示词模板很关键尤其是约束模型不要编造。我用的模板大概是你是A股研报分析助手。请基于以下检索到的研报片段回答问题。 要求 1. 回答只能引用检索片段中出现的观点并注明来源机构与发布时间。 2. 如果检索片段不足以支撑结论明确回答“暂无相关研报支持该观点”。 3. 不给出任何投资建议只做信息整理。 检索片段 {document} 问题 {question}工程上必须设置两个兜底逻辑。第一检索不到内容时直接让模型说“暂无相关信息”而不是硬着头皮编。第二参考来源用单独字段返回给前端展示标题、机构、时间一目了然。加了这两个逻辑之后对话结果的可用性提升非常明显基本不会再出现“一本正经胡说八道”的局面。5. 从爬虫到对话的系统串联与部署5.1 整体架构和目录设计单机能跑通的系统没必要一开始就上微服务。我的最终架构分成四层但都放进同一个项目里采集层定期运行爬虫脚本和行情更新脚本存储层SQLite保存结构化的研报和行情chroma保存向量片段服务层FastAPI提供查询接口、行情接口、对话接口展示层Streamlit搭建的Web界面目录结构是这样的stock-assistant/ ├── crawler/ │ ├── report_spider.py # 研报爬虫 │ └── quote_updater.py # 行情更新 ├── database/ │ ├── models.py # 表结构定义 │ └── connection.py # SQLite连接 ├── rag/ │ ├── embed.py # 向量化、入库 │ ├── retriever.py # 检索逻辑 │ └── chat_engine.py # 对话生成 ├── api/ │ └── main.py # FastAPI路由 └── ui/ └── app.py # Streamlit界面5.2 核心链路代码FastAPI端最核心的对话接口长这样from fastapi import FastAPI from pydantic import BaseModel from rag.retriever import retrieve_reports from rag.chat_engine import generate_answer app FastAPI() class ChatRequest(BaseModel): question: str app.post(/api/chat) def chat(req: ChatRequest): chunks, sources retrieve_reports(req.question, top_k5) answer generate_answer(req.question, chunks) return { answer: answer, sources: sources }Streamlit界面则很简单页面顶部显示输入框下面显示回答和来源左侧可以选股票代码看行情图右侧有每日更新的研报列表。用Streamlit最大的好处是不用写任何前端代码半小时就能出一个可用的分析界面。5.3 实测效果及性能评估整个系统跑通之后我积累的数据情况是研报元数据约4万条行情数据约120万行向量片段约14万个。对话响应时间在本地推理环境下大约3到5秒其中检索占不到0.2秒大头基本都耗在模型生成上。查询行情接口的响应在100毫秒以内完全够用。性能瓶颈在向量库的写入阶段——首次全量向量化4万条研报摘要在CPU上跑了近两个小时。我的建议是向量化这种重活可以设计成分批处理机制每次只处理新增的研报而不是每次都重新embed全部。增量向量化要按发布时间或者ID做断点防止重复消耗算力。这个优化做完之后每天新增的几十条研报只需要一分钟不到就能完成入库。6. 数据链路中的真实踩坑与避坑经验6.1 限流被封的教训第一次跑完整流程时我为了一时爽直接上了线程池并发爬列表页最后被公开站点限流封了IP。那天晚上排查了很久最后把并发删掉改成串行加随机延时才恢复。这件事让我意识到对公开数据源保持克制就是对自己系统的保护。之后我只保留了3秒以上的随机延时和断点续爬功能——如果爬虫中途挂了重启时能从上次记录的位置继续而不是从头再来。断点续爬的实现就是记录每个列表页的页码每抓完一页写入本地进度文件这样成本很低收益却很大。6.2 PDF研报解析乱码的坑有些公开发布渠道的研报是PDF格式摘要不直接展示在页面上。为了把摘要也收进来我引入过PDF解析。结果发现PDF转文本最大的问题不是技术选型而是金融PDF里经常有两栏排版、页眉页脚、表格混排直接抽取出来的文本顺序全是乱的。后来我的方案是一律优先抓HTML页面上的纯文本摘要实在拿不到摘要的才用PDF解析而且解析后必须人工抽查样本。不要盲目信任PDF解析结果尤其是那种页眉里包含股票代码或机构名称的内容特别容易被误当成正文。这个教训我印象很深刻当时解析出来的文本里有一堆重复的页脚信息放进向量库以后检索效果非常差。6.3 RAG幻觉的治理对话上线第一天我问它“某股票有没有机构给出卖出评级”它非常自信地回答“没有最近评级均为买入或增持”。但我去数据库一查明明有一条卖出评级。问题出在检索环节向量检索的top 5返回里没有包含那条卖出研报模型只基于给定的5条片段作答自然就说“没有”。这是一个典型的RAG评价盲区。我用了两个办法解决提高检索召回量从top 5调整到top 10输出时在提示词里要求区分“明确证据”和“推测信息”。在对话结果里强制附上“检索来源”让用户自己判断可信度而不是把大模型的输出当成结论。后来我更进一步把“卖出”这种强评级的词汇作为关键词加入检索权重只要问题里出现评级关键词就提前筛掉不含该评级的候选片段。这个办法能大幅减少类似幻觉。6.4 行情数据的复权与停牌坑复权问题我在早期差点犯了大错。如果我直接用不复权的历史价格做收益率统计遇到除权除息日股价会凭空跳空下跌导致分析结果完全失真。后来我统一用tushare的复权因子做了后复权处理所有统计都以复权价格计算。另外停牌股在行情表里会出现交易日期缺失做时间序列分析时不能直接用pandas默认的连续date_range硬灌要用实际交易日历对齐。A股有节假日休市安排最靠谱的方式是直接用行情数据里出现过的日期做索引而不是拿自然日去推测。6.5 关于合规的一点最终提醒最后再说一个老生常谈但很容易被忽视的问题。自己做投研工具数据源务必选公开的、合法授权的。爬虫可以帮你省去手工复制粘贴的时间但前提是它只用在你理直气壮能看的内容上。对需要登录、需要付费、明确禁止抓取的数据哪怕技术上都绕得开也不建议碰。这个项目我用了很久数据源一直很稳定这本身就是一种收益。做完这套系统我最大的体会是工具的核心价值不在于“用了多厉害的大模型”而在于数据清洗做得到不到位。研报爬虫、实时行情、知识库、RAG对话每一个环节单拿出来都不算新鲜但把它们的衔接缝补好从零把数据流水线跑通才是真正花时间的地方。如果你也想做一个类似的A股智能助手我建议从规模最小的垂直场景开始先把“某只股票最近一个月被哪些机构覆盖”这种简单问题完整跑通再逐步加行业维度、对话分析、行情联动。一口吃不成胖子但一口一口吃这个系统会变得比任何现成研报平台都顺手。