signal-platform/app/services/sec_fundamentals_importer.py

"""SEC fundamentals importer (workstream A, phase A3).

A ``SourceImporter`` (see ``app/services/data_import.py``) that populates the
immutable ``fundamental_snapshots`` from SEC Company Facts and back-fills
``tickers.cik/sic/sic_description``. EDGAR-daily-index driven: it fetches
companyfacts only for tracked issuers that filed since the last run (full-history
backfill on the first run / for newly-added issuers). Shadow only — nothing reads
snapshots until A4.

Guardrails (design + reviews):
- ``detect_revision`` caches the resolved universe + the exact tracked index rows
  and composes the revision from them; ``stage`` consumes those same cached inputs
  (it does not refetch the index/universe) so promoted data always matches the
  computed revision.
- Resolution is read-only in ``stage`` (proposals only); ticker writes happen in
  ``promote`` via ``apply_ticker_updates``.
- ``validate`` runs the **index↔Company-Facts consistency gate** before any write:
  a tracked XBRL index accession missing from Company Facts fails the run (the two
  are separate SEC products that can lag) so we retry rather than record a
  null/partial snapshot. Non-XBRL amendments are skipped with a recorded reason.
- ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable),
  reports differing existing accessions, and applies ticker updates in the same
  transaction.
"""

from __future__ import annotations

import logging
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from typing import Any, Callable

from sqlalchemy import func, select

from app.database import insert_for_session
from app.models.data_import_run import DataImportRun
from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.system_event import SystemEvent
from app.services import sec_facts_parser as parser
from app.services import sec_universe
from app.services.data_import import STATUS_PROMOTED, ValidationResult
from app.services.sec_client import SecClient, SecError, cik10
from app.services.sec_facts_parser import FilingMeta, SnapshotRow
from app.services.sec_universe import ResolvedUniverse

logger = logging.getLogger(__name__)

SOURCE = "sec_facts"
_XBRL_FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A"}
# On the one-time backfill, require this fraction of tracked issuers to yield at
# least one snapshot (guards a broken fetch/parse from promoting a hollow table).
MIN_BACKFILL_COVERAGE = 0.5

_SNAPSHOT_COLS = (
    "cik", "accession", "form", "filed_date", "accepted_at", "period_start",
    "period_end", "fiscal_year", "fiscal_period", "revenue", "net_income",
    "operating_income", "diluted_eps", "cfo", "capex", "depreciation_amortization",
    "cash_and_st_investments", "total_debt", "shares_outstanding",
    "shares_outstanding_date",
)
# Compare ALL source fields (every column except the accession key) to flag a
# differing existing accession — immutable, so we report, never mutate.
_COMPARE_COLS = tuple(c for c in _SNAPSHOT_COLS if c != "accession")


@dataclass
class StagedFundamentals:
    resolved: ResolvedUniverse
    sic_updates: list[tuple[int, str | None, str | None]] = field(default_factory=list)
    rows: list[SnapshotRow] = field(default_factory=list)
    skipped_filings: list[dict[str, str]] = field(default_factory=list)
    field_issues: list[dict[str, str]] = field(default_factory=list)
    skipped_non_xbrl: list[dict[str, str]] = field(default_factory=list)
    missing_xbrl: list[dict[str, str]] = field(default_factory=list)
    invalid_payloads: list[dict[str, str]] = field(default_factory=list)
    existing_accessions: set[str] = field(default_factory=set)
    discrepancies: list[dict[str, Any]] = field(default_factory=list)
    backfill: bool = False
    issuers_fetched: int = 0
    issuers_with_rows: int = 0


def _now() -> datetime:
    return datetime.now(timezone.utc)


