凌晨两点，某中型私募基金的投研总监老张盯着屏幕——明天要给客户路演，但 47 份年报的核心段落还没摘出来。手动复制粘贴到凌晨，眼睛都快瞎了。
这是 2023 年的故事。2026 年的今天，我帮老张他们搭建了一套 证券投研摘要平台，核心流程全自动化：PDF 财报进去，结构化摘要 + 风险提示 + 投资亮点 + 合规审计日志全出来，准确率比人工高，还不用熬夜。
本文完整记录这套系统的技术架构、代码实现、成本核算，以及我在实际部署中踩过的坑。适合券商自营、公募基金、阳光私募、上市公司 IR 部门参考。
一、场景痛点与方案设计
1.1 投研场景的核心挑战
- 文档体量巨大：一份年报 PDF 通常 150-400 页，单次解析需要 128K 以上上下文窗口
- 批量处理需求：每季度财报季，分析师需要在 3-5 天内完成 50-100 家公司的初步筛选
- 合规留痕强制要求：证监会要求研报生成过程可追溯、可审计，修改需留痕
- 成本敏感：传统方案用 GPT-4o 解析长文档，单份成本超过 ¥8，按月计算吓人
1.2 整体技术架构
财报 PDF → PDF 解析器 → 文本清洗 → DeepSeek 长文本分析 → 结构化摘要
↓
合规审计日志 (MySQL)
↓
摘要导出 (JSON/Excel)选择 HolySheep AI 作为底层 LLM 调用平台，核心原因：DeepSeek V3.2 output 价格仅 $0.42/MTok，是 Claude Sonnet 4.5 的 1/35，用在我们每月 2000 万 Token 的场景，月度成本从 ¥46 万降到 ¥2.3 万，ROI 直接拉满。
二、核心代码实现
2.1 环境准备与依赖安装
# Python 3.11+
pip install openai pdfplumber python-docx pymysql python-dateutil rich2.2 PDF 财报解析模块
import pdfplumber
import re
from typing import List, Dict
class AnnualReportParser:
"""年报 PDF 解析器 - 提取文本并做基础清洗"""
def __init__(self, min_paragraph_len: int = 100):
self.min_paragraph_len = min_paragraph_len
# 需要重点关注的章节
self.key_sections = [
'经营情况讨论与分析',
'主要经营业务',
'财务报表',
'重大事项',
'未来展望',
'风险因素'
]
def extract_text(self, pdf_path: str) -> Dict[str, str]:
"""提取 PDF 各页面文本，按章节组织"""
sections = {section: [] for section in self.key_sections}
sections['全文'] = []
current_section = '全文'
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
text = page.extract_text()
if not text:
continue
# 检测章节标题切换
for section in self.key_sections:
if section in text[:200]: # 只检查页面开头
current_section = section
break
sections[current_section].append(text)
sections['全文'].append(text)
# 合并并清洗
result = {}
for section, texts in sections.items():
combined = '\n'.join(texts)
result[section] = self._clean_text(combined)
return result
def _clean_text(self, text: str) -> str:
"""基础文本清洗"""
# 移除多余空白
text = re.sub(r'\s+', ' ', text)
# 移除页码和页眉页脚常见模式
text = re.sub(r'\n\d+\s*/?\s*\d+\n', '\n', text)
# 移除连续短句（通常是表格行）
lines = [l.strip() for l in text.split('\n') if len(l.strip()) > self.min_paragraph_len]
return '\n'.join(lines)
使用示例
parser = AnnualReportParser()
report_data = parser.extract_text('/data/600519_2024_annual.pdf')
main_business_text = report_data.get('主要经营业务', '')[:8000] # 取前 8000 字符
print(f"提取文本量: {len(main_business_text)} 字符")2.3 DeepSeek 财报分析 + 合规留痕
import openai
from datetime import datetime
import json
import pymysql
from typing import Optional
HolySheep API 配置
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class InvestmentResearchAnalyzer:
"""投研摘要生成器 - 支持批量处理和合规留痕"""
def __init__(self, api_key: str, base_url: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url=base_url
)
self.db_config = {
'host': 'localhost',
'user': 'research_user',
'password': 'your_password',
'database': 'investment_research',
'charset': 'utf8mb4'
}
# 系统提示词 - 引导模型输出结构化摘要
self.analysis_prompt = """你是一位专业的证券分析师，擅长从年报中提取关键投资信息。
请从以下财报文本中提取并生成结构化摘要，格式严格遵循：
公司基本信息
- 公司名称：[提取]
- 报告期：[YYYY年]
核心财务数据
- 营业收入：[数值]亿元，同比[增长/下降]%
- 净利润：[数值]亿元，同比[增长/下降]%
- 毛利率：[数值]%
- ROE：[数值]%
业务亮点
1. [亮点1]
2. [亮点2]
3. [亮点3]
主要风险
1. [风险1]
2. [风险2]
投资评级建议
[维持买入/增持/中性/减持/卖出]，理由：[简述]
请确保数据准确，只输出客观事实，不要添加任何投资建议。"""
def analyze_report(self, company_name: str, report_text: str,
user_id: int, batch_id: Optional[str] = None) -> Dict:
"""分析单份财报并记录审计日志"""
# 生成批次 ID
if not batch_id:
batch_id = f"BATCH_{datetime.now().strftime('%Y%m%d%H%M%S')}"
# 调用 DeepSeek V3.2 进行分析
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": self.analysis_prompt},
{"role": "user", "content": f"公司名称：{company_name}\n\n财报内容：\n{report_text[:60000]}"}
],
temperature=0.1, # 低温度确保准确性
max_tokens=4000,
timeout=120
)
# 提取结果
analysis_result = response.choices[0].message.content
usage = response.usage
# 计算成本 (HolySheep DeepSeek V3.2: $0.42/MTok output)
cost_usd = (usage.completion_tokens / 1_000_000) * 0.42
cost_cny = cost_usd * 7.3 # HolySheep 汇率 1:7.3
# 写入合规审计日志
audit_log_id = self._save_audit_log(
batch_id=batch_id,
user_id=user_id,
company_name=company_name,
model="deepseek-v3.2",
input_tokens=usage.prompt_tokens,
output_tokens=usage.completion_tokens,
cost_usd=cost_usd,
request_content=report_text[:5000], # 只记录前 5000 字符
response_content=analysis_result[:3000]
)
return {
'batch_id': batch_id,
'audit_log_id': audit_log_id,
'result': analysis_result,
'usage': {
'prompt_tokens': usage.prompt_tokens,
'completion_tokens': usage.completion_tokens,
'cost_usd': round(cost_usd, 4),
'cost_cny': round(cost_cny, 2)
},
'timestamp': datetime.now().isoformat()
}
def batch_analyze(self, reports: List[Dict], user_id: int) -> List[Dict]:
"""批量分析多份财报"""
results = []
batch_id = f"BATCH_{datetime.now().strftime('%Y%m%d%H%M%S')}"
for i, report in enumerate(reports):
print(f"处理第 {i+1}/{len(reports)} 份: {report['company_name']}")
try:
result = self.analyze_report(
company_name=report['company_name'],
report_text=report['content'],
user_id=user_id,
batch_id=batch_id
)
results.append(result)
except Exception as e:
print(f"处理失败: {report['company_name']}, 错误: {str(e)}")
results.append({
'company_name': report['company_name'],
'status': 'failed',
'error': str(e)
})
return results
def _save_audit_log(self, batch_id: str, user_id: int, company_name: str,
model: str, input_tokens: int, output_tokens: int,
cost_usd: float, request_content: str,
response_content: str) -> int:
"""保存合规审计日志到数据库"""
conn = pymysql.connect(**self.db_config)
try:
with conn.cursor() as cursor:
sql = """
INSERT INTO audit_logs (
batch_id, user_id, company_name, model_name,
input_tokens, output_tokens, cost_usd,
request_snapshot, response_snapshot,
created_at, modified_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
now = datetime.now()
cursor.execute(sql, (
batch_id, user_id, company_name, model,
input_tokens, output_tokens, cost_usd,
request_content, response_content,
now, now
))
conn.commit()
return cursor.lastrowid
finally:
conn.close()
使用示例
analyzer = InvestmentResearchAnalyzer(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
单份分析
single_result = analyzer.analyze_report(
company_name="贵州茅台",
report_text=report_data.get('主要经营业务', ''),
user_id=1
)
print(f"单份成本: ¥{single_result['usage']['cost_cny']}")
批量分析
batch_reports = [
{'company_name': '宁德时代', 'content': '...' * 1000},
{'company_name': '比亚迪', 'content': '...' * 1000},
# ... 更多公司
]
batch_results = analyzer.batch_analyze(batch_reports, user_id=1)三、成本对比：HolySheep vs 官方 API
以我实际运行数据为例：每月处理 500 份财报，每份约 8 万 Token 输入 + 4000 Token 输出。
| 指标 | OpenAI GPT-4o | Anthropic Claude 3.5 | HolySheep DeepSeek V3.2 |
|---|---|---|---|
| Output 价格 | $15/MTok | $15/MTok | $0.42/MTok |
| 月度 Input Token | 40 亿 | 40 亿 | |
| 月度 Output Token | 2 亿 | 2 亿 | |
| 月度 API 成本 | ¥219,000 | ¥219,000 | ¥6,132 |
| 年度成本 | ¥262.8 万 | ¥262.8 万 | ¥7.36 万 |
| 节省比例 | - | - | 97% |
| 国内延迟 | 200-500ms | 200-500ms | <50ms |
| 充值方式 | 国际信用卡 | 国际信用卡 | 微信/支付宝 |
HolySheep 的汇率是官方报价 ¥7.3=$1，无损结算。我对比测试过实际到账金额，充值 ¥730 确实到账 $100，没有任何隐形损耗。
四、完整批量处理脚本
import os
import json
from pathlib import Path
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
class BatchReportProcessor:
"""批量财报处理流水线"""
def __init__(self, holysheep_api_key: str):
self.analyzer = InvestmentResearchAnalyzer(
api_key=holysheep_api_key,
base_url="https://api.holysheep.ai/v1"
)
self.parser = AnnualReportParser()
def process_directory(self, input_dir: str, output_dir: str,
max_workers: int = 5) -> Dict:
"""处理目录下所有 PDF 文件"""
input_path = Path(input_dir)
pdf_files = list(input_path.glob('*.pdf'))
print(f"发现 {len(pdf_files)} 份 PDF 文件")
all_results = []
total_cost_cny = 0
start_time = datetime.now()
# 使用线程池并发处理（注意 API 速率限制）
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {}
for pdf_file in pdf_files:
future = executor.submit(
self._process_single_file,
pdf_file,
output_dir
)
futures[future] = pdf_file.name
for future in as_completed(futures):
filename = futures[future]
try:
result = future.result()
all_results.append(result)
total_cost_cny += result['usage']['cost_cny']
print(f"✓ 完成: {filename}, 成本: ¥{result['usage']['cost_cny']:.2f}")
except Exception as e:
print(f"✗ 失败: {filename}, 错误: {str(e)}")
elapsed = (datetime.now() - start_time).total_seconds()
# 生成汇总报告
summary = {
'total_files': len(pdf_files),
'success_count': len([r for r in all_results if r.get('status') != 'failed']),
'failed_count': len([r for r in all_results if r.get('status') == 'failed']),
'total_cost_cny': round(total_cost_cny, 2),
'elapsed_seconds': round(elapsed, 1),
'avg_cost_per_file': round(total_cost_cny / len(all_results), 2) if all_results else 0,
'timestamp': datetime.now().isoformat()
}
# 保存结果
summary_path = Path(output_dir) / f"summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(summary_path, 'w', encoding='utf-8') as f:
json.dump({'summary': summary, 'details': all_results}, f, ensure_ascii=False, indent=2)
return summary
def _process_single_file(self, pdf_path: Path, output_dir: str) -> Dict:
"""处理单个 PDF 文件"""
# 解析 PDF
report_data = self.parser.extract_text(str(pdf_path))
# 提取公司名称（从文件名或内容）
company_name = pdf_path.stem.split('_')[0]
# 合并关键章节
key_content = '\n\n'.join([
report_data.get('主要经营业务', ''),
report_data.get('经营情况讨论与分析', '')[:5000]
])
# 调用分析
result = self.analyzer.analyze_report(
company_name=company_name,
report_text=key_content,
user_id=1,
batch_id=f"DIR_{datetime.now().strftime('%Y%m%d')}"
)
# 保存单文件结果
output_path = Path(output_dir) / f"{pdf_path.stem}_summary.md"
with open(output_path, 'w', encoding='utf-8') as f:
f.write(f"# {company_name} 财报摘要\n\n")
f.write(f"生成时间: {result['timestamp']}\n\n")
f.write(result['result'])
result['output_file'] = str(output_path)
return result
实际运行
if __name__ == '__main__':
processor = BatchReportProcessor(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY"
)
summary = processor.process_directory(
input_dir='/data/annual_reports/2024/',
output_dir='/data/output/summaries/',
max_workers=3 # 并发数根据账号配额调整
)
print("\n========== 处理完成 ==========")
print(f"总文件数: {summary['total_files']}")
print(f"成功: {summary['success_count']}, 失败: {summary['failed_count']}")
print(f"总成本: ¥{summary['total_cost_cny']}")
print(f"平均成本: ¥{summary['avg_cost_per_file']}/份")
print(f"耗时: {summary['elapsed_seconds']}秒")五、常见报错排查
错误 1：PDF 解析乱码或内容为空
# 问题：某些 PDF 使用图片嵌入文字，pdfplumber 无法提取
解决方案：使用 OCR 预处理
from pdf2image import convert_from_path
import pytesseract
def extract_with_ocr(pdf_path: str) -> str:
"""带 OCR 的 PDF 解析"""
images = convert_from_path(pdf_path, dpi=200)
full_text = []
for image in images:
text = pytesseract.image_to_string(image, lang='chi_sim')
full_text.append(text)
return '\n'.join(full_text)
或者尝试使用 PyMuPDF (fitz) 作为备选解析器
import fitz # PyMuPDF
def extract_with_pymupdf(pdf_path: str) -> str:
"""PyMuPDF 解析"""
doc = fitz.open(pdf_path)
texts = []
for page in doc:
texts.append(page.get_text())
return '\n'.join(texts)错误 2：API 调用超时 (timeout=120)
# 问题：大批量调用时偶发超时
解决方案：增加重试机制 + 指数退避
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=60))
def analyze_with_retry(self, company_name: str, report_text: str, user_id: int):
"""带重试的分析方法"""
return self.analyze_report(company_name, report_text, user_id)
另一个方案：使用批量接口（如果 HolySheep 支持）
检查 API 文档是否有 batch endpoint
错误 3：Token 超出模型上下文限制
# 问题：年报文本超过模型上下文窗口
解决方案：分段处理 + 结果聚合
def chunked_analysis(self, company_name: str, full_text: str,
chunk_size: int = 50000, overlap: int = 2000) -> str:
"""分段分析长文本"""
chunks = []
start = 0
while start < len(full_text):
end = start + chunk_size
chunk = full_text[start:end]
chunks.append(chunk)
start = end - overlap # 重叠处理避免截断重要段落
# 逐段分析
section_results = []
for i, chunk in enumerate(chunks):
result = self._analyze_chunk(company_name, chunk, section_index=i)
section_results.append(result)
# 合并结果
return self._merge_sections(section_results)错误 4：数据库审计日志写入失败
# 问题：MySQL 连接池耗尽或字符编码问题
解决方案：连接池 + UTF-8 强制指定
import pymysql
from DBUtils.PooledDB import PooledDB
class AuditLogger:
def __init__(self):
self.pool = PooledDB(
creator=pymysql,
maxconnections=10,
mincached=2,
host='localhost',
user='research_user',
password='your_password',
database='investment_research',
charset='utf8mb4', # 必须是 utf8mb4 支持生僻字
cursorclass=pymysql.cursors.DictCursor
)
def log_with_fallback(self, log_data: Dict) -> bool:
"""带降级处理的日志写入"""
try:
conn = self.pool.connection()
cursor = conn.cursor()
cursor.execute(self._insert_sql, tuple(log_data.values()))
conn.commit()
return True
except Exception as e:
# 降级：写入本地文件
self._save_to_file(log_data)
print(f"数据库写入失败，已保存到本地: {e}")
return False
finally:
try:
cursor.close()
conn.close()
except:
pass六、适合谁与不适合谁
| 场景 | 推荐程度 | 原因 |
|---|---|---|
| 券商自营/公募基金研报 | ⭐⭐⭐⭐⭐ | 批量处理 + 合规留痕需求完美匹配 |
| 阳光私募日常研究 | ⭐⭐⭐⭐⭐ | 成本敏感，需快速筛选标的 |
| 上市公司 IR 部门 | ⭐⭐⭐⭐ | 生成投资者问答素材，自动生成会议纪要 |
| 咨询公司行研报告 | ⭐⭐⭐⭐ | 快速产出结构化分析 |
| 个人投资者学习 | ⭐⭐⭐ | 可用，但量小时优势不明显 |
| 法律/审计文件处理 | ⭐⭐⭐ | 可复用，需调整 Prompt |
| 实时行情 + 新闻分析 | ⭐⭐ | 建议用 Gemini 2.5 Flash 补充 |
不适合的场景
- 需要毫秒级实时响应的交易系统：建议用专门低延迟方案
- 对数据准确性 100% 零容忍的监管报送：AI 生成内容必须人工复核
- 海外上市公司财报：英文内容建议用 Claude 或 GPT-4
七、价格与回本测算
实际运行成本（以中型私募为例）
| 成本项 | 月用量 | 单价 | 月成本 |
|---|---|---|---|
| 年报解析（DeepSeek V3.2） | 100 份 × 8 万 Token | $0.42/MTok | ¥2,452 |
| 补充分析（Gemini 2.5 Flash） | 50 次 × 10 万 Token | $2.50/MTok | ¥912 |
| PDF 解析 + 存储 | 100GB | - | ¥50 |
| 数据库 + 服务器 | - | - | ¥200 |
| 月度总成本 | ¥3,614 |
回本测算
- 人工成本对比：分析师手动完成 100 份年报摘要约需 20 人天，按 ¥2,000/人天 = ¥40,000
- AI 替代后：系统自动完成 + 2 小时人工审核 = ¥800 人力成本
- 月度节省：约 ¥39,200
- 回本周期：开发投入 ¥5 万，约 1.3 个月回本
八、为什么选 HolySheep
我在选型时对比了 5 家中转平台，最终锁定 HolySheep，核心原因：
- 价格屠夫：DeepSeek V3.2 定价 $0.42/MTok，比第二名便宜 40%，比官方 API 便宜 97%
- 汇率无损：¥7.3 = $1，实测充值 ¥730 到账 $100.00，没有克扣
- 国内直连：上海测试延迟 <30ms，北京 <45ms，不用搭魔法
- 充值便捷：微信/支付宝秒到账，不用折腾国际信用卡
- 模型丰富：DeepSeek/GPT-4.1/Claude/Gemini 都有，一个 Key 切换
- 注册有礼：点击注册 送免费额度，实测到账 ¥50
九、购买建议与 CTA
我的建议
如果你符合以下任一条件，强烈建议立即上线这套系统：
- 每月处理超过 20 份财报 PDF
- 有合规审计留痕的监管要求
- 希望将分析师从重复劳动中解放出来做深度研究
目前 HolySheep 的 DeepSeek V3.2 是性价比最优解，配合 Claude Sonnet 4.5 做英文财报补充，足够应对大多数投研场景。
起步建议
- 先用 免费注册 领取赠额
- 下载我的代码，本地跑通单份解析流程
- 小批量测试 10 份，验证输出质量
- 确认满足需求后，再上批量处理
技术问题欢迎评论区交流，看到会回。