""" Earnings Surprise service — SEC EDGAR XBRL 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. """ import logging from datetime import datetime, timezone from typing import Dict, List, Optional, Tuple from sqlalchemy import select, desc, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.dialects.postgresql import insert as pg_insert 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 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 # ------------------------------------------------------------------ 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}") 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", {}) # Try each EPS concept in priority order eps_entries = [] 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") 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 except ValueError: continue eps_entries.append({ "end": end, "val": float(val), "filed": filed, "form": form, "concept": concept, }) if eps_entries: break # Use the first concept that has data return eps_entries # ------------------------------------------------------------------ # Index to DB with QoQ surprise # ------------------------------------------------------------------ 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: count = await db.execute( select(func.count(EarningsSurprise.id)).where( EarningsSurprise.ticker == ticker ) ) 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 ) ) await db.flush() self._http.set_deadline(60.0) try: eps_entries = await self._fetch_eps_from_xbrl(ticker) finally: self._http.clear_deadline() if not eps_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). unique: List[Dict] = [] for e in eps_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 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 rows = [] for i, entry in enumerate(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"] prev_eps = unique[i - 1]["val"] if i > 0 else 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) 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 "surprise": surprise, "surprise_percentage": surprise_pct, "data_source": "SEC_XBRL", }) if not rows: return 0 # Batch upsert inserted = 0 for i in range(0, len(rows), _CHUNK): chunk = rows[i:i + _CHUNK] stmt = pg_insert(EarningsSurprise).values(chunk) stmt = stmt.on_conflict_do_update( constraint="uq_earnings_surprise", set_={ "reported_date": stmt.excluded.reported_date, "reported_eps": stmt.excluded.reported_eps, "estimated_eps": stmt.excluded.estimated_eps, "surprise": stmt.excluded.surprise, "surprise_percentage": stmt.excluded.surprise_percentage, "data_source": stmt.excluded.data_source, }, ) result = await db.execute(stmt) inserted += result.rowcount await db.commit() logger.info(f"Earnings: upserted {inserted} quarters for {ticker} (XBRL)") return inserted # ------------------------------------------------------------------ # 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 ) ) if (count_q.scalar() or 0) == 0: await self.index_earnings(db, ticker) result = await db.execute( select(EarningsSurprise) .where(EarningsSurprise.ticker == ticker) .order_by(desc(EarningsSurprise.fiscal_date_ending)) .limit(quarters) ) rows = result.scalars().all() # Compute streak streak = 0 if rows: first_sign = None for r in rows: if r.surprise is None: break if first_sign is None: first_sign = r.surprise > 0 if (r.surprise > 0) == first_sign: streak += 1 if first_sign else -1 else: break if first_sign is False: streak = -abs(streak) 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