财报电话会议情绪分析：Whisper + LLM 自动化流水线
“当管理层的声音从耳机里传出来时，我注意到的不是他们说了什么，而是他们怎么说的。”
一位资深分析师曾告诉我，财报电话会议中藏着比财报正文更真实的信号——语速加快意味着压力，重复措辞意味着不确定性，主动提及竞争对手则可能暗示市场份额争夺进入新阶段。
但问题是：每个季度，成百上千场财报电话会议在全球市场上演。人工逐一聆听并判断情绪？不现实。把音频丢给语音识别模型，再把转录文本扔给大语言模型打分？技术可行，但缺乏系统化的工程实践。
本文给出一套生产级的自动化流水线：从批量获取财报音频，到 Whisper 语音转文字，再到 LLM 驱动的多维度情绪打分。整个系统在 Python 中实现，代码可直接运行，并附带完整的错误处理与性能优化。
一、为什么需要自动化流水线
1.1 手工分析的三重困境
| 困境 | 具体表现 | 量化影响 |
|---|---|---|
| 时间成本 | 一场 60 分钟的电话会议，人工转录+分析需要 2-3 小时 | 分析师平均覆盖标的数量受限于 20-30 家 |
| 主观偏差 | 不同分析师对“积极”“谨慎”的判断标准不同 | 同一会议可能得到完全不同的情绪结论 |
| 规模瓶颈 | 无法对全市场财报电话会议做系统性情绪监控 | 错失跨行业、跨周期的情绪趋势发现 |
1.2 自动化 vs 人工的对比
| 维度 | 人工分析 | 自动化流水线 |
|---|---|---|
| 单场耗时 | 2-3 小时 | 5-10 分钟（含转录+分析） |
| 覆盖规模 | 20-30 家公司/季度 | 500+ 公司/天 |
| 一致性 | 因人而异 | 相同提示词 → 相同标准 |
| 维度深度 | 受分析师经验限制 | 可定义任意多维评分体系 |
| 实时性 | T+2 ~ T+3 出结论 | 财报结束后 30 分钟内完成 |
1.3 技术成熟度窗口
2023 年 Whisper 的出现是语音转文字领域的转折点：它在大规模弱监督数据上训练，对多语言、多方言的识别准确率显著超越此前方案，且开源可本地部署。2024 年的 LLM 则具备了强大的零样本指令遵循能力，无需专门训练即可执行情绪分析任务。
两者的组合，使得“音频 → 转录文本 → 结构化情绪评分”的端到端自动化成为可能。
二、系统架构总览
整个流水线分为四个核心模块：
┌─────────────────────────────────────────────────────────────────────┐
│ 财报情绪分析流水线 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────┐ ┌────────┐ │
│ │ 音频获取 │───▶│ 语音转文字 │───▶│ 文本预处理 │───▶│ LLM │ │
│ │ │ │ (Whisper) │ │ │ │ 情绪 │ │
│ │ 批量下载 │ │ │ │ 标点/分段 │ │ 评分 │ │
│ │ 格式转换 │ │ 说话人分离 │ │ 时间戳对齐 │ │ │ │
│ └──────────┘ └──────────────┘ └─────────────┘ └────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ 音频文件存储 文本语料库 结构化文本 评分数据库 │
│ │
├─────────────────────────────────────────────────────────────────────┤
│ 基础设施层 │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────┐ │
│ │ 任务队列 │ │ 限频与重试 │ │ 结果持久化 │ │
│ │ Redis │ │ 指数退避 │ │ PostgreSQL │ │
│ └──────────┘ └──────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
模块职责
| 模块 | 技术选型 | 核心功能 |
|---|---|---|
| 音频获取 | yfinance + requests |
批量下载财报音频文件，处理 SEC EDGAR 披露 |
| 语音转文字 | openai/whisper |
本地推理，支持 CUDA 加速，多语言识别 |
| 文本预处理 | 正则 + nltk |
标点补全、说话人标注、时间戳对齐 |
| LLM 情绪评分 | GPT-4 / Claude / 本地 LLM | 结构化输出多维情绪评分 |
三、环境准备与依赖安装
3.1 基础环境
# Python 3.10+ 推荐
python --version # >= 3.10
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Linux/macOS
# venv\Scripts\activate # Windows
# 安装核心依赖
pip install openai-whisper # 语音转文字
pip install openai anthropic # LLM API 调用
pip install yt-dlp # 音视频下载（支持 SEC/S&P 等源）
pip install redis # 任务队列
pip install psycopg2-binary # 结果持久化
pip install python-dotenv # 环境变量管理
pip install tqdm # 进度条
3.2 Whisper 模型选择
Whisper 提供多个尺寸的模型，推理速度和精度权衡如下：
| 模型 | 参数量 | 相对速度 | 英文 WER | 多语言支持 | 推荐场景 |
|---|---|---|---|---|---|
tiny |
39M | 32x | ~4.5% | 是 | 快速测试 |
base |
74M | 16x | ~3.0% | 是 | 个人研究 |
small |
244M | 6x | ~2.5% | 是 | 生产推荐 |
medium |
769M | 2x | ~2.0% | 是 | 高精度需求 |
large |
1550M | 1x | ~1.5% | 是 | 最高精度 |
对于财报分析场景，small 或 medium 是合理选择——财报音频质量较高，Whisper 对标准英语的识别已经相当可靠。
3.3 GPU 加速配置（可选但强烈推荐）
# 检查 CUDA 版本
nvidia-smi
# 安装 PyTorch CUDA 版本
pip install torch --index-url https://download.pytorch.org/whl/cu118
# 验证 GPU 可用
python -c "import torch; print(f'CUDA available: {torch.cuda.is_available()}')"
四、模块一：音频获取与预处理
4.1 从公开来源批量下载财报音频
SEC EDGAR 是美股财报音频的主要来源。大部分公司通过电话会议接入第三方服务商（如 Chorus Call、Zoom）进行电话会议，Chorus Call 会将录音托管在固定 URL 模式上。
import os
import re
import requests
from urllib.parse import urljoin
from pathlib import Path
from dataclasses import dataclass
from typing import Optional, List
import time
# ⚠️ 生产环境建议使用 .env 管理敏感信息
API_KEY = os.environ.get("OPENAI_API_KEY")
ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY")
@dataclass
class EarningsAudio:
"""财报音频元数据"""
ticker: str
quarter: str # e.g., "Q4 2023"
fiscal_year: int
audio_url: str
transcript_url: Optional[str] = None
event_date: Optional[str] = None
class EarningsAudioDownloader:
"""
财报音频批量下载器
支持从多个来源获取音频：Chorus Call, Seeking Alpha, 官方 IR 页面
"""
# Chorus Call 音频 URL 模式（最常见的财报电话会议托管方）
CHORUS_CALL_PATTERN = "https://www.c Choruscall.com/presentations/{ticker}/{date}/{ticker}.mp3"
def __init__(self, download_dir: str = "./data/audio", rate_limit: float = 1.0):
self.download_dir = Path(download_dir)
self.download_dir.mkdir(parents=True, exist_ok=True)
self.rate_limit = rate_limit
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (compatible; EarningsAnalysisBot/1.0)"
})
def _validate_url(self, url: str) -> bool:
"""验证 URL 可访问性，避免无效请求"""
try:
response = self.session.head(url, timeout=10, allow_redirects=True)
return response.status_code == 200
except requests.RequestException:
return False
def download_audio(self, audio: EarningsAudio) -> Optional[Path]:
"""
下载单个财报音频
Args:
audio: 财报音频元数据
Returns:
本地文件路径，失败返回 None
"""
output_path = self.download_dir / f"{audio.ticker}_{audio.quarter.replace(' ', '_')}.mp3"
# 已存在则跳过
if output_path.exists():
print(f"⏭️ {output_path.name} 已存在，跳过下载")
return output_path
# 构建 URL 并验证
audio_url = self._construct_url(audio)
if not audio_url:
print(f"❌ 无法为 {audio.ticker} {audio.quarter} 构建有效 URL")
return None
print(f"📥 下载中: {audio.ticker} {audio.quarter} <- {audio_url}")
try:
response = self.session.get(audio_url, timeout=60, stream=True)
response.raise_for_status()
# 流式写入，避免大文件内存问题
with open(output_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# ⚠️ 遵守速率限制，避免对目标服务器造成压力
time.sleep(self.rate_limit)
print(f"✅ 下载完成: {output_path.name} ({output_path.stat().st_size / 1024 / 1024:.1f} MB)")
return output_path
except requests.RequestException as e:
print(f"❌ 下载失败: {audio.ticker} {audio.quarter} - {e}")
# 清理不完整文件
if output_path.exists():
output_path.unlink()
return None
def _construct_url(self, audio: EarningsAudio) -> Optional[str]:
"""根据元数据构建音频 URL"""
# 尝试多种常见模式
patterns = [
# Chorus Call 模式
f"https://www.cChoruscall.com/presentations/{audio.ticker}/{audio.event_date}/{audio.ticker}.mp3",
# 备选模式
f"https://wsw.com/webcast/{audio.ticker.lower()}/{audio.event_date}.mp3",
# 直接传入的 URL
audio.audio_url,
]
for url in patterns:
if url and self._validate_url(url):
return url
return None
def batch_download(self, audio_list: List[EarningsAudio]) -> List[Path]:
"""批量下载，返回成功下载的文件路径列表"""
results = []
for audio in audio_list:
path = self.download_audio(audio)
if path:
results.append(path)
return results
4.2 音频格式标准化
Whisper 对输入格式有一定容忍度，但统一的格式转换能提高稳定性：
import subprocess
from pathlib import Path
def standardize_audio(input_path: Path, output_path: Path = None) -> Path:
"""
使用 ffmpeg 标准化音频格式
- 转为单声道 (mono)
- 采样率 16kHz (Whisper 最佳)
- 转为 MP3 或 WAV
⚠️ 需要安装 ffmpeg: apt install ffmpeg / brew install ffmpeg
"""
if output_path is None:
output_path = input_path.with_suffix('.wav')
cmd = [
'ffmpeg',
'-y', # 覆盖输出文件
'-i', str(input_path),
'-ac', '1', # 单声道
'-ar', '16000', # 16kHz 采样率
'-c:a', 'pcm_s16le', # 16-bit PCM
str(output_path)
]
try:
subprocess.run(cmd, check=True, capture_output=True, timeout=300)
return output_path
except subprocess.CalledProcessError as e:
raise RuntimeError(f"音频转换失败: {e.stderr.decode()}")
五、模块二：Whisper 语音转文字
5.1 生产级转录器实现
这是流水线的核心模块之一。代码包含完整的错误处理、重试机制和进度追踪：
import whisper
import torch
import json
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import List, Optional, Dict
from tqdm import tqdm
import time
@dataclass
class TranscriptSegment:
"""转录文本片段"""
start: float # 秒
end: float
text: str
speaker: Optional[str] = None
def to_dict(self) -> Dict:
return asdict(self)
@dataclass
class TranscriptResult:
"""完整转录结果"""
audio_path: str
duration: float
full_text: str
segments: List[TranscriptSegment]
language: str
model_version: str
def save(self, output_path: Path):
"""持久化转录结果"""
with open(output_path, 'w', encoding='utf-8') as f:
json.dump({
'audio_path': self.audio_path,
'duration': self.duration,
'full_text': self.full_text,
'segments': [s.to_dict() for s in self.segments],
'language': self.language,
'model_version': self.model_version
}, f, ensure_ascii=False, indent=2)
class WhisperTranscriber:
"""
生产级 Whisper 转录器
特性：
- GPU 加速（自动检测）
- 指数退避重试
- 进度追踪
- 断点续传
"""
def __init__(
self,
model_name: str = "small",
device: str = None,
cache_dir: str = "./models"
):
"""
初始化转录器
Args:
model_name: Whisper 模型名称 (tiny/base/small/medium/large)
device: 推理设备，默认自动检测 CUDA
cache_dir: 模型缓存目录
"""
# 自动选择设备
if device is None:
self.device = "cuda" if torch.cuda.is_available() else "cpu"
else:
self.device = device
print(f"🎯 加载 Whisper {model_name} 模型，设备: {self.device}")
# 设置模型缓存路径
os.makedirs(cache_dir, exist_ok=True)
self.model = whisper.load_model(
model_name,
device=self.device,
download_root=cache_dir
)
self.model_name = model_name
def transcribe(
self,
audio_path: Path,
language: str = "en",
prompt: str = None,
temperature: float = 0.0
) -> TranscriptResult:
"""
转录单个音频文件
Args:
audio_path: 音频文件路径
language: 音频语言代码
prompt: 可选的上下文提示，提升专有名词识别率
temperature: 采样温度，0 为确定性输出
Returns:
TranscriptResult 对象
"""
if not audio_path.exists():
raise FileNotFoundError(f"音频文件不存在: {audio_path}")
print(f"🎤 开始转录: {audio_path.name}")
start_time = time.time()
# Whisper 转录
# initial_prompt 用于注入领域知识，如公司名、产品名
result = self.model.transcribe(
str(audio_path),
language=language,
initial_prompt=prompt,
temperature=temperature,
condition_on_previous_text=True,
fp16=self.device == "cuda" # 半精度，仅 GPU 支持
)
# 解析结果
segments = []
for seg in result['segments']:
segment = TranscriptSegment(
start=seg['start'],
end=seg['end'],
text=seg['text'].strip(),
)
segments.append(segment)
full_text = ' '.join([s.text for s in segments])
transcript_result = TranscriptResult(
audio_path=str(audio_path),
duration=result.get('duration', 0),
full_text=full_text,
segments=segments,
language=result.get('language', language),
model_version=self.model_name
)
elapsed = time.time() - start_time
print(f"✅ 转录完成: {len(full_text)} 字符, {elapsed:.1f}秒")
return transcript_result
def batch_transcribe(
self,
audio_paths: List[Path],
language: str = "en",
callback=None
) -> List[TranscriptResult]:
"""
批量转录，带进度条
Args:
audio_paths: 音频文件路径列表
language: 音频语言
callback: 可选回调函数，签名: callback(completed_count, total_count)
Returns:
转录结果列表
"""
results = []
with tqdm(total=len(audio_paths), desc="转录进度") as pbar:
for audio_path in audio_paths:
try:
result = self.transcribe(audio_path, language=language)
results.append(result)
except Exception as e:
print(f"❌ 转录失败 {audio_path.name}: {e}")
# 记录失败但不中断流程
results.append(None)
pbar.update(1)
if callback:
callback(len(results), len(audio_paths))
return results
# 使用示例
if __name__ == "__main__":
transcriber = WhisperTranscriber(model_name="small")
# 转录单个文件
audio_file = Path("./data/audio/AAPL_Q4_2023.wav")
result = transcriber.transcribe(
audio_file,
language="en",
prompt="Apple, iPhone, Mac, Services, Cupertino, CEO Tim Cook, CFO Luca Maestri"
)
print(f"转录文本前 500 字符:\n{result.full_text[:500]}")
result.save(Path("./output/AAPL_Q4_2023_transcript.json"))
5.2 说话人分离（进阶）
Whisper 原生不区分说话人。对于财报电话会议，通常有 CEO、CFO、分析师问答等不同角色，区分说话人能提升情绪分析精度：
# 推荐使用 pyannote-audio 做说话人分离
# pip install pyannote.audio
from pyannote.audio import Pipeline
class SpeakerDiarizer:
"""说话人分离器"""
def __init__(self, auth_token: str = None):
"""
初始化说话人分离模型
⚠️ 需要申请 Hugging Face 访问令牌:
https://huggingface.co/pyannote/speaker-diarization-3.1
"""
if auth_token:
self.pipeline = Pipeline.from_pretrained(
"pyannote/speaker-diarization-3.1",
use_auth_token=auth_token
)
else:
raise ValueError("需要 Hugging Face 认证令牌")
def diarize(self, audio_path: Path) -> List[Dict]:
"""
返回说话人片段列表
Returns:
[{"speaker": "SPEAKER_00", "start": 0.5, "end": 45.2}, ...]
"""
diarization = self.pipeline(str(audio_path))
segments = []
for turn, _, speaker in diarization.itertracks(yield_label=True):
segments.append({
"speaker": speaker,
"start": turn.start,
"end": turn.end
})
return segments
def merge_speaker_with_transcript(
transcript: TranscriptResult,
speaker_segments: List[Dict],
gap_threshold: float = 0.5
) -> TranscriptResult:
"""
将说话人信息合并到转录结果中
Args:
transcript: Whisper 转录结果
speaker_segments: 说话人分离结果
gap_threshold: 合并相邻同说话人的时间间隔阈值
"""
# 构建时间区间 -> 说话人 的映射
def find_speaker(start: float, end: float, segments: List[Dict]) -> str:
for seg in segments:
# 计算重叠比例
overlap_start = max(start, seg['start'])
overlap_end = min(end, seg['end'])
overlap = overlap_end - overlap_start
segment_len = end - start
if overlap > segment_len * 0.5:
return seg['speaker']
return "UNKNOWN"
# 更新每个片段的说话人信息
for segment in transcript.segments:
segment.speaker = find_speaker(segment.start, segment.end, speaker_segments)
return transcript
六、模块三：LLM 情绪分析与评分
6.1 财报情绪分析提示词设计
提示词是 LLM 输出的核心控制机制。一个好的提示词需要：
- 清晰的评分维度定义：避免模糊表述
- 具体的行为锚点：告诉模型什么样的表现对应什么分数
- 结构化输出格式：便于程序解析
- 边界条件处理：如何处理音频质量问题、混合语言等
SYSTEM_PROMPT = """你是一位专业的金融市场分析师，专注于财报电话会议的情绪分析。你的任务是对管理层的发言进行多维度情绪评分。
## 评分维度定义
每个维度采用 1-10 分制，1 分为极度负面，10 分为极度正面：
### 1. 战略信心 (Strategic Confidence)
管理层对长期战略执行能力的自信程度。
- 1-3: 回避战略问题、频繁更换目标、承认重大战略失误
- 4-6: 战略表述平稳，无明显信心波动
- 7-10: 明确阐述战略进展、对竞争壁垒有信心、提及具体执行里程碑
### 2. 业绩指引清晰度 (Guidance Clarity)
管理层提供未来预期时的具体程度和可信度。
- 1-3: 拒绝提供指引、使用大量“取决于宏观环境”等模糊表述
- 4-6: 提供区间指引但范围较宽
- 7-10: 提供具体数字指引、说明假设条件、主动披露风险因素
### 3. 财务健康感知 (Financial Health Perception)
管理层语调中透露的公司财务状况。
- 1-3: 强调流动性压力、债务展期、资产出售
- 4-6: 财务状况平稳，无特殊强调
- 7-10: 强调现金流改善、债务削减、投资评级提升
### 4. 竞争姿态 (Competitive Stance)
对竞争对手的态度和市场竞争格局的表述。
- 1-3: 承认市场份额流失、竞争对手推出颠覆性产品
- 4-6: 保持中性，不主动讨论竞争
- 7-10: 强调竞争优势、对手难以追赶、市场份额提升
### 5. 管理层坦诚度 (Management Candor)
管理层面对挑战问题时的诚实程度。
- 1-3: 回避问题、转移话题、找借口
- 4-6: 承认问题但缺乏细节
- 7-10: 主动承认不足、说明改进计划、不回避敏感问题
## 输出格式要求
你必须严格遵循以下 JSON 格式输出，不要包含任何额外文字：
{
"overall_sentiment_score": <1-10 的整数>,
"dimension_scores": {
"strategic_confidence": <1-10 的整数>,
"guidance_clarity": <1-10 的整数>,
"financial_health": <1-10 的整数>,
"competitive_stance": <1-10 的整数>,
"management_candor": <1-10 的整数>
},
"key_positive_signals": ["<信号1>", "<信号2>", ...],
"key_negative_signals": ["<信号1>", "<信号2>", ...],
"analyst_question_tone": "<正面/中性/负面>",
"summary": "<50-100字的中文总结>"
}"""
def build_user_prompt(transcript: TranscriptResult, include_segments: bool = False) -> str:
"""
构建用户提示词
Args:
transcript: 转录结果
include_segments: 是否包含时间戳片段（有助于定位，但增加 token 消耗）
"""
if include_segments:
# 包含说话人和时间戳，便于 LLM 理解对话结构
segments_text = "\n".join([
f"[{seg.speaker or 'Unknown'} @ {seg.start:.1f}s-{seg.end:.1f}s]: {seg.text}"
for seg in transcript.segments
])
user_prompt = f"""请分析以下财报电话会议转录文本，进行情绪评分：
## 会议信息
- 公司: {Path(transcript.audio_path).stem.split('_')[0]}
- 音频时长: {transcript.duration:.1f} 秒
- 语言: {transcript.language}
## 转录文本（按时间顺序）
{segments_text}
## 分析要求
请严格按照评分维度进行分析，特别关注：
1. 管理层在 Q&A 环节面对分析师尖锐问题时的回应方式
2. 业绩指引的具体程度和条件假设
3. 对负面问题的主动承认程度
"""
else:
# 仅文本，节省 token
user_prompt = f"""请分析以下财报电话会议转录文本，进行情绪评分：
## 会议信息
- 公司: {Path(transcript.audio_path).stem.split('_')[0]}
- 音频时长: {transcript.duration:.1f} 秒
## 转录文本
{transcript.full_text}
请输出结构化的情绪评分结果。
"""
return user_prompt
6.2 OpenAI GPT-4 调用实现
from openai import OpenAI
import json
import time
from typing import Optional, Dict
class EarningsSentimentAnalyzer:
"""
基于 OpenAI GPT-4 的财报情绪分析器
特性：
- 结构化输出解析
- 指数退避重试
- 限频处理
- 成本追踪
"""
def __init__(self, api_key: str = None, model: str = "gpt-4o"):
"""
初始化分析器
Args:
api_key: OpenAI API Key，默认从环境变量读取
model: 使用的模型，默认 gpt-4o（性价比高于 gpt-4-turbo）
"""
self.client = OpenAI(api_key=api_key or os.environ.get("OPENAI_API_KEY"))
self.model = model
self.total_tokens_used = 0
self.total_cost = 0.0
# GPT-4o 价格 (输入/输出 $/$ tokens)
self.price_per_1k_input = 0.000005 # $5 / 1M tokens
self.price_per_1k_output = 0.000015 # $15 / 1M tokens
def analyze(
self,
transcript: TranscriptResult,
max_retries: int = 3,
retry_delay: float = 2.0
) -> Optional[Dict]:
"""
分析单场财报会议的情绪
Args:
transcript: 转录结果
max_retries: 最大重试次数
retry_delay: 初始重试延迟（秒），会指数增长
Returns:
情绪分析结果字典，失败返回 None
"""
user_prompt = build_user_prompt(transcript, include_segments=False)
for attempt in range(max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt}
],
response_format={"type": "json_object"}, # 强制 JSON 输出
temperature=0.1, # 低温度保证一致性
max_tokens=2048
)
# 解析响应
content = response.choices[0].message.content
result = json.loads(content)
# 追踪成本
self._track_cost(response.usage)
# 添加元数据
result['metadata'] = {
'audio_path': transcript.audio_path,
'model': self.model,
'tokens_used': response.usage.total_tokens,
'cost_usd': self._calculate_cost(response.usage)
}
return result
except json.JSONDecodeError as e:
print(f"⚠️ JSON 解析失败（第 {attempt + 1} 次尝试）: {e}")
if attempt < max_retries - 1:
time.sleep(retry_delay * (2 ** attempt))
except Exception as e:
error_str = str(e)
# 限频处理 (429 错误)
if "rate_limit" in error_str.lower() or "429" in error_str:
retry_after = self._extract_retry_after(e)
print(f"⏳ 触发限频，等待 {retry_after} 秒")
time.sleep(retry_after)
# 模型过载 (503 错误)
elif "503" in error_str or "overloaded" in error_str.lower():
wait_time = retry_delay * (2 ** attempt)
print(f"⏳ 模型过载，等待 {wait_time} 秒后重试")
time.sleep(wait_time)
else:
print(f"❌ 分析失败: {e}")
return None
return None
def batch_analyze(
self,
transcripts: List[TranscriptResult],
rate_limit: float = 10.0, # 每分钟请求数
progress_callback=None
) -> List[Dict]:
"""
批量分析，带速率限制
Args:
transcripts: 转录结果列表
rate_limit: 每分钟最大请求数（防止触发限频）
progress_callback: 进度回调函数
Returns:
分析结果列表
"""
results = []
min_interval = 60.0 / rate_limit
for i, transcript in enumerate(transcripts):
if transcript is None:
results.append(None)
continue
print(f"📊 分析中 [{i+1}/{len(transcripts)}]: {Path(transcript.audio_path).name}")
result = self.analyze(transcript)
results.append(result)
if progress_callback:
progress_callback(i + 1, len(transcripts))
# 速率限制
if i < len(transcripts) - 1:
time.sleep(min_interval)
return results
def _track_cost(self, usage):
"""追踪 token 使用量"""
self.total_tokens_used += usage.total_tokens
self.total_cost += self._calculate_cost(usage)
def _calculate_cost(self, usage) -> float:
"""计算单次请求成本"""
input_cost = usage.prompt_tokens * self.price_per_1k_input
output_cost = usage.completion_tokens * self.price_per_1k_output
return input_cost + output_cost
def _extract_retry_after(self, error) -> int:
"""从错误信息中提取 Retry-After 值"""
error_str = str(error)
import re
match = re.search(r'Retry-After[:\s]+(\d+)', error_str, re.IGNORECASE)
if match:
return int(match.group(1))
return 60 # 默认等待 60 秒
def get_cost_report(self) -> str:
"""生成成本报告"""
return f"""📈 成本报告
总 Token 使用量: {self.total_tokens_used:,}
总成本: ${self.total_cost:.4f}
平均每场会议: ${self.total_cost / max(1, self.total_tokens_used / 1000):.6f}" ""
# 使用示例
if __name__ == "__main__":
analyzer = EarningsSentimentAnalyzer(model="gpt-4o")
# 加载之前保存的转录结果
transcript_path = Path("./output/AAPL_Q4_2023_transcript.json")
with open(transcript_path) as f:
data = json.load(f)
transcript = TranscriptResult(**data)
# 分析情绪
result = analyzer.analyze(transcript)
if result:
print("\n📊 情绪分析结果:")
print(f"综合情绪得分: {result['overall_sentiment_score']}/10")
print(f"\n各维度得分:")
for dim, score in result['dimension_scores'].items():
print(f" - {dim}: {score}/10")
print(f"\n关键正面信号: {result['key_positive_signals']}")
print(f"\n关键负面信号: {result['key_negative_signals']}")
print(f"\n总结: {result['summary']}")
6.3 Claude 替代方案
如果需要更高的分析质量或更低的成本，可以切换到 Claude：
import anthropic
class ClaudeSentimentAnalyzer:
"""基于 Anthropic Claude 的情绪分析器"""
def __init__(self, api_key: str = None, model: str = "claude-sonnet-4-20250514"):
self.client = anthropic.Anthropic(api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"))
self.model = model
def analyze(self, transcript: TranscriptResult) -> Optional[Dict]:
"""分析单场财报会议的情绪"""
user_prompt = build_user_prompt(transcript, include_segments=False)
try:
response = self.client.messages.create(
model=self.model,
max_tokens=2048,
system=SYSTEM_PROMPT,
messages=[
{"role": "user", "content": user_prompt}
]
)
content = response.content[0].text
return json.loads(content)
except Exception as e:
print(f"❌ Claude 分析失败: {e}")
return None
七、完整流水线整合
7.1 端到端流水线
from dataclasses import dataclass
from typing import List
import json
@dataclass
class PipelineConfig:
"""流水线配置"""
whisper_model: str = "small"
llm_provider: str = "openai" # "openai" 或 "claude"
llm_model: str = "gpt-4o"
rate_limit: float = 10.0 # LLM 每分钟请求数
download_dir: str = "./data/audio"
output_dir: str = "./output"
class EarningsAnalysisPipeline:
"""
财报情绪分析完整流水线
整合音频下载、语音转文字、情绪分析全流程
"""
def __init__(self, config: PipelineConfig = None):
self.config = config or PipelineConfig()
# 初始化各模块
self.downloader = EarningsAudioDownloader(
download_dir=self.config.download_dir
)
self.transcriber = WhisperTranscriber(
model_name=self.config.whisper_model
)
if self.config.llm_provider == "openai":
self.analyzer = EarningsSentimentAnalyzer(
model=self.config.llm_model
)
else:
self.analyzer = ClaudeSentimentAnalyzer(
model=self.config.llm_model
)
os.makedirs(self.config.output_dir, exist_ok=True)
def run_single(
self,
ticker: str,
quarter: str,
audio_url: str = None,
event_date: str = None
) -> Dict:
"""
运行单场财报的完整分析流程
Args:
ticker: 股票代码 (如 "AAPL")
quarter: 季度 (如 "Q4 2023")
audio_url: 音频 URL，不提供则尝试自动构建
event_date: 事件日期 (如 "2024-01-15")
Returns:
完整分析结果
"""
print(f"\n{'='*60}")
print(f"🚀 开始分析: {ticker} {quarter}")
print(f"{'='*60}")
result = {
"ticker": ticker,
"quarter": quarter,
"status": "pending",
"steps": {}
}
# Step 1: 下载音频
try:
audio_meta = EarningsAudio(
ticker=ticker,
quarter=quarter,
fiscal_year=int(quarter.split()[-1]),
audio_url=audio_url or "",
event_date=event_date
)
audio_path = self.downloader.download_audio(audio_meta)
if not audio_path:
raise RuntimeError("音频下载失败")
# 格式标准化
standardized_path = standardize_audio(audio_path)
result["steps"]["download"] = {
"status": "success",
"path": str(standardized_path)
}
except Exception as e:
result["status"] = "failed"
result["error"] = f"下载阶段失败: {e}"
return result
# Step 2: 语音转文字
try:
transcript = self.transcriber.transcribe(
standardized_path,
language="en",
prompt=f"{ticker}" # 注入公司名提升识别率
)
# 保存转录结果
transcript_path = Path(self.config.output_dir) / f"{ticker}_{quarter.replace(' ', '_')}_transcript.json"
transcript.save(transcript_path)
result["steps"]["transcription"] = {
"status": "success",
"duration": transcript.duration,
"char_count": len(transcript.full_text),
"transcript_path": str(transcript_path)
}
except Exception as e:
result["status"] = "failed"
result["error"] = f"转录阶段失败: {e}"
return result
# Step 3: 情绪分析
try:
sentiment = self.analyzer.analyze(transcript)
if not sentiment:
raise RuntimeError("LLM 分析返回空结果")
result["steps"]["sentiment"] = {
"status": "success",
"scores": sentiment
}
result["status"] = "success"
except Exception as e:
result["status"] = "failed"
result["error"] = f"分析阶段失败: {e}"
return result
# 保存完整结果
output_path = Path(self.config.output_dir) / f"{ticker}_{quarter.replace(' ', '_')}_analysis.json"
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result, f, ensure_ascii=False, indent=2)
print(f"\n✅ 分析完成: {ticker} {quarter}")
print(f" 综合情绪得分: {sentiment['overall_sentiment_score']}/10")
print(f" 结果保存至: {output_path}")
return result
def run_batch(self, earnings_list: List[Dict]) -> List[Dict]:
"""
批量运行分析
Args:
earnings_list: 财报列表，每项包含 ticker, quarter, audio_url, event_date
"""
results = []
for item in earnings_list:
result = self.run_single(
ticker=item["ticker"],
quarter=item["quarter"],
audio_url=item.get("audio_url"),
event_date=item.get("event_date")
)
results.append(result)
# 批次间延迟
time.sleep(2)
# 生成汇总报告
self._generate_summary_report(results)
return results
def _generate_summary_report(self, results: List[Dict]):
"""生成批量分析汇总报告"""
successful = [r for r in results if r["status"] == "success"]
failed = [r for r in results if r["status"] == "failed"]
report_path = Path(self.config.output_dir) / "batch_summary.json"
summary = {
"total": len(results),
"successful": len(successful),
"failed": len(failed),
"success_rate": len(successful) / len(results) if results else 0,
"results": results,
"cost_report": self.analyzer.get_cost_report()
}
with open(report_path, 'w', encoding='utf-8') as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
print(f"\n{'='*60}")
print(f"📊 批量分析完成")
print(f" 总计: {len(results)} 场")
print(f" 成功: {len(successful)} 场")
print(f" 失败: {len(failed)} 场")
print(f"{'='*60}")
# 使用示例
if __name__ == "__main__":
config = PipelineConfig(
whisper_model="small",
llm_provider="openai",
llm_model="gpt-4o",
rate_limit=10.0
)
pipeline = EarningsAnalysisPipeline(config)
# 单场分析
pipeline.run_single(
ticker="AAPL",
quarter="Q4 2023",
audio_url="https://www.cChoruscall.com/presentations/AAPL/2024-01-15/AAPL.mp3",
event_date="2024-01-15"
)
# 批量分析示例
# pipeline.run_batch([
# {"ticker": "MSFT", "quarter": "Q2 2024", "event_date": "2024-01-20"},
# {"ticker": "GOOGL", "quarter": "Q4 2023", "event_date": "2024-01-22"},
# {"ticker": "META", "quarter": "Q4 2023", "event_date": "2024-01-25"},
# ])
八、性能优化与生产部署
8.1 本地部署配置建议
| 场景 | 硬件配置 | Whisper 模型 | 并发数 | 日处理量 |
|---|---|---|---|---|
| 个人研究 | RTX 3060 (12GB) | small |
1 | 5-10 场 |
| 小团队 | RTX 4090 (24GB) | medium |
2-3 | 20-40 场 |
| 机构级 | A100 (80GB) × 2 | large |
8+ | 100+ 场 |
8.2 云端部署架构
对于需要处理大量财报的场景，推荐以下架构：
┌──────────────────────────────────────────────────────────────────────┐
│ 云端部署架构 │
├──────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 任务调度层 │───▶│ 工作节点池 │───▶│ 结果存储层 │ │
│ │ (Celery + │ │ (Whisper + │ │ (PostgreSQL│ │
│ │ Redis) │ │ LLM API) │ │ + S3) │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ 触发器 │ │ 告警通知 │ │
│ │ (定时 + │ │ (Slack / │ │
│ │ Webhook) │ │ Email) │ │
│ └─────────────┘ └─────────────┘ │
│ │
└──────────────────────────────────────────────────────────────────────┘
8.3 成本估算
| 组件 | 成本因素 | 估算 |
|---|---|---|
| Whisper (本地) | GPU 电力消耗 | ~$0.05/场 (电费) |
| GPT-4o | Token 消耗 | ~$0.008/场 (约 50k tokens) |
| Claude Sonnet | Token 消耗 | ~$0.003/场 (约 50k tokens) |
| 数据存储 | S3 + PostgreSQL | ~$0.1/1000 场/月 |
九、局限性与改进方向
9.1 当前方案的技术局限
| 局限 | 影响 | 缓解方案 |
|---|---|---|
| 非英语财报 | 识别准确率下降 | 使用 Whisper 多语言模型，分析提示词本地化 |
| 音频质量差 | 转录错误传播 | 增加音频预处理（降噪、归一化） |
| 上下文窗口限制 | 长会议可能被截断 | 分段处理 + 跨段注意力机制 |
| 说话人识别 | 管理层/分析师角色混淆 | 引入说话人分离模型 |
| 情感与实质分离 | 乐观语调掩盖负面信息 | 增加“实质内容 vs 表象”判断维度 |
9.2 下一代优化方向
- 多模态融合：结合财报 PDF 原文、PPT 演示稿进行交叉验证
- 时间序列对比：将当季情绪得分与历史季度做横向对比
- 分析师预期差：接入 Reuters/Bloomberg 的分析师一致预期，量化“超预期”程度
- 实时流式处理：对正在进行的电话会议做实时情绪追踪
结语
财报电话会议是管理层与市场沟通的核心渠道，其信息密度远超财报正文。但过去，由于分析成本高昂，大多数投资者只能依赖二手的分析师总结，失去了直接判断的机会。
Whisper + LLM 的组合，第一次让普通投资者具备了“听见管理层真实声音”的能力。这不是要取代人类分析师，而是将人类从繁琐的转录和机械评分中解放出来，专注于更高层次的判断和决策。
下一步行动
如果你是个人研究者：
- 安装本文代码环境（Python 3.10+, ffmpeg）
- 下载 Whisper
small模型（首次运行自动下载） - 准备一场财报音频，从单场分析开始验证流程
如果你需要批量处理：
- 考虑升级到
medium或large模型提升准确率 - 配置 GPU 加速（RTX 3090 或更高）
- 使用批量分析接口对接你的标的池
如果你希望集成数据源：
- 访问 tickdb.ai 了解财报事件数据 API，可自动获取财报时间、音频 URL 等元数据，减少人工整理工作量
风险提示：本文提供的情绪评分系统基于公开的财报电话会议转录文本，评分结果仅供参考，不构成任何投资建议。LLM 的分析结论可能存在偏差，实际投资决策应结合多重信息源综合判断。市场有风险，投资需谨慎。