批量下载A股上市公司报告 Skill

功能概述

批量下载PDF（默认）: 按行业/公司/年份下载年报、中报、季报
提取MD&A（可选）: 从年报PDF中提取"核心竞争力分析"和"公司未来发展的展望"文本，输出CSV

触发条件

"下载XX公司年报/中报/季报"
"批量下载XX行业报告"
"提取XX年报的MD&A" / "提取核心竞争力分析"
"下载报告并提取管理层讨论与分析"

输入

公司列表（股票代码+名称）或行业分类
年份范围（如2021-2025）
报告类型（年报/中报/一季报/三季报，默认全部）
可选: 是否提取MD&A文本（核心竞争力分析 + 未来发展展望），仅对年报有效
输出目录（默认桌面）

完整工作流

阶段1: 获取公司名单

场景A: 用户提供明确名单 → 直接使用
场景B: 用户说"XX行业所有A股" → 用 akshare 获取全A股后按行业筛选

# 深交所
import akshare as ak
df_sz = ak.stock_info_sz_name_code()  # 含"所属行业"列, 格式如"C 制造业"
chiznext = df_sz[df_sz["A股代码"].str.startswith(("300","301"))]
manufacturing = chiznext[chiznext["所属行业"].str.startswith("C")]

# 上交所 (如果需要)
df_sh = ak.stock_info_sh_name_code()

证监会行业代码: C=制造业, J=金融业, K=房地产, I=信息技术, ...
注意: akshare 的 stock_zh_a_spot_em() 走东方财富API，容易限流封IP，优先用上面两个。

阶段2: 搜索报告链接

数据源: 巨潮资讯网 (cninfo.com.cn) — 证监会指定信息披露网站

API调用:

import requests

url = "https://www.cninfo.com.cn/new/hisAnnouncement/query"
headers = {
    "User-Agent": "Mozilla/5.0",
    "Content-Type": "application/x-www-form-urlencoded",
    "Origin": "https://www.cninfo.com.cn",
    "Referer": "https://www.cninfo.com.cn/new/commonUrl?url=disclosure/list/notice",
}

# 关键: 公司名必须字间加空格!
spaced_name = " ".join(list("五粮液"))  # → "五 粮 液"

data = {
    "pageNum": 1,
    "pageSize": 10,
    "tabName": "fulltext",
    # 注意: 不设 column/plate 参数!
    "searchkey": f"{spaced_name} 年度报告",  # 公司名+报告类型, 不带年份数字
    "seDate": "2022-01-01~2022-04-30",       # 年报在次年1-4月
}

resp = requests.post(url, data=data, headers=headers, timeout=15)
result = resp.json()

日期范围规则:

| 报告类型 | 披露截止日 | 推荐搜索窗口 | |----------|-----------|-------------| | 年报(year Y) | Y+1年4月30日 | {Y+1}-01-01 ~ {Y+1}-04-30 | | 中报/一季报/三季报 | 当年 | {Y}-01-01 ~ {Y+1}-06-30 |

过滤逻辑 (搜索返回结果后的二次筛选):

BAD_KW = ["社会责任", "ESG", "英文", "English", "摘要", "更正", "补充",
          "问询", "审核", "修订", "预告", "快报", "提示性", "说明会",
          "半年",   # 搜季报时排除半年报干扰
          "年度",   # 搜季报时排除年报干扰
          ]

for ann in announcements:
    if ann["secCode"] != target_code:   # 1. 匹配股票代码
        continue
    if report_keyword not in ann["announcementTitle"]:  # 2. 匹配报告类型
        continue
    if str(year) not in ann["announcementTitle"]:       # 3. 匹配年份
        continue
    if any(kw in ann["announcementTitle"] for kw in BAD_KW):  # 4. 排除非正文
        continue
    # → 找到了

请求频率: 每次 >= 0.3秒间隔，避免限流。CNINFO 限流比东方财富宽松得多。

阶段3: 下载PDF

pdf_url = "https://static.cninfo.com.cn/" + ann["adjunctUrl"]

dl_headers = {
    "User-Agent": "Mozilla/5.0",
    "Referer": "https://www.cninfo.com.cn/",
}

resp = requests.get(pdf_url, headers=dl_headers, timeout=60)
if resp.status_code == 200 and len(resp.content) > 10000:
    with open(f"{code}_{name}_{year}_{label}.pdf", "wb") as f:
        f.write(resp.content)

断点续传: 用 JSON 缓存已下载的文件名，启动时加载跳过。

阶段4（可选）: 提取MD&A文本

仅对年报有效。从PDF中提取"管理层讨论与分析"章节的两个子节：

提取内容:

a. 核心竞争力分析
b. 公司未来发展的展望

技术方案:

方案1: pypdf + 多进程（推荐） — 已大规模验证

from pypdf import PdfReader
from concurrent.futures import ProcessPoolExecutor
import io, re

def extract_text(pdf_data):
    """pypdf提取文本，纯Python稳定不崩"""
    parts = []
    reader = PdfReader(io.BytesIO(pdf_data))
    for page in reader.pages:
        t = page.extract_text()
        if t: parts.append(t)
    return "\n".join(parts)

