signal-platform/app/services/fundamentals_quality_service.py

"""Actionability gate for incomplete SEC fundamentals."""

from __future__ import annotations

import json
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

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

from app.models.data_import_run import DataImportRun
from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.sec_filing_gap import SecFilingGap
from app.models.ticker import Ticker

_SEC_FORMS = ("10-K", "10-Q", "10-K/A", "10-Q/A")

# How recent the issuer's own newest filing must be for an *escalated* gap to
# stop pausing setups. A gap pauses an issuer until it is either resolved or
# superseded by a later ingested filing — which assumes the gap is temporary.
# It is not always: SEC's per-company Company-Facts files can go stale
# indefinitely (2026-08, 43 large caps whose Q2 10-Qs the frames API carried but
# whose companyfacts files never received), and since the supersede rule needs a
# *successfully ingested* later filing, a stale file also swallows the next
# quarter. The pause is then open-ended rather than seasonal.
#
# So the pause hands off to the alert: once `filing_gap_aged` has escalated a gap
# to an operator (`escalated_at`), the issuer resumes on the fundamentals it does
# have — provided those are recent. An issuer with nothing this fresh has no
# usable fundamentals at all and stays paused, which is the case the gate was
# built for. The retry queue is untouched: `active_gaps` still returns these, so
# the importer keeps retrying and a recovered filing still resolves normally.
GAP_GATE_RECENT_FILING_DAYS = 180


@dataclass(frozen=True)
class SetupQuality:
    eligible: bool
    code: str | None = None
    message: str | None = None


async def active_gaps(
    db: AsyncSession,
    ciks: set[str] | None = None,
) -> list[SecFilingGap]:
    """Unresolved gaps that have not been superseded by a later filing."""
    matching_snapshot = exists().where(
        FundamentalSnapshot.accession == SecFilingGap.accession
    )
    gap_date = func.coalesce(
        SecFilingGap.index_date,
        func.date(SecFilingGap.first_seen_at),
    )
    later_snapshot = exists().where(
        FundamentalSnapshot.cik == SecFilingGap.cik,
        FundamentalSnapshot.form.in_(_SEC_FORMS),
        FundamentalSnapshot.filed_date > gap_date,
    )
    stmt = select(SecFilingGap).where(
        ~matching_snapshot,
        ~later_snapshot,
    )
    if ciks is not None:
        if not ciks:
            return []
        stmt = stmt.where(SecFilingGap.cik.in_(ciks))
    return list((await db.execute(stmt)).scalars().all())


async def gap_exempt_ciks(
    db: AsyncSession, gaps: list[SecFilingGap]
) -> set[str]:
    """CIKs whose gaps have stopped pausing setups (see GAP_GATE_RECENT_FILING_DAYS).

    Every one of a CIK's active gaps must be escalated: one fresh gap alongside an
    old one still means a filing we might yet ingest, which is worth pausing for.

    Public because the importer alerts on this exact transition (a CIK dropping
    out of this set is a pause coming back on) and the rule must not exist twice.
    """
    by_cik: dict[str, list[SecFilingGap]] = defaultdict(list)
    for gap in gaps:
        by_cik[gap.cik].append(gap)
    escalated = {
        cik
        for cik, items in by_cik.items()
        if all(gap.escalated_at is not None for gap in items)
    }
    if not escalated:
        return set()
    cutoff = (
        datetime.now(timezone.utc) - timedelta(days=GAP_GATE_RECENT_FILING_DAYS)
    ).date()
    rows = await db.execute(
        select(FundamentalSnapshot.cik)
        .where(
            FundamentalSnapshot.cik.in_(escalated),
            FundamentalSnapshot.form.in_(_SEC_FORMS),
            FundamentalSnapshot.filed_date >= cutoff,
        )
        .distinct()
    )
    return set(rows.scalars())


