"""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. - ``reparse=True`` is the one exception to immutability, and it is deliberate: it restages every accession with the current parser and **rewrites** the rows that now reconstruct differently. Immutability protects SEC's record (one row per accession, amendments retained) — but the stored row is *our* reconstruction, so after a parser fix, keeping it is preserving a stale cache, not history. Manually invoked through ``scripts/reparse_fundamentals.py``; never scheduled. """ 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 select, update 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", "weighted_avg_diluted_shares", ) # 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) # Tracked issuers whose registrant has NO XBRL 10-K/10-Q at all: they can # never yield a snapshot, so this is a resolution problem (a ticker pointed # at a successor shell), not missing data. See sec_universe.CIK_OVERRIDES_KEY. no_xbrl_filings: list[dict[str, Any]] = field(default_factory=list) 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, reparse: bool = False, ) -> None: self._client_factory = client_factory or (lambda: SecClient()) self.today = today or _now().date() # Reparse: re-derive every stored accession with the CURRENT parser and # rewrite the ones that now reconstruct differently. Snapshots are # immutable with respect to SEC (one row per accession, amendments kept), # but the stored row is *our reconstruction* — when a parser bug is fixed, # leaving it stale is not immutability, it is a stale cache. Manually # invoked via scripts/reparse_fundamentals.py; never scheduled. self.reparse = reparse # 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") # Reparse needs every accession restaged, not just those filed since # the last run — the facts a fixed parser now accepts were never # stored, so a reparse cannot be served from the database. if last_processed is None or self.reparse: 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 not xbrl_meta: staged.no_xbrl_filings.append( {"cik": cik10(cik), "name": sub.get("name"), "tickers": sub.get("tickers")} ) 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}) # fiscalYearEnd (MMDD) is what lets the parser derive period identity from # reportDate instead of SEC's unreliable fy/fp fields. result = parser.parse_snapshots( cf, xbrl_meta, accns, fiscal_year_end=sub.get("fiscal_year_end") ) 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), "no_xbrl_filings": staged.no_xbrl_filings[:50], "no_xbrl_filings_count": len(staged.no_xbrl_filings), "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 updated = 0 # Only accessions whose reconstruction actually changed are rewritten; # an unchanged stored row is left completely alone. changed = {d["accession"] for d in staged.discrepancies} if self.reparse else set() for row in staged.rows: if row.accession in staged.existing_accessions: if row.accession in changed: # Write the FULL column set (_row_values covers _SNAPSHOT_COLS) # so a rewritten row is never half old-parse, half new-parse. # created_at stays at the original insert; import_run_id # attributes the rewrite. values = _row_values(row, run_id) values.pop("created_at", None) await db.execute( update(FundamentalSnapshot) .where(FundamentalSnapshot.accession == row.accession) .values(**values) ) updated += 1 continue # otherwise 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]) disposition = ( f"REWRITTEN by reparse run {run_id}" if self.reparse else "kept immutable" ) db.add(SystemEvent( severity="warning", source="sec_facts", code="snapshot_reparse" if self.reparse else "snapshot_discrepancy", message=( f"{len(staged.discrepancies)} stored accession(s) reconstructed " f"differently; {disposition}: {accns}" )[:4000], dedup_key=f"sec_facts:discrepancy:{run_id}", created_at=_now(), )) # A tracked issuer whose registrant has no XBRL filings can never produce a # snapshot, and it is restaged on every run forever. That is a resolution # problem, not missing data, and it is silent without this. if staged.no_xbrl_filings: named = ", ".join( f"{e['cik']} ({e.get('name') or '?'})" for e in staged.no_xbrl_filings[:10] ) db.add(SystemEvent( severity="warning", source="sec_facts", code="no_xbrl_filings", message=( f"{len(staged.no_xbrl_filings)} tracked issuer(s) resolved to a " f"registrant with no XBRL 10-K/10-Q. Either a successor shell " f"(pin the real filer via the '{sec_universe.CIK_OVERRIDES_KEY}' " f"setting) or a new registrant that has not filed its first " f"10-K/10-Q yet, which needs nothing and clears itself: {named}" )[:4000], dedup_key=f"sec_facts:no_xbrl_filings:{run_id}", created_at=_now(), )) ticker_counts = await sec_universe.apply_ticker_updates( db, staged.resolved, staged.sic_updates ) return { "inserted": inserted, "updated": updated, "existing_unchanged": len(staged.existing_accessions) - updated, "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 row.""" return [ col for col in _COMPARE_COLS if not _same_value(getattr(row, col), getattr(old, col)) ] def _same_value(parsed: Any, stored: Any) -> bool: """Compare a freshly parsed value against its stored round-trip. Datetimes need care: every timestamp here is UTC by construction, but ``DateTime(timezone=True)`` only preserves tzinfo on Postgres — SQLite hands back a naive value. Comparing representations would report an unchanged row as differing, which would both spam the discrepancy warning and make a reparse rewrite every row it touched. Compare instants instead. """ if isinstance(parsed, datetime) and isinstance(stored, datetime): return _as_utc(parsed) == _as_utc(stored) return parsed == stored def _as_utc(value: datetime) -> datetime: return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc)