feat(sec): reparse path, CIK overrides, resolution validation

Operational plumbing to land the parser fixes and to make silent resolution
failures visible.

- Reparse: SecFundamentalsImporter(reparse=True) restages every accession with
  the current parser and rewrites the ones that now reconstruct differently,
  writing the full column set so a row is never half old-parse. Snapshots stay
  immutable with respect to SEC; the stored row is our reconstruction, and after
  a parser fix keeping it is a stale cache, not history. run_import(force=True)
  bypasses the unchanged-revision no-op, since the staleness is on our side, not
  the source's. Exposed as scripts/reparse_fundamentals.py, dry-run by default.
- CIK overrides: sec_universe reads a {symbol: cik} pin from
  SystemSetting['sec_cik_overrides'], applied ahead of company_tickers.json, for
  when SEC maps a ticker to a successor shell with no filings (XOM -> a zero-
  filing "ExxonMobil Holdings Corp" while every 10-K/Q is under CIK 34088).
- Resolution validation: a tracked issuer resolving to a registrant with no XBRL
  filings now records no_xbrl_filings and raises a warning naming the CIKs and
  the override setting, instead of silently yielding nothing on every run.
- _diff_fields compares datetime instants, not representations: accepted_at
  round-trips naive from SQLite but tz-aware from Postgres, which otherwise made
  a reparse of identical data report every row as changed (and false-positived
  the pre-existing discrepancy warning).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/services/data_import.py

@@ -145,11 +145,17 @@ async def run_import(
     importer: SourceImporter,
     *,
     engine: AsyncEngine | None = None,
+    force: bool = False,
 ) -> DataImportRun | None:
     """Run one import for ``importer``.
 
     Returns the recorded ``DataImportRun`` (promoted / no_op / failed), or None
     when the per-source advisory lock is already held (another run is active).
+
+    ``force`` runs even when the revision is unchanged. The revision tracks the
+    *source*, so a re-import driven by a change on our side — a parser fix that
+    makes stored rows stale — is a no_op under the normal gate. Manually invoked
+    only; scheduled jobs must leave it False so an unchanged source stays a no_op.
     """
     engine = engine or app_engine
     source = importer.source
@@ -189,7 +195,7 @@ async def run_import(
                 revision = await importer.detect_revision(session)
                 run.revision = revision
                 last_rev = await _last_promoted_revision(session, source)
-                if revision is not None and revision == last_rev:
+                if not force and revision is not None and revision == last_rev:
                     run.status = STATUS_NO_OP
                     run.completed_at = _now()
                     await session.commit()

app/services/sec_fundamentals_importer.py

@@ -21,6 +21,12 @@ Guardrails (design + reviews):
 - ``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
@@ -31,7 +37,7 @@ from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
 from typing import Any, Callable
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 
 from app.database import insert_for_session
 from app.models.data_import_run import DataImportRun
@@ -57,7 +63,7 @@ _SNAPSHOT_COLS = (
     "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",
+    "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.
@@ -75,6 +81,10 @@ class StagedFundamentals:
     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
@@ -93,9 +103,17 @@ class SecFundamentalsImporter:
         *,
         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]] = []
@@ -111,7 +129,10 @@ class SecFundamentalsImporter:
             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:
+            # 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:
@@ -175,6 +196,10 @@ class SecFundamentalsImporter:
             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)
@@ -191,7 +216,11 @@ class SecFundamentalsImporter:
                     # 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)
+        # 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)
@@ -241,6 +270,8 @@ class SecFundamentalsImporter:
             "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),
@@ -257,9 +288,26 @@ class SecFundamentalsImporter:
 
     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:
-                continue  # immutable — keep the original row
+                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)
@@ -269,24 +317,48 @@ class SecFundamentalsImporter:
         # 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_discrepancy",
+                code="snapshot_reparse" if self.reparse else "snapshot_discrepancy",
                 message=(
                     f"{len(staged.discrepancies)} stored accession(s) reconstructed "
-                    f"differently; kept immutable: {accns}"
+                    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 — pin the right CIK via the "
+                    f"'{sec_universe.CIK_OVERRIDES_KEY}' setting: {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,
-            "existing_unchanged": len(staged.existing_accessions),
+            "updated": updated,
+            "existing_unchanged": len(staged.existing_accessions) - updated,
             "discrepancies": len(staged.discrepancies),
             **ticker_counts,
         }
