"""Tracked-universe CIK/SIC resolution and the SEC importer's composite revision. Resolves the app's tracked tickers to SEC issuers (CIK) and prepares ``tickers.cik/sic/sic_description`` back-fills. Also builds the **universe fingerprint** in the importer's composite revision, so adding a ticker changes the revision and forces a run instead of being ``no_op``'d away or starved waiting for its issuer to file (A3 design, Decision 1 review fix). **Transaction contract:** resolution is read-only — `resolve_ciks` and `fetch_sic_updates` compute *proposed* updates and mutate nothing. They run in the importer's `stage` (which must not write, or a failed validation would leak changes on the framework's failure commit). The proposals are applied only in `promote`, via `apply_ticker_updates`, atomically with the snapshot inserts. """ from __future__ import annotations import hashlib import json import logging from dataclasses import dataclass, field from typing import Iterable from sqlalchemy import select, update from app.models.ticker import Ticker from app.services import settings_store, ticker_service 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: """Read-only result of CIK resolution. `cik_updates` are proposed writes (ticker_id → new cik string) applied later in promote.""" symbol_to_cik: dict[str, int] = field(default_factory=dict) cik_to_ticker_ids: dict[int, list[int]] = field(default_factory=dict) cik_updates: list[tuple[int, str]] = field(default_factory=list) 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( ticker_service.active_only(select(Ticker.id, Ticker.symbol, Ticker.cik)) ) ).all() result = ResolvedUniverse() for tid, symbol, current_cik in rows: if not symbol: continue sym = normalise_symbol(symbol) 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 result.cik_to_ticker_ids.setdefault(cik, []).append(tid) if current_cik != f"{cik:010d}": result.cik_updates.append((tid, f"{cik:010d}")) logger.info( "resolve_ciks: %d resolved, %d proposed cik updates", len(result.symbol_to_cik), len(result.cik_updates), ) 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]]: """Fetch SIC for each CIK (recent-only submissions, no history shards) and return proposed `(ticker_id, sic, sic_description)` writes. **Read-only** — no DB mutation. Callers pass only the CIKs that need it (e.g. missing a SIC).""" updates: list[tuple[int, str | None, str | None]] = [] for cik, ticker_ids in cik_to_ticker_ids.items(): sub = await client.submissions(cik, include_history=False) sic = str(sub["sic"]) if sub.get("sic") else None desc = sub.get("sic_description") for tid in ticker_ids: updates.append((tid, sic, desc)) return updates async def apply_ticker_updates( db, resolved: ResolvedUniverse, sic_updates: list[tuple[int, str | None, str | None]] | None = None, ) -> dict[str, int]: """Apply the proposed cik / sic writes. **The only writer** — call inside promote so it commits atomically with the snapshot inserts.""" for tid, cik in resolved.cik_updates: await db.execute(update(Ticker).where(Ticker.id == tid).values(cik=cik)) for tid, sic, desc in sic_updates or []: await db.execute( update(Ticker).where(Ticker.id == tid).values(sic=sic, sic_description=desc) ) return {"cik_updates": len(resolved.cik_updates), "sic_updates": len(sic_updates or [])} def universe_fingerprint(symbol_to_cik: dict[str, int]) -> str: """Stable short hash of the tracked symbol->CIK set. Changes whenever a ticker is added/removed or its CIK mapping changes.""" canonical = ";".join(f"{sym}:{cik}" for sym, cik in sorted(symbol_to_cik.items())) return hashlib.blake2b(canonical.encode("utf-8"), digest_size=12).hexdigest() def index_content_hash(index_rows: Iterable[dict]) -> str: """Order-independent hash of the tracked index accessions consumed this run.""" keys = sorted(f"{r['cik']}/{r['accession']}" for r in index_rows) return hashlib.blake2b("|".join(keys).encode("utf-8"), digest_size=12).hexdigest() def compose_revision(index_date, content_hash: str, symbol_to_cik: dict[str, int]) -> str: """Composite revision = processed index date + index-content hash + universe fingerprint. Equal across runs ⇒ nothing new ⇒ no_op. Rejects a missing index date rather than emitting a `None:...` revision that could false-match.""" if index_date is None: raise ValueError("compose_revision requires a non-null index date") return f"{index_date}:{content_hash}:{universe_fingerprint(symbol_to_cik)}"