第二章:多源信息获取——研报数据源分类与实战策略

做研报Agent,第一步就是搞数据。说白了,你连数据都拿不到,后面那些花里胡哨的摘要、分析全是空谈。

我个人习惯把研报数据源分成四大类:PDF、网页、数据库、API。每一类都有它的脾气,也有对应的驯服方法。今天咱们就挨个盘一遍。

2.1 四大数据源分类与特点

先看一张图,帮你快速建立整体认知:

研报数据源分类体系 四大数据源 PDF 格式固定 文本+图表混合 解析难度:中 网页 动态加载 反爬虫多 解析难度:高 数据库 结构化数据 查询效率高 解析难度:低 API 标准化接口 有调用限制 解析难度:低 核心挑战:格式多样性 + 反爬虫 + 数据一致性 目标:统一结构化存储,供后续摘要生成 多源信息融合 → 统一摘要生成

核心原则:不同数据源,用不同的策略。没有银弹。

2.2 PDF研报——最顽固的老大哥

PDF是研报的经典载体。券商、研究所出的正式报告,90%都是PDF。但PDF这玩意儿,天生就不是给机器读的。

我记得刚入行时,接了个项目要批量解析5000份PDF研报。我天真地以为用PyPDF2一把梭就行,结果跑出来一堆乱码和表格错位。嗯,从那以后我再也不敢小看PDF了。

2.2.1 PDF解析策略

工具 适用场景 优点 缺点
PyMuPDF (fitz) 文本型PDF 速度快,保留结构 对扫描件无效
pdfplumber 表格密集型 表格提取精准 大文件慢
OCR (Tesseract) 扫描件/图片型PDF 能处理任何PDF 速度慢,有识别误差
camelot 复杂表格 基于视觉的表格解析 依赖Ghostscript

我的经验:先判断PDF类型。用PyMuPDF快速扫描,如果提取的文本长度小于页面字符数的30%,大概率是扫描件,直接上OCR。

# 一个实用的PDF解析判断逻辑
import fitz  # PyMuPDF

def detect_pdf_type(pdf_path):
    doc = fitz.open(pdf_path)
    total_chars = 0
    for page in doc:
        text = page.get_text()
        total_chars += len(text)
    doc.close()
    
    # 粗略判断:如果平均每页少于200字符,可能是扫描件
    if total_chars / len(doc) < 200:
        return "scan"  # 需要OCR
    else:
        return "text"  # 直接解析

2.3 网页研报——动态加载与反爬虫的博弈

网页端的研报,比如东方财富、同花顺、雪球这些平台,数据更新快,但也是最难搞的。为什么?因为人家有反爬虫啊。

你想想看,一个网站每天几百万访问量,要是谁都能随便爬,服务器早崩了。所以反爬虫是合理的,但我们做数据采集也得有办法。

2.3.1 爬虫策略

  • 静态页面:用requests + BeautifulSoup,简单直接
  • 动态加载页面:用Selenium或Playwright模拟浏览器
  • 接口直连:抓包分析,直接调后端API(我最喜欢的方式)

注意:千万不要用Selenium去爬大量数据。我见过有人开50个Chrome实例,结果服务器内存直接爆了。Selenium适合小规模、需要交互的场景。

2.3.2 反爬虫应对三板斧

我曾经被一个券商网站封了整整一个IP段,就是因为没做好伪装。后来我总结了一套策略:

  1. 请求头伪装:User-Agent、Referer、Cookie一个都不能少
  2. 请求频率控制:随机延时 + 限速,别像个机器人一样精准
  3. 代理IP池:准备一批高质量代理,轮换使用
# 一个带反爬虫应对的请求示例
import requests
import time
import random

def safe_request(url, headers=None, proxies=None):
    default_headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Accept": "text/html,application/xhtml+xml",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Referer": "https://www.example.com/",
    }
    if headers:
        default_headers.update(headers)
    
    # 随机延时 1-3 秒
    time.sleep(random.uniform(1, 3))
    
    try:
        resp = requests.get(url, headers=default_headers, proxies=proxies, timeout=10)
        resp.raise_for_status()
        return resp
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

避坑指南:我曾经遇到过网站返回200但内容是空壳的情况。后来发现是网站检测到了爬虫特征,返回了假数据。解决办法是:检查返回内容的长度和关键词,如果异常就换IP重试。

2.4 数据库与API——结构化数据的优雅获取

数据库和API,说白了就是别人帮你把数据整理好了,你直接去拿就行。但这里也有坑。

2.4.1 数据库接入

很多金融数据服务商提供数据库直连,比如Wind、Choice、聚宽。直接用SQL查询,效率最高。

# 以MySQL为例,查询某只股票的研报数据
import pymysql

conn = pymysql.connect(
    host="your_host",
    user="your_user",
    password="your_password",
    database="research_db"
)

cursor = conn.cursor()
sql = """
SELECT report_title, publish_date, content 
FROM research_reports 
WHERE stock_code = '600519' 
  AND publish_date >= '2024-01-01'
LIMIT 10
"""
cursor.execute(sql)
results = cursor.fetchall()
cursor.close()
conn.close()

2.4.2 API调用

API是最省心的方式,但通常有调用频率限制和收费。我建议:

  • 先看文档,搞清楚限流策略
  • 做好缓存,避免重复请求
  • 加异常重试机制
# API调用示例(带重试)
import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_report_via_api(api_key, report_id):
    url = f"https://api.example.com/v1/reports/{report_id}"
    headers = {"Authorization": f"Bearer {api_key}"}
    resp = requests.get(url, headers=headers, timeout=10)
    resp.raise_for_status()
    return resp.json()

我的建议:如果条件允许,优先用API和数据库。爬虫虽然免费,但维护成本高。你想想看,网站改个版,你的爬虫可能就废了。API和数据库相对稳定得多。

2.5 多源数据统一策略

数据拿到手了,但格式五花八门。PDF是文本块,网页是HTML,数据库是表格,API是JSON。怎么统一?

我的做法是:定义一个统一的数据模型,所有源都转换成这个模型。

# 统一研报数据模型
class ResearchReport:
    def __init__(self, source, source_type):
        self.source = source          # 来源标识
        self.source_type = source_type # pdf/web/db/api
        self.title = ""
        self.publish_date = ""
        self.author = ""
        self.content = ""             # 纯文本内容
        self.tables = []              # 表格数据
        self.metadata = {}            # 额外元数据
    
    def to_dict(self):
        return {
            "source": self.source,
            "title": self.title,
            "date": self.publish_date,
            "content": self.content,
            "tables": self.tables,
        }

每个数据源写一个适配器,负责把原始数据转换成这个模型。这样后续的摘要生成模块就不用关心数据是从哪来的了。

小技巧:在适配器里加一个校验函数,检查必填字段是否完整。我遇到过API返回的数据缺少标题,结果摘要生成直接报错。加个校验就能提前发现问题。

好了,多源信息获取这块就聊到这儿。数据拿到了,下一步就是怎么从这些乱七八糟的内容里提炼出精华——那就是摘要生成的事了。


公众号:蓝海资料掘金营,微信deep3321