class SecFundamentalsImporter:
    source = SOURCE

    def __init__(
        self,
        *,
        client_factory: Callable[[], SecClient] | None = None,
        today: date | None = None,
    ) -> None:
        self._client_factory = client_factory or (lambda: SecClient())
        self.today = today or _now().date()
        # cached by detect_revision, consumed by stage:
        self._resolved: ResolvedUniverse | None = None
        self._index_rows: list[dict[str, Any]] = []
        self._latest_index_date: date | None = None
        self._backfill = False

    # -- SourceImporter protocol -------------------------------------------

    async def detect_revision(self, db) -> str | None:
        async with self._client_factory() as client:
            self._resolved = await sec_universe.resolve_ciks(db, client)
            last_processed = await self._last_processed_index_date(db)
            self._latest_index_date = await client.latest_index_date(self.today)
            if self._latest_index_date is None:
                raise SecError("no EDGAR daily index available")
            if last_processed is None:
                self._backfill = True
                self._index_rows = []
            else:
                self._backfill = False
                self._index_rows = await self._collect_index_rows(
                    client, last_processed, self._latest_index_date
                )
        content = sec_universe.index_content_hash(self._index_rows)
        return sec_universe.compose_revision(
            self._latest_index_date, content, self._resolved.symbol_to_cik
        )

    async def stage(self, db) -> StagedFundamentals:
        assert self._resolved is not None, "detect_revision must run first"
        resolved = self._resolved
        staged = StagedFundamentals(resolved=resolved, backfill=self._backfill)

        cik_to_tids = resolved.cik_to_ticker_ids
        filed_by_cik: dict[int, list[str]] = defaultdict(list)
        for r in self._index_rows:
            if r["cik"] in cik_to_tids:
                filed_by_cik[r["cik"]].append(r["accession"])

        existing = await self._ciks_with_snapshots(db, set(cik_to_tids))
        if self._backfill:
            backfill_ciks = set(cik_to_tids)
        else:
            # Newly added issuers (resolved but no snapshots yet) get a full-history
            # backfill; issuers that already have history are handled incrementally.
            backfill_ciks = {c for c in cik_to_tids if c not in existing}
        incremental_ciks = set(filed_by_cik) - backfill_ciks

        async with self._client_factory() as client:
            for cik in sorted(backfill_ciks | incremental_ciks):
                is_backfill = cik in backfill_ciks
                await self._stage_issuer(client, cik, is_backfill, filed_by_cik, staged)

        # Read-only discrepancy detection: an accession we reconstructed that is
        # already stored, differing in ANY source field (immutable → report in
        # validation, event on promote, never mutate). Also gives promote the
        # existing set so its insert count is dialect-independent.
        if staged.rows:
            existing = await self._existing_by_accession(db, [r.accession for r in staged.rows])
            staged.existing_accessions = set(existing)
            for row in staged.rows:
                old = existing.get(row.accession)
                if old is not None:
                    fields = _diff_fields(row, old)
                    if fields:
                        staged.discrepancies.append({"accession": row.accession, "fields": fields})
        return staged

    async def _stage_issuer(self, client, cik, is_backfill, filed_by_cik, staged) -> None:
        cf = await client.companyfacts(cik)
        bad = _companyfacts_structure_error(cf)
        if bad is not None:
            # Malformed payload (missing facts/units structure) — record separately
            # and fail validation, rather than letting it degrade to skipped rows.
            staged.invalid_payloads.append({"cik": cik10(cik), "reason": bad})
            staged.issuers_fetched += 1
            return
        sub = await client.submissions(cik, include_history=is_backfill)
        xbrl_meta, nonxbrl = _filing_meta(sub)

        if is_backfill:
            accns = set(xbrl_meta)
        else:
            present = parser.companyfacts_accessions(cf)
            accns = set()
            for accn in filed_by_cik.get(cik, []):
                if accn in nonxbrl:
                    staged.skipped_non_xbrl.append({"cik": cik10(cik), "accession": accn})
                elif accn in xbrl_meta and accn in present:
                    accns.add(accn)
                else:
                    # XBRL (or unknown) filing not yet in Company Facts → the products
                    # have lagged; fail+retry rather than record nothing for it.
                    staged.missing_xbrl.append({"cik": cik10(cik), "accession": accn})

        result = parser.parse_snapshots(cf, xbrl_meta, accns)
        staged.rows.extend(result.rows)
        staged.skipped_filings.extend(result.skipped_filings)
        staged.field_issues.extend(result.field_issues)
        staged.issuers_fetched += 1
        if result.rows:
            staged.issuers_with_rows += 1

        # SIC proposal for this issuer's tickers (read-only; applied in promote).
        sic = str(sub["sic"]) if sub.get("sic") else None
        desc = sub.get("sic_description")
        for tid in staged.resolved.cik_to_ticker_ids.get(cik, []):
            staged.sic_updates.append((tid, sic, desc))

    async def validate(self, db, staged: StagedFundamentals) -> ValidationResult:
        messages: list[str] = []

        # Consistency gate — before any write.
        if staged.missing_xbrl:
            messages.append(
                f"{len(staged.missing_xbrl)} tracked XBRL filing(s) not yet in "
                "Company Facts (index/facts lag) — retry"
            )
        # Malformed companyfacts payloads must fail, not degrade to skipped rows.
        if staged.invalid_payloads:
            messages.append(
                f"{len(staged.invalid_payloads)} issuer(s) returned a malformed "
                "companyfacts payload (missing facts structure)"
            )

        accns = [r.accession for r in staged.rows]
        if len(accns) != len(set(accns)):
            messages.append("duplicate accession in staged snapshots")

        if staged.backfill:
            n_issuers = len(staged.resolved.cik_to_ticker_ids)
            coverage = staged.issuers_with_rows / n_issuers if n_issuers else 0.0
            if coverage < MIN_BACKFILL_COVERAGE:
                messages.append(
                    f"backfill coverage {coverage:.0%} < {MIN_BACKFILL_COVERAGE:.0%}"
                )

        summary = {
            "backfill": staged.backfill,
            "issuers_fetched": staged.issuers_fetched,
            "issuers_with_rows": staged.issuers_with_rows,
            "snapshot_rows": len(staged.rows),
            "skipped_filings": len(staged.skipped_filings),
            "field_issues": len(staged.field_issues),
            "skipped_non_xbrl": len(staged.skipped_non_xbrl),
            "missing_xbrl": len(staged.missing_xbrl),
            "invalid_payloads": staged.invalid_payloads,
            "cik_updates": len(staged.resolved.cik_updates),
            # differing existing accessions (immutable — kept, reported here)
            "discrepancies": staged.discrepancies[:50],
            "discrepancy_count": len(staged.discrepancies),
        }
        return ValidationResult(
            ok=not messages,
            summary=summary,
            source_max_date=self._latest_index_date,
            messages=messages,
        )

    async def promote(self, db, staged: StagedFundamentals, run_id: int) -> dict[str, int]:
        inserted = 0
        for row in staged.rows:
            if row.accession in staged.existing_accessions:
                continue  # immutable — keep the original row
            stmt = insert_for_session(db, FundamentalSnapshot).values(**_row_values(row, run_id))
            stmt = stmt.on_conflict_do_nothing(index_elements=["accession"])  # race belt-and-suspenders
            await db.execute(stmt)
            inserted += 1

        # Warn (in-transaction, so it commits atomically with the promotion) when
        # any existing accession reconstructed differently — kept immutable.
        if staged.discrepancies:
            accns = ", ".join(d["accession"] for d in staged.discrepancies[:10])
            db.add(SystemEvent(
                severity="warning",
                source="sec_facts",
                code="snapshot_discrepancy",
                message=(
                    f"{len(staged.discrepancies)} stored accession(s) reconstructed "
                    f"differently; kept immutable: {accns}"
                )[:4000],
                dedup_key=f"sec_facts:discrepancy:{run_id}",
                created_at=_now(),
            ))

        ticker_counts = await sec_universe.apply_ticker_updates(
            db, staged.resolved, staged.sic_updates
        )
        return {
            "inserted": inserted,
            "existing_unchanged": len(staged.existing_accessions),
            "discrepancies": len(staged.discrepancies),
            **ticker_counts,
        }

    # -- helpers -----------------------------------------------------------

    async def _last_processed_index_date(self, db) -> date | None:
        return (
            await db.execute(
                select(DataImportRun.source_max_date)
                .where(DataImportRun.source == SOURCE, DataImportRun.status == STATUS_PROMOTED)
                .order_by(DataImportRun.id.desc())
                .limit(1)
            )
        ).scalar_one_or_none()

    async def _collect_index_rows(
        self, client: SecClient, last_processed: date, latest: date
    ) -> list[dict[str, Any]]:
        # Walk EVERY unprocessed date. No cap — dropping the older part of a long
        # outage while still advancing source_max_date would permanently lose
        # those filings. A large gap is one-time cost, not silent data loss.
        tracked = set(self._resolved.cik_to_ticker_ids) if self._resolved else set()
        gap = (latest - last_processed).days
        if gap > 60:
            logger.warning("sec_facts: %d-day index gap since %s; walking all", gap, last_processed)
        rows: list[dict[str, Any]] = []
        day = last_processed + timedelta(days=1)
        while day <= latest:
            for r in await client.daily_index(day):
                if r["form"] in _XBRL_FORMS and r["cik"] in tracked:
                    rows.append(r)
            day += timedelta(days=1)
        return rows

    async def _ciks_with_snapshots(self, db, ciks: set[int]) -> set[int]:
        if not ciks:
            return set()
        cik_strs = [cik10(c) for c in ciks]
        found = (
            await db.execute(
                select(FundamentalSnapshot.cik)
                .where(FundamentalSnapshot.cik.in_(cik_strs))
                .distinct()
            )
        ).scalars().all()
        return {int(c) for c in found}

    async def _existing_by_accession(self, db, accessions: list[str]) -> dict[str, FundamentalSnapshot]:
        if not accessions:
            return {}
        rows = (
            await db.execute(
                select(FundamentalSnapshot).where(FundamentalSnapshot.accession.in_(accessions))
            )
        ).scalars().all()
        return {r.accession: r for r in rows}


def _companyfacts_structure_error(cf: Any) -> str | None:
    """None if the payload is structurally sound, else a reason string. Checks the
    top-level ``facts`` mapping AND that every concept carries a ``units`` mapping —
    a missing/non-dict units would silently drop that concept's facts otherwise."""
    if not isinstance(cf, dict) or not isinstance(cf.get("facts"), dict):
        return "missing facts structure"
    for concepts in cf["facts"].values():
        if not isinstance(concepts, dict):
            return "malformed taxonomy structure"
        for body in concepts.values():
            if not isinstance(body, dict) or not isinstance(body.get("units"), dict):
                return "missing units structure"
    return None


def _filing_meta(sub: dict[str, Any]) -> tuple[dict[str, FilingMeta], set[str]]:
    """(xbrl_meta, nonxbrl_accessions) from a submissions payload. xbrl_meta only
    includes 10-K/10-Q(/A) filings that are XBRL and have full period metadata."""
    xbrl: dict[str, FilingMeta] = {}
    nonxbrl: set[str] = set()
    for f in sub.get("filings", []):
        if f["form"] not in _XBRL_FORMS:
            continue
        if not f.get("is_xbrl"):
            nonxbrl.add(f["accession"])
            continue
        if not (f.get("report_date") and f.get("filing_date") and f.get("acceptance_datetime")):
            continue
        xbrl[f["accession"]] = FilingMeta(
            report_date=date.fromisoformat(f["report_date"]),
            filing_date=date.fromisoformat(f["filing_date"]),
            accepted_at=_parse_dt(f["acceptance_datetime"]),
            form=f["form"],
        )
    return xbrl, nonxbrl


def _parse_dt(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def _row_values(row: SnapshotRow, run_id: int) -> dict[str, Any]:
    values = {col: getattr(row, col) for col in _SNAPSHOT_COLS}
    values["import_run_id"] = run_id
    values["created_at"] = _now()
    return values


def _diff_fields(row: SnapshotRow, old: FundamentalSnapshot) -> list[str]:
    """Source fields where a re-parsed row differs from the stored (immutable) row."""
    return [col for col in _COMPARE_COLS if getattr(row, col) != getattr(old, col)]