From 5939be7b7fe08e59661078308d9bcb6dcec56a82 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Wed, 22 Jul 2026 14:49:33 +0200 Subject: [PATCH] =?UTF-8?q?fix(sec):=20A3=20slice-1=20review=20=E2=80=94?= =?UTF-8?q?=20read-only=20resolution,=20error=20propagation,=20fair-access?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/services/sec_client.py | 105 +++++++++++++++++++------ app/services/sec_universe.py | 132 ++++++++++++++++++++------------ tests/unit/test_sec_client.py | 68 +++++++++++++++- tests/unit/test_sec_universe.py | 57 +++++++++----- 4 files changed, 267 insertions(+), 95 deletions(-) diff --git a/app/services/sec_client.py b/app/services/sec_client.py index b25cf79..0f1bb9e 100644 --- a/app/services/sec_client.py +++ b/app/services/sec_client.py @@ -21,6 +21,7 @@ from __future__ import annotations import asyncio import logging import os +import re from datetime import date, datetime from pathlib import Path from typing import Any @@ -44,13 +45,31 @@ _FORMS_10 = frozenset({"10-K", "10-Q", "10-K/A", "10-Q/A"}) class SecError(ProviderError): - """SEC request failed.""" + """SEC request failed (403, exhausted 429/5xx, timeout, transport, parse).""" class SecForbiddenError(SecError): """SEC returned 403 — User-Agent/pattern rejected. Alert and stop.""" +class SecNotFoundError(SecError): + """SEC returned 404 — the resource does not exist (e.g. no index for a day). + + The *only* error a caller may treat as 'missing' — every other SecError + (403, exhausted retries, 5xx, timeout) must propagate so a fetch failure is + never mistaken for an empty result.""" + + +def _looks_like_contact_email(ua: str) -> bool: + if "example.com" in ua.lower() or "set-a-real-email" in ua.lower(): + return False + return re.search(r"[^@\s]+@[^@\s]+\.[^@\s]+", ua) is not None + + +# SEC asks callers to stay well under 10 req/s; enforce a floor on real clients. +_MIN_PROD_SPACING = 0.11 + + def cik10(cik: int | str) -> str: """Zero-pad a CIK to the 10-digit form SEC URLs use (320193 -> 0000320193).""" return str(int(cik)).zfill(10) @@ -81,7 +100,24 @@ class SecClient: self._lock = asyncio.Lock() self._last_request = 0.0 + def _validate_fair_access(self) -> None: + """On a real (non-mocked) client, enforce SEC fair-access preconditions + so we can't accidentally hammer SEC or get 403'd: a genuine contact-email + User-Agent and a spacing floor. Mock transports skip this (tests use 0).""" + if not _looks_like_contact_email(self._ua): + raise SecError( + "sec_user_agent must contain a real contact email (got " + f"{self._ua!r}) — SEC fair-access requires it" + ) + if self._spacing < _MIN_PROD_SPACING: + raise SecError( + f"sec_request_spacing_seconds {self._spacing} is below the " + f"{_MIN_PROD_SPACING}s fair-access floor" + ) + async def __aenter__(self) -> "SecClient": + if self._transport is None: + self._validate_fair_access() self._client = httpx.AsyncClient( headers={"User-Agent": self._ua, "Accept-Encoding": "gzip, deflate"}, timeout=self._timeout, @@ -108,22 +144,34 @@ class SecClient: attempt = 0 while True: await self._throttle() - resp = await self._client.get(url) - if resp.status_code == 403: + try: + resp = await self._client.get(url) + except (httpx.TimeoutException, httpx.TransportError) as exc: + attempt += 1 + if attempt > self._max_retries: + raise SecError(f"SEC network error for {url}: {exc}") from exc + await asyncio.sleep(min(2.0**attempt, 30.0)) + continue + + code = resp.status_code + if code == 403: raise SecForbiddenError( f"SEC 403 for {url} — User-Agent/pattern rejected; set a real " "sec_user_agent contact email" ) - if resp.status_code == 429: + if code == 404: + raise SecNotFoundError(f"SEC 404 for {url}") + # 429 and 5xx are transient — retry with backoff, honoring Retry-After. + if code == 429 or 500 <= code < 600: attempt += 1 if attempt > self._max_retries: - raise SecError(f"SEC 429 after {self._max_retries} retries: {url}") - backoff = min(2.0**attempt, 30.0) - logger.warning("SEC 429 for %s — backoff %.1fs (attempt %d)", url, backoff, attempt) - await asyncio.sleep(backoff) + raise SecError(f"SEC {code} after {self._max_retries} retries: {url}") + delay = _retry_after_seconds(resp) or min(2.0**attempt, 30.0) + logger.warning("SEC %d for %s — backoff %.1fs (attempt %d)", code, url, delay, attempt) + await asyncio.sleep(delay) continue - if resp.status_code >= 400: - raise SecError(f"SEC {resp.status_code} for {url}") + if code >= 400: + raise SecError(f"SEC {code} for {url}") return resp async def get_json(self, url: str) -> Any: @@ -144,18 +192,20 @@ class SecClient: out[sym] = int(row["cik_str"]) return out - async def submissions(self, cik: int | str) -> dict[str, Any]: - """Issuer metadata + the FULL merged filing history. + async def submissions(self, cik: int | str, *, include_history: bool = False) -> dict[str, Any]: + """Issuer metadata + filing list. ``filings.recent`` caps at 1000; older accessions live in - ``filings.files[]`` shards. This merges them so backfill sees every - filing's reportDate / acceptanceDateTime / isXBRL. + ``filings.files[]`` shards. Only ``include_history=True`` (the one-time + full backfill) fetches those shards — SIC refresh and incremental runs + use the recent list alone and make no extra requests. """ base = await self.get_json(f"{_DATA}/submissions/CIK{cik10(cik)}.json") filings = _rows_from_arrays(base["filings"]["recent"]) - for shard in base["filings"].get("files") or []: - shard_data = await self.get_json(f"{_DATA}/submissions/{shard['name']}") - filings.extend(_rows_from_arrays(shard_data)) + if include_history: + for shard in base["filings"].get("files") or []: + shard_data = await self.get_json(f"{_DATA}/submissions/{shard['name']}") + filings.extend(_rows_from_arrays(shard_data)) return { "cik": int(base["cik"]), "name": base.get("name"), @@ -178,8 +228,8 @@ class SecClient: url = f"{_WWW}/Archives/edgar/daily-index/{year}/QTR{qtr}/index.json" try: idx = await self.get_json(url) - except SecError: - continue + except SecNotFoundError: + continue # quarter dir absent — only 404 is "missing" dates = [ d for item in idx.get("directory", {}).get("item", []) @@ -201,9 +251,9 @@ class SecClient: url = f"{_WWW}/Archives/edgar/daily-index/{day.year}/QTR{qtr}/form.{day:%Y%m%d}.idx" try: text = await self.get_text(url) - except SecError as exc: - logger.info("no daily index for %s (%s)", day, exc) - return [] + except SecNotFoundError: + logger.info("no daily index for %s (404)", day) + return [] # weekend/holiday/not-yet-published; other errors propagate return _parse_form_index(text) @@ -252,6 +302,17 @@ def _parse_form_index(text: str) -> list[dict[str, Any]]: return rows +def _retry_after_seconds(resp: httpx.Response) -> float | None: + """Parse a numeric-seconds Retry-After header (SEC uses seconds), capped.""" + raw = resp.headers.get("Retry-After") + if not raw: + return None + try: + return min(float(raw), 60.0) + except (TypeError, ValueError): + return None + + def _index_file_date(name: str) -> date | None: if name.startswith("form.") and name.endswith(".idx"): try: diff --git a/app/services/sec_universe.py b/app/services/sec_universe.py index 15d4596..9d80bd0 100644 --- a/app/services/sec_universe.py +++ b/app/services/sec_universe.py @@ -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)}" diff --git a/tests/unit/test_sec_client.py b/tests/unit/test_sec_client.py index f98d538..dc5d78b 100644 --- a/tests/unit/test_sec_client.py +++ b/tests/unit/test_sec_client.py @@ -88,9 +88,9 @@ async def test_company_tickers_normalised_and_multiclass(): assert m["BRK-B"] == 1067983 # dash form -async def test_submissions_merges_shards_and_filters_forms(): +async def test_submissions_history_merges_shards_and_filters_forms(): async with _client() as c: - sub = await c.submissions(320193) + sub = await c.submissions(320193, include_history=True) assert sub["sic"] == "3571" and sub["fiscal_year_end"] == "0926" accns = {f["accession"] for f in sub["filings"]} # 10-Q + 10-K from recent, 10-Q from the shard; the 8-K is filtered out @@ -99,6 +99,20 @@ async def test_submissions_merges_shards_and_filters_forms(): assert older["report_date"] == "1993-12-31" and older["is_xbrl"] is False +async def test_submissions_recent_only_skips_shard_requests(): + seen = [] + + def handler(request): + seen.append(str(request.url)) + return _handler(request) + + async with SecClient(transport=httpx.MockTransport(handler), spacing_seconds=0) as c: + sub = await c.submissions(320193) # include_history defaults False + assert not any("submissions-001" in u for u in seen) # no shard fetch + accns = {f["accession"] for f in sub["filings"]} + assert accns == {"0000320193-26-000013", "0000320193-26-000006"} # recent only + + async def test_daily_index_parses_10kq_rows(): async with _client() as c: rows = await c.daily_index(date(2026, 7, 21)) @@ -157,3 +171,53 @@ async def test_429_gives_up_after_max_retries(monkeypatch): async with SecClient(transport=httpx.MockTransport(handler), spacing_seconds=0, max_retries=2) as c: with pytest.raises(SecError): await c.get_json("https://data.sec.gov/z") + + +def _status_client(status): + def handler(request): + return httpx.Response(status) + # max_retries=0 so 5xx/429 raise immediately (no retry sleeps) + return SecClient(transport=httpx.MockTransport(handler), spacing_seconds=0, max_retries=0) + + +async def test_index_methods_propagate_403(): + # A 403 must NOT be mistaken for "no index". + async with _status_client(403) as c: + with pytest.raises(sc.SecForbiddenError): + await c.daily_index(date(2026, 7, 21)) + async with _status_client(403) as c: + with pytest.raises(sc.SecForbiddenError): + await c.latest_index_date(today=date(2026, 7, 22)) + + +async def test_index_methods_propagate_500(): + async with _status_client(500) as c: + with pytest.raises(SecError): + await c.daily_index(date(2026, 7, 21)) + async with _status_client(500) as c: + with pytest.raises(SecError): + await c.latest_index_date(today=date(2026, 7, 22)) + + +async def test_only_404_is_treated_as_missing(): + async with _status_client(404) as c: + assert await c.daily_index(date(2026, 7, 21)) == [] + assert await c.latest_index_date(today=date(2026, 7, 22)) is None + + +async def test_fair_access_validation_on_real_client(): + # Placeholder email rejected. + with pytest.raises(SecError): + async with SecClient(user_agent="signal-platform (contact: you@example.com)"): + pass + # Non-email UA rejected. + with pytest.raises(SecError): + async with SecClient(user_agent="signal-platform"): + pass + # Valid UA but unsafe production spacing rejected. + with pytest.raises(SecError): + async with SecClient(user_agent="signal-platform real@person.io", spacing_seconds=0.0): + pass + # Valid UA + safe spacing opens fine. + async with SecClient(user_agent="signal-platform real@person.io", spacing_seconds=0.2): + pass diff --git a/tests/unit/test_sec_universe.py b/tests/unit/test_sec_universe.py index 69f85fb..170b719 100644 --- a/tests/unit/test_sec_universe.py +++ b/tests/unit/test_sec_universe.py @@ -1,4 +1,5 @@ -"""Tests for CIK/SIC resolution and the composite-revision fingerprint.""" +"""Tests for CIK/SIC resolution (read-only), apply-in-promote, and the +composite-revision fingerprint.""" from __future__ import annotations @@ -40,11 +41,11 @@ class FakeSecClient: async def company_tickers(self): return dict(self._tickers) - async def submissions(self, cik): + async def submissions(self, cik, *, include_history=False): return self._submissions[int(cik)] -async def test_resolve_ciks_sets_cik_and_returns_mapping(factory): +async def test_resolve_ciks_is_read_only_and_proposes_updates(factory): async with factory() as s: for sym in ["AAPL", "GOOGL", "GOOG", "ZZZZ"]: # ZZZZ not in SEC s.add(Ticker(symbol=sym)) @@ -52,35 +53,47 @@ async def test_resolve_ciks_sets_cik_and_returns_mapping(factory): client = FakeSecClient({"AAPL": 320193, "GOOGL": 1652044, "GOOG": 1652044}) async with factory() as s: - mapping = await su.resolve_ciks(s, client) - await s.commit() + resolved = await su.resolve_ciks(s, client) + assert not s.dirty and not s.new # NOTHING mutated during resolution + await s.rollback() - assert mapping == {"AAPL": 320193, "GOOGL": 1652044, "GOOG": 1652044} + assert resolved.symbol_to_cik == {"AAPL": 320193, "GOOGL": 1652044, "GOOG": 1652044} + assert len(resolved.cik_updates) == 3 # AAPL, GOOGL, GOOG (ZZZZ unresolved) + assert set(resolved.cik_to_ticker_ids) == {320193, 1652044} + + # Read-only really means the DB is untouched until apply. async with factory() as s: ciks = {t.symbol: t.cik for t in (await s.execute(select(Ticker))).scalars()} - assert ciks["AAPL"] == "0000320193" - assert ciks["GOOGL"] == ciks["GOOG"] == "0001652044" # multi-class share CIK - assert ciks["ZZZZ"] is None # unresolved stays null + assert all(v is None for v in ciks.values()) -async def test_refresh_sic_updates_all_tickers_of_a_cik(factory): +async def test_apply_ticker_updates_writes_cik_and_sic(factory): async with factory() as s: for sym in ["GOOGL", "GOOG"]: - s.add(Ticker(symbol=sym, cik="0001652044")) + s.add(Ticker(symbol=sym)) await s.commit() client = FakeSecClient( - {}, submissions={1652044: {"sic": "7370", "sic_description": "Services-Computer"}} + {"GOOGL": 1652044, "GOOG": 1652044}, + submissions={1652044: {"sic": "7370", "sic_description": "Services-Computer"}}, ) async with factory() as s: - n = await su.refresh_sic(s, client, [1652044]) + resolved = await su.resolve_ciks(s, client) + sic_updates = await su.fetch_sic_updates(client, resolved.cik_to_ticker_ids) + counts = await su.apply_ticker_updates(s, resolved, sic_updates) await s.commit() - assert n == 1 + assert counts == {"cik_updates": 2, "sic_updates": 2} async with factory() as s: - rows = {t.symbol: (t.sic, t.sic_description) for t in (await s.execute(select(Ticker))).scalars()} - assert rows["GOOGL"] == ("7370", "Services-Computer") - assert rows["GOOG"] == ("7370", "Services-Computer") + rows = {t.symbol: (t.cik, t.sic, t.sic_description) for t in (await s.execute(select(Ticker))).scalars()} + assert rows["GOOGL"] == ("0001652044", "7370", "Services-Computer") + assert rows["GOOG"] == ("0001652044", "7370", "Services-Computer") + + +async def test_fetch_sic_updates_is_read_only(factory): + client = FakeSecClient({}, submissions={1: {"sic": "1", "sic_description": "x"}}) + updates = await su.fetch_sic_updates(client, {1: [10, 11]}) + assert updates == [(10, "1", "x"), (11, "1", "x")] # proposals only, no DB touched def test_universe_fingerprint_changes_on_membership(): @@ -93,11 +106,13 @@ def test_universe_fingerprint_changes_on_membership(): assert a != remapped # changed CIK mapping forces a new revision +def test_compose_revision_rejects_missing_index_date(): + with pytest.raises(ValueError): + su.compose_revision(None, "abc", {"AAPL": 320193}) + + def test_compose_revision_and_index_hash(): - rows = [ - {"cik": 320193, "accession": "a-1"}, - {"cik": 66740, "accession": "b-2"}, - ] + rows = [{"cik": 320193, "accession": "a-1"}, {"cik": 66740, "accession": "b-2"}] h1 = su.index_content_hash(rows) h2 = su.index_content_hash(list(reversed(rows))) assert h1 == h2 # order-independent