#!/usr/bin/env python """SEC EDGAR enrichment for the seed registry (docs/SEEDS.md). Matches registry companies to `company_tickers.json` (ticker first, then normalised name) and fills *missing* `sec_cik`, `ticker`, `exchange`, `hq_city`/`hq_region` and an industry hint (SIC → taxonomy slug) from the submissions JSON. Existing values are never overwritten. Polite: SEC-required User-Agent, ≤ 5 requests/s, responses cached under data/seed/edgar/. Usage: .venv/bin/python scripts/seed_edgar.py [--max-submissions N] [--dry-run] """ from __future__ import annotations import argparse import json import logging import sys import time from collections import Counter, defaultdict from pathlib import Path from typing import Any import httpx ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from companyatlas.ids import normalize_alias from companyatlas.registry.industries import map_sic log = logging.getLogger("seed_edgar") USER_AGENT = "CompanyAtlasBot/0.1 contact@spboucher.ai" TICKERS_URL = "https://www.sec.gov/files/company_tickers.json" SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik:010d}.json" MIN_INTERVAL_S = 0.21 # ≤ 5 requests per second (SEC fair-access policy is 10/s) CACHE_DIR = ROOT / "data" / "seed" / "edgar" OUT_DIR = ROOT / "registry" / "companies" US_STATES = {"AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA", "HI", "ID", "IL", "IN", "IA", "KS", "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS", "MO", "MT", "NE", "NV", "NH", "NJ", "NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI", "SC", "SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY", "DC", "PR"} class Edgar: def __init__(self) -> None: self.client = httpx.Client(headers={"User-Agent": USER_AGENT, "Accept-Encoding": "gzip, deflate"}, timeout=30, follow_redirects=True) self.last = 0.0 self.requests = 0 CACHE_DIR.mkdir(parents=True, exist_ok=True) def get_json(self, url: str, cache_name: str) -> Any | None: path = CACHE_DIR / cache_name if path.exists(): return json.loads(path.read_text(encoding="utf-8")) for attempt in range(4): wait = MIN_INTERVAL_S - (time.monotonic() - self.last) if wait > 0: time.sleep(wait) try: r = self.client.get(url) except httpx.HTTPError as e: log.warning("edgar %s: %s", url, e) time.sleep(2 * (attempt + 1)) continue finally: self.last = time.monotonic() self.requests += 1 if r.status_code == 200: path.write_text(r.text, encoding="utf-8") return r.json() if r.status_code == 404: path.write_text("null", encoding="utf-8") return None log.warning("edgar %s → HTTP %d, backing off", url, r.status_code) time.sleep(5 * (attempt + 1)) return None def load_registry() -> dict[Path, list[dict[str, Any]]]: out: dict[Path, list[dict[str, Any]]] = {} for path in sorted(OUT_DIR.glob("wikidata-*.ndjson")): with path.open(encoding="utf-8") as f: out[path] = [json.loads(line) for line in f if line.strip()] return out def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--max-submissions", type=int, default=8000, help="cap on submissions JSON fetches (≈ 30 min at 5 req/s when cold)") ap.add_argument("--dry-run", action="store_true") ap.add_argument("-v", "--verbose", action="store_true") args = ap.parse_args() logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logging.getLogger("httpx").setLevel(logging.WARNING) edgar = Edgar() tickers = edgar.get_json(TICKERS_URL, "company_tickers.json") if not tickers: log.error("could not fetch %s", TICKERS_URL) return 1 by_ticker: dict[str, list[dict[str, Any]]] = defaultdict(list) by_name: dict[str, set[int]] = defaultdict(set) for e in tickers.values(): by_ticker[e["ticker"].upper()].append(e) by_name[normalize_alias(e["title"])].add(int(e["cik_str"])) log.info("edgar tickers: %d entries, %d names", len(tickers), len(by_name)) files = load_registry() stats: Counter[str] = Counter() submissions_fetched = 0 for path, rows in files.items(): for r in rows: cik: int | None = int(r["sec_cik"]) if r.get("sec_cik") and str(r["sec_cik"]).isdigit() else None if cik is None: cands: set[int] = set() if r.get("ticker") and r["ticker"].upper() in by_ticker: for e in by_ticker[r["ticker"].upper()]: if normalize_alias(e["title"])[:6] == normalize_alias(r["display_name"])[:6] or r.get("country") == "US": cands.add(int(e["cik_str"])) if not cands: names = [r["display_name"], r.get("legal_name") or ""] + list(r.get("aliases") or []) for nm in names: if nm and normalize_alias(nm) in by_name: cands |= by_name[normalize_alias(nm)] if len(cands) > 1 and r.get("country") != "US": cands = set() # ambiguous non-US name: do not guess if len(cands) == 1: cik = cands.pop() stats["matched_name_or_ticker"] += 1 elif len(cands) > 1: stats["ambiguous"] += 1 continue else: continue else: stats["had_cik"] += 1 r.setdefault("industry_labels", []) changed = False if not r.get("sec_cik"): r["sec_cik"] = str(cik) changed = True needs_sub = not r.get("exchange") or not r.get("industries") or not r.get("hq_city") or not r.get("ticker") if needs_sub and submissions_fetched < args.max_submissions: sub = edgar.get_json(SUBMISSIONS_URL.format(cik=cik), f"CIK{cik:010d}.json") submissions_fetched += 1 if sub: if not r.get("ticker") and sub.get("tickers"): r["ticker"] = sub["tickers"][0] changed = True if not r.get("exchange") and sub.get("exchanges"): r["exchange"] = next((x for x in sub["exchanges"] if x), None) changed = changed or bool(r["exchange"]) if r.get("ticker"): r["public_company"] = True sic = sub.get("sic") slug = map_sic(sic) if sic and sub.get("sicDescription") and f"SIC {sic} {sub['sicDescription']}" not in r["industry_labels"]: r["industry_labels"].append(f"SIC {sic} {sub['sicDescription']}") changed = True if slug and not r.get("industries"): r["industries"] = [slug] stats["industry_from_sic"] += 1 changed = True biz = (sub.get("addresses") or {}).get("business") or {} if not r.get("hq_city") and biz.get("city"): r["hq_city"] = biz["city"].title() changed = True if not r.get("hq_region") and biz.get("stateOrCountry") in US_STATES: r["hq_region"] = biz["stateOrCountry"] changed = True if changed: r["source_edgar"] = {"cik": str(cik), "enriched_at": time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime())} stats["rows_enriched"] += 1 if not args.dry_run: with path.open("w", encoding="utf-8") as f: for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n") log.info("edgar done: %s (%d HTTP requests, %d submissions)", dict(stats), edgar.requests, submissions_fetched) (ROOT / "data" / "seed" / "edgar-report.json").write_text(json.dumps({"stats": stats, "submissions": submissions_fetched}, indent=1)) if not args.dry_run: total = sum(len(rows) for rows in files.values()) with_cik = sum(1 for rows in files.values() for r in rows if r.get("sec_cik")) no_ind = sum(1 for rows in files.values() for r in rows if not r.get("industries")) update_readme(f"* EDGAR ({time.strftime('%Y-%m-%d', time.gmtime())}): {stats['matched_name_or_ticker']:,} companies newly matched to a CIK " f"(ticker, then exact normalised name), {stats['had_cik']:,} already had one → {with_cik:,} of {total:,} with a SEC CIK; " f"{stats['rows_enriched']:,} rows enriched, {stats['industry_from_sic']:,} industries from SIC ({no_ind:,} still without an " f"industry mapping), {submissions_fetched:,} submissions JSON read.", stats, submissions_fetched) return 0 def update_readme(line: str, stats: Counter[str], submissions: int) -> None: """Replace (or append) the EDGAR line of the README harvest report and record the counters in data/seed/harvest-stats.json.""" readme = OUT_DIR / "README.md" if readme.exists(): lines = [ln for ln in readme.read_text(encoding="utf-8").splitlines() if not ln.startswith("* EDGAR (")] while lines and not lines[-1].strip(): lines.pop() readme.write_text("\n".join([*lines, line]) + "\n", encoding="utf-8") stats_path = ROOT / "data" / "seed" / "harvest-stats.json" data = json.loads(stats_path.read_text(encoding="utf-8")) if stats_path.exists() else {} data["edgar"] = {**stats, "submissions": submissions, "at": time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime())} stats_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") if __name__ == "__main__": sys.exit(main())