From e121c7ab40fbaf92d1795549d793ee9082649b1d Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Wed, 25 Mar 2026 12:16:15 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Earnings=20Surprise=EB=A5=BC=20yfinance?= =?UTF-8?q?-plus=EB=A1=9C=20=EC=A0=84=ED=99=98=20(Alpha=20Vantage=20?= =?UTF-8?q?=EC=A0=9C=EA=B1=B0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - yfinance Ticker.earnings_dates에서 EPS Estimate + Reported EPS + Surprise(%) 직접 제공 - Alpha Vantage 의존성 완전 제거 (API 키 불필요, rate limit 없음) - ~25분기(6년+) 커버리지, 진짜 애널리스트 컨센서스 기반 - AAPL: 12분기 연속 beat, avg +4.36% - MSFT: 8분기 연속 beat, avg +4.50% --- app/api/v1/endpoints/earnings.py | 16 +- app/services/earnings_service.py | 306 ++++++++----------------------- 2 files changed, 85 insertions(+), 237 deletions(-) diff --git a/app/api/v1/endpoints/earnings.py b/app/api/v1/endpoints/earnings.py index 09d2fb8..af15016 100644 --- a/app/api/v1/endpoints/earnings.py +++ b/app/api/v1/endpoints/earnings.py @@ -22,20 +22,20 @@ logger = logging.getLogger("app.api.v1.earnings") response_model=EarningsSurpriseResponse, summary="Get earnings surprise history", description=( - "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만 반환." + "Quarterly EPS surprise: reported vs analyst consensus estimate.\n\n" + "**데이터 소스**: yfinance-plus (`Ticker.earnings_dates`). API 키 불필요.\n" + "**커버리지**: ~25분기 (6년+). 첫 조회 시 자동 인덱싱.\n\n" + "**surprise** = reported_eps - estimated_eps.\n" + "**surprise_percentage** = (surprise / estimated) × 100.\n" + "**streak**: 연속 beat (양수) 또는 miss (음수) 횟수." ), ) @with_cache(namespace="earnings:surprise", ttl=None, key_params=["symbol", "quarters"]) async def get_earnings_surprise( symbol: str, response: Response, - quarters: int = Query(8, ge=1, le=80, description="Number of recent quarters (max 80 = ~20 years)"), - force_refresh: bool = Query(False, description="Bypass cache and re-fetch from SEC EDGAR"), + quarters: int = Query(8, ge=1, le=40, description="Number of recent quarters (max ~25 available)"), + force_refresh: bool = Query(False, description="Bypass cache and re-fetch from yfinance"), db: AsyncSession = Depends(get_db), ): svc = EarningsService() diff --git a/app/services/earnings_service.py b/app/services/earnings_service.py index 8f99327..d4ef403 100644 --- a/app/services/earnings_service.py +++ b/app/services/earnings_service.py @@ -1,152 +1,84 @@ """ -Earnings Surprise service — SEC EDGAR XBRL + Alpha Vantage +Earnings Surprise service — yfinance-plus -- Reported EPS: SEC EDGAR XBRL companyfacts (무료, 2009+) -- Estimated EPS: Alpha Vantage EARNINGS endpoint (무료 키, 500 calls/day) -- Surprise = reported - estimated (진짜 애널리스트 컨센서스 대비) +Uses Ticker.earnings_dates which provides: +- EPS Estimate (analyst consensus) +- Reported EPS (actual) +- Surprise(%) (pre-calculated) -Alpha Vantage 키가 없으면 estimated_eps 없이 reported_eps만 저장. +No external API key required. ~25 quarters of history per ticker. """ -import asyncio import logging -import time as _time -from datetime import datetime, timedelta, timezone +from datetime import datetime, 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__) -_EPS_CONCEPTS = [ - "EarningsPerShareDiluted", - "EarningsPerShareBasic", -] _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: - def __init__(self): - self._http = SECHttpClient("Stock Oracle Earnings Service") - # ------------------------------------------------------------------ - # 1) SEC EDGAR XBRL — reported EPS + # Fetch from yfinance-plus # ------------------------------------------------------------------ - async def _fetch_eps_from_xbrl(self, ticker: str) -> List[Dict]: - cik = await self._http.get_company_cik(ticker) - if not cik: - raise ValueError(f"Could not find CIK for ticker {ticker}") + def _fetch_earnings_from_yfinance(self, ticker: str) -> List[Dict]: + """Fetch earnings dates with EPS estimate/actual from yfinance-plus. + + This is a synchronous call (yfinance uses requests internally). + """ + import warnings + warnings.filterwarnings("ignore", category=DeprecationWarning) + warnings.filterwarnings("ignore", category=FutureWarning) - url = f"{self._http.sec_base_data}/api/xbrl/companyfacts/CIK{cik.zfill(10)}.json" - data = await self._http.fetch_json(url) + from yfinance_plus import Ticker + t = Ticker(ticker.upper()) + df = t.earnings_dates - us_gaap = data.get("facts", {}).get("us-gaap", {}) + if df is None or df.empty: + return [] - eps_entries: List[Dict] = [] - for concept in _EPS_CONCEPTS: - if concept not in us_gaap: + rows = [] + for idx, row in df.iterrows(): + # idx is the earnings date (Timestamp with timezone) + reported_date = idx.to_pydatetime() + if reported_date.tzinfo: + reported_date = reported_date.astimezone(timezone.utc) + else: + reported_date = reported_date.replace(tzinfo=timezone.utc) + + estimated_eps = _safe_float(row.get("EPS Estimate")) + reported_eps = _safe_float(row.get("Reported EPS")) + surprise_pct = _safe_float(row.get("Surprise(%)")) + + # Skip future earnings (no reported EPS yet) + if reported_eps is None: continue - 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 - if form not in ("10-Q", "10-K", "10-Q/A", "10-K/A"): - continue - if start and end: - try: - 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": entry.get("filed"), "form": form, - }) - if eps_entries: - break - return eps_entries - # ------------------------------------------------------------------ - # 2) Alpha Vantage — estimated EPS - # ------------------------------------------------------------------ + surprise = None + if reported_eps is not None and estimated_eps is not None: + surprise = round(reported_eps - estimated_eps, 6) - 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 + rows.append({ + "reported_date": reported_date, + "reported_eps": reported_eps, + "estimated_eps": estimated_eps, + "surprise": surprise, + "surprise_percentage": surprise_pct, + }) + + return rows # ------------------------------------------------------------------ - # 3) Index: XBRL + AV merge + # Index to DB # ------------------------------------------------------------------ async def index_earnings( @@ -170,88 +102,32 @@ class EarningsService: ) await db.flush() - # Fetch XBRL reported EPS - self._http.set_deadline(60.0) - try: - xbrl_entries = await self._fetch_eps_from_xbrl(ticker) - finally: - self._http.clear_deadline() - - if not xbrl_entries: - logger.warning(f"Earnings: no XBRL EPS data for {ticker}") + import asyncio + raw = await asyncio.to_thread(self._fetch_earnings_from_yfinance, ticker) + if not raw: + logger.warning(f"Earnings: no yfinance data for {ticker}") return 0 - # Dedup & sort - xbrl_entries.sort(key=lambda x: x["end"]) - unique: List[Dict] = [] - for e in xbrl_entries: - if unique: - 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) - - # Fetch Alpha Vantage estimates (best-effort) - av_estimates = await self._fetch_estimates_from_av(ticker) - - # Build rows — merge XBRL actual + AV estimate - rows = [] - for entry in unique: - try: - 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) - except ValueError: - pass - - reported_eps = entry["val"] - - # 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 - 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({ + db_rows = [] + for r in raw: + # Use reported_date as fiscal_date_ending (earnings announcement date) + db_rows.append({ "ticker": ticker, - "fiscal_date_ending": fiscal_date, - "reported_date": reported_date, - "reported_eps": reported_eps, - "estimated_eps": estimated_eps, - "surprise": surprise, - "surprise_percentage": surprise_pct, - "data_source": data_source, + "fiscal_date_ending": r["reported_date"], + "reported_date": r["reported_date"], + "reported_eps": r["reported_eps"], + "estimated_eps": r["estimated_eps"], + "surprise": r["surprise"], + "surprise_percentage": r["surprise_percentage"], + "data_source": "YFINANCE", }) - if not rows: + if not db_rows: return 0 inserted = 0 - for i in range(0, len(rows), _CHUNK): - chunk = rows[i:i + _CHUNK] + for i in range(0, len(db_rows), _CHUNK): + chunk = db_rows[i:i + _CHUNK] stmt = pg_insert(EarningsSurprise).values(chunk) stmt = stmt.on_conflict_do_update( constraint="uq_earnings_surprise", @@ -268,12 +144,11 @@ class EarningsService: inserted += result.rowcount await db.commit() - 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)") + logger.info(f"Earnings: upserted {inserted} quarters for {ticker}") return inserted # ------------------------------------------------------------------ - # 4) Query + # Query # ------------------------------------------------------------------ async def get_earnings_surprise( @@ -282,7 +157,9 @@ class EarningsService: 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) @@ -295,7 +172,7 @@ class EarningsService: ) rows = result.scalars().all() - # Streak: consecutive beats or misses + # Streak: consecutive beats (surprise > 0) or misses streak = 0 if rows: first_sign = None @@ -317,41 +194,12 @@ class EarningsService: 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 == "": + if val is None: return None try: - return float(val) + import math + f = float(val) + return None if math.isnan(f) else f except (ValueError, TypeError): return None