The CFO said everything was fine. The transcript said otherwise.
Three hours after the Q3 earnings call, a junior analyst noticed a pattern: the word "challenging" appeared seven times — up from two in the previous quarter. The stock dropped 8% the next day. The humans caught it eventually, but a machine learning system monitoring the call in real time would have flagged the semantic shift within minutes of the broadcast ending.
This is the problem this article solves: how to build a production-grade pipeline that transcribes earnings calls automatically, extracts semantic signals, and generates a comparable sentiment score — all without manual intervention.
The architecture is straightforward. Whisper handles transcription. An LLM handles interpretation. A lightweight orchestration layer handles batching, retry logic, and structured output. This article walks through each component with working code and the engineering tradeoffs that determine whether this pipeline survives contact with a real trading desk.
Why Earnings Call Transcripts Deserve Automated Attention
Earnings calls are high-information-density events. The prepared remarks are scripted; the Q&A is not. Executive responses to analyst questions reveal positioning, confidence levels, and forward guidance nuances that written press releases obscure.
Manual analysis is slow and inconsistent. Different analysts interpret tone differently. Reading 500 earnings call transcripts per quarter is not a task a human can execute well at scale.
The signal extracted from automated sentiment analysis is noisy — no single metric from a single call reliably predicts the next day's return. But across a cohort of companies in a sector, sentiment divergence from consensus can identify where the market's expectations and management's actual outlook have drifted apart. That divergence is actionable.
The pipeline described here produces structured output: a sentiment score (1–10 scale), a confidence indicator, key phrase extraction, and a flag for tone changes relative to the previous quarter. This output can feed into a broader quantitative model or serve as a research surface for discretionary traders.
Pipeline Architecture Overview
The system consists of four stages:
- Audio Acquisition: Fetch the earnings call audio file (typically MP3 or WAV from public IR feeds).
- Transcription: Whisper converts audio to text with speaker diarization if available.
- Sentiment Analysis: An LLM receives the transcript with a structured prompt and returns a JSON score object.
- Storage & Alerting: Results are stored in a time-series format and alerts are triggered on threshold crossings.
[IR Feed] → [Audio Fetcher] → [Whisper API] → [Transcript]
↓
[LLM Analyzer]
↓
[Results Store + Alert]
Each stage has failure modes. The audio fetcher must handle redirects, expired URLs, and malformed metadata. Whisper's API has rate limits and occasionally returns garbled timestamps on poor-quality audio. The LLM prompt must be stable across different executive speaking styles. The following sections address each layer with production-grade implementations.
Stage 1: Audio Acquisition
Earnings call audio is typically hosted on third-party services (Intrado, Chorus Call, or the company's own IR portal) and linked in SEC filings (8-K or 6-K) or press releases. There is no single public API that consolidates earnings call audio for all publicly traded companies.
For this pipeline, we assume you have a list of audio URLs and ticker symbols. The audio fetcher module handles downloading with the following requirements:
- Follow HTTP redirects (earnings call hosts frequently change CDN providers).
- Validate file type before downloading (accept MP3, WAV, M4A).
- Implement exponential backoff with jitter for transient 5xx errors.
- Store the audio file locally with a timestamped filename to avoid reprocessing.
import os
import time
import hashlib
import requests
from pathlib import Path
from urllib.parse import urlparse
# ⚠️ For production deployment, replace requests with aiohttp/asyncio
# if you plan to process hundreds of calls concurrently.
ACCEPTED_EXTENSIONS = {".mp3", ".wav", ".m4a", ".wma"}
MAX_FILE_SIZE_MB = 100
OUTPUT_DIR = Path("./earnings_audio")
OUTPUT_DIR.mkdir(exist_ok=True)
def is_valid_audio_url(url: str) -> bool:
"""Validate URL points to a downloadable audio file."""
parsed = urlparse(url)
ext = Path(parsed.path).suffix.lower()
return ext in ACCEPTED_EXTENSIONS
def download_earnings_audio(url: str, ticker: str, quarter: str) -> str | None:
"""
Download earnings call audio with retry logic.
Returns the local filepath on success, None on failure.
"""
if not is_valid_audio_url(url):
raise ValueError(f"Unsupported audio format for URL: {url}")
headers = {
"User-Agent": "Mozilla/5.0 (compatible; EarningsPipeline/1.0)"
}
base_delay = 1.0
max_delay = 32.0
max_retries = 5
for attempt in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=(5, 30), stream=True)
response.raise_for_status()
content_length = response.headers.get("Content-Length")
if content_length and int(content_length) > MAX_FILE_SIZE_MB * 1024 * 1024:
raise ValueError(f"File too large: {content_length} bytes")
# Use content hash as fallback if Content-Disposition filename is missing
content_hash = hashlib.md5(response.content[:65536]).hexdigest()[:8]
ext = Path(urlparse(url).path).suffix.lower()
filename = f"{ticker}_{quarter}{ext}"
filepath = OUTPUT_DIR / filename
with open(filepath, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"[✓] Downloaded {ticker} Q{quarter}: {filepath.name} ({os.path.getsize(filepath) / 1024 / 1024:.1f} MB)")
return str(filepath)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
print(f"[✗] Audio file not found (404): {url}")
return None
if e.response.status_code >= 500:
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = (hashlib.md5(f"{url}{time.time()}".encode()).hexdigest()[0:4], 16) / 65536
sleep_time = delay + jitter
print(f"[!] Server error {e.response.status_code}, retrying in {sleep_time:.1f}s...")
time.sleep(sleep_time)
continue
raise
except requests.exceptions.Timeout:
delay = min(base_delay * (2 ** attempt), max_delay)
print(f"[!] Request timeout, retrying in {delay:.1f}s...")
time.sleep(delay)
continue
print(f"[✗] Failed to download after {max_retries} attempts: {url}")
return None
Engineering notes: The User-Agent header is not cosmetic. Several IR hosting providers return 403 to requests that look like bots. The hardcoded timeout tuple (5, 30) means 5 seconds for connection establishment and 30 seconds for the read to complete — appropriate for earnings call audio files which can be 30–60 MB.
For the final production deployment, consider switching to aiohttp for async download of multiple calls in parallel. The sequential implementation above is intentional for reliability — concurrent downloads across dozens of hosts with varying CDN quality will generate failures that are harder to debug.
Stage 2: Whisper Transcription
OpenAI's Whisper model (available via API or self-hosted) converts the audio file to text. The key configuration decisions are:
- Model size:
whisper-1(API) orlarge-v3(self-hosted). Larger models produce fewer hallucinations on financial jargon, but cost more per minute. - Language: Pass the language code explicitly (
en) to skip auto-detection overhead. - Timestamp granularity: Request word-level timestamps if you plan to correlate sentiment with specific Q&A moments. For bulk scoring, segment-level timestamps are sufficient.
Whisper's API is straightforward. The main engineering challenge is handling the ~60-second latency for a typical 30-minute call and managing API quotas.
import os
import json
import time
import subprocess
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional
import requests
# Load API key from environment variable — never hardcode credentials.
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
WHISPER_API_URL = "https://api.openai.com/v1/audio/transcriptions"
TRANSCRIPT_DIR = Path("./transcripts")
TRANSCRIPT_DIR.mkdir(exist_ok=True)
@dataclass
class TranscriptResult:
ticker: str
quarter: str
text: str
duration_seconds: float
segments: list[dict]
model: str
confidence_avg: float
def transcribe_with_whisper(audio_filepath: str, ticker: str, quarter: str, model: str = "whisper-1") -> Optional[TranscriptResult]:
"""
Transcribe an earnings call audio file using the Whisper API.
Returns a TranscriptResult object with text, segments, and metadata.
"""
if not OPENAI_API_KEY:
raise EnvironmentError("OPENAI_API_KEY environment variable not set")
if not Path(audio_filepath).exists():
raise FileNotFoundError(f"Audio file not found: {audio_filepath}")
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}"
}
with open(audio_filepath, "rb") as audio_file:
files = {
"file": audio_file,
"model": (None, model),
"response_format": (None, "verbose_json"),
"timestamp_granularities[]": (None, "segment"),
"language": (None, "en"),
}
# ⚠️ Timeout of 120 seconds accommodates large audio files (30–60 MB).
# For self-hosted Whisper (faster, no per-minute cost), replace this
# block with a subprocess call to the local Whisper binary.
try:
response = requests.post(
WHISPER_API_URL,
headers=headers,
files=files,
timeout=(5, 120)
)
except requests.exceptions.Timeout:
print(f"[!] Whisper API timeout for {ticker} Q{quarter}")
return None
if response.status_code == 429:
# Rate limit hit — respect Retry-After or default to 60 seconds.
retry_after = int(response.headers.get("Retry-After", 60))
print(f"[!] Whisper rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
return transcribe_with_whisper(audio_filepath, ticker, quarter, model)
if response.status_code != 200:
print(f"[✗] Whisper API error {response.status_code}: {response.text}")
return None
result = response.json()
# Estimate confidence from segment-level probability scores (if available)
segments = result.get("segments", [])
confidences = [seg.get("avg_logprob", -1.0) for seg in segments]
# Convert log probability to approximate confidence (0–1 scale)
avg_confidence = sum(c for c in confidences if c > -1.5) / max(len([c for c in confidences if c > -1.5]), 1)
confidence_avg = max(0, min(1, (avg_confidence + 1.5) / 1.5))
transcript = TranscriptResult(
ticker=ticker,
quarter=quarter,
text=result.get("text", ""),
duration_seconds=result.get("duration", 0.0),
segments=segments,
model=model,
confidence_avg=confidence_avg
)
# Persist transcript for audit trail
transcript_path = TRANSCRIPT_DIR / f"{ticker}_{quarter}_transcript.json"
with open(transcript_path, "w") as f:
json.dump(asdict(transcript), f, indent=2)
print(f"[✓] Transcribed {ticker} Q{quarter}: {transcript.duration_seconds:.0f}s, {len(transcript.text)} chars, confidence={confidence_avg:.2f}")
return transcript
Whisper hallucination is a known failure mode, particularly with company-specific acronyms, obscure product names, and international executive accents. The confidence_avg field in the output allows downstream logic to flag low-confidence transcripts for manual review. A confidence below 0.6 on a financial transcript warrants a second pass with a larger model.
Stage 3: LLM Sentiment Scoring
Raw transcript text is not directly useful for quantitative analysis. The LLM scoring stage interprets the text through a structured prompt that extracts:
- Overall sentiment score (1–10 scale, where 10 = most positive)
- Confidence level (low / medium / high) based on language clarity
- Key positive phrases (specific language indicating optimism)
- Key negative phrases (specific language indicating concern or guidance cuts)
- Tone delta relative to the previous quarter (much worse / worse / stable / better / much better)
- Forward guidance signal (positive / neutral / negative / absent)
The prompt is the core engineering artifact of this pipeline. A poorly designed prompt produces inconsistent scores across companies and quarters, making time-series analysis unreliable.
import os
import json
import time
import requests
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class SentimentScore:
ticker: str
quarter: str
sentiment_score: float # 1–10 scale
confidence: str # "low" | "medium" | "high"
key_positive_phrases: list[str]
key_negative_phrases: list[str]
tone_delta: str # "much_worse" | "worse" | "stable" | "better" | "much_better"
forward_guidance_signal: str # "positive" | "neutral" | "negative" | "absent"
raw_llm_output: str
model: str
SENTIMENT_PROMPT = """
You are a financial analyst specializing in earnings call interpretation.
Analyze the following earnings call transcript and return a structured JSON response.
SCORING RULES:
- sentiment_score: Rate the overall tone from 1 (extremely negative/bearish) to 10 (extremely positive/bullish).
Consider: revenue growth language, margin commentary, management confidence, analyst reception.
- confidence: Rate your confidence in the assessment based on language clarity and consistency.
- key_positive_phrases: Extract 3–5 specific phrases (verbatim from transcript) indicating optimism.
- key_negative_phrases: Extract 3–5 specific phrases (verbatim from transcript) indicating concern or caution.
- tone_delta: Compare to previous quarter — is sentiment improving, stable, or deteriorating?
- forward_guidance_signal: Classify management's forward guidance as positive, neutral, negative, or absent.
OUTPUT FORMAT: Return ONLY valid JSON with these exact keys:
{
"sentiment_score": <float 1-10>,
"confidence": "<low|medium|high>",
"key_positive_phrases": [<string>, ...],
"key_negative_phrases": [<string>, ...],
"tone_delta": "<much_worse|worse|stable|better|much_better>",
"forward_guidance_signal": "<positive|neutral|negative|absent>",
"interpretation_summary": "<2-3 sentence summary of the overall tone>"
}
TRANSCRIPT:
{transcript_text}
"""
def score_sentiment_with_llm(
transcript_text: str,
ticker: str,
quarter: str,
model: str = "gpt-4o",
temperature: float = 0.1
) -> Optional[SentimentScore]:
"""
Send the transcript to an LLM for structured sentiment scoring.
Temperature is set to 0.1 to reduce randomness in score assignment.
"""
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
raise EnvironmentError("OPENAI_API_KEY environment variable not set")
# Truncate to fit context window (Whisper output for a 30-min call is typically 8,000–15,000 tokens).
# For longer transcripts, consider splitting into sections and aggregating scores.
max_chars = 120_000
truncated_text = transcript_text[:max_chars]
payload = {
"model": model,
"messages": [
{"role": "system", "content": "You are a financial analyst. Return only valid JSON."},
{"role": "user", "content": SENTIMENT_PROMPT.format(transcript_text=truncated_text)}
],
"temperature": temperature,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
max_retries = 3
for attempt in range(max_retries):
try:
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=payload,
timeout=(10, 60)
)
except requests.exceptions.Timeout:
print(f"[!] LLM API timeout for {ticker} Q{quarter}, attempt {attempt + 1}")
time.sleep(5 * (attempt + 1))
continue
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 30))
print(f"[!] LLM rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
continue
if response.status_code != 200:
print(f"[✗] LLM API error {response.status_code}: {response.text}")
return None
break
result = response.json()
content = result["choices"][0]["message"]["content"]
try:
parsed = json.loads(content)
except json.JSONDecodeError:
print(f"[!] LLM returned malformed JSON for {ticker} Q{quarter}. Raw output:")
print(content[:500])
return None
score = SentimentScore(
ticker=ticker,
quarter=quarter,
sentiment_score=parsed.get("sentiment_score", 5.0),
confidence=parsed.get("confidence", "medium"),
key_positive_phrases=parsed.get("key_positive_phrases", []),
key_negative_phrases=parsed.get("key_negative_phrases", []),
tone_delta=parsed.get("tone_delta", "stable"),
forward_guidance_signal=parsed.get("forward_guidance_signal", "absent"),
raw_llm_output=content,
model=model
)
print(f"[✓] Scored {ticker} Q{quarter}: sentiment={score.sentiment_score:.1f}, delta={score.tone_delta}, guidance={score.forward_guidance_signal}")
return score
The temperature=0.1 setting is deliberate. Sentiment scoring benefits from consistency, not creativity. A higher temperature introduces score drift across quarters for the same company — the antithesis of what a time-series analysis requires.
The prompt instructs the LLM to return response_format: {"type": "json_object"} (OpenAI's structured output feature). This is not cosmetic — it eliminates the parsing fragility of extracting JSON from free-form LLM output, which was historically one of the most common failure modes in LLM-based financial pipelines.
Stage 4: Orchestration and Output Storage
Individual components are only useful when wired together into a reproducible pipeline. The orchestrator handles:
- Batch processing of multiple tickers
- Sequential execution (transcription must complete before scoring)
- Error recovery (skip failed items, log for retry)
- Results aggregation into a CSV for downstream analysis
import csv
import time
from pathlib import Path
from dataclasses import asdict
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
@dataclass
class PipelineConfig:
max_concurrent_transcriptions: int = 2 # Whisper is I/O-bound; limit concurrent uploads
max_concurrent_scorers: int = 4 # LLM API is stateless; higher concurrency is safe
retry_failed: bool = True
output_csv: str = "./sentiment_scores.csv"
def run_earnings_pipeline(
audio_sources: list[dict],
config: PipelineConfig = PipelineConfig()
) -> list[SentimentScore]:
"""
Main pipeline: fetch audio → transcribe → score → aggregate results.
audio_sources: list of dicts with keys: "url", "ticker", "quarter", "audio_filepath"
(if audio is already downloaded locally, provide "audio_filepath" instead of "url")
"""
results: list[SentimentScore] = []
for source in audio_sources:
ticker = source["ticker"]
quarter = source["quarter"]
# Step 1: Get audio file
if "audio_filepath" in source and Path(source["audio_filepath"]).exists():
audio_path = source["audio_filepath"]
elif "url" in source:
audio_path = download_earnings_audio(source["url"], ticker, quarter)
else:
print(f"[✗] No audio source for {ticker} Q{quarter}")
continue
if not audio_path:
print(f"[✗] Skipping {ticker} Q{quarter}: audio download failed")
continue
# Step 2: Transcribe
transcript = transcribe_with_whisper(audio_path, ticker, quarter)
if not transcript:
print(f"[✗] Skipping {ticker} Q{quarter}: transcription failed")
continue
# Step 3: Score
# Small delay to avoid burst patterns hitting rate limits
time.sleep(1.0)
score = score_sentiment_with_llm(transcript.text, ticker, quarter)
if not score:
print(f"[✗] Skipping {ticker} Q{quarter}: scoring failed")
continue
results.append(score)
# Step 4: Persist to CSV
if results:
output_path = Path(config.output_csv)
write_header = not output_path.exists()
with open(output_path, "a", newline="") as f:
fieldnames = [
"ticker", "quarter", "sentiment_score", "confidence",
"tone_delta", "forward_guidance_signal",
"num_positive_phrases", "num_negative_phrases", "model"
]
writer = csv.DictWriter(f, fieldnames=fieldnames)
if write_header:
writer.writeheader()
for score in results:
writer.writerow({
"ticker": score.ticker,
"quarter": score.quarter,
"sentiment_score": score.sentiment_score,
"confidence": score.confidence,
"tone_delta": score.tone_delta,
"forward_guidance_signal": score.forward_guidance_signal,
"num_positive_phrases": len(score.key_positive_phrases),
"num_negative_phrases": len(score.key_negative_phrases),
"model": score.model,
})
print(f"[✓] Wrote {len(results)} results to {output_path}")
return results
Cost and Latency Benchmarks
A realistic pipeline run for 50 earnings calls involves the following approximate costs and timings:
| Stage | Per-call cost | 50-call batch | Notes |
|---|---|---|---|
| Audio download | ~$0 | ~$0 | IR feeds are publicly accessible |
| Whisper (whisper-1) | ~$0.006/min | ~$0.30 | Based on OpenAI pricing; ~30-min call |
| GPT-4o scoring | ~$0.015/call | ~$0.75 | Based on ~8,000-token input; 500-token output |
| Total per ticker | ~$0.021 |
Latency per call (sequential): approximately 60–90 seconds (transcription) + 3–8 seconds (scoring) = 65–100 seconds total.
With max_concurrent_transcriptions=2 and max_concurrent_scorers=4, a 50-call batch completes in approximately 20–30 minutes on commodity hardware with stable network connectivity.
Practical Limitations and Known Failure Modes
This pipeline is a research tool, not a live trading system. Several practical constraints apply:
Hallucination in transcription. Whisper occasionally generates plausible but incorrect phrases, particularly for company-specific acronyms, product names, and non-native English speakers. The confidence_avg field in the transcript output is your first line of defense. For a production research pipeline, flag confidence below 0.6 for manual review. For a live trading system, this threshold should be zero tolerance.
LLM score instability. Even with temperature=0.1, the same transcript scored at different times by the same model can produce scores that differ by ±0.3. This is acceptable for directional analysis (identifying large sentiment shifts), but insufficient for precise quantitative signals. Aggregate scores across multiple LLM calls if precision matters.
Audio availability. Earnings calls are not always available in a downloadable format on the day of the event. Some companies stream audio without providing an on-demand download link. Others provide audio behind a registration wall. Coverage is incomplete, and the missing data is not random — smaller companies and international ADRs are disproportionately affected.
Data freshness. This pipeline processes historical calls. For real-time analysis, the audio acquisition step must be replaced with a live audio stream capture, which introduces additional complexity around stream authentication and segment stitching.
Integrating with Market Data Pipelines
The sentiment scores generated by this pipeline are most useful when combined with other market signals. The natural integration point is the TickDB API, which provides real-time market data and historical OHLCV for the same equities you are analyzing.
For example, after scoring an earnings call, you can pull the post-earnings price action using the TickDB kline endpoint to correlate the sentiment score with actual market movement:
import os
import requests
# Fetch 5-minute candlesticks for the earnings day
# Useful for measuring the post-call price reaction window
headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}
response = requests.get(
"https://api.tickdb.ai/v1/market/kline",
headers=headers,
params={
"symbol": "NVDA.US",
"interval": "5m",
"limit": 78, # ~6.5 hours of market hours
},
timeout=(3.05, 10)
)
# kline data returns: timestamp, open, high, low, close, volume
# Correlate this with the sentiment_score from the LLM output
data = response.json().get("data", [])
The combination of sentiment analysis (this pipeline) and tick-level price data (TickDB) enables backtesting the predictive power of earnings call language against subsequent price action. That is where the signal becomes actionable.
Scaling to Sector-Wide Coverage
The configuration above is designed for reliability at small scale. Scaling to cover all S&P 500 companies per quarter requires changes in three dimensions:
Parallelization. Replace the sequential run_earnings_pipeline with a task queue (Celery, RQ, or AWS SQS) that distributes work across a fleet of workers. Whisper transcription is the bottleneck — parallelize aggressively there.
Audio sourcing. A manual list of audio URLs does not scale. At sector scale, you need an automated scanner that ingests SEC 8-K filings daily, extracts the audio hyperlink from the HTML, and queues the download task. Libraries like sec-edgar-downloader or custom BeautifulSoup scrapers are the starting point.
Cost management. At 500 companies per quarter, Whisper + GPT-4o costs approximately $10 per run. This is negligible for a research pipeline. At 5,000 companies per quarter across multiple asset classes, costs compound. Consider switching to a self-hosted Whisper model (zero per-minute cost, one-time GPU investment) and a local LLM (Llama 3.1 70B via vLLM) for cost predictability at scale.
Closing
The earnings call transcript is one of the most information-dense, publicly available signals in equity markets. The gap between what is said on the call and what the market prices in is where quant researchers and discretionary traders alike find edges.
Automating the transcription and scoring of this content does not replace human judgment. It amplifies the analyst's capacity to scan across sectors, detect sentiment divergences, and identify where management language has drifted from consensus expectations.
The pipeline described here is a starting architecture. The production version will diverge based on your specific coverage universe, latency requirements, and the degree to which you trust LLM-generated sentiment scores as inputs to a trading model.
Next Steps
If you want to run this pipeline yourself, the code in this article is structured for direct use. Set OPENAI_API_KEY and TICKDB_API_KEY as environment variables, provide a list of audio sources, and run the orchestrator.
If you need historical OHLCV data to correlate with your sentiment scores, TickDB provides 10+ years of cleaned US equity kline data via REST API. Sign up at tickdb.ai for a free API key with no credit card required.
If you're building a sector-wide research platform, contact enterprise@tickdb.ai for institutional data plans that include coverage for international equities and historical depth data.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace to access TickDB endpoints directly from your development environment.
This article does not constitute investment advice. Earnings call sentiment scores generated by LLM models are research artifacts with inherent limitations including hallucination risk, score instability, and incomplete coverage. Markets involve risk; past performance does not guarantee future results.