diff --git a/alembic/versions/f7a8b9c0d1e2_add_dividend_calendar.py b/alembic/versions/f7a8b9c0d1e2_add_dividend_calendar.py new file mode 100644 index 0000000..3fd7ced --- /dev/null +++ b/alembic/versions/f7a8b9c0d1e2_add_dividend_calendar.py @@ -0,0 +1,57 @@ +"""add dividend_calendar table + +Revision ID: f7a8b9c0d1e2 +Revises: e6f7a8b9c0d1 +Create Date: 2026-04-03 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "f7a8b9c0d1e2" +down_revision: Union[str, Sequence[str], None] = "e6f7a8b9c0d1" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + conn = op.get_bind() + + if not conn.dialect.has_table(conn, "dividend_calendar"): + op.create_table( + "dividend_calendar", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("ticker", sa.String(10), nullable=False), + sa.Column("ex_dividend_date", postgresql.TIMESTAMP(timezone=True), nullable=False), + sa.Column("amount", sa.Float(), nullable=False), + sa.Column("declaration_date", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.Column("record_date", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.Column("payment_date", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.Column("currency", sa.String(10), server_default="USD"), + sa.Column("dividend_type", sa.String(20), server_default="regular"), + sa.Column("frequency", sa.String(20), nullable=True), + sa.Column("as_of_date", postgresql.TIMESTAMP(timezone=True), nullable=False), + sa.Column("source", sa.String(50), nullable=False), + sa.Column("source_file_date", postgresql.TIMESTAMP(timezone=True), nullable=True), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True)), + sa.Column("updated_at", postgresql.TIMESTAMP(timezone=True)), + sa.UniqueConstraint( + "ticker", "ex_dividend_date", "as_of_date", "source", + name="uq_dividend_calendar", + ), + ) + op.create_index("ix_dividend_calendar_ticker", "dividend_calendar", ["ticker"]) + op.create_index("idx_dividend_ticker_exdate", "dividend_calendar", ["ticker", "ex_dividend_date"]) + op.create_index("idx_dividend_exdate", "dividend_calendar", ["ex_dividend_date"]) + op.create_index("idx_dividend_as_of", "dividend_calendar", ["as_of_date"]) + op.create_index( + "idx_dividend_pit_query", + "dividend_calendar", + ["ticker", "ex_dividend_date", "as_of_date"], + ) + + +def downgrade() -> None: + op.drop_table("dividend_calendar") diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 8d85e77..ff1c090 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -3,7 +3,7 @@ API v1 router """ from fastapi import APIRouter -from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention, insider, earnings, universe +from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention, insider, earnings, universe, dividends api_router = APIRouter() @@ -29,4 +29,5 @@ api_router.include_router(screener.router, prefix="/screener", tags=["screener"] api_router.include_router(attention.router, prefix="/attention", tags=["attention"]) api_router.include_router(insider.router, prefix="/insider", tags=["insider"]) api_router.include_router(earnings.router, prefix="/earnings", tags=["earnings"]) -api_router.include_router(universe.router, prefix="/universe", tags=["universe"]) \ No newline at end of file +api_router.include_router(universe.router, prefix="/universe", tags=["universe"]) +api_router.include_router(dividends.router, prefix="/dividends", tags=["dividends"]) \ No newline at end of file diff --git a/app/api/v1/endpoints/dividends.py b/app/api/v1/endpoints/dividends.py new file mode 100644 index 0000000..8a06c53 --- /dev/null +++ b/app/api/v1/endpoints/dividends.py @@ -0,0 +1,202 @@ +""" +PIT Ex-Dividend Calendar endpoints +""" + +import logging +from datetime import date, datetime, timedelta, timezone +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import Response +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import get_db +from app.schemas.dividend import ( + DividendCalendarEntry, + DividendUpcomingResponse, + DividendHistoryResponse, + DividendIngestRequest, + DividendIngestResponse, +) +from app.services.dividend_service import DividendService +from app.utils.cache import with_cache + +router = APIRouter() +logger = logging.getLogger("app.api.v1.dividends") + + +@router.get( + "/upcoming", + response_model=DividendUpcomingResponse, + summary="PIT upcoming ex-dividend calendar", + description=( + "Point-in-Time 배당락 캘린더. `as_of_date` 기준으로 당시 알려져 있었던 배당 일정 중 " + "`from_ex_date` ~ `to_ex_date` 범위의 ex-date를 반환.\n\n" + "**PIT 의미**: 같은 (ticker, ex_date)에 여러 revision이 있으면 " + "`as_of_date <= query_as_of_date` 조건 내에서 가장 최신 revision만 반환.\n\n" + "**데이터 소스**: yfinance-plus. API 키 불필요. " + "symbols 파라미터 없이 조회 시 이미 인덱싱된 종목 전체 반환.\n\n" + "**백필**: `POST /dividends/admin/ingest` 로 원하는 종목 선인덱싱 가능." + ), +) +@with_cache( + namespace="dividend:upcoming", + ttl=3600, + key_params=["as_of_date", "from_ex_date", "to_ex_date", "symbols", "limit"], +) +async def get_upcoming_dividends( + response: Response, + as_of_date: Optional[date] = Query( + None, description="PIT 기준일 (YYYY-MM-DD). 생략 시 오늘." + ), + from_ex_date: Optional[date] = Query( + None, description="Ex-date 시작 (YYYY-MM-DD). 생략 시 오늘." + ), + to_ex_date: Optional[date] = Query( + None, description="Ex-date 끝 (YYYY-MM-DD). 생략 시 오늘 + 60일." + ), + symbols: Optional[List[str]] = Query( + None, description="종목 필터 (e.g. ?symbols=AAPL&symbols=MSFT). 생략 시 전체." + ), + limit: int = Query(500, ge=1, le=5000, description="최대 반환 개수"), + force_refresh: bool = Query(False, description="캐시 무시"), + db: AsyncSession = Depends(get_db), +): + today = date.today() + as_of = as_of_date or today + from_ex = from_ex_date or today + to_ex = to_ex_date or (today + timedelta(days=60)) + + if from_ex > to_ex: + raise HTTPException(status_code=400, detail="from_ex_date must be <= to_ex_date") + + as_of_dt = datetime(as_of.year, as_of.month, as_of.day, 23, 59, 59, tzinfo=timezone.utc) + from_ex_dt = datetime(from_ex.year, from_ex.month, from_ex.day, tzinfo=timezone.utc) + to_ex_dt = datetime(to_ex.year, to_ex.month, to_ex.day, 23, 59, 59, tzinfo=timezone.utc) + + svc = DividendService() + try: + rows, total = await svc.get_upcoming_dividends( + db, + as_of_date=as_of_dt, + from_ex_date=from_ex_dt, + to_ex_date=to_ex_dt, + symbols=symbols, + limit=limit, + ) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + except Exception as e: + logger.error(f"Dividend upcoming query error: {e}") + raise HTTPException(status_code=502, detail=f"Failed to query dividends: {e}") + + entries = [DividendCalendarEntry.from_orm_obj(r) for r in rows] + + return DividendUpcomingResponse( + dividends=entries, + total_count=total, + metadata={ + "as_of_date": as_of.isoformat(), + "from_ex_date": from_ex.isoformat(), + "to_ex_date": to_ex.isoformat(), + "symbols_filter": symbols, + "results_returned": len(entries), + "pit_semantics": "DISTINCT ON (ticker, ex_date) ORDER BY as_of_date DESC", + }, + ) + + +@router.get( + "/history/{symbol}", + response_model=DividendHistoryResponse, + summary="종목별 배당 이력", + description=( + "단일 종목의 전체 배당 이력. yfinance 데이터가 없으면 자동 인덱싱.\n\n" + "각 ex-date별 최신 revision을 반환 (ex-date 내림차순).\n\n" + "`annual_yield_estimate`: 최근 12개월 배당 합산액 (주가 대비 yield는 클라이언트 계산 필요)." + ), +) +@with_cache( + namespace="dividend:history", + ttl=None, + key_params=["symbol", "limit"], +) +async def get_dividend_history( + symbol: str, + response: Response, + limit: int = Query(100, ge=1, le=1000, description="최대 반환 개수"), + force_refresh: bool = Query(False, description="캐시 무시 + yfinance 재조회"), + db: AsyncSession = Depends(get_db), +): + svc = DividendService() + + if force_refresh: + try: + await svc.index_dividends(db, symbol, force_refresh=True) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + except Exception as e: + logger.error(f"Dividend force refresh error for {symbol}: {e}") + raise HTTPException(status_code=502, detail=f"Re-fetch failed: {e}") + + try: + rows, total, annual_yield = await svc.get_dividend_history( + db, ticker=symbol, limit=limit + ) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + except Exception as e: + logger.error(f"Dividend history error for {symbol}: {e}") + raise HTTPException(status_code=502, detail=f"Failed to fetch dividend history: {e}") + + entries = [DividendCalendarEntry.from_orm_obj(r) for r in rows] + + return DividendHistoryResponse( + symbol=symbol.upper(), + dividends=entries, + total_count=total, + annual_yield_estimate=annual_yield, + metadata={ + "data_source": "yfinance", + "limit": limit, + "results_returned": len(entries), + "annual_yield_note": "TTM: ex_date가 최근 365일 이내인 배당 합산액", + }, + ) + + +@router.post( + "/admin/ingest", + response_model=DividendIngestResponse, + summary="배당 데이터 벌크 인제스트", + description=( + "yfinance에서 지정 종목 배당 이력을 가져와 DB에 저장.\n\n" + "**예시**:\n" + "- `{\"symbols\": [\"AAPL\", \"MSFT\", \"JNJ\"]}` — 신규 종목 인덱싱\n" + "- `{\"symbols\": [...], \"force_refresh\": true}` — 기존 데이터 재인제스트\n\n" + "종목당 약 25년치 이력. 100종목 기준 5~10분 소요 (yfinance rate limit).\n\n" + "이미 인덱싱된 종목은 `force_refresh: false`일 때 건너뜀 (멱등성)." + ), +) +async def ingest_dividends( + body: DividendIngestRequest, + db: AsyncSession = Depends(get_db), +): + svc = DividendService() + try: + result = await svc.bulk_ingest( + db, + symbols=[s.upper().strip() for s in body.symbols], + force_refresh=body.force_refresh, + ) + except Exception as e: + logger.error(f"Dividend bulk ingest error: {e}") + raise HTTPException(status_code=502, detail=f"Bulk ingest failed: {e}") + + return DividendIngestResponse( + symbols_processed=result["symbols_processed"], + total_records_upserted=result["total_records_upserted"], + failed_symbols=result["failed_symbols"], + status="completed", + metadata={"force_refresh": body.force_refresh}, + ) diff --git a/app/api/v1/endpoints/earnings.py b/app/api/v1/endpoints/earnings.py index af15016..8e241fa 100644 --- a/app/api/v1/endpoints/earnings.py +++ b/app/api/v1/endpoints/earnings.py @@ -1,15 +1,25 @@ """ -Earnings Surprise endpoints — SEC EDGAR XBRL EPS data +Earnings endpoints — surprise history + future calendar """ +import asyncio import logging +from datetime import date, datetime, timezone +from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import Response from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db -from app.schemas.earnings import EarningsSurpriseEntry, EarningsSurpriseResponse +from app.schemas.earnings import ( + EarningsSurpriseEntry, + EarningsSurpriseResponse, + EarningsCalendarEntry, + EarningsCalendarResponse, + BulkEarningsCalendarRequest, + BulkEarningsCalendarResponse, +) from app.services.earnings_service import EarningsService from app.utils.cache import with_cache @@ -17,6 +27,151 @@ router = APIRouter() logger = logging.getLogger("app.api.v1.earnings") +@router.get( + "/calendar/{symbol}", + response_model=EarningsCalendarResponse, + summary="Get upcoming earnings dates for a symbol", + description=( + "Upcoming earnings announcement dates with EPS estimates.\n\n" + "**데이터 소스**: yfinance-plus (`Ticker.earnings_dates`). API 키 불필요.\n" + "**earnings_time**: `pre_market` / `post_market` / `during_market` / `unknown`.\n\n" + "**PIT (Point-in-Time) backtesting**: `as_of_date`를 지정하면 해당 날짜 기준 " + "upcoming earnings를 반환합니다. 이미 보고된 earnings도 당시엔 예정이었으므로 " + "`reported_eps`가 채워진 상태로 반환됩니다.\n\n" + "**Note**: Revenue estimates are not available from this source." + ), +) +@with_cache(namespace="earnings:calendar", ttl=3600, key_params=["symbol", "days_ahead", "limit", "as_of_date"]) +async def get_earnings_calendar( + symbol: str, + response: Response, + days_ahead: int = Query(30, ge=1, le=365, description="Days to look ahead from as_of_date (or today)"), + limit: int = Query(4, ge=1, le=20, description="Max earnings dates to return"), + as_of_date: Optional[date] = Query(None, description="PIT date for backtesting (YYYY-MM-DD). Defaults to today."), + force_refresh: bool = Query(False, description="Bypass cache and re-fetch from yfinance"), +): + as_of_dt = ( + datetime(as_of_date.year, as_of_date.month, as_of_date.day, tzinfo=timezone.utc) + if as_of_date else None + ) + + svc = EarningsService() + try: + rows = await svc.get_future_earnings( + ticker=symbol, days_ahead=days_ahead, limit=limit, as_of_date=as_of_dt + ) + except Exception as e: + logger.error(f"Earnings calendar error for {symbol}: {e}") + raise HTTPException(status_code=502, detail=f"Failed to fetch earnings calendar: {e}") + + now = datetime.now(timezone.utc) + entries = [ + EarningsCalendarEntry( + symbol=symbol.upper(), + earnings_date=r["earnings_date"], + earnings_time=r["earnings_time"], + estimated_eps=r["estimated_eps"], + reported_eps=r["reported_eps"], + source="yfinance", + fetched_at=now, + ) + for r in rows + ] + + return EarningsCalendarResponse( + symbol=symbol.upper(), + upcoming_earnings=entries, + metadata={ + "data_source": "yfinance", + "as_of_date": as_of_date.isoformat() if as_of_date else "today", + "days_ahead": days_ahead, + "limit": limit, + "results_returned": len(entries), + "note": "Revenue estimates not available from this source.", + }, + ) + + +@router.post( + "/calendar/bulk", + response_model=BulkEarningsCalendarResponse, + summary="Bulk future earnings calendar", + description=( + "Fetch upcoming earnings dates for multiple symbols (max 50).\n\n" + "Returns a flat list of calendar entries sorted by `earnings_date` ascending.\n" + "Useful for checking upcoming earnings of sector peers or candidates." + ), +) +async def get_bulk_earnings_calendar( + request: BulkEarningsCalendarRequest, + response: Response, +): + as_of_dt = ( + datetime(request.as_of_date.year, request.as_of_date.month, request.as_of_date.day, tzinfo=timezone.utc) + if request.as_of_date else None + ) + + svc = EarningsService() + now = datetime.now(timezone.utc) + semaphore = asyncio.Semaphore(10) + + async def process_one(sym: str) -> List[EarningsCalendarEntry]: + async with semaphore: + try: + rows = await svc.get_future_earnings( + ticker=sym, + days_ahead=request.days_ahead, + limit=request.limit, + as_of_date=as_of_dt, + ) + return [ + EarningsCalendarEntry( + symbol=sym.upper(), + earnings_date=r["earnings_date"], + earnings_time=r["earnings_time"], + estimated_eps=r["estimated_eps"], + reported_eps=r["reported_eps"], + source="yfinance", + fetched_at=now, + ) + for r in rows + ] + except Exception as e: + logger.warning(f"Earnings calendar bulk: {sym} failed: {e}") + return [] + + tasks = [process_one(sym) for sym in request.symbols] + try: + results = await asyncio.wait_for( + asyncio.gather(*tasks, return_exceptions=True), + timeout=300, + ) + except asyncio.TimeoutError: + raise HTTPException(status_code=504, detail="Bulk earnings calendar request timed out.") + + entries: List[EarningsCalendarEntry] = [] + failed = 0 + for r in results: + if isinstance(r, Exception): + failed += 1 + elif isinstance(r, list): + entries.extend(r) + + entries.sort(key=lambda e: e.earnings_date) + + return BulkEarningsCalendarResponse( + entries=entries, + metadata={ + "symbols_requested": len(request.symbols), + "symbols_failed": failed, + "total_entries": len(entries), + "as_of_date": request.as_of_date.isoformat() if request.as_of_date else "today", + "days_ahead": request.days_ahead, + "per_symbol_limit": request.limit, + }, + ) + + @router.get( "/surprise/{symbol}", response_model=EarningsSurpriseResponse, diff --git a/app/models/dividend_calendar.py b/app/models/dividend_calendar.py new file mode 100644 index 0000000..5376050 --- /dev/null +++ b/app/models/dividend_calendar.py @@ -0,0 +1,53 @@ +""" +PIT Ex-Dividend Calendar model — revisioned dividend announcements + +Stores multiple revisions of the same (ticker, ex_dividend_date) with different +as_of_date values, enabling Point-in-Time backtesting without lookahead bias. +""" + +from datetime import datetime, timezone +import uuid + +from sqlalchemy import Column, String, Float, Index, UniqueConstraint +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP + +from app.core.database import Base + + +class DividendCalendar(Base): + __tablename__ = "dividend_calendar" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + ticker = Column(String(10), nullable=False, index=True) + ex_dividend_date = Column(TIMESTAMP(timezone=True), nullable=False) + amount = Column(Float, nullable=False) + declaration_date = Column(TIMESTAMP(timezone=True), nullable=True) + record_date = Column(TIMESTAMP(timezone=True), nullable=True) + payment_date = Column(TIMESTAMP(timezone=True), nullable=True) + currency = Column(String(10), default="USD") + dividend_type = Column(String(20), default="regular") # regular | special + frequency = Column(String(20), nullable=True) # quarterly | semi-annual | annual | monthly | irregular + as_of_date = Column(TIMESTAMP(timezone=True), nullable=False) # PIT key + source = Column(String(50), nullable=False) # "yfinance", "finra_orf", etc. + source_file_date = Column(TIMESTAMP(timezone=True), nullable=True) + + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + updated_at = Column( + TIMESTAMP(timezone=True), + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) + + __table_args__ = ( + UniqueConstraint( + "ticker", "ex_dividend_date", "as_of_date", "source", + name="uq_dividend_calendar", + ), + Index("idx_dividend_ticker_exdate", "ticker", "ex_dividend_date"), + Index("idx_dividend_exdate", "ex_dividend_date"), + Index("idx_dividend_as_of", "as_of_date"), + # Composite index directly supports DISTINCT ON (ticker, ex_dividend_date) ORDER BY as_of_date DESC + Index("idx_dividend_pit_query", "ticker", "ex_dividend_date", "as_of_date"), + ) diff --git a/app/schemas/dividend.py b/app/schemas/dividend.py new file mode 100644 index 0000000..4ad67d1 --- /dev/null +++ b/app/schemas/dividend.py @@ -0,0 +1,76 @@ +""" +PIT Dividend Calendar schemas +""" + +from datetime import date, datetime +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class DividendCalendarEntry(BaseModel): + model_config = ConfigDict(from_attributes=True) + + ticker: str + ex_dividend_date: date + amount: float + declaration_date: Optional[date] = None + record_date: Optional[date] = None + payment_date: Optional[date] = None + currency: str = "USD" + dividend_type: str = "regular" + frequency: Optional[str] = None + as_of_date: date + source: str + + @classmethod + def from_orm_obj(cls, obj) -> "DividendCalendarEntry": + def _to_date(val): + if val is None: + return None + return val.date() if hasattr(val, "date") else val + + return cls( + ticker=obj.ticker, + ex_dividend_date=_to_date(obj.ex_dividend_date), + amount=obj.amount, + declaration_date=_to_date(obj.declaration_date), + record_date=_to_date(obj.record_date), + payment_date=_to_date(obj.payment_date), + currency=obj.currency or "USD", + dividend_type=obj.dividend_type or "regular", + frequency=obj.frequency, + as_of_date=_to_date(obj.as_of_date), + source=obj.source, + ) + + +class DividendUpcomingResponse(BaseModel): + """Response for PIT upcoming dividends query.""" + dividends: List[DividendCalendarEntry] + total_count: int + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class DividendHistoryResponse(BaseModel): + """Response for single-symbol dividend history.""" + symbol: str + dividends: List[DividendCalendarEntry] + total_count: int + annual_yield_estimate: Optional[float] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class DividendIngestRequest(BaseModel): + """Request body for bulk backfill ingest.""" + symbols: List[str] = Field(..., min_length=1, max_length=200) + force_refresh: bool = Field(False, description="Re-ingest even if data exists") + + +class DividendIngestResponse(BaseModel): + """Response for admin ingest endpoint.""" + symbols_processed: int + total_records_upserted: int + failed_symbols: List[str] = Field(default_factory=list) + status: str + metadata: Dict[str, Any] = Field(default_factory=dict) diff --git a/app/schemas/earnings.py b/app/schemas/earnings.py index 7b9ea57..977c64f 100644 --- a/app/schemas/earnings.py +++ b/app/schemas/earnings.py @@ -2,7 +2,7 @@ Earnings Surprise schemas """ -from datetime import date +from datetime import date, datetime from typing import Any, Dict, List, Optional from pydantic import BaseModel, ConfigDict, Field @@ -47,3 +47,33 @@ class EarningsSurpriseResponse(BaseModel): streak: int = 0 # positive = consecutive beats, negative = consecutive misses avg_surprise_pct: Optional[float] = None metadata: Dict[str, Any] = Field(default_factory=dict) + + +class EarningsCalendarEntry(BaseModel): + symbol: str + earnings_date: datetime + earnings_time: str # "pre_market" | "post_market" | "during_market" | "unknown" + estimated_eps: Optional[float] = None + reported_eps: Optional[float] = None # non-null when as_of_date is historical + source: str = "yfinance" + fetched_at: datetime + + +class EarningsCalendarResponse(BaseModel): + symbol: str + upcoming_earnings: List[EarningsCalendarEntry] + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class BulkEarningsCalendarRequest(BaseModel): + symbols: List[str] = Field(..., min_length=1, max_length=50) + days_ahead: int = Field(30, ge=1, le=365) + limit: int = Field(4, ge=1, le=20) + as_of_date: Optional[date] = Field( + None, description="PIT date for backtesting (YYYY-MM-DD). Defaults to today." + ) + + +class BulkEarningsCalendarResponse(BaseModel): + entries: List[EarningsCalendarEntry] + metadata: Dict[str, Any] = Field(default_factory=dict) diff --git a/app/services/dividend_service.py b/app/services/dividend_service.py new file mode 100644 index 0000000..96c1ecf --- /dev/null +++ b/app/services/dividend_service.py @@ -0,0 +1,538 @@ +""" +PIT Dividend Calendar service — yfinance-plus + +PIT rules: + - Historical dividends: as_of_date = ex_dividend_date - 30 days + (approximation: dividends are typically declared 2-4 weeks before ex-date) + - Upcoming/future dividends: as_of_date = ingestion timestamp + (captures when we first observed the announcement) + +Special dividend detection: + - Heuristic: amount >= 2.5 × median of most recent 12 payments + - Applied per-ticker at fetch time; stored in dividend_type column +""" + +import logging +import math +from collections import Counter +from datetime import date, datetime, timedelta, timezone +from typing import Dict, List, Optional, Tuple + +from sqlalchemy import select, and_, func, desc, delete +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.dialects.postgresql import insert as pg_insert + +from app.models.dividend_calendar import DividendCalendar + +logger = logging.getLogger(__name__) + +_CHUNK = 2000 # floor(32767 / 16 columns) = 2047, rounded down + + +class DividendService: + + # ------------------------------------------------------------------ + # yfinance data extraction (synchronous — called via asyncio.to_thread) + # ------------------------------------------------------------------ + + def _fetch_dividends_from_yfinance(self, ticker: str) -> List[Dict]: + """Fetch dividend data from yfinance-plus. Returns list of row dicts.""" + import warnings + warnings.filterwarnings("ignore", category=DeprecationWarning) + warnings.filterwarnings("ignore", category=FutureWarning) + + try: + from yfinance_plus import Ticker + except ImportError: + import yfinance as yf + Ticker = yf.Ticker + + ticker = ticker.upper() + t = Ticker(ticker) + now_utc = datetime.now(timezone.utc) + rows: List[Dict] = [] + + # ---- 1. Historical dividends ---- + try: + divs = t.dividends # pandas Series: DatetimeIndex -> float + if divs is not None and not divs.empty: + for dt_idx, amount in divs.items(): + amount_f = _safe_float(amount) + if amount_f is None or amount_f <= 0: + continue + + ex_date = _normalize_ex_date(dt_idx) + if ex_date is None: + continue + + # PIT: dividends are typically declared 2-4 weeks before ex-date. + # Without declaration_date from yfinance, approximate with -30 days. + as_of = ex_date - timedelta(days=30) + + rows.append({ + "ticker": ticker, + "ex_dividend_date": ex_date, + "amount": round(amount_f, 6), + "declaration_date": None, + "record_date": None, + "payment_date": None, + "currency": "USD", + "dividend_type": "regular", + "frequency": None, + "as_of_date": as_of, + "source": "yfinance", + "source_file_date": None, + }) + except Exception as e: + logger.warning(f"Dividend: t.dividends failed for {ticker}: {e}") + + # ---- 2. Upcoming dividend from calendar + info (fetched once) ---- + # Fetch info once and reuse in step 3 to avoid duplicate API calls. + info: Dict = {} + try: + info = t.info or {} + except Exception: + pass + + try: + cal = t.calendar + if cal is not None and isinstance(cal, dict): + ex_date_val = cal.get("Ex-Dividend Date") + div_date_val = cal.get("Dividend Date") + + if ex_date_val is not None: + ex_date = _normalize_ex_date(ex_date_val) + if ex_date and ex_date > now_utc: + last_div = _safe_float(info.get("lastDividendValue")) + div_rate = _safe_float(info.get("dividendRate")) + amount = last_div or (div_rate / 4 if div_rate else None) + + if amount and amount > 0: + payment_date = _to_utc_datetime(div_date_val) + rows.append({ + "ticker": ticker, + "ex_dividend_date": ex_date, + "amount": round(amount, 6), + "declaration_date": None, + "record_date": None, + "payment_date": payment_date, + "currency": "USD", + "dividend_type": "regular", + "frequency": None, + "as_of_date": now_utc, # PIT: known NOW (ingestion time) + "source": "yfinance", + "source_file_date": None, + }) + except Exception as e: + logger.warning(f"Dividend: t.calendar failed for {ticker}: {e}") + + # ---- 3. Enrich with frequency (info already fetched above) ---- + try: + freq_str = _infer_frequency(info) or _infer_frequency_from_history(rows) + if freq_str: + for row in rows: + row["frequency"] = freq_str + except Exception: + pass + + # ---- 4. Flag special dividends ---- + _flag_special_dividends(rows) + + return rows + + # ------------------------------------------------------------------ + # Ingest a single ticker to DB + # ------------------------------------------------------------------ + + async def index_dividends( + self, + db: AsyncSession, + ticker: str, + force_refresh: bool = False, + ) -> int: + """Fetch from yfinance and upsert to DB. Returns number of rows upserted. + + When force_refresh=True: fetch first, then delete+insert in one transaction + to avoid data loss if the fetch fails. + """ + ticker = ticker.upper() + + if not force_refresh: + count_q = await db.execute( + select(func.count(DividendCalendar.id)).where( + DividendCalendar.ticker == ticker + ) + ) + if (count_q.scalar() or 0) > 0: + return 0 + + # Fetch from yfinance BEFORE deleting existing rows. + # This prevents data loss if the fetch fails. + import asyncio + try: + raw = await asyncio.to_thread(self._fetch_dividends_from_yfinance, ticker) + except Exception as e: + logger.error(f"Dividend: yfinance fetch failed for {ticker}: {e}") + raise ValueError(f"Could not fetch dividend data for {ticker}: {e}") + + if force_refresh: + # Safe to delete now that we have fresh data + await db.execute( + delete(DividendCalendar).where(DividendCalendar.ticker == ticker) + ) + + if not raw: + if force_refresh: + await db.commit() + logger.info(f"Dividend: no data returned by yfinance for {ticker}") + return 0 + + inserted = 0 + for i in range(0, len(raw), _CHUNK): + chunk = raw[i:i + _CHUNK] + stmt = pg_insert(DividendCalendar).values(chunk) + stmt = stmt.on_conflict_do_update( + constraint="uq_dividend_calendar", + set_={ + "amount": stmt.excluded.amount, + "declaration_date": stmt.excluded.declaration_date, + "record_date": stmt.excluded.record_date, + "payment_date": stmt.excluded.payment_date, + "currency": stmt.excluded.currency, + "dividend_type": stmt.excluded.dividend_type, + "frequency": stmt.excluded.frequency, + "source_file_date": stmt.excluded.source_file_date, + "updated_at": func.now(), + }, + ) + result = await db.execute(stmt) + inserted += result.rowcount + + await db.commit() + logger.info(f"Dividend: upserted {inserted} records for {ticker}") + return inserted + + # ------------------------------------------------------------------ + # Bulk ingest (admin backfill) + # ------------------------------------------------------------------ + + async def bulk_ingest( + self, + db: AsyncSession, + symbols: List[str], + force_refresh: bool = False, + ) -> Dict: + """Ingest dividends for multiple symbols sequentially. Returns summary.""" + total_upserted = 0 + failed: List[str] = [] + + for sym in symbols: + try: + count = await self.index_dividends(db, sym, force_refresh=force_refresh) + total_upserted += count + except Exception as e: + logger.error(f"Dividend: bulk ingest failed for {sym}: {e}") + failed.append(sym) + + return { + "symbols_processed": len(symbols), + "total_records_upserted": total_upserted, + "failed_symbols": failed, + } + + # ------------------------------------------------------------------ + # PIT query — upcoming dividends + # ------------------------------------------------------------------ + + async def get_upcoming_dividends( + self, + db: AsyncSession, + as_of_date: datetime, + from_ex_date: datetime, + to_ex_date: datetime, + symbols: Optional[List[str]] = None, + limit: int = 500, + ) -> Tuple[List[DividendCalendar], int]: + """ + PIT upcoming dividends query. + + Uses DISTINCT ON (ticker, ex_dividend_date) + ORDER BY as_of_date DESC + to return the latest-known revision for each dividend event as of as_of_date. + + When specific symbols are requested but not yet in the DB, auto-indexes + them from yfinance (same behaviour as get_dividend_history). + + SQL equivalent: + SELECT DISTINCT ON (ticker, ex_dividend_date) * + FROM dividend_calendar + WHERE as_of_date <= :as_of_date + AND ex_dividend_date BETWEEN :from_ex_date AND :to_ex_date + ORDER BY ticker, ex_dividend_date, as_of_date DESC + """ + # Auto-index any requested symbols not yet in the DB (deduped) + if symbols: + upper_syms = list(dict.fromkeys(s.upper() for s in symbols)) # dedup, preserve order + existing_q = await db.execute( + select(DividendCalendar.ticker.distinct()).where( + DividendCalendar.ticker.in_(upper_syms) + ) + ) + existing = {r for r in existing_q.scalars().all()} + missing = [s for s in upper_syms if s not in existing] + for sym in missing: + try: + await self.index_dividends(db, sym) + except Exception as e: + logger.warning(f"Dividend: auto-index failed for {sym}: {e}") + + conditions = [ + DividendCalendar.as_of_date <= as_of_date, + DividendCalendar.ex_dividend_date >= from_ex_date, + DividendCalendar.ex_dividend_date <= to_ex_date, + ] + if symbols: + conditions.append(DividendCalendar.ticker.in_([s.upper() for s in symbols])) + + # DISTINCT ON via SQLAlchemy .distinct(col1, col2) — PostgreSQL only + pit_stmt = ( + select(DividendCalendar) + .where(and_(*conditions)) + .order_by( + DividendCalendar.ticker, + DividendCalendar.ex_dividend_date, + desc(DividendCalendar.as_of_date), + ) + .distinct(DividendCalendar.ticker, DividendCalendar.ex_dividend_date) + ) + + # Total count via subquery + count_stmt = select(func.count()).select_from(pit_stmt.subquery()) + total = (await db.execute(count_stmt)).scalar() or 0 + + # Fetch with limit + result = await db.execute(pit_stmt.limit(limit)) + rows = result.scalars().all() + + return rows, total + + # ------------------------------------------------------------------ + # History query (per symbol) + # ------------------------------------------------------------------ + + async def get_dividend_history( + self, + db: AsyncSession, + ticker: str, + limit: int = 100, + ) -> Tuple[List[DividendCalendar], int, Optional[float]]: + """ + Dividend history for a single symbol. + + Returns latest-known revision per ex_date (desc), total count, + and trailing 12-month dividend sum for yield estimation. + Auto-indexes from yfinance if no data exists. + """ + ticker = ticker.upper() + + count_q = await db.execute( + select(func.count(DividendCalendar.id)).where( + DividendCalendar.ticker == ticker + ) + ) + if (count_q.scalar() or 0) == 0: + await self.index_dividends(db, ticker) + + now_utc = datetime.now(timezone.utc) + + # PIT query: latest revision per ex_date as of now + pit_stmt = ( + select(DividendCalendar) + .where( + and_( + DividendCalendar.ticker == ticker, + DividendCalendar.as_of_date <= now_utc, + ) + ) + .order_by( + DividendCalendar.ticker, + DividendCalendar.ex_dividend_date, + desc(DividendCalendar.as_of_date), + ) + .distinct(DividendCalendar.ticker, DividendCalendar.ex_dividend_date) + ) + + result = await db.execute(pit_stmt) + all_rows = result.scalars().all() + + # Sort by ex_date desc for the response + all_rows_sorted = sorted(all_rows, key=lambda r: r.ex_dividend_date, reverse=True) + total = len(all_rows_sorted) + rows = all_rows_sorted[:limit] + + # TTM: sum of amounts with ex_date in last 365 days + ttm_cutoff = now_utc - timedelta(days=365) + ttm_sum = sum( + r.amount for r in all_rows_sorted + if r.ex_dividend_date >= ttm_cutoff + ) + annual_yield = round(ttm_sum, 4) if ttm_sum > 0 else None + + return rows, total, annual_yield + + +# ------------------------------------------------------------------ +# Utility helpers +# ------------------------------------------------------------------ + +def _safe_float(val) -> Optional[float]: + """Convert to float, returning None for None/NaN/±Inf.""" + if val is None: + return None + try: + f = float(val) + if math.isnan(f) or math.isinf(f): + return None + return f + except (ValueError, TypeError): + return None + + +def _to_utc_datetime(val) -> Optional[datetime]: + """Convert date/datetime/Timestamp/str to UTC-aware datetime.""" + if val is None: + return None + if isinstance(val, datetime): + return val.astimezone(timezone.utc) if val.tzinfo else val.replace(tzinfo=timezone.utc) + if hasattr(val, "to_pydatetime"): + return _to_utc_datetime(val.to_pydatetime()) + if isinstance(val, date) and not isinstance(val, datetime): + return datetime(val.year, val.month, val.day, tzinfo=timezone.utc) + if isinstance(val, str): + try: + dt = datetime.fromisoformat(val) + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + except ValueError: + return None + return None + + +def _normalize_ex_date(val) -> Optional[datetime]: + """Convert an ex-dividend date to UTC midnight. + + yfinance returns dates in US/Eastern time (e.g. 2025-02-10 05:00:00+00 which + is 2025-02-10 00:00 EST). We normalize to UTC midnight of the Eastern calendar + date to prevent timezone artifacts from creating duplicate rows. + """ + dt = _to_utc_datetime(val) + if dt is None: + return None + try: + import zoneinfo + eastern = zoneinfo.ZoneInfo("America/New_York") + local_dt = dt.astimezone(eastern) + return datetime(local_dt.year, local_dt.month, local_dt.day, tzinfo=timezone.utc) + except Exception: + return datetime(dt.year, dt.month, dt.day, tzinfo=timezone.utc) + + +def _infer_frequency(info: dict) -> Optional[str]: + """Infer dividend frequency from yfinance info dict.""" + freq_hint = info.get("dividendFrequency") + if freq_hint: + fh = str(freq_hint).lower() + mapping = {"1": "annual", "2": "semi-annual", "4": "quarterly", "12": "monthly"} + return mapping.get(fh, fh) + + rate = _safe_float(info.get("dividendRate")) + last_val = _safe_float(info.get("lastDividendValue")) + if rate and last_val and last_val > 0: + ratio = rate / last_val + if 3.5 <= ratio <= 4.5: + return "quarterly" + if 1.8 <= ratio <= 2.2: + return "semi-annual" + if 0.8 <= ratio <= 1.2: + return "annual" + if 11.0 <= ratio <= 13.0: + return "monthly" + return None + + +def _infer_frequency_from_history(rows: List[Dict]) -> Optional[str]: + """Infer payment frequency by counting payments per year in recent history. + + Uses the most recent 3 years of data to avoid frequency changes in older + history skewing the estimate. Only considers regular dividends (called + before _flag_special_dividends, so all rows are "regular" at this point). + """ + if len(rows) < 3: + return None + + # Sort by ex_date, use last 3 years + dated = sorted( + [r for r in rows if r["amount"] > 0], + key=lambda r: r["ex_dividend_date"], + ) + if not dated: + return None + + cutoff = dated[-1]["ex_dividend_date"] - timedelta(days=3 * 365) + recent = [r for r in dated if r["ex_dividend_date"] >= cutoff] + if len(recent) < 3: + return None + + # Count payments per calendar year + years: Counter = Counter() + for r in recent: + dt = r["ex_dividend_date"] + year = dt.year if hasattr(dt, "year") else dt + years[year] += 1 + + if not years: + return None + + # Median payments-per-year to reduce skew from partial years at boundaries + counts = sorted(years.values()) + median_count = counts[len(counts) // 2] + + if median_count >= 10: + return "monthly" + if median_count >= 3: + return "quarterly" + if median_count >= 2: + return "semi-annual" + if median_count == 1: + return "annual" + return None + + +def _flag_special_dividends(rows: List[Dict]) -> None: + """Heuristic: flag dividends that are clear outliers relative to recent history. + + Uses the median of the most recent 12 payments as the baseline, so long-term + dividend growers (MSFT, JNJ) are not incorrectly flagged. Requires at least + 4 recent payments to avoid false positives. + """ + if len(rows) < 4: + return + + dated = sorted( + [r for r in rows if r["amount"] > 0], + key=lambda r: r["ex_dividend_date"], + ) + if not dated: + return + + baseline_window = dated[-12:] + if len(baseline_window) < 4: + return + + amounts = sorted(r["amount"] for r in baseline_window) + median = amounts[len(amounts) // 2] + if median <= 0: + return + + threshold = median * 2.5 + for row in rows: + if row["amount"] >= threshold: + row["dividend_type"] = "special" diff --git a/app/services/earnings_service.py b/app/services/earnings_service.py index d4ef403..7d3ff7e 100644 --- a/app/services/earnings_service.py +++ b/app/services/earnings_service.py @@ -10,7 +10,7 @@ No external API key required. ~25 quarters of history per ticker. """ import logging -from datetime import datetime, timezone +from datetime import datetime, timezone, timedelta from typing import Dict, List, Optional, Tuple from sqlalchemy import select, desc, func @@ -151,6 +151,70 @@ class EarningsService: # 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]: @@ -203,3 +267,29 @@ def _safe_float(val) -> Optional[float]: 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"