File size: 3,532 Bytes
35676b4 e6496c0 35676b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | import logging
import httpx
from ingestion.alphavantage import AV_BASE, _next_key, _key_count, _is_rate_limited
logger = logging.getLogger(__name__)
def _parse_transcript(data: dict) -> str:
raw = data.get("transcript", "")
if isinstance(raw, list):
lines = []
for item in raw:
speaker = item.get("speaker", "")
title = item.get("title", "")
label = f"{speaker} ({title})" if title else speaker
lines.append(f"{label}: {item.get('content', '')}")
return "\n".join(lines)
if raw is None:
return ""
return str(raw)
def fetch_transcript(ticker: str, quarter: str) -> str:
"""Fetch earnings call transcript from Alpha Vantage.
Args:
ticker: Stock symbol e.g. "AAPL"
quarter: Alpha Vantage format e.g. "2024Q4"
Returns:
Transcript text, or empty string if unavailable.
"""
rate_limited_keys = 0
n_keys = max(_key_count(), 1)
for attempt in range(n_keys):
key = _next_key()
key_label = f"key[...{key[-4:]}]"
try:
resp = httpx.get(
AV_BASE,
params={
"function": "EARNINGS_CALL_TRANSCRIPT",
"symbol": ticker.upper(),
"quarter": quarter,
"apikey": key,
},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
if "Information" in data:
rate_limited_keys += 1
logger.warning(
"[transcript] %s %s %s — rate limit hit (attempt %d/%d): %s",
ticker, quarter, key_label, attempt + 1, n_keys, data["Information"],
)
continue
if "Note" in data:
rate_limited_keys += 1
logger.warning(
"[transcript] %s %s %s — API note/rate limit (attempt %d/%d): %s",
ticker, quarter, key_label, attempt + 1, n_keys, data["Note"],
)
continue
if "Error Message" in data:
logger.warning(
"[transcript] %s %s %s — API error: %s",
ticker, quarter, key_label, data["Error Message"],
)
return ""
text = _parse_transcript(data)
if text:
logger.info(
"[transcript] %s %s — fetched %d chars via %s",
ticker, quarter, len(text), key_label,
)
else:
logger.info(
"[transcript] %s %s — API returned empty transcript (no content available) via %s",
ticker, quarter, key_label,
)
return text
except ValueError:
raise
except Exception as e:
logger.warning(
"[transcript] %s %s %s — request error (attempt %d/%d): %s",
ticker, quarter, key_label, attempt + 1, n_keys, e,
)
continue
if rate_limited_keys == n_keys:
logger.error(
"[transcript] %s %s — ALL %d keys exhausted by rate limits. No transcript fetched.",
ticker, quarter, n_keys,
)
else:
logger.error(
"[transcript] %s %s — failed after %d attempt(s). No transcript fetched.",
ticker, quarter, n_keys,
)
return ""
|