signal-platform/app/services/fundamentals_candidate_service.py

"""Local SEC/Dolt candidate values for the fundamentals compat cache.

This is the read path behind the ``fundamental_data`` refresh. It never contacts
SEC or Dolt: every input comes from PostgreSQL, so price- and earnings-driven
values can still refresh when an upstream import is unchanged or unavailable.
"""

from __future__ import annotations

import math
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Any
from zoneinfo import ZoneInfo

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.earnings_event import EarningsEvent
from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.ohlcv import OHLCVRecord
from app.models.ticker import Ticker
from app.services import fundamentals_derivation as deriv
from app.services import ticker_service


@dataclass(frozen=True)
class CandidateFundamentals:
    ticker_id: int
    symbol: str
    cik: str | None
    pe_ratio: float | None
    revenue_growth: float | None
    earnings_surprise: float | None
    market_cap: float | None
    next_earnings_date: date | None
    price_date: date | None
    unavailable_fields: dict[str, str] = field(default_factory=dict)


async def build_candidates(
    db: AsyncSession,
    *,
    today: date | None = None,
) -> list[CandidateFundamentals]:
    """Derive current cache candidates using only already-stored data."""
    today = today or datetime.now(ZoneInfo("America/New_York")).date()
    tickers = list(
        (
            await db.execute(
                ticker_service.active_only(select(Ticker).order_by(Ticker.symbol))
            )
        ).scalars()
    )
    if not tickers:
        return []

    ticker_ids = [ticker.id for ticker in tickers]
    ciks = sorted({ticker.cik for ticker in tickers if ticker.cik})
    derived_by_cik = await _derived_by_cik(db, ciks)
    closes_by_ticker = await _latest_closes(db, ticker_ids)
    surprise_by_ticker, next_by_ticker = await _earnings_values(
        db, ticker_ids, today
    )

    out: list[CandidateFundamentals] = []
    for ticker in tickers:
        derived = derived_by_cik.get(ticker.cik) if ticker.cik else None
        close = closes_by_ticker.get(ticker.id)
        price = close[0] if close is not None else None
        price_date = close[1] if close is not None else None
        growth_series = (
            derived.metrics.get("revenue_growth_yoy")
            if derived is not None
            else None
        )

        pe_ratio = (
            _pe(price, derived.ttm_diluted_eps)
            if derived is not None
            else None
        )
        revenue_growth = (
            float(growth_series.value)
            if growth_series is not None and _finite(growth_series.value)
            else None
        )
        earnings_surprise = surprise_by_ticker.get(ticker.id)
        market_cap = (
            _market_cap(price, derived.shares_outstanding)
            if derived is not None
            else None
        )
        next_earnings_date = next_by_ticker.get(ticker.id)

        out.append(
            CandidateFundamentals(
                ticker_id=ticker.id,
                symbol=ticker.symbol,
                cik=ticker.cik,
                pe_ratio=pe_ratio,
                revenue_growth=revenue_growth,
                earnings_surprise=earnings_surprise,
                market_cap=market_cap,
                next_earnings_date=next_earnings_date,
                price_date=price_date,
                unavailable_fields=_availability_metadata(
                    derived=derived,
                    price=price,
                    pe_ratio=pe_ratio,
                    revenue_growth=revenue_growth,
                    earnings_surprise=earnings_surprise,
                    market_cap=market_cap,
                    next_earnings_date=next_earnings_date,
                ),
            )
        )
    return out


async def _derived_by_cik(
    db: AsyncSession, ciks: list[str]
) -> dict[str, deriv.DerivedFundamentals]:
    if not ciks:
        return {}
    grouped: dict[str, list[FundamentalSnapshot]] = defaultdict(list)
    rows = (
        await db.execute(
            select(FundamentalSnapshot).where(FundamentalSnapshot.cik.in_(ciks))
        )
    ).scalars()
    for row in rows:
        grouped[row.cik].append(row)
    return {cik: deriv.derive(grouped.get(cik, [])) for cik in ciks}


async def _latest_closes(
    db: AsyncSession, ticker_ids: list[int]
) -> dict[int, tuple[float, date]]:
    latest = (
        select(
            OHLCVRecord.ticker_id,
            func.max(OHLCVRecord.date).label("max_date"),
        )
        .where(OHLCVRecord.ticker_id.in_(ticker_ids))
        .group_by(OHLCVRecord.ticker_id)
        .subquery()
    )
    rows = (
        await db.execute(
            select(
                OHLCVRecord.ticker_id,
                OHLCVRecord.close,
                OHLCVRecord.date,
            ).join(
                latest,
                (OHLCVRecord.ticker_id == latest.c.ticker_id)
                & (OHLCVRecord.date == latest.c.max_date),
            )
        )
    ).all()
    return {
        ticker_id: (float(close), close_date)
        for ticker_id, close, close_date in rows
        if _finite(close)
    }