async def _latest_validation(db: AsyncSession) -> dict:
    payload = (
        await db.execute(
            select(DataImportRun.validation_json)
            .where(
                DataImportRun.source == "sec_facts",
                DataImportRun.validation_json.is_not(None),
            )
            .order_by(DataImportRun.id.desc())
            .limit(1)
        )
    ).scalar_one_or_none()
    if not payload:
        return {}
    try:
        summary = json.loads(payload)
    except (TypeError, ValueError):
        return {}
    return summary if isinstance(summary, dict) else {}


async def blocked_reasons_by_cik(
    db: AsyncSession,
    ciks: set[str] | None = None,
) -> dict[str, str]:
    """Current SEC blocker code by CIK; no historical audit scan."""
    if ciks is not None and not ciks:
        return {}

    gaps = await active_gaps(db, ciks)
    # Escalated gaps on issuers that still have recent fundamentals no longer
    # pause setups, on either path below — the summary mirrors the same filings.
    exempt = await gap_exempt_ciks(db, gaps)
    reasons = {
        gap.cik: "sec_filing_gap" for gap in gaps if gap.cik not in exempt
    }
    summary = await _latest_validation(db)

    def wanted(cik: str) -> bool:
        return ciks is None or cik in ciks

    # New summaries carry the complete compact CIK set while the detailed lists
    # stay capped for audit readability. Detailed entries supply the reason.
    for cik in summary.get("setup_blocked_ciks") or []:
        normalized = str(cik) if cik else ""
        if normalized and wanted(normalized) and normalized not in exempt:
            reasons.setdefault(normalized, "sec_filing_gap")
    for item in summary.get("missing_xbrl") or []:
        normalized = str(item.get("cik") or "")
        if normalized and wanted(normalized) and normalized not in exempt:
            reasons.setdefault(normalized, "sec_filing_gap")
    for cik in summary.get("no_xbrl_ciks") or []:
        normalized = str(cik) if cik else ""
        if normalized and wanted(normalized):
            reasons[normalized] = "no_xbrl_filings"
    for item in summary.get("no_xbrl_filings") or []:
        normalized = str(item.get("cik") or "")
        if normalized and wanted(normalized):
            reasons[normalized] = "no_xbrl_filings"
    return reasons


async def blocked_ciks(db: AsyncSession) -> set[str]:
    return set(await blocked_reasons_by_cik(db))


async def blocked_ticker_ids(db: AsyncSession) -> set[int]:
    ciks = await blocked_ciks(db)
    if not ciks:
        return set()
    rows = await db.execute(select(Ticker.id).where(Ticker.cik.in_(ciks)))
    return {int(ticker_id) for ticker_id in rows.scalars()}


async def ticker_quality(db: AsyncSession, symbol: str) -> SetupQuality:
    ticker = (
        await db.execute(
            select(Ticker).where(Ticker.symbol == symbol.strip().upper())
        )
    ).scalar_one_or_none()
    if ticker is None or not ticker.cik:
        return SetupQuality(eligible=True)
    reason = (await blocked_reasons_by_cik(db, {ticker.cik})).get(ticker.cik)
    if reason == "no_xbrl_filings":
        return SetupQuality(
            eligible=False,
            code=reason,
            message=(
                "No SEC 10-K/10-Q is available for this registrant, so new setups "
                "are paused. New registrants clear automatically after their first "
                "filing; a successor shell needs an SEC CIK override."
            ),
        )
    if reason:
        return SetupQuality(
            eligible=False,
            code=reason,
            message=(
                "A recent SEC filing is still being reconciled, so new setups are "
                "paused. The scheduled fundamentals import retries it automatically."
            ),
        )
    return SetupQuality(eligible=True)


async def ticker_is_eligible(db: AsyncSession, ticker_id: int) -> bool:
    cik = (
        await db.execute(select(Ticker.cik).where(Ticker.id == ticker_id))
    ).scalar_one_or_none()
    if not cik:
        return True
    return cik not in await blocked_reasons_by_cik(db, {cik})