From 99a998b2f1625fb9037336e09f0fb4eb1feedbcb Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Wed, 25 Mar 2026 08:20:57 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Earnings=20Surprise=EC=97=90=20Alpha=20?= =?UTF-8?q?Vantage=20=EC=95=A0=EB=84=90=EB=A6=AC=EC=8A=A4=ED=8A=B8=20?= =?UTF-8?q?=EC=B6=94=EC=A0=95=EC=B9=98=20=ED=86=B5=ED=95=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - XBRL reported EPS + Alpha Vantage estimated EPS 합산 - surprise = reported - estimated (진짜 컨센서스 대비) - fiscal 날짜 ±15일 fuzzy match (Apple 등 비표준 fiscal calendar 대응) - AV 키 미설정 시 reported EPS만 반환 (graceful degradation) - AV rate limiter: 5 req/min token bucket - force_refresh 시 기존 데이터 삭제 후 재인덱싱 --- app/api/v1/endpoints/earnings.py | 12 +- app/services/earnings_service.py | 255 +++++++++++++++++++++---------- 2 files changed, 180 insertions(+), 87 deletions(-) diff --git a/app/api/v1/endpoints/earnings.py b/app/api/v1/endpoints/earnings.py index 67a7caf..09d2fb8 100644 --- a/app/api/v1/endpoints/earnings.py +++ b/app/api/v1/endpoints/earnings.py @@ -22,12 +22,12 @@ logger = logging.getLogger("app.api.v1.earnings") response_model=EarningsSurpriseResponse, summary="Get earnings surprise history", description=( - "Quarterly EPS surprise data from SEC EDGAR XBRL.\n\n" - "**데이터 소스**: SEC EDGAR companyfacts API (무료, API 키 불필요).\n" - "**커버리지**: 2009년~ (대부분 종목). 첫 조회 시 자동 인덱싱.\n\n" - "**surprise**: 현재 분기 EPS - 이전 분기 EPS (QoQ 변화).\n" - "**estimated_eps**: 이전 분기 EPS (비교 기준).\n" - "**streak**: 연속 QoQ 증가 (양수) 또는 감소 (음수) 횟수." + "Quarterly EPS surprise: reported (SEC XBRL) vs estimated (Alpha Vantage consensus).\n\n" + "**데이터 소스**:\n" + "- Reported EPS: SEC EDGAR XBRL (무료, 2009년~)\n" + "- Estimated EPS: Alpha Vantage EARNINGS API (`ALPHA_VANTAGE_API_KEY` 설정 시)\n\n" + "**surprise** = reported_eps - estimated_eps (애널리스트 컨센서스 대비).\n" + "API 키 미설정 시 estimated_eps 없이 reported_eps만 반환." ), ) @with_cache(namespace="earnings:surprise", ttl=None, key_params=["symbol", "quarters"]) diff --git a/app/services/earnings_service.py b/app/services/earnings_service.py index 5d6eaa2..8f99327 100644 --- a/app/services/earnings_service.py +++ b/app/services/earnings_service.py @@ -1,46 +1,75 @@ """ -Earnings Surprise service — SEC EDGAR XBRL +Earnings Surprise service — SEC EDGAR XBRL + Alpha Vantage -Fetches quarterly EPS from SEC EDGAR companyfacts API and computes -QoQ surprise (current EPS - previous quarter EPS). -Coverage: 2009+ for most companies. No API key required. +- Reported EPS: SEC EDGAR XBRL companyfacts (무료, 2009+) +- Estimated EPS: Alpha Vantage EARNINGS endpoint (무료 키, 500 calls/day) +- Surprise = reported - estimated (진짜 애널리스트 컨센서스 대비) + +Alpha Vantage 키가 없으면 estimated_eps 없이 reported_eps만 저장. """ +import asyncio import logging -from datetime import datetime, timezone +import time as _time +from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional, Tuple +import aiohttp from sqlalchemy import select, desc, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.dialects.postgresql import insert as pg_insert +from app.core.config import settings +from app.core.http_client import get_http_session from app.models.earnings_surprise import EarningsSurprise from app.services.sec_http_client import SECHttpClient logger = logging.getLogger(__name__) -# XBRL EPS concepts in priority order (diluted preferred over basic) _EPS_CONCEPTS = [ "EarningsPerShareDiluted", "EarningsPerShareBasic", ] - -# Chunk size for batch insert _CHUNK = 2000 +_AV_BASE = "https://www.alphavantage.co/query" + + +# --------------------------------------------------------------------------- +# Alpha Vantage rate limiter (5 req/min) +# --------------------------------------------------------------------------- +class _RateLimiter: + def __init__(self, rate: float = 5.0 / 60.0, capacity: float = 1.0): + self._rate = rate + self._capacity = capacity + self._tokens = capacity + self._last = _time.monotonic() + self._lock = asyncio.Lock() + + async def acquire(self) -> None: + while True: + async with self._lock: + now = _time.monotonic() + self._tokens = min(self._capacity, self._tokens + (now - self._last) * self._rate) + self._last = now + if self._tokens >= 1.0: + self._tokens -= 1.0 + return + wait = (1.0 - self._tokens) / self._rate + await asyncio.sleep(wait) + +_av_limiter = _RateLimiter() class EarningsService: - """SEC EDGAR XBRL-based Earnings Surprise service.""" def __init__(self): self._http = SECHttpClient("Stock Oracle Earnings Service") # ------------------------------------------------------------------ - # Fetch EPS from SEC EDGAR XBRL + # 1) SEC EDGAR XBRL — reported EPS # ------------------------------------------------------------------ async def _fetch_eps_from_xbrl(self, ticker: str) -> List[Dict]: - """Fetch quarterly EPS data from SEC EDGAR companyfacts XBRL API.""" cik = await self._http.get_company_cik(ticker) if not cik: raise ValueError(f"Could not find CIK for ticker {ticker}") @@ -48,62 +77,81 @@ class EarningsService: url = f"{self._http.sec_base_data}/api/xbrl/companyfacts/CIK{cik.zfill(10)}.json" data = await self._http.fetch_json(url) - facts = data.get("facts", {}) - us_gaap = facts.get("us-gaap", {}) + us_gaap = data.get("facts", {}).get("us-gaap", {}) - # Try each EPS concept in priority order - eps_entries = [] + eps_entries: List[Dict] = [] for concept in _EPS_CONCEPTS: if concept not in us_gaap: continue - units = us_gaap[concept].get("units", {}) - usd_per_share = units.get("USD/shares", []) - if not usd_per_share: - continue - - for entry in usd_per_share: - start = entry.get("start") - end = entry.get("end") - val = entry.get("val") - filed = entry.get("filed") + for entry in us_gaap[concept].get("units", {}).get("USD/shares", []): + start, end, val = entry.get("start"), entry.get("end"), entry.get("val") form = entry.get("form", "") - if not end or val is None: continue - # Only quarterly filings (10-Q) and annual (10-K) if form not in ("10-Q", "10-K", "10-Q/A", "10-K/A"): continue - # Skip annual periods for quarterly analysis (period > 100 days) if start and end: try: - s = datetime.strptime(start, "%Y-%m-%d") - e = datetime.strptime(end, "%Y-%m-%d") - if (e - s).days > 100: - continue # Annual or semi-annual period, skip + if (datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days > 100: + continue except ValueError: continue - eps_entries.append({ - "end": end, - "val": float(val), - "filed": filed, - "form": form, - "concept": concept, + "end": end, "val": float(val), + "filed": entry.get("filed"), "form": form, }) - if eps_entries: - break # Use the first concept that has data - + break return eps_entries # ------------------------------------------------------------------ - # Index to DB with QoQ surprise + # 2) Alpha Vantage — estimated EPS + # ------------------------------------------------------------------ + + async def _fetch_estimates_from_av(self, ticker: str) -> Dict[str, Dict]: + """Fetch estimated EPS from Alpha Vantage. Returns {fiscal_date -> {estimatedEPS, ...}}.""" + api_key = settings.ALPHA_VANTAGE_API_KEY + if not api_key: + return {} + + await _av_limiter.acquire() + + url = f"{_AV_BASE}?function=EARNINGS&symbol={ticker.upper()}&apikey={api_key}" + try: + session = await get_http_session() + async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp: + if resp.status != 200: + return {} + data = await resp.json(content_type=None) + if "Error Message" in data or "Note" in data: + logger.warning(f"AV earnings error for {ticker}: {data.get('Error Message') or data.get('Note')}") + return {} + except Exception as e: + logger.warning(f"AV earnings fetch error for {ticker}: {e}") + return {} + + # Build lookup: fiscal_date_ending -> estimates + estimates: Dict[str, Dict] = {} + for q in data.get("quarterlyEarnings", []): + fd = q.get("fiscalDateEnding") + if not fd: + continue + estimates[fd] = { + "estimated_eps": _safe_float(q.get("estimatedEPS")), + "reported_date": q.get("reportedDate"), + "av_reported_eps": _safe_float(q.get("reportedEPS")), + "av_surprise": _safe_float(q.get("surprise")), + "av_surprise_pct": _safe_float(q.get("surprisePercentage")), + } + return estimates + + # ------------------------------------------------------------------ + # 3) Index: XBRL + AV merge # ------------------------------------------------------------------ async def index_earnings( self, db: AsyncSession, ticker: str, force_refresh: bool = False ) -> int: - """Fetch EPS from SEC XBRL, compute QoQ surprise, and upsert into DB.""" ticker = ticker.upper() if not force_refresh: @@ -115,7 +163,6 @@ class EarningsService: if (count.scalar() or 0) > 0: return 0 else: - # Delete existing data for clean re-index await db.execute( EarningsSurprise.__table__.delete().where( EarningsSurprise.ticker == ticker @@ -123,76 +170,85 @@ class EarningsService: ) await db.flush() + # Fetch XBRL reported EPS self._http.set_deadline(60.0) try: - eps_entries = await self._fetch_eps_from_xbrl(ticker) + xbrl_entries = await self._fetch_eps_from_xbrl(ticker) finally: self._http.clear_deadline() - if not eps_entries: + if not xbrl_entries: logger.warning(f"Earnings: no XBRL EPS data for {ticker}") return 0 - # Sort by date ascending - eps_entries.sort(key=lambda x: x["end"]) - - # Deduplicate: entries within 15 days are the same quarter. - # Keep the one filed from 10-Q (quarterly) over 10-K (annual). + # Dedup & sort + xbrl_entries.sort(key=lambda x: x["end"]) unique: List[Dict] = [] - for e in eps_entries: + for e in xbrl_entries: if unique: - prev_date = datetime.strptime(unique[-1]["end"], "%Y-%m-%d") - curr_date = datetime.strptime(e["end"], "%Y-%m-%d") - if abs((curr_date - prev_date).days) <= 15: - # Same quarter — prefer 10-Q over 10-K + prev_d = datetime.strptime(unique[-1]["end"], "%Y-%m-%d") + curr_d = datetime.strptime(e["end"], "%Y-%m-%d") + if abs((curr_d - prev_d).days) <= 15: if e["form"].startswith("10-Q") and not unique[-1]["form"].startswith("10-Q"): unique[-1] = e continue unique.append(e) - # Build rows with QoQ surprise + # Fetch Alpha Vantage estimates (best-effort) + av_estimates = await self._fetch_estimates_from_av(ticker) + + # Build rows — merge XBRL actual + AV estimate rows = [] - for i, entry in enumerate(unique): + for entry in unique: try: - fiscal_date = datetime.strptime(entry["end"], "%Y-%m-%d").replace( - tzinfo=timezone.utc - ) + fiscal_date = datetime.strptime(entry["end"], "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError: continue reported_date = None if entry.get("filed"): try: - reported_date = datetime.strptime(entry["filed"], "%Y-%m-%d").replace( - tzinfo=timezone.utc - ) + reported_date = datetime.strptime(entry["filed"], "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError: pass reported_eps = entry["val"] - prev_eps = unique[i - 1]["val"] if i > 0 else None + + # Match AV estimate by date (try exact, then ±5 days) + av = _match_av_estimate(entry["end"], av_estimates) + + estimated_eps = None surprise = None surprise_pct = None - if prev_eps is not None: - surprise = round(reported_eps - prev_eps, 6) - if prev_eps != 0: - surprise_pct = round((surprise / abs(prev_eps)) * 100, 4) + data_source = "SEC_XBRL" + + if av and av.get("estimated_eps") is not None: + estimated_eps = av["estimated_eps"] + surprise = round(reported_eps - estimated_eps, 6) + if estimated_eps != 0: + surprise_pct = round((surprise / abs(estimated_eps)) * 100, 4) + data_source = "SEC_XBRL+AV" + # Use AV reported_date if we don't have one + if not reported_date and av.get("reported_date"): + try: + reported_date = datetime.strptime(av["reported_date"], "%Y-%m-%d").replace(tzinfo=timezone.utc) + except ValueError: + pass rows.append({ "ticker": ticker, "fiscal_date_ending": fiscal_date, "reported_date": reported_date, "reported_eps": reported_eps, - "estimated_eps": prev_eps, # previous quarter as baseline + "estimated_eps": estimated_eps, "surprise": surprise, "surprise_percentage": surprise_pct, - "data_source": "SEC_XBRL", + "data_source": data_source, }) if not rows: return 0 - # Batch upsert inserted = 0 for i in range(0, len(rows), _CHUNK): chunk = rows[i:i + _CHUNK] @@ -212,23 +268,21 @@ class EarningsService: inserted += result.rowcount await db.commit() - logger.info(f"Earnings: upserted {inserted} quarters for {ticker} (XBRL)") + av_matched = sum(1 for r in rows if r["data_source"] == "SEC_XBRL+AV") + logger.info(f"Earnings: upserted {inserted} quarters for {ticker} ({av_matched} with AV estimates)") return inserted # ------------------------------------------------------------------ - # Query + # 4) Query # ------------------------------------------------------------------ async def get_earnings_surprise( self, db: AsyncSession, ticker: str, quarters: int = 8 ) -> Tuple[List[EarningsSurprise], Dict]: - """Get earnings surprise data. Auto-indexes if no data.""" ticker = ticker.upper() count_q = await db.execute( - select(func.count(EarningsSurprise.id)).where( - EarningsSurprise.ticker == ticker - ) + select(func.count(EarningsSurprise.id)).where(EarningsSurprise.ticker == ticker) ) if (count_q.scalar() or 0) == 0: await self.index_earnings(db, ticker) @@ -241,7 +295,7 @@ class EarningsService: ) rows = result.scalars().all() - # Compute streak + # Streak: consecutive beats or misses streak = 0 if rows: first_sign = None @@ -260,5 +314,44 @@ class EarningsService: pcts = [r.surprise_percentage for r in rows if r.surprise_percentage is not None] avg_pct = round(sum(pcts) / len(pcts), 4) if pcts else None - stats = {"streak": streak, "avg_surprise_pct": avg_pct} - return rows, stats + return rows, {"streak": streak, "avg_surprise_pct": avg_pct} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _match_av_estimate(xbrl_date: str, av_estimates: Dict[str, Dict]) -> Optional[Dict]: + """Match XBRL fiscal date to Alpha Vantage estimate. + + Apple fiscal quarter ends on last Saturday (e.g., 12-28) while AV uses + calendar quarter end (12-31), so we need a wide fuzzy match window. + Strategy: same quarter = same year+month pair within ±15 days. + """ + if not av_estimates: + return None + if xbrl_date in av_estimates: + return av_estimates[xbrl_date] + try: + xd = datetime.strptime(xbrl_date, "%Y-%m-%d") + except ValueError: + return None + best, best_delta = None, 999 + for av_date_str, av_data in av_estimates.items(): + try: + ad = datetime.strptime(av_date_str, "%Y-%m-%d") + delta = abs((xd - ad).days) + if delta <= 15 and delta < best_delta: + best, best_delta = av_data, delta + except ValueError: + continue + return best + + +def _safe_float(val) -> Optional[float]: + if val is None or val == "None" or val == "": + return None + try: + return float(val) + except (ValueError, TypeError): + return None