spb/hfmarketdata

1#!/usr/bin/env python32"""Backfill SEC EDGAR fundamentals for the whole universe (resumable).34    # production (M3U96b)5    cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata venv/bin/python scripts/edgar_backfill.py --workers 46    # subset / dry run7    venv/bin/python scripts/edgar_backfill.py --tickers AAPL,MSFT,SHAK8    # from the SEC bulk archive (one 1+ GB download instead of ~7 600 API calls):9    #   curl -A "$HFMD_SEC_USER_AGENT" -o /tmp/companyfacts.zip https://www.sec.gov/Archives/edgar/daily-index/xbrl/companyfacts.zip10    venv/bin/python scripts/edgar_backfill.py --from-zip /tmp/companyfacts.zip1112Steps: sync the CIK↔ticker universe (SEC lists ∩ price lake) → for every CIK fetch companyfacts + submissions13(token bucket ≤ 10 req/s, gzip cache under data_root/edgar/raw) → facts Parquet lake → standardized statements →14coverage → screener row. Progress is kept in data_root/edgar/backfill_manifest.json so a rerun only processes15CIKs that failed or were never done (use --force to redo). Ends with the bulk Parquet files and the health state.1617Author: Simon-Pierre Boucher <contact@spboucher.ai>18"""19from __future__ import annotations2021import argparse22import json23import logging24import sys25import time26import zipfile27from concurrent.futures import ThreadPoolExecutor, as_completed28from pathlib import Path2930HERE = Path(__file__).resolve().parent31sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))3233from core.config import settings  # noqa: E4023435log = logging.getLogger("edgar_backfill")363738def manifest_path() -> Path:39    return settings.data_root / "edgar" / "backfill_manifest.json"404142def load_manifest() -> dict:43    p = manifest_path()44    return json.loads(p.read_text()) if p.exists() else {"started_at": None, "companies": {}}454647def save_manifest(m: dict) -> None:48    p = manifest_path()49    p.parent.mkdir(parents=True, exist_ok=True)50    tmp = p.with_suffix(".json.tmp")51    tmp.write_text(json.dumps(m, default=str))52    tmp.replace(p)535455def main() -> int:56    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)57    ap.add_argument("--tickers", help="comma-separated tickers (default: whole universe)")58    ap.add_argument("--ciks", help="comma-separated CIKs")59    ap.add_argument("--limit", type=int, help="process at most N companies")60    ap.add_argument("--workers", type=int, default=4, help="parallel companies (network is bounded by the 10 req/s bucket)")61    ap.add_argument("--rate", type=float, default=10.0, help="requests per second (SEC max 10)")62    ap.add_argument("--force", action="store_true", help="redo companies already done in the manifest")63    ap.add_argument("--refresh", action="store_true", help="ignore the raw JSON cache (refetch companyfacts/submissions)")64    ap.add_argument("--no-metalinks", action="store_true", help="skip MetaLinks.json (custom extension logging)")65    ap.add_argument("--from-zip", help="path to SEC companyfacts.zip (CIK##########.json inside) — no companyfacts API calls")66    ap.add_argument("--skip-universe", action="store_true", help="do not refresh edgar_companies from the SEC lists")67    ap.add_argument("--no-bulk", action="store_true", help="skip the bulk Parquet rebuild at the end")68    ap.add_argument("-v", "--verbose", action="store_true")69    args = ap.parse_args()70    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")7172    from fundamentals import ingest, utcnow73    from fundamentals.edgar_client import EdgarClient74    from fundamentals.models import init_db7576    init_db()77    client = EdgarClient(rate=args.rate)78    if not args.skip_universe:79        st = ingest.sync_universe(client)80        log.info("universe: lake=%s sec_tickers=%s companies=%s added=%s updated=%s delisted=%s unmatched(sample)=%s",81                 st.lake_tickers, st.sec_tickers, st.companies, st.added, st.updated, st.delisted, st.unmatched[:10])82    ciks = ingest.tracked_ciks()83    if args.ciks:84        want = {int(c) for c in args.ciks.split(",")}85        ciks = [c for c in ciks if c in want]86    if args.tickers:87        from fundamentals.service import resolve_company88        ciks = [resolve_company(t.strip()).cik for t in args.tickers.split(",") if t.strip()]89    manifest = load_manifest()90    manifest["started_at"] = manifest.get("started_at") or utcnow().isoformat()91    done = manifest["companies"]92    todo = [c for c in ciks if args.force or done.get(str(c), {}).get("status") not in ("ok", "no_facts")]93    if args.limit:94        todo = todo[:args.limit]95    log.info("%d companies tracked, %d to process", len(ciks), len(todo))96    ingest._set_state("backfill", last_run_at=utcnow(), companies_total=len(ciks),97                      companies_done=sum(1 for c in ciks if done.get(str(c), {}).get("status") == "ok"))9899    zf = zipfile.ZipFile(args.from_zip) if args.from_zip else None100101    def work(cik: int):102        cf = None103        if zf is not None:104            name = f"CIK{cik:010d}.json"105            try:106                cf = json.loads(zf.read(name))107            except KeyError:108                cf = {"cik": cik, "facts": {}}109        return ingest.ingest_company(client, cik, refresh=args.refresh, with_metalinks=not args.no_metalinks, companyfacts=cf)110111    t0 = time.time()112    n_ok = n_err = 0113    with ThreadPoolExecutor(max_workers=max(1, args.workers)) as ex:114        futures = {ex.submit(work, c): c for c in todo}115        for i, fut in enumerate(as_completed(futures), start=1):116            cik = futures[fut]117            try:118                r = fut.result()119            except Exception as e:  # pragma: no cover120                r = None121                log.exception("cik %s crashed: %s", cik, e)122            if r is not None and r.error == ingest.NO_FACTS:123                # no XBRL facts on EDGAR (funds, trusts, paper filers): permanent — not retried on resume124                done[str(cik)] = {"status": "no_facts", "ticker": r.ticker, "at": utcnow().isoformat()}125            elif r is None or r.error:126                n_err += 1127                done[str(cik)] = {"status": "error", "error": r.error if r else "crash", "at": utcnow().isoformat()}128            else:129                n_ok += 1130                done[str(cik)] = {"status": "ok", "ticker": r.ticker, "rows": r.rows, "facts": r.facts, "completeness": r.completeness,131                                  "extensions": r.extensions, "at": utcnow().isoformat(), "seconds": round(r.seconds, 2)}132                log.info("[%d/%d] %s cik=%s facts=%s rows=%s derived=%s restated=%s completeness=%s ext=%s %.1fs",133                         i, len(todo), r.ticker, cik, r.facts, r.rows, r.derived_rows, r.restated_rows, r.completeness, r.extensions, r.seconds)134            if i % 25 == 0 or i == len(todo):135                save_manifest(manifest)136                ingest._set_state("backfill", companies_done=sum(1 for v in done.values() if v.get("status") == "ok"),137                                  failures_add=0, requests_made=client.stats.requests,138                                  failure_samples=[v.get("error") for v in done.values() if v.get("status") == "error"][:10])139                elapsed = time.time() - t0140                log.info("progress %d/%d — %.0f s elapsed, %.1f s/company, requests=%d cache_hits=%d retries=%d",141                         i, len(todo), elapsed, elapsed / i, client.stats.requests, client.stats.cache_hits, client.stats.retries)142    save_manifest(manifest)143    ingest._set_state("backfill", last_success_at=utcnow(), failures=n_err, requests_made=client.stats.requests)144    ingest._update_mapping_failure_rate()145    if not args.no_bulk:146        from bulk.build import build_all147        metas = build_all()148        log.info("bulk files rebuilt: %s", [(m["year"], m["rows"]) for m in metas])149    log.info("done: ok=%d errors=%d in %.0f s (%s)", n_ok, n_err, time.time() - t0, client.stats)150    return 0 if n_err == 0 else 1151152153if __name__ == "__main__":154    sys.exit(main())155