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

feat: Earnings Surprise를 yfinance-plus로 전환 (Alpha Vantage 제거)

Browse Source 
- 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%
main
I Luk Kim 5 months ago
parent 99a998b2f1
commit e121c7ab40
2 changed files with 85 additions and 237 deletions
Show Stats Download Patch File Download Diff File

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

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

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

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

Write Preview
Loading…
Cancel
Save