async def _earnings_values(
    db: AsyncSession,
    ticker_ids: list[int],
    today: date,
) -> tuple[dict[int, float], dict[int, date]]:
    rows = (
        await db.execute(
            select(EarningsEvent)
            .where(EarningsEvent.ticker_id.in_(ticker_ids))
            .order_by(EarningsEvent.ticker_id, EarningsEvent.announce_date.desc())
        )
    ).scalars()
    surprises: dict[int, float] = {}
    upcoming: dict[int, date] = {}
    for row in rows:
        if row.announce_date >= today:
            current = upcoming.get(row.ticker_id)
            if current is None or row.announce_date < current:
                upcoming[row.ticker_id] = row.announce_date
            continue
        if row.ticker_id in surprises:
            continue
        surprise = _surprise(row.eps_estimate, row.eps_actual)
        if surprise is not None:
            surprises[row.ticker_id] = surprise
    return surprises, upcoming


def _availability_metadata(
    *,
    derived: deriv.DerivedFundamentals | None,
    price: float | None,
    pe_ratio: float | None,
    revenue_growth: float | None,
    earnings_surprise: float | None,
    market_cap: float | None,
    next_earnings_date: date | None,
) -> dict[str, str]:
    metadata: dict[str, str] = {}

    if pe_ratio is not None:
        metadata["source_pe_ratio"] = "sec_facts+ohlcv_records"
    elif derived is None or derived.latest_period_end is None:
        metadata["pe_ratio"] = "no SEC fundamental snapshots"
    elif not _finite(price) or price <= 0:
        metadata["pe_ratio"] = "no usable PostgreSQL close"
    elif derived.ttm_diluted_eps_caveat:
        metadata["pe_ratio"] = derived.ttm_diluted_eps_caveat
    else:
        metadata["pe_ratio"] = "no positive SEC-derived TTM diluted EPS"

    if revenue_growth is not None:
        metadata["source_revenue_growth"] = "sec_facts"
    else:
        metadata["revenue_growth"] = "SEC-derived TTM revenue growth unavailable"

    if earnings_surprise is not None:
        metadata["source_earnings_surprise"] = "dolt_earnings"
    else:
        metadata["earnings_surprise"] = (
            "no completed earnings event with actual and nonzero estimate"
        )

    if market_cap is not None:
        metadata["source_market_cap"] = "sec_facts+ohlcv_records"
        if derived is not None and derived.shares_outstanding_estimated:
            metadata["market_cap_estimated"] = (
                "shares use the SEC weighted-average diluted fallback"
            )
    elif derived is None or derived.latest_period_end is None:
        metadata["market_cap"] = "no SEC fundamental snapshots"
    elif not _finite(price) or price <= 0:
        metadata["market_cap"] = "no usable PostgreSQL close"
    else:
        metadata["market_cap"] = "SEC-derived shares outstanding unavailable"

    if next_earnings_date is not None:
        metadata["source_next_earnings_date"] = "dolt_earnings"
    else:
        metadata["next_earnings_date"] = "no upcoming earnings event"
    return metadata


def _surprise(
    estimate: float | None,
    actual: float | None,
) -> float | None:
    if not _finite(estimate) or not _finite(actual) or estimate == 0:
        return None
    return (float(actual) - float(estimate)) / abs(float(estimate)) * 100.0


def _pe(price: float | None, ttm_eps: float | None) -> float | None:
    if (
        not _finite(price)
        or price <= 0
        or not _finite(ttm_eps)
        or ttm_eps <= 0
    ):
        return None
    return float(price) / float(ttm_eps)


def _market_cap(
    price: float | None,
    shares_outstanding: float | None,
) -> float | None:
    if (
        not _finite(price)
        or price <= 0
        or not _finite(shares_outstanding)
        or shares_outstanding <= 0
    ):
        return None
    return float(price) * float(shares_outstanding)


def _finite(value: Any) -> bool:
    return (
        isinstance(value, (int, float))
        and not isinstance(value, bool)
        and math.isfinite(value)
    )