signal-platform/app/services/sec_universe.py

"""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 logging
from dataclasses import dataclass, field
from typing import Iterable

from sqlalchemy import select, update

from app.models.ticker import Ticker
from app.services.earnings_alignment import normalise_symbol
from app.services.sec_client import SecClient

logger = logging.getLogger(__name__)


@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()
    rows = (await db.execute(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 = 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 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)}"