dennisthiessenandClaude Opus 5
1d4ed39fd2
Detection could retire an actively traded symbol — silently, since it then
vanishes from every signal. Three causes:
- Form 25 is filed per security class. An issuer removing its notes, preferred
or warrants files one while the common keeps trading. The filing's own
descriptionClassSecurity distinguishes them, so the primary document is now
fetched and read; anything not recognisably common equity is rejected, as is
anything unreadable (pre-2009 filings have no primary_doc.xml). Fail closed.
- Form 15 ends a reporting obligation and is no evidence trading stopped. The
whole family is dropped.
- A historical filing for a long-gone class could retire a symbol whose bars ran
years later, stamping the old date. Filings before the last bar (less a 30-day
lead for the exchange) are now ignored.
Rule 12d2-2 makes removal effective ten days after filing, so delisted_on is the
effective date rather than the filing date.
bootstrap_universe(prune_missing=True) still ran a cascading delete over
delisted rows, undoing the retention this branch exists for; it now skips them
and reports kept_delisted so the count is explicable.
clear_delisted had no route, which made "safe to automate because it is
reversible" false — reversal needed SQL. POST/DELETE /tickers/{symbol}/delisting
now mark and un-mark, giving an operator a non-destructive alternative to the
cascading DELETE that was the only option.
Shared-CIK siblings (GOOG/GOOGL) stay safe by construction: the probe is
per-symbol and gated on that symbol's own staleness, so a class that still
trades is never probed.
Not addressed: pruning a symbol merely dropped from the index still destroys its
history — the same survivorship problem in a different costume, needing a
tracked/membership state separate from delisting.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
511 lines
20 KiB
Python
511 lines
20 KiB
Python
"""Async SEC EDGAR client for the fundamentals importer (workstream A).
|
|
|
|
All access is batch (never at request time). This wraps the three SEC products
|
|
the A3 design uses — `company_tickers.json`, `submissions/`, `companyfacts/`, and
|
|
the daily filing index — behind one client that honors SEC's fair-access policy:
|
|
|
|
- an identifying ``User-Agent`` with a contact email on every request (config);
|
|
- request spacing well under the 10 req/s limit;
|
|
- exponential backoff + retry on 429;
|
|
- **403 → alert and stop** (raise ``SecForbiddenError``), never a retry-loop — a
|
|
403 means the UA or request pattern is wrong and retrying won't fix it. The one
|
|
exception is S3's ``AccessDenied`` on an ``/Archives/`` path, which is how the
|
|
bucket reports an absent file (``_is_absent_archive_key``).
|
|
|
|
Parsing lives here (index fixed-width, submissions pagination); DB writes and the
|
|
snapshot mapping live in the importer. No conditional GETs — the companyfacts
|
|
endpoint exposes no ETag/Last-Modified (verified), which is why the importer is
|
|
daily-index driven rather than polling archives.
|
|
"""
|
|
|
|
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
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
from app.exceptions import ProviderError
|
|
from app.services.earnings_alignment import normalise_symbol
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_WWW = "https://www.sec.gov"
|
|
_DATA = "https://data.sec.gov"
|
|
|
|
# Resolve CA bundle for explicit httpx verify (matches app/providers/alpaca.py).
|
|
_CA = os.environ.get("SSL_CERT_FILE", "")
|
|
_CA_VERIFY: str | bool = _CA if _CA and Path(_CA).exists() else True
|
|
|
|
_FORMS_10 = frozenset({"10-K", "10-Q", "10-K/A", "10-Q/A"})
|
|
|
|
# Notification of removal from listing. "25" is issuer-filed, "25-NSE" exchange-
|
|
# filed. The Form 15 family is deliberately absent: it ends a *reporting*
|
|
# obligation and does not mean the security stopped trading.
|
|
_DELISTING_FORMS = frozenset({"25", "25-NSE"})
|
|
|
|
# ``descriptionClassSecurity`` is free text ("Common Stock", "Class A Common
|
|
# Stock, $0.01 par value", "6.25% Notes due 2030", "Warrants", "Depositary
|
|
# Shares"). Only a common-equity class means the ticker itself stopped trading.
|
|
_NON_COMMON_CLASS = re.compile(
|
|
r"\b(note|bond|debenture|preferred|warrant|right|unit|depositary|"
|
|
r"subordinated|debt|trust)s?\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _is_common_stock(description: str) -> bool:
|
|
"""Does this Form 25 security class describe common equity?
|
|
|
|
Requires an explicit common-stock match AND no debt/preferred/warrant marker,
|
|
so "Depositary Shares each representing 1/1000th of Preferred" cannot pass on
|
|
the word "shares" alone. Unrecognised text is rejected — a symbol is retired
|
|
on this answer, so ambiguity must not read as yes.
|
|
"""
|
|
if _NON_COMMON_CLASS.search(description):
|
|
return False
|
|
return re.search(r"\bcommon\s+(stock|share)", description, re.IGNORECASE) is not None
|
|
|
|
|
|
class SecError(ProviderError):
|
|
"""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):
|
|
"""The resource does not exist (e.g. no daily index published for a day).
|
|
|
|
The *only* error a caller may treat as 'missing' — every other SecError
|
|
(fair-access rejection, exhausted retries, 5xx, timeout) must propagate so a
|
|
fetch failure is never mistaken for an empty result.
|
|
|
|
Raised for a 404, and for the one 403 that also means "absent": see
|
|
``_is_absent_archive_key``."""
|
|
|
|
|
|
def _is_absent_archive_key(url: str, resp: httpx.Response) -> bool:
|
|
"""True when a 403 means "this file does not exist", not "you are blocked".
|
|
|
|
``www.sec.gov/Archives`` is served straight out of an S3 bucket that grants
|
|
no ``s3:ListBucket``, so a missing key cannot be answered with 404 — S3
|
|
returns **403 with its ``AccessDenied`` XML** instead. SEC publishes a daily
|
|
index only for business days, so every weekend and market holiday inside an
|
|
incremental walk lands on exactly this response (verified 2026-07-30:
|
|
``form.20260725.idx``, a Saturday, 403s while the Friday and Monday files
|
|
return 200 on the same User-Agent).
|
|
|
|
A genuine fair-access rejection is distinguishable and must stay fatal: it is
|
|
SEC's WAF interstitial — ``text/html``, "Your Request Originates from an
|
|
Undeclared Automated Tool" — and it is returned for files that *do* exist,
|
|
on any path. Hence the narrow gate: the Archives prefix plus S3's own error
|
|
document. Nothing else may be downgraded to "missing"."""
|
|
try:
|
|
parsed = httpx.URL(url)
|
|
except (TypeError, ValueError): # pragma: no cover — url comes from us
|
|
return False
|
|
if parsed.host != "www.sec.gov" or not parsed.path.startswith("/Archives/"):
|
|
return False
|
|
if "xml" not in resp.headers.get("Content-Type", "").lower():
|
|
return False
|
|
try:
|
|
return "<Code>AccessDenied</Code>" in resp.text
|
|
except (UnicodeDecodeError, httpx.HTTPError): # pragma: no cover
|
|
return False
|
|
|
|
|
|
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)
|
|
|
|
|
|
class SecClient:
|
|
"""Fair-access SEC HTTP client. Use as ``async with SecClient() as c:``."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
user_agent: str | None = None,
|
|
spacing_seconds: float | None = None,
|
|
max_retries: int | None = None,
|
|
timeout: float | None = None,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
) -> None:
|
|
self._ua = user_agent or settings.sec_user_agent
|
|
self._spacing = (
|
|
spacing_seconds if spacing_seconds is not None else settings.sec_request_spacing_seconds
|
|
)
|
|
self._max_retries = (
|
|
max_retries if max_retries is not None else settings.sec_max_retries
|
|
)
|
|
self._timeout = timeout if timeout is not None else settings.sec_request_timeout_seconds
|
|
self._transport = transport # injectable for tests
|
|
self._client: httpx.AsyncClient | None = None
|
|
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,
|
|
verify=_CA_VERIFY,
|
|
transport=self._transport,
|
|
)
|
|
return self
|
|
|
|
async def __aexit__(self, *exc) -> None:
|
|
if self._client is not None:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
async def _throttle(self) -> None:
|
|
async with self._lock:
|
|
now = asyncio.get_event_loop().time()
|
|
wait = self._spacing - (now - self._last_request)
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
self._last_request = asyncio.get_event_loop().time()
|
|
|
|
async def _get(self, url: str) -> httpx.Response:
|
|
assert self._client is not None, "use `async with SecClient()`"
|
|
attempt = 0
|
|
while True:
|
|
await self._throttle()
|
|
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:
|
|
if _is_absent_archive_key(url, resp):
|
|
raise SecNotFoundError(f"SEC 403/AccessDenied (absent) for {url}")
|
|
raise SecForbiddenError(
|
|
f"SEC 403 for {url} — User-Agent/pattern rejected; set a real "
|
|
"sec_user_agent contact email"
|
|
)
|
|
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 {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 code >= 400:
|
|
raise SecError(f"SEC {code} for {url}")
|
|
return resp
|
|
|
|
async def get_json(self, url: str) -> Any:
|
|
return (await self._get(url)).json()
|
|
|
|
async def get_text(self, url: str) -> str:
|
|
return (await self._get(url)).text
|
|
|
|
# -- domain fetchers ---------------------------------------------------
|
|
|
|
async def company_tickers(self) -> dict[str, int]:
|
|
"""Map normalised ticker -> CIK (int). Multi-class tickers share a CIK."""
|
|
data = await self.get_json(f"{_WWW}/files/company_tickers.json")
|
|
out: dict[str, int] = {}
|
|
for row in data.values():
|
|
sym = normalise_symbol(row.get("ticker"))
|
|
if sym:
|
|
out[sym] = int(row["cik_str"])
|
|
return out
|
|
|
|
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. 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"])
|
|
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"),
|
|
"sic": base.get("sic"),
|
|
"sic_description": base.get("sicDescription"),
|
|
"fiscal_year_end": base.get("fiscalYearEnd"),
|
|
"tickers": base.get("tickers") or [],
|
|
"filings": filings,
|
|
}
|
|
|
|
async def delisting_filing(
|
|
self, cik: int | str, *, not_before: date | None = None
|
|
) -> dict[str, Any] | None:
|
|
"""Newest Form 25 removing this issuer's COMMON stock from listing.
|
|
|
|
Deliberately narrow, because the caller retires a symbol on the answer:
|
|
|
|
- **Form 25 only.** The Form 15 family terminates a reporting obligation
|
|
(often just a class falling under the holder threshold) and is no
|
|
evidence that trading stopped.
|
|
- **Class-checked.** Form 25 is filed per security class — an issuer
|
|
delisting its notes, preferred, warrants or an ADR class while the
|
|
common keeps trading files one too. The filing's own
|
|
``descriptionClassSecurity`` is what separates those, so the primary
|
|
document is fetched and read rather than trusting the form type.
|
|
- **``not_before``** rejects a historical filing for some long-gone
|
|
class. Without it a 2019 Form 25 would retire a symbol whose bars
|
|
stopped in 2026, and stamp 2019 as the date.
|
|
|
|
Anything unreadable — no primary document (pre-2009 filings have none),
|
|
malformed XML, unrecognised class — returns ``None``. Fail closed: the
|
|
caller keeps warning instead of retiring on a guess.
|
|
|
|
Reads ``filings.recent`` directly; ``submissions()`` keeps only the
|
|
10-K/10-Q family, so Form 25 never survives its parser.
|
|
"""
|
|
base = await self.get_json(f"{_DATA}/submissions/CIK{cik10(cik)}.json")
|
|
arrays = (base.get("filings") or {}).get("recent") or {}
|
|
forms = arrays.get("form") or []
|
|
dates = arrays.get("filingDate") or []
|
|
accessions = arrays.get("accessionNumber") or []
|
|
docs = arrays.get("primaryDocument") or []
|
|
|
|
candidates: list[tuple[date, str, str, str]] = []
|
|
for i, form in enumerate(forms):
|
|
if form not in _DELISTING_FORMS or i >= len(dates) or not dates[i]:
|
|
continue
|
|
try:
|
|
filed = date.fromisoformat(dates[i])
|
|
except ValueError:
|
|
continue
|
|
if not_before is not None and filed < not_before:
|
|
continue
|
|
if i >= len(accessions) or not accessions[i]:
|
|
continue
|
|
candidates.append((filed, form, accessions[i], docs[i] if i < len(docs) else ""))
|
|
|
|
for filed, form, accession, _doc in sorted(candidates, reverse=True):
|
|
security = await self._form25_security_class(cik, accession)
|
|
if security is None:
|
|
continue
|
|
if not _is_common_stock(security):
|
|
continue
|
|
return {
|
|
"form": form,
|
|
"filing_date": filed,
|
|
"security_class": security,
|
|
}
|
|
return None
|
|
|
|
async def _form25_security_class(
|
|
self, cik: int | str, accession: str
|
|
) -> str | None:
|
|
"""``descriptionClassSecurity`` from a Form 25's primary XML, or None.
|
|
|
|
The rendered ``primaryDocument`` is an XSL view of this file; the raw
|
|
``primary_doc.xml`` beside it is the structured original.
|
|
"""
|
|
folder = accession.replace("-", "")
|
|
url = (
|
|
f"{_WWW}/Archives/edgar/data/{int(cik)}/{folder}/primary_doc.xml"
|
|
)
|
|
try:
|
|
body = await self.get_text(url)
|
|
except SecNotFoundError:
|
|
return None
|
|
match = re.search(
|
|
r"<descriptionClassSecurity>(.*?)</descriptionClassSecurity>",
|
|
body,
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
if match is None:
|
|
return None
|
|
return " ".join(match.group(1).split()) or None
|
|
|
|
async def companyfacts(self, cik: int | str) -> dict[str, Any]:
|
|
"""Raw companyfacts JSON ({cik, entityName, facts})."""
|
|
return await self.get_json(f"{_DATA}/api/xbrl/companyfacts/CIK{cik10(cik)}.json")
|
|
|
|
async def latest_index_date(self, today: date | None = None) -> date | None:
|
|
"""The most recent published daily-index date (drives the revision). Checks
|
|
the current quarter, falling back to the previous one at a quarter boundary."""
|
|
today = today or date.today()
|
|
for year, qtr in _quarters_back(today, 2):
|
|
url = f"{_WWW}/Archives/edgar/daily-index/{year}/QTR{qtr}/index.json"
|
|
try:
|
|
idx = await self.get_json(url)
|
|
except SecNotFoundError:
|
|
continue # quarter dir absent — only 404 is "missing"
|
|
dates = [
|
|
d
|
|
for item in idx.get("directory", {}).get("item", [])
|
|
if (d := _index_file_date(item.get("name", ""))) is not None
|
|
and d <= today
|
|
]
|
|
if dates:
|
|
return max(dates)
|
|
return None
|
|
|
|
async def daily_index(self, day: date) -> list[dict[str, Any]]:
|
|
"""Parse the daily form index into 10-K/10-Q(/A) rows for all issuers.
|
|
|
|
Returns [{form, cik, accession, company}]. The caller filters to the
|
|
tracked universe. A missing index (weekend/holiday/not-yet-published)
|
|
returns [] rather than raising.
|
|
"""
|
|
qtr = (day.month - 1) // 3 + 1
|
|
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 SecNotFoundError:
|
|
# Absent on a weekend is routine (SEC publishes business days only); on a
|
|
# weekday it is either a market holiday or something worth a look — a SEC
|
|
# hiccup, or a rejection page misread as absent, would otherwise let the
|
|
# importer advance past real filings silently. Log-level only, no alert:
|
|
# cheaper than carrying a holiday calendar just to stay quiet ~10 days/yr.
|
|
logger.log(
|
|
logging.INFO if day.weekday() >= 5 else logging.WARNING,
|
|
"no daily index published for %s (%s)",
|
|
day,
|
|
f"{day:%a}",
|
|
)
|
|
return [] # weekend/holiday/not-yet-published; other errors propagate
|
|
return _parse_form_index(text)
|
|
|
|
|
|
def _rows_from_arrays(arrays: dict[str, list]) -> list[dict[str, Any]]:
|
|
"""Turn SEC's parallel-array filing block into row dicts (keeping only 10-K/10-Q
|
|
family filings — the ones that carry XBRL fundamentals)."""
|
|
forms = arrays.get("form", [])
|
|
out: list[dict[str, Any]] = []
|
|
for i, form in enumerate(forms):
|
|
if form not in _FORMS_10:
|
|
continue
|
|
out.append(
|
|
{
|
|
"accession": arrays["accessionNumber"][i],
|
|
"form": form,
|
|
"report_date": arrays["reportDate"][i] or None,
|
|
"filing_date": arrays["filingDate"][i] or None,
|
|
"acceptance_datetime": arrays["acceptanceDateTime"][i] or None,
|
|
"is_xbrl": bool(arrays.get("isXBRL", [0] * len(forms))[i]),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def _parse_form_index(text: str) -> list[dict[str, Any]]:
|
|
"""Parse a daily ``form.YYYYMMDD.idx`` (fixed columns: Form / Company / CIK /
|
|
Date Filed / File Name-with-accession)."""
|
|
rows: list[dict[str, Any]] = []
|
|
started = False
|
|
for line in text.splitlines():
|
|
if not started:
|
|
if set(line.strip()) == {"-"}: # the dashed separator row
|
|
started = True
|
|
continue
|
|
parts = line.split()
|
|
if len(parts) < 5:
|
|
continue
|
|
form = parts[0]
|
|
if form not in _FORMS_10:
|
|
continue
|
|
path = parts[-1] # edgar/data/<cik>/<accession>.txt
|
|
cik = _cik_from_path(path)
|
|
accession = _accession_from_path(path)
|
|
if cik is None or accession is None:
|
|
continue
|
|
rows.append({"form": form, "cik": cik, "accession": accession, "path": path})
|
|
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:
|
|
return datetime.strptime(name[5:13], "%Y%m%d").date()
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _quarters_back(today: date, n: int) -> list[tuple[int, int]]:
|
|
"""(year, quarter) for `today`'s quarter and the previous n-1, newest first."""
|
|
q = (today.month - 1) // 3 + 1
|
|
out = []
|
|
y = today.year
|
|
for _ in range(n):
|
|
out.append((y, q))
|
|
q -= 1
|
|
if q == 0:
|
|
q = 4
|
|
y -= 1
|
|
return out
|
|
|
|
|
|
def _cik_from_path(path: str) -> int | None:
|
|
segs = path.split("/")
|
|
if len(segs) >= 3 and segs[2].isdigit():
|
|
return int(segs[2])
|
|
return None
|
|
|
|
|
|
def _accession_from_path(path: str) -> str | None:
|
|
stem = path.rsplit("/", 1)[-1]
|
|
if stem.endswith(".txt"):
|
|
stem = stem[:-4]
|
|
return stem or None
|