"""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.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 # Bound how far back the incremental index walk goes if the job hasn't run in a # while (each day = one small request); older gaps are logged, not silently lost. _MAX_INDEX_WALK_DAYS = 45 _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", ) # Fields compared to flag a differing existing accession (immutable → report, not mutate). _DISCREPANCY_COLS = ("period_end", "fiscal_year", "fiscal_period", "revenue", "net_income") @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) 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) return staged async def _stage_issuer(self, client, cik, is_backfill, filed_by_cik, staged) -> None: cf = await client.companyfacts(cik) 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" ) 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), "cik_updates": len(staged.resolved.cik_updates), } 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 discrepancies = 0 if staged.rows: existing = await self._existing_by_accession(db, [r.accession for r in staged.rows]) for row in staged.rows: old = existing.get(row.accession) if old is not None: if _differs(row, old): discrepancies += 1 logger.warning( "sec_facts: accession %s reconstructed differently than " "stored (immutable — not overwriting)", row.accession ) continue # ON CONFLICT DO NOTHING (below) leaves it untouched stmt = insert_for_session(db, FundamentalSnapshot).values( **_row_values(row, run_id) ) stmt = stmt.on_conflict_do_nothing(index_elements=["accession"]) await db.execute(stmt) inserted += 1 ticker_counts = await sec_universe.apply_ticker_updates( db, staged.resolved, staged.sic_updates ) return { "inserted": inserted, "existing_unchanged": len(staged.rows) - inserted, "discrepancies": 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]]: tracked = set(self._resolved.cik_to_ticker_ids) if self._resolved else set() start = max(last_processed + timedelta(days=1), latest - timedelta(days=_MAX_INDEX_WALK_DAYS)) if start > last_processed + timedelta(days=1): logger.warning("sec_facts: index gap > %d days; walking from %s", _MAX_INDEX_WALK_DAYS, start) rows: list[dict[str, Any]] = [] day = start 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 _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 _differs(row: SnapshotRow, old: FundamentalSnapshot) -> bool: return any(getattr(row, col) != getattr(old, col) for col in _DISCREPANCY_COLS)