fix(sec): A3 slice-1 review — read-only resolution, error propagation, fair-access
Addresses the slice-1 review: 1. Resolution is now read-only (A1 transaction contract). resolve_ciks / fetch_sic_updates compute proposals and mutate nothing; a new apply_ticker_updates issues the writes, called only in promote — so a failed validation can't leak ticker changes on the framework's failure commit. 2. Only 404 means "missing". Added SecNotFoundError; daily_index / latest_index_date catch only that. 403, exhausted 429, 5xx, timeouts, and transport/parse errors now propagate instead of looking like "no index". 3. Fair-access enforced when opening a REAL client (transport=None): reject blank/placeholder/non-email User-Agent and sub-0.11s spacing. Mock transports skip it (tests use 0 spacing). 4. submissions(include_history=False) by default — only the one-time full backfill fetches the history shards; SIC/incremental work makes no extra requests. Plus: retry transient 5xx/network errors and honor Retry-After during the 1 GB backfill; compose_revision rejects a missing index date (no "None:..." revision). Re-verified live vs real SEC (fair-access validation passes, shard merge intact). Tests: 18 (added error propagation, read-only resolution, fair-access, recent-only submissions, reject-None revision). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1,19 +1,26 @@
|
||||
"""Tracked-universe CIK/SIC resolution and the SEC importer's composite revision.
|
||||
|
||||
Resolves the app's tracked tickers to SEC issuers (CIK) and back-fills
|
||||
``tickers.cik/sic/sic_description``. Also builds the **universe fingerprint** that
|
||||
goes into the importer's composite revision, so that 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).
|
||||
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
|
||||
from sqlalchemy import select, update
|
||||
|
||||
from app.models.ticker import Ticker
|
||||
from app.services.earnings_alignment import normalise_symbol
|
||||
@@ -22,47 +29,72 @@ from app.services.sec_client import SecClient
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def resolve_ciks(db, client: SecClient) -> dict[str, int]:
|
||||
"""Resolve tracked tickers to CIKs via company_tickers.json and persist
|
||||
``tickers.cik`` where it changed. Returns {normalised symbol: cik} for the
|
||||
tracked tickers that resolved (multi-class tickers share a CIK)."""
|
||||
ticker_to_cik = await client.company_tickers()
|
||||
rows = (await db.execute(select(Ticker))).scalars().all()
|
||||
@dataclass
|
||||
class ResolvedUniverse:
|
||||
"""Read-only result of CIK resolution. `cik_updates` are proposed writes
|
||||
(ticker_id → new cik string) applied later in promote."""
|
||||
|
||||
resolved: dict[str, int] = {}
|
||||
changed = 0
|
||||
for t in rows:
|
||||
if not t.symbol:
|
||||
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(t.symbol)
|
||||
sym = normalise_symbol(symbol)
|
||||
cik = ticker_to_cik.get(sym)
|
||||
if cik is None:
|
||||
continue # e.g. ADRs / non-SEC issuers — snapshots simply absent
|
||||
resolved[sym] = cik
|
||||
cik_str = f"{cik:010d}"
|
||||
if t.cik != cik_str:
|
||||
t.cik = cik_str
|
||||
changed += 1
|
||||
logger.info("resolve_ciks: %d tracked resolved, %d cik updates", len(resolved), changed)
|
||||
return resolved
|
||||
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 refresh_sic(db, client: SecClient, ciks: Iterable[int]) -> int:
|
||||
"""Fetch submissions for the given CIKs and set sic/sic_description on every
|
||||
tracked ticker sharing each CIK. Returns the number of CIKs refreshed. Callers
|
||||
pass only the CIKs that need it (e.g. those still missing a SIC) to stay light."""
|
||||
refreshed = 0
|
||||
for cik in sorted(set(int(c) for c in ciks)):
|
||||
sub = await client.submissions(cik)
|
||||
cik_str = f"{cik:010d}"
|
||||
rows = (
|
||||
await db.execute(select(Ticker).where(Ticker.cik == cik_str))
|
||||
).scalars().all()
|
||||
for t in rows:
|
||||
t.sic = str(sub["sic"]) if sub.get("sic") else None
|
||||
t.sic_description = sub.get("sic_description")
|
||||
refreshed += 1
|
||||
return refreshed
|
||||
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:
|
||||
@@ -72,16 +104,16 @@ def universe_fingerprint(symbol_to_cik: dict[str, int]) -> str:
|
||||
return hashlib.blake2b(canonical.encode("utf-8"), digest_size=12).hexdigest()
|
||||
|
||||
|
||||
def compose_revision(
|
||||
index_date, index_content_hash: str, symbol_to_cik: dict[str, int]
|
||||
) -> str:
|
||||
"""The importer's composite revision: latest processed index date + a hash of
|
||||
the index content consumed this run + the universe fingerprint. Equal revision
|
||||
across runs ⇒ nothing new to import ⇒ no_op."""
|
||||
return f"{index_date}:{index_content_hash}:{universe_fingerprint(symbol_to_cik)}"
|
||||
|
||||
|
||||
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)}"
|
||||
|
||||
Reference in New Issue
Block a user