You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
296 lines
9.9 KiB
Python
296 lines
9.9 KiB
Python
"""
|
|
Earnings Surprise service — yfinance-plus
|
|
|
|
Uses Ticker.earnings_dates which provides:
|
|
- EPS Estimate (analyst consensus)
|
|
- Reported EPS (actual)
|
|
- Surprise(%) (pre-calculated)
|
|
|
|
No external API key required. ~25 quarters of history per ticker.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_CHUNK = 2000
|
|
|
|
|
|
class EarningsService:
|
|
|
|
# ------------------------------------------------------------------
|
|
# Fetch from yfinance-plus
|
|
# ------------------------------------------------------------------
|
|
|
|
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)
|
|
|
|
from yfinance_plus import Ticker
|
|
t = Ticker(ticker.upper())
|
|
df = t.earnings_dates
|
|
|
|
if df is None or df.empty:
|
|
return []
|
|
|
|
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
|
|
|
|
surprise = None
|
|
if reported_eps is not None and estimated_eps is not None:
|
|
surprise = round(reported_eps - estimated_eps, 6)
|
|
|
|
rows.append({
|
|
"reported_date": reported_date,
|
|
"reported_eps": reported_eps,
|
|
"estimated_eps": estimated_eps,
|
|
"surprise": surprise,
|
|
"surprise_percentage": surprise_pct,
|
|
})
|
|
|
|
return rows
|
|
|
|
# ------------------------------------------------------------------
|
|
# Index to DB
|
|
# ------------------------------------------------------------------
|
|
|
|
async def index_earnings(
|
|
self, db: AsyncSession, ticker: str, force_refresh: bool = False
|
|
) -> int:
|
|
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:
|
|
await db.execute(
|
|
EarningsSurprise.__table__.delete().where(
|
|
EarningsSurprise.ticker == ticker
|
|
)
|
|
)
|
|
await db.flush()
|
|
|
|
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
|
|
|
|
db_rows = []
|
|
for r in raw:
|
|
# Use reported_date as fiscal_date_ending (earnings announcement date)
|
|
db_rows.append({
|
|
"ticker": ticker,
|
|
"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 db_rows:
|
|
return 0
|
|
|
|
inserted = 0
|
|
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",
|
|
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}")
|
|
return inserted
|
|
|
|
# ------------------------------------------------------------------
|
|
# Query
|
|
# ------------------------------------------------------------------
|
|
|
|
# ------------------------------------------------------------------
|
|
# Future earnings calendar
|
|
# ------------------------------------------------------------------
|
|
|
|
def _fetch_future_earnings_from_yfinance(
|
|
self, ticker: str, days_ahead: int, limit: int,
|
|
as_of_date: Optional[datetime] = None,
|
|
) -> List[Dict]:
|
|
"""Fetch earnings calendar entries from yfinance-plus with PIT support.
|
|
|
|
When as_of_date is given (historical backtesting), returns earnings that
|
|
were upcoming as of that date — regardless of whether they have since been
|
|
reported. When omitted, defaults to now (live calendar).
|
|
|
|
Filter: as_of < dt <= as_of + timedelta(days=days_ahead)
|
|
No reported_eps check — date range alone determines "upcoming".
|
|
"""
|
|
import warnings
|
|
warnings.filterwarnings("ignore", category=DeprecationWarning)
|
|
warnings.filterwarnings("ignore", category=FutureWarning)
|
|
|
|
from yfinance_plus import Ticker
|
|
t = Ticker(ticker.upper())
|
|
df = t.earnings_dates
|
|
|
|
if df is None or df.empty:
|
|
return []
|
|
|
|
as_of = as_of_date if as_of_date is not None else datetime.now(timezone.utc)
|
|
cutoff = as_of + timedelta(days=days_ahead)
|
|
|
|
rows = []
|
|
for idx, row in df.iterrows():
|
|
dt = idx.to_pydatetime()
|
|
if dt.tzinfo:
|
|
dt = dt.astimezone(timezone.utc)
|
|
else:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
|
|
# PIT filter: earnings must fall strictly after as_of and within window
|
|
if dt <= as_of or dt > cutoff:
|
|
continue
|
|
|
|
rows.append({
|
|
"earnings_date": dt,
|
|
"earnings_time": _infer_earnings_time(dt),
|
|
"estimated_eps": _safe_float(row.get("EPS Estimate")),
|
|
"reported_eps": _safe_float(row.get("Reported EPS")),
|
|
})
|
|
|
|
rows.sort(key=lambda r: r["earnings_date"])
|
|
return rows[:limit]
|
|
|
|
async def get_future_earnings(
|
|
self, ticker: str, days_ahead: int = 30, limit: int = 4,
|
|
as_of_date: Optional[datetime] = None,
|
|
) -> List[Dict]:
|
|
"""Return earnings calendar entries for a ticker (PIT-aware)."""
|
|
import asyncio
|
|
return await asyncio.to_thread(
|
|
self._fetch_future_earnings_from_yfinance,
|
|
ticker.upper(), days_ahead, limit, as_of_date,
|
|
)
|
|
|
|
async def get_earnings_surprise(
|
|
self, db: AsyncSession, ticker: str, quarters: int = 8
|
|
) -> Tuple[List[EarningsSurprise], Dict]:
|
|
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()
|
|
|
|
# Streak: consecutive beats (surprise > 0) or misses
|
|
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
|
|
|
|
return rows, {"streak": streak, "avg_surprise_pct": avg_pct}
|
|
|
|
|
|
def _safe_float(val) -> Optional[float]:
|
|
if val is None:
|
|
return None
|
|
try:
|
|
import math
|
|
f = float(val)
|
|
return None if math.isnan(f) else f
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _infer_earnings_time(dt_utc: datetime) -> str:
|
|
"""Infer pre/post market timing from earnings datetime.
|
|
|
|
yfinance often uses midnight (00:00) when actual time is unknown.
|
|
Converts to US/Eastern and checks market hours.
|
|
"""
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
dt_et = dt_utc.astimezone(ZoneInfo("America/New_York"))
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
hour, minute = dt_et.hour, dt_et.minute
|
|
|
|
# Midnight = yfinance doesn't know the time
|
|
if hour == 0 and minute == 0:
|
|
return "unknown"
|
|
|
|
if (hour, minute) < (9, 30):
|
|
return "pre_market"
|
|
elif (hour, minute) >= (16, 0):
|
|
return "post_market"
|
|
else:
|
|
return "during_market"
|