@@ -395,5 +467,26 @@ def _row_values(row: SnapshotRow, run_id: int) -> dict[str, Any]:
 
 
 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)]
+    """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)

app/services/sec_universe.py

@@ -16,6 +16,7 @@ changes on the framework's failure commit). The proposals are applied only in
 from __future__ import annotations
 
 import hashlib
+import json
 import logging
 from dataclasses import dataclass, field
 from typing import Iterable
@@ -23,11 +24,21 @@ from typing import Iterable
 from sqlalchemy import select, update
 
 from app.models.ticker import Ticker
+from app.services import settings_store
 from app.services.earnings_alignment import normalise_symbol
 from app.services.sec_client import SecClient
 
 logger = logging.getLogger(__name__)
 
+# JSON {symbol: cik} pinning a ticker to a specific registrant, overriding
+# company_tickers.json. Needed when SEC maps a ticker to a successor entity that
+# has not filed: XOM points at CIK 2115436 "ExxonMobil Holdings Corp" (zero XBRL
+# filings) while every 10-K/10-Q — including one filed 2026-05-04 — is still under
+# CIK 34088. Which registrant is the real filer is a judgement about a corporate
+# event, so it is pinned explicitly rather than guessed. The importer's
+# `no_xbrl_filings` warning is what tells you a pin is needed.
+CIK_OVERRIDES_KEY = "sec_cik_overrides"
+
 
 @dataclass
 class ResolvedUniverse:
@@ -43,6 +54,7 @@ async def resolve_ciks(db, client: SecClient) -> ResolvedUniverse:
     """Resolve tracked tickers to CIKs via company_tickers.json. **Read-only** —
     returns the mapping + proposed `tickers.cik` writes; mutates nothing."""
     ticker_to_cik = await client.company_tickers()
+    overrides = await cik_overrides(db)
     rows = (await db.execute(select(Ticker.id, Ticker.symbol, Ticker.cik))).all()
 
     result = ResolvedUniverse()
@@ -50,7 +62,7 @@ async def resolve_ciks(db, client: SecClient) -> ResolvedUniverse:
         if not symbol:
             continue
         sym = normalise_symbol(symbol)
-        cik = ticker_to_cik.get(sym)
+        cik = overrides.get(sym) or ticker_to_cik.get(sym)
         if cik is None:
             continue  # ADRs / non-SEC issuers — snapshots simply absent
         result.symbol_to_cik[sym] = cik
@@ -65,6 +77,34 @@ async def resolve_ciks(db, client: SecClient) -> ResolvedUniverse:
     return result
 
 
+async def cik_overrides(db) -> dict[str, int]:
+    """Manual ``{symbol: cik}`` pins from ``SystemSetting[CIK_OVERRIDES_KEY]``.
+
+    A malformed setting must never take the importer down, so anything unparseable
+    is logged and ignored — the run then falls back to company_tickers.json.
+    """
+    raw = await settings_store.get_value(db, CIK_OVERRIDES_KEY)
+    if not raw:
+        return {}
+    try:
+        loaded = json.loads(raw)
+    except (TypeError, ValueError):
+        logger.warning("%s is not valid JSON — ignoring CIK overrides", CIK_OVERRIDES_KEY)
+        return {}
+    if not isinstance(loaded, dict):
+        logger.warning("%s must be a {symbol: cik} object — ignoring", CIK_OVERRIDES_KEY)
+        return {}
+    out: dict[str, int] = {}
+    for symbol, cik in loaded.items():
+        try:
+            out[normalise_symbol(str(symbol))] = int(cik)
+        except (TypeError, ValueError):
+            logger.warning("%s: bad entry %r -> %r — ignoring", CIK_OVERRIDES_KEY, symbol, cik)
+    if out:
+        logger.info("resolve_ciks: %d CIK override(s) applied: %s", len(out), sorted(out))
+    return out
+
+
 async def fetch_sic_updates(
     client: SecClient, cik_to_ticker_ids: dict[int, Iterable[int]]
 ) -> list[tuple[int, str | None, str | None]]: