yirugi
/
stock-oracle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: Earnings Surprise에 Alpha Vantage 애널리스트 추정치 통합

Browse Source 
- 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 시 기존 데이터 삭제 후 재인덱싱
main
I Luk Kim 5 months ago
parent 971c0f053f
commit 99a998b2f1
2 changed files with 180 additions and 87 deletions
Show Stats Download Patch File Download Diff File

12
app/api/v1/endpoints/earnings.py
Unescape Escape View File

@ -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"])

255
app/services/earnings_service.py
Unescape Escape View File

@ -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

Write Preview
Loading…
Cancel
Save