1"""Instrument master and identifier crosswalk: FinData entity/instrument ↔ ticker ↔ SEC CIK/SIC.23SEC data are public; requests carry the fair-access User-Agent (no secret) and are throttled to <= 8 req/s.4"""5from __future__ import annotations67import gzip8import json9import os10import time11from pathlib import Path1213import httpx14import polars as pl1516SEC_TICKERS_URL = "https://www.sec.gov/files/company_tickers.json"17SEC_SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik:010d}.json"1819# SIC divisions (US Department of Labor): used as the coarse "sector" for fixed effects and demeaning.20SIC_DIVISIONS = [21 (100, 999, "A_agriculture"), (1000, 1499, "B_mining"), (1500, 1799, "C_construction"), (2000, 3999, "D_manufacturing"),22 (4000, 4999, "E_transport_utilities"), (5000, 5199, "F_wholesale"), (5200, 5999, "G_retail"),23 (6000, 6799, "H_finance_insurance_realestate"), (7000, 8999, "I_services"), (9100, 9729, "J_public_admin"),24]252627def sic_division(sic: int | None) -> str:28 if sic is None:29 return "unknown"30 for lo, hi, name in SIC_DIVISIONS:31 if lo <= sic <= hi:32 return name33 return "unknown"343536def _ua() -> str:37 return os.environ.get("SEC_USER_AGENT", "UQO WP39 research (mailto:simon-pierre.boucher@uqo.ca)")383940def _get_json(client: httpx.Client, url: str) -> dict:41 r = client.get(url, headers={"User-Agent": _ua(), "Accept-Encoding": "gzip, deflate"})42 r.raise_for_status()43 raw = r.content44 if raw[:2] == b"\x1f\x8b":45 raw = gzip.decompress(raw)46 return json.loads(raw)474849def fetch_sec_tickers(cache_dir: Path) -> pl.DataFrame:50 """Current SEC ticker→CIK table (snapshot cached with its download date)."""51 cache_dir.mkdir(parents=True, exist_ok=True)52 f = cache_dir / "sec_company_tickers.json"53 if not f.exists():54 with httpx.Client(timeout=60) as c:55 d = _get_json(c, SEC_TICKERS_URL)56 f.write_text(json.dumps({"downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "data": d}))57 d = json.loads(f.read_text())["data"]58 rows = [{"ticker": v["ticker"].upper(), "cik": int(v["cik_str"]), "sec_name": v["title"]} for v in d.values()]59 return pl.DataFrame(rows)606162def fetch_sec_submissions(ciks: list[int], cache_dir: Path, rate_per_s: float = 8.0, log=None, workers: int = 6) -> pl.DataFrame:63 """SIC code, exchanges, fiscal year end and former names per CIK (cached per CIK; global token-bucket rate limit)."""64 import threading65 from concurrent.futures import ThreadPoolExecutor6667 out_dir = cache_dir / "sec_submissions"68 out_dir.mkdir(parents=True, exist_ok=True)69 lock, state = threading.Lock(), {"last": 0.0}70 min_gap = 1.0 / rate_per_s7172 def fetch_one(cik: int) -> None:73 f = out_dir / f"CIK{cik:010d}.json"74 if f.exists():75 return76 with lock:77 wait = min_gap - (time.time() - state["last"])78 if wait > 0:79 time.sleep(wait)80 state["last"] = time.time()81 with httpx.Client(timeout=60) as c:82 try:83 d = _get_json(c, SEC_SUBMISSIONS_URL.format(cik=cik))84 except httpx.HTTPStatusError as exc:85 d = {"error": exc.response.status_code}86 except Exception as exc: # noqa: BLE00187 d = {"error": type(exc).__name__}88 slim = {k: d.get(k) for k in ("cik", "sic", "sicDescription", "tickers", "exchanges", "fiscalYearEnd",89 "formerNames", "name", "stateOfIncorporation", "error")}90 f.write_text(json.dumps(slim))9192 with ThreadPoolExecutor(max_workers=workers) as ex:93 for i, _ in enumerate(ex.map(fetch_one, ciks)):94 if log and i % 500 == 0:95 log.info("SEC submissions %d/%d", i, len(ciks))96 rows = []97 for cik in ciks:98 d = json.loads((out_dir / f"CIK{cik:010d}.json").read_text())99 if d.get("error") and not d.get("sic"):100 (out_dir / f"CIK{cik:010d}.json").unlink(missing_ok=True) if d.get("error") not in (404,) else None101 sic = int(d["sic"]) if d.get("sic") not in (None, "") else None102 rows.append({"cik": cik, "sic": sic, "sic_description": d.get("sicDescription"),103 "sic_division": sic_division(sic), "sec_tickers": d.get("tickers") or [],104 "exchanges": d.get("exchanges") or [], "fiscal_year_end": d.get("fiscalYearEnd"),105 "n_former_names": len(d.get("formerNames") or []), "sec_error": d.get("error")})106 return pl.DataFrame(rows)107108109def build_crosswalk(entities: pl.DataFrame, sec_tickers: pl.DataFrame, submissions: pl.DataFrame | None) -> pl.DataFrame:110 """entities: entity_id, security_id, ticker (from FinData). Effective-date ranges are the fundamentals span."""111 xw = entities.join(sec_tickers.unique(subset=["ticker"]), on="ticker", how="left")112 if submissions is not None:113 xw = xw.join(submissions, on="cik", how="left")114 xw = xw.with_columns([115 pl.col("cik").is_not_null().alias("has_cik"),116 pl.when(pl.col("cik").is_null()).then(pl.lit("unknown")).otherwise(pl.col("sic_division").fill_null("unknown")).alias("sic_division")117 if submissions is not None else pl.lit("unknown").alias("sic_division"),118 ])119 return xw120