def find_mda(text):
    """定位MD&A章节（跳过目录区前3000字符避免交叉引用）"""
    skip = min(3000, len(text) // 10)
    for m in re.finditer(r'第[三四五六]节\s*管理层讨论[与和]分析', text[skip:]):
        start = skip + m.start()
        return text[start:start + 100000]  # 取10万字
    return None

def extract_subsections(mda_text):
    """提取核心竞争力和未来发展展望"""
    core, outlook = "", ""
    m = re.search(r'核心竞争力分析', mda_text)
    if m:
        # 到下一个子章节为止
        next_sec = re.search(r'\n\s*[（(][一二三四五六七八九十]+[)）]\s*\S',
                            mda_text[m.end():m.end()+15000])
        end = m.end() + next_sec.start() if next_sec else m.end() + 12000
        core = mda_text[m.end():end].strip()
    
    m = re.search(r'公司未来发展[的之]展望', mda_text)
    if m:
        # 到下一个大节为止
        next_major = re.search(r'\n\s*第[一二三四五六七八九十]+\s*[节章]',
                              mda_text[m.end():m.end()+20000])
        end = m.end() + next_major.start() if next_major else m.end() + 15000
        outlook = mda_text[m.end():end].strip()
    
    return core, outlook

方案2: 避免的坑 — 不要用这些库:

pdfplumber: 单家25秒，太慢（978家预估5-6小时）
pypdfium2: C++扩展，28倍快但遇到畸形PDF会segfault崩溃，Python无法捕获

并发策略: ProcessPoolExecutor(max_workers=8) — 突破GIL

线程池 (ThreadPoolExecutor): 受GIL限制，实测仅4家/分钟
进程池 (ProcessPoolExecutor): 真正并行，实测43家/分钟

输出CSV格式: | 股票代码 | 股票名称 | 年报标题 | 年报日期 | 核心竞争力分析 | 公司未来发展的展望 | cc字数 | fo字数 | 状态 | |---------|---------|---------|---------|-------------|------------------|--------|--------|------|

核心技术难点与解决方案

坑1: CNINFO 搜索引擎分词 (最坑)

现象: 搜"五粮液年度报告"返回0条，搜"五粮液年度报告"返回2条。
原因: CNINFO将无空格的公司名当做单个token，加空格后才能与后续关键词正确组合。
解决: 所有公司名搜索前强制字间加空格:

spaced = " ".join(list(name.replace(" ", "")))

坑2: 年报标题不含公司名

现象: 五粮液/洋河等公司的年报标题是"2021年年度报告"，不包含公司名。如果过滤逻辑依赖"公司名 in 标题"会漏掉。
解决: 只用 secCode 做公司匹配，不用标题中的公司名。

坑3: "年度社会责任报告" 伪装成年报

现象: 搜索"年度报告"时，CNINFO优先返回"年度社会责任报告"(CSR)。
解决: BAD_KW中加入"社会责任"、"ESG"排除。

坑4: column/plate 参数导致搜不到

现象: 设了 column="sse" 或 plate="sh" 后部分公司搜不到（如五粮液）。
解决: 不传这两个参数，CNINFO会自动匹配。

坑5: CNINFO 的 stock 参数无效

现象: 无论传 stock="000858" 还是 stock="000858,SZ" 都返回0结果。
解决: 放弃 stock 参数，只用 searchkey 全文搜索。

坑6: 东方财富 push2 API 限流

现象: RemoteDisconnected, 连续几次请求后IP被封。
解决: 获取公司名单用 akshare 的深交所官方数据；搜报告用 CNINFO（更宽松）。

坑7: 2021-2022年季报搜索不全

现象: 2023-2025的季报能搜到，2021-2022的大量缺失。
原因: CNINFO搜索索引对较早公告的覆盖不完整（非披露缺失，是索引缺失）。
解决: 目前无解，属正常损耗，年报和中报不受影响。

坑8: pdfplumber太慢 (MD&A提取)

现象: 单份年报PDF文本提取耗时25秒，978家预估5-6小时。
解决: 换 pypdf（10秒/家，2.6x加速），配合多进程（43家/分钟）。

坑9: pypdfium2闪退 (MD&A提取)

现象: C扩展库遇到某些PDF直接segfault，exit code 3，Python try/except完全无效。
解决: 用纯Python的 pypdf 替代，牺牲一点速度换稳定性。

坑10: 线程版MD&A提取只有4家/分钟

现象: ThreadPoolExecutor 15线程实际吞吐仅4家/分钟。
原因: pypdf文本提取是CPU密集型，Python GIL限制了线程并行。
解决: 换 ProcessPoolExecutor（8进程），真正突破GIL，达到43家/分钟。

输出规范

文件命名: {股票代码}_{公司名}_{年份}_{报告类型}.pdf
示例: 000858_五粮液_2023_年报.pdf
可选统计: JSON记录每家公司各类型下载数量

批量下载并分析所需A股上市公司报告