diff --git a/app/services/data_import.py b/app/services/data_import.py index 11f1adf..f65992b 100644 --- a/app/services/data_import.py +++ b/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() diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 144497b..ccd0944 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/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) diff --git a/app/services/sec_universe.py b/app/services/sec_universe.py index 9d80bd0..a81efde 100644 --- a/app/services/sec_universe.py +++ b/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]]: diff --git a/scripts/reparse_fundamentals.py b/scripts/reparse_fundamentals.py new file mode 100644 index 0000000..f4fecc1 --- /dev/null +++ b/scripts/reparse_fundamentals.py @@ -0,0 +1,120 @@ +"""Re-derive every stored SEC snapshot with the current parser. + +Snapshots are immutable per accession, so a parser fix does not reach rows that +are already stored: a normal import skips them and only logs a +``snapshot_discrepancy``. This script is the deliberate, manual exception -- +it restages every accession from SEC Company Facts and rewrites the rows whose +reconstruction changed. + +**Dry run by default.** Nothing is written unless ``--apply`` is passed. The dry +run stages and validates exactly as the real run does (both are read-only) and +reports the full blast radius: how many rows would change, which fields, and +per-symbol before/after samples. + +Cost: a reparse cannot be served from the database -- the facts a fixed parser now +accepts were never stored -- so it refetches Company Facts for every tracked issuer +under the SEC fair-access throttle. Expect a long run and a lot of network. + +Scope note: this rewrites ``fundamental_snapshots`` only. As of the A5 gate those +rows feed the fundamentals API/UI and the parity report; scoring still reads the +legacy ``fundamental_data`` table, so a reparse does not move composite scores or +backtests until the cutover happens. + +Examples +-------- + # dry run: report what would change, write nothing + python scripts/reparse_fundamentals.py + + # dry run, showing more per-field detail + python scripts/reparse_fundamentals.py --samples 40 + + # actually rewrite the changed rows + python scripts/reparse_fundamentals.py --apply +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from collections import Counter +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.database import async_session_factory # noqa: E402 +from app.services.data_import import run_import # noqa: E402 +from app.services.sec_fundamentals_importer import SecFundamentalsImporter # noqa: E402 + + +def _parse_args() -> argparse.Namespace: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--apply", action="store_true", + help="rewrite changed rows (default: dry run, writes nothing)") + ap.add_argument("--samples", type=int, default=20, + help="how many changed accessions to show in detail (default 20)") + return ap.parse_args() + + +async def _dry_run(samples: int) -> int: + importer = SecFundamentalsImporter(reparse=True) + async with async_session_factory() as db: + print("staging every tracked issuer from SEC Company Facts (this is the slow part)...") + revision = await importer.detect_revision(db) + staged = await importer.stage(db) + result = await importer.validate(db, staged) + + print(f"\nrevision : {revision}") + print(f"issuers fetched : {staged.issuers_fetched}") + print(f"rows reconstructed : {len(staged.rows)}") + print(f"already stored : {len(staged.existing_accessions)}") + print(f"WOULD BE REWRITTEN : {len(staged.discrepancies)}") + print(f"new inserts : {len(staged.rows) - len(staged.existing_accessions)}") + print(f"validation ok : {result.ok}") + if not result.ok: + print(f"validation messages : {result.messages}") + + if staged.discrepancies: + field_counts = Counter(f for d in staged.discrepancies for f in d["fields"]) + print("\nchanged fields (accession count per field):") + for name, count in field_counts.most_common(): + print(f" {name:28s} {count}") + + by_accession = {r.accession: r for r in staged.rows} + print(f"\nfirst {min(samples, len(staged.discrepancies))} changed accessions:") + for d in staged.discrepancies[:samples]: + row = by_accession.get(d["accession"]) + where = f"{row.cik} {row.fiscal_year} {row.fiscal_period}" if row else "?" + print(f" {d['accession']} {where:28s} {', '.join(d['fields'])}") + + print( + "\nDRY RUN -- nothing was written." + "\nCheck that the changes are the *kinds* you expect (recovered nulls," + "\ncorrected values) and sample issuers you did not anticipate before" + "\nre-running with --apply." + ) + return 0 if result.ok else 1 + + +async def _apply() -> int: + # force=True: the revision tracks SEC, which has not changed — the staleness + # is on our side, so the normal no-op gate would skip this. + run = await run_import(SecFundamentalsImporter(reparse=True), force=True) + if run is None: + print("another sec_facts import holds the lock; nothing done") + return 1 + print(f"run {run.id}: status={run.status}") + print(f" revision : {run.revision}") + print(f" row_counts : {run.row_counts_json}") + if run.error_details: + print(f" error : {run.error_details}") + return 0 if run.status == "promoted" else 1 + + +def main() -> int: + args = _parse_args() + return asyncio.run(_apply() if args.apply else _dry_run(args.samples)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index 4e7191f..bbd71c0 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -411,3 +411,181 @@ async def test_discrepancy_in_shares_is_detected_and_reported(engine): assert k.shares_outstanding == 999.0 and k.import_run_id == 1 # immutable — not overwritten events = (await s.execute(select(SystemEvent).where(SystemEvent.code == "snapshot_discrepancy"))).scalars().all() assert len(events) == 1 and events[0].severity == "warning" + + +# --- reparse: rewriting rows a fixed parser reconstructs differently -------- + +# A 4-4-5 filer's YTD-Q3 span (36 weeks = 251 days). The old 20-day tolerance +# around 273 rejected it and stored revenue=None; 25 accepts it. Reparsing with +# the fixed parser is exactly the situation this mode exists for. +CF_Q3_445 = _rev("2025-09-01", "2026-05-10", 207431, 2026, "Q3", "Q3F") +SUB_445 = [_filing("Q3F", "10-Q", "2026-05-10", "2026-06-01", "2026-06-01T10:01:00.000Z")] + + +def _445_client(): + return FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_Q3_445], [_shares("2026-05-15", 100, "Q3F", 2026, "Q3")])}, + submissions={320193: _submissions(SUB_445)}, + latest_index=date(2026, 6, 1), + ) + + +async def _import_with_old_tolerance(engine, monkeypatch): + """Seed the DB the way the pre-fix parser did: Q3 revenue rejected -> null.""" + from app.services import sec_facts_parser + + monkeypatch.setattr(sec_facts_parser, "_YTD_TOLERANCE_DAYS", 20) + run = await run_import(_importer(_445_client(), today=date(2026, 6, 2)), engine=engine) + monkeypatch.undo() + return run + + +async def test_reparse_rewrites_rows_the_fixed_parser_reads_differently(engine, monkeypatch): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + first = await _import_with_old_tolerance(engine, monkeypatch) + + async with factory() as s: + stale = (await s.execute(select(FundamentalSnapshot))).scalar_one() + assert stale.revenue is None, "precondition: the old parser stored a null" + + # Reparse with the current (fixed) parser. force=True because SEC has not + # changed -- the staleness is on our side, so the revision gate would no-op. + run = await run_import( + SecFundamentalsImporter( + client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True + ), + engine=engine, + force=True, + ) + + assert run.status == STATUS_PROMOTED + assert '"updated": 1' in run.row_counts_json + async with factory() as s: + fixed = (await s.execute(select(FundamentalSnapshot))).scalar_one() + assert fixed.revenue == 207431 # rewritten in place + assert fixed.accession == stale.accession + assert fixed.import_run_id == run.id # rewrite is attributable + assert fixed.import_run_id != first.id + + +async def test_reparse_leaves_unchanged_rows_untouched(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + first = await run_import(_importer(_445_client(), today=date(2026, 6, 2)), engine=engine) + + run = await run_import( + SecFundamentalsImporter( + client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True + ), + engine=engine, + force=True, + ) + + assert '"updated": 0' in run.row_counts_json + async with factory() as s: + row = (await s.execute(select(FundamentalSnapshot))).scalar_one() + assert row.import_run_id == first.id # provenance preserved, no needless rewrite + + +async def test_without_reparse_a_differing_row_stays_immutable(engine, monkeypatch): + """The default contract is unchanged: report the discrepancy, never mutate.""" + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + await _import_with_old_tolerance(engine, monkeypatch) + + importer = SecFundamentalsImporter( + client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True + ) + async with _factory(engine)() as db: + await importer.detect_revision(db) + staged = await importer.stage(db) + importer.reparse = False # same staged diff, default disposition + counts = await importer.promote(db, staged, run_id=999) + await db.commit() + + assert staged.discrepancies, "the diff should still be detected and reported" + assert counts["updated"] == 0 + async with factory() as s: + row = (await s.execute(select(FundamentalSnapshot))).scalar_one() + assert row.revenue is None # untouched + + +async def test_force_bypasses_the_unchanged_revision_no_op(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + await run_import(_importer(_445_client(), today=date(2026, 6, 2)), engine=engine) + + same = _importer(_445_client(), today=date(2026, 6, 2)) + assert (await run_import(same, engine=engine)).status == "no_op" + + forced = SecFundamentalsImporter( + client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True + ) + assert (await run_import(forced, engine=engine, force=True)).status == STATUS_PROMOTED + + +# --- CIK resolution: successor registrants with no filings ----------------- + +async def test_issuer_with_no_xbrl_filings_is_reported_not_silent(engine): + """XOM resolved to CIK 2115436 'ExxonMobil Holdings Corp', which has zero + filings, so it produced no snapshots and nothing said why.""" + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + client = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K], [SH_K])}, + submissions={320193: {**_submissions([]), "name": "Shell Holdings Corp"}}, + latest_index=date(2026, 1, 31), + ) + importer = _importer(client) + async with factory() as db: + await importer.detect_revision(db) + staged = await importer.stage(db) + result = await importer.validate(db, staged) + + assert result.summary["no_xbrl_filings_count"] == 1 + assert staged.no_xbrl_filings[0]["cik"] == "0000320193" + assert staged.no_xbrl_filings[0]["name"] == "Shell Holdings Corp" + + +async def test_cik_override_pins_a_ticker_to_the_real_filer(engine): + from app.models.settings import SystemSetting + from app.services.sec_universe import CIK_OVERRIDES_KEY, resolve_ciks + + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + async with factory() as s: + s.add(SystemSetting(key=CIK_OVERRIDES_KEY, value='{"AAPL": 34088}')) + await s.commit() + + client = FakeSecClient( + tickers={"AAPL": 320193}, # SEC points at the wrong registrant + companyfacts={}, submissions={}, latest_index=date(2026, 1, 31), + ) + async with factory() as db: + resolved = await resolve_ciks(db, client) + + assert resolved.symbol_to_cik["AAPL"] == 34088 + assert resolved.cik_updates == [(1, "0000034088")] + + +async def test_malformed_cik_override_is_ignored_not_fatal(engine): + from app.models.settings import SystemSetting + from app.services.sec_universe import CIK_OVERRIDES_KEY, resolve_ciks + + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + async with factory() as s: + s.add(SystemSetting(key=CIK_OVERRIDES_KEY, value="not json at all")) + await s.commit() + + client = FakeSecClient( + tickers={"AAPL": 320193}, companyfacts={}, submissions={}, + latest_index=date(2026, 1, 31), + ) + async with factory() as db: + resolved = await resolve_ciks(db, client) + + assert resolved.symbol_to_cik["AAPL"] == 320193 # fell back to company_tickers