fundamentals: scripts edgar_backfill (manifeste reprenable, --from-zip), edgar_incremental (cycle 2 min, événements Redis) et edgar_reconcile (échantillon de 20 sociétés)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 20 days ago (Sep 5, 2026) parent 7b0f0c1

3 changed files +370 −0

added scripts/edgar_backfill.py +154 −0

		`@@ -0,0 +1,154 @@`
	1	+#!/usr/bin/env python3
	2	+"""Backfill SEC EDGAR fundamentals for the whole universe (resumable).
	3	+
	4	+ # production (M3U96b)
	5	+ cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata venv/bin/python scripts/edgar_backfill.py --workers 4
	6	+ # subset / dry run
	7	+ venv/bin/python scripts/edgar_backfill.py --tickers AAPL,MSFT,SHAK
	8	+ # 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.zip
	10	+ venv/bin/python scripts/edgar_backfill.py --from-zip /tmp/companyfacts.zip
	11	+
	12	+Steps: sync the CIK↔ticker universe (SEC lists ∩ price lake) → for every CIK fetch companyfacts + submissions
	13	+(token bucket ≤ 10 req/s, gzip cache under data_root/edgar/raw) → facts Parquet lake → standardized statements →
	14	+coverage → screener row. Progress is kept in data_root/edgar/backfill_manifest.json so a rerun only processes
	15	+CIKs that failed or were never done (use --force to redo). Ends with the bulk Parquet files and the health state.
	16	+
	17	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	18	+"""
	19	+from __future__ import annotations
	20	+
	21	+import argparse
	22	+import json
	23	+import logging
	24	+import sys
	25	+import time
	26	+import zipfile
	27	+from concurrent.futures import ThreadPoolExecutor, as_completed
	28	+from pathlib import Path
	29	+
	30	+HERE = Path(__file__).resolve().parent
	31	+sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))
	32	+
	33	+from core.config import settings # noqa: E402
	34	+
	35	+log = logging.getLogger("edgar_backfill")
	36	+
	37	+
	38	+def manifest_path() -> Path:
	39	+ return settings.data_root / "edgar" / "backfill_manifest.json"
	40	+
	41	+
	42	+def load_manifest() -> dict:
	43	+ p = manifest_path()
	44	+ return json.loads(p.read_text()) if p.exists() else {"started_at": None, "companies": {}}
	45	+
	46	+
	47	+def 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)
	53	+
	54	+
	55	+def 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")
	71	+
	72	+ from fundamentals import ingest, utcnow
	73	+ from fundamentals.edgar_client import EdgarClient
	74	+ from fundamentals.models import init_db
	75	+
	76	+ 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_company
	88	+ 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"))
	98	+
	99	+ zf = zipfile.ZipFile(args.from_zip) if args.from_zip else None
	100	+
	101	+ def work(cik: int):
	102	+ cf = None
	103	+ 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)
	110	+
	111	+ t0 = time.time()
	112	+ n_ok = n_err = 0
	113	+ 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 cover
	120	+ r = None
	121	+ log.exception("cik %s crashed: %s", cik, e)
	122	+ if r is not None and r.error == "companyfacts_404":
	123	+ # no XBRL facts on EDGAR (funds, trusts, paper filers): permanent — not retried on resume
	124	+ done[str(cik)] = {"status": "no_facts", "ticker": r.ticker, "at": utcnow().isoformat()}
	125	+ elif r is None or r.error:
	126	+ n_err += 1
	127	+ done[str(cik)] = {"status": "error", "error": r.error if r else "crash", "at": utcnow().isoformat()}
	128	+ else:
	129	+ n_ok += 1
	130	+ 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() - t0
	140	+ 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_all
	147	+ 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 1
	151	+
	152	+
	153	+if __name__ == "__main__":
	154	+ sys.exit(main())

added scripts/edgar_incremental.py +85 −0

		`@@ -0,0 +1,85 @@`
	1	+#!/usr/bin/env python3
	2	+"""Incremental EDGAR poller: every 2 minutes, detect new filings of tracked CIKs and re-normalise them.
	3	+
	4	+ # production (M3U96b) — run under PM2 next to the API:
	5	+ cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata pm2 start venv/bin/python --name edgar-incremental -- scripts/edgar_incremental.py
	6	+ # one cycle (cron / debugging)
	7	+ venv/bin/python scripts/edgar_incremental.py --once
	8	+
	9	+Sources (see fundamentals.ingest.poll_new_filings): the EDGAR Atom feed `browse-edgar?action=getcurrent` for
	10	+10-K / 10-Q / 8-K / 20-F (live, ~4 requests per cycle) + the daily master index of today/yesterday as a safety
	11	+net. For every new accession of a tracked CIK: companyfacts + submissions are refetched (cache bypass),
	12	+statements re-normalised (new versions, restatements), coverage + screener row refreshed, and a `filing`
	13	+event is published on Redis (`filings` channel + `filings:stream` buffer) for the WebSocket. Lag, failures and
	14	+the mapping failure rate are written to `fund_ingest_state` (GET /v1/fundamentals/_health).
	15	+
	16	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	17	+"""
	18	+from __future__ import annotations
	19	+
	20	+import argparse
	21	+import logging
	22	+import signal
	23	+import sys
	24	+import time
	25	+from pathlib import Path
	26	+
	27	+HERE = Path(__file__).resolve().parent
	28	+sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))
	29	+
	30	+log = logging.getLogger("edgar_incremental")
	31	+_stop = False
	32	+
	33	+
	34	+def _sig(*_):
	35	+ global _stop
	36	+ _stop = True
	37	+
	38	+
	39	+def main() -> int:
	40	+ ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
	41	+ ap.add_argument("--interval", type=float, default=120.0, help="seconds between polls (default 120)")
	42	+ ap.add_argument("--once", action="store_true", help="run a single cycle and exit")
	43	+ ap.add_argument("--forms", default="10-K,10-Q,8-K,20-F", help="Atom feeds to poll")
	44	+ ap.add_argument("--no-daily-index", action="store_true")
	45	+ ap.add_argument("--rate", type=float, default=10.0)
	46	+ ap.add_argument("-v", "--verbose", action="store_true")
	47	+ args = ap.parse_args()
	48	+ logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
	49	+ signal.signal(signal.SIGTERM, _sig)
	50	+ signal.signal(signal.SIGINT, _sig)
	51	+
	52	+ from fundamentals import ingest, utcnow
	53	+ from fundamentals.edgar_client import EdgarClient
	54	+ from fundamentals.models import init_db
	55	+
	56	+ init_db()
	57	+ client = EdgarClient(rate=args.rate)
	58	+ forms = tuple(f.strip() for f in args.forms.split(",") if f.strip())
	59	+ while not _stop:
	60	+ t0 = time.time()
	61	+ try:
	62	+ summary = ingest.poll_new_filings(client, forms=forms, include_daily_index=not args.no_daily_index)
	63	+ log.info("cycle: seen=%s new=%s affected=%s events=%s errors=%s (%.1fs)", summary["seen"], summary["new"],
	64	+ summary["affected_ciks"], summary["events"], summary["errors"], time.time() - t0)
	65	+ if summary["affected_ciks"]:
	66	+ try:
	67	+ from bulk.build import available_years, build_year
	68	+ for y in available_years()[-2:]: # keep the two most recent yearly extracts fresh
	69	+ build_year(y)
	70	+ except Exception as e: # pragma: no cover
	71	+ log.warning("bulk rebuild failed: %s", e)
	72	+ except Exception as e:
	73	+ log.exception("cycle failed: %s", e)
	74	+ ingest._set_state("incremental", last_run_at=utcnow(), failures_add=1, failure_samples=[str(e)])
	75	+ if args.once:
	76	+ break
	77	+ for _ in range(int(max(1.0, args.interval - (time.time() - t0)))):
	78	+ if _stop:
	79	+ break
	80	+ time.sleep(1)
	81	+ return 0
	82	+
	83	+
	84	+if __name__ == "__main__":
	85	+ sys.exit(main())

added scripts/edgar_reconcile.py +131 −0

		`@@ -0,0 +1,131 @@`
	1	+#!/usr/bin/env python3
	2	+"""Reconciliation: re-check a random sample of companies against EDGAR and report discrepancies.
	3	+
	4	+ venv/bin/python scripts/edgar_reconcile.py # 20 random tracked companies
	5	+ venv/bin/python scripts/edgar_reconcile.py --sample 50 --json /tmp/reconcile.json
	6	+ venv/bin/python scripts/edgar_reconcile.py --tickers AAPL,MSFT
	7	+
	8	+For each sampled CIK the companyfacts document is refetched (cache bypassed), normalised in memory with the
	9	+current mapping and compared — latest version of every period, every public account — with what the database
	10	+serves. Differences (value changed, period missing on either side) are listed per company; the summary is
	11	+stored in `fund_ingest_state` (key `reconcile`) and exposed on GET /v1/fundamentals/_health. Exit code 1 when
	12	+any discrepancy is found (so it can be wired to an alert). Use `--fix` to re-ingest the companies that differ.
	13	+
	14	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	15	+"""
	16	+from __future__ import annotations
	17	+
	18	+import argparse
	19	+import json
	20	+import logging
	21	+import math
	22	+import random
	23	+import sys
	24	+from pathlib import Path
	25	+
	26	+HERE = Path(__file__).resolve().parent
	27	+sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))
	28	+
	29	+log = logging.getLogger("edgar_reconcile")
	30	+
	31	+
	32	+def compare(cik: int, ticker: str, client) -> dict:
	33	+ from sqlalchemy import select
	34	+
	35	+ from core.db import session
	36	+ from fundamentals import mapping as M
	37	+ from fundamentals import normalize as N
	38	+ from fundamentals.models import EdgarCompany, fund_statements
	39	+
	40	+ cf = client.companyfacts(cik, refresh=True)
	41	+ sub = client.submissions(cik, refresh=True)
	42	+ with session() as s:
	43	+ c = s.get(EdgarCompany, cik)
	44	+ db_rows = [dict(r._mapping) for r in s.execute(select(fund_statements).where(fund_statements.c.cik == cik))]
	45	+ if cf is None:
	46	+ return {"cik": cik, "ticker": ticker, "error": "companyfacts_404", "discrepancies": []}
	47	+ fresh = N.normalize_company(cik, ticker, N.facts_frame(cf), N.filings_from_submissions(sub),
	48	+ (sub or {}).get("fiscalYearEnd") or (c.fiscal_year_end if c else None))
	49	+ a = {(r["statement"], r["fiscal_year"], r["fiscal_quarter"]): r for r in N.select_as_of(fresh.rows, None)}
	50	+ b = {(r["statement"], r["fiscal_year"], r["fiscal_quarter"]): r for r in N.select_as_of(db_rows, None)}
	51	+ diffs = []
	52	+ for key in sorted(set(a) \| set(b)):
	53	+ if key not in a:
	54	+ diffs.append({"period": key, "issue": "period_missing_on_edgar_side"})
	55	+ continue
	56	+ if key not in b:
	57	+ diffs.append({"period": key, "issue": "period_missing_in_db"})
	58	+ continue
	59	+ for acc in M.PUBLIC_ACCOUNTS:
	60	+ if M.ACCOUNT_BY_NAME[acc].statement != key[0]:
	61	+ continue
	62	+ x, y = a[key].get(acc), b[key].get(acc)
	63	+ if x is None and y is None:
	64	+ continue
	65	+ if x is None or y is None or not math.isclose(x, y, rel_tol=1e-6, abs_tol=0.5):
	66	+ diffs.append({"period": key, "account": acc, "edgar": x, "db": y, "issue": "value_differs"})
	67	+ from fundamentals.edgar_client import TRACKED_FORMS
	68	+ known = {r["accn"] for r in db_rows}
	69	+ new = sorted(a for a, f in N.filings_from_submissions(sub).items() if f.form in TRACKED_FORMS and a not in known) if sub else []
	70	+ return {"cik": cik, "ticker": ticker, "periods_edgar": len(a), "periods_db": len(b), "discrepancies": diffs,
	71	+ "new_filings": new[:5]}
	72	+
	73	+
	74	+def main() -> int:
	75	+ ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
	76	+ ap.add_argument("--sample", type=int, default=20)
	77	+ ap.add_argument("--tickers", help="explicit tickers instead of a random sample")
	78	+ ap.add_argument("--seed", type=int)
	79	+ ap.add_argument("--json", help="write the full report to this path")
	80	+ ap.add_argument("--fix", action="store_true", help="re-ingest companies with discrepancies")
	81	+ ap.add_argument("-v", "--verbose", action="store_true")
	82	+ args = ap.parse_args()
	83	+ logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
	84	+
	85	+ from sqlalchemy import select
	86	+
	87	+ from core.db import session
	88	+ from fundamentals import ingest, utcnow
	89	+ from fundamentals.edgar_client import EdgarClient
	90	+ from fundamentals.models import EdgarCompany, init_db
	91	+
	92	+ init_db()
	93	+ client = EdgarClient()
	94	+ with session() as s:
	95	+ companies = [(c.cik, c.ticker) for c in s.scalars(select(EdgarCompany).where(EdgarCompany.normalized_at.isnot(None)))]
	96	+ if args.tickers:
	97	+ want = {t.strip().upper() for t in args.tickers.split(",")}
	98	+ sample = [x for x in companies if x[1] in want]
	99	+ else:
	100	+ rng = random.Random(args.seed)
	101	+ sample = rng.sample(companies, min(args.sample, len(companies)))
	102	+ report = {"run_at": utcnow().isoformat(), "sample": len(sample), "companies": []}
	103	+ bad = 0
	104	+ for cik, ticker in sample:
	105	+ try:
	106	+ r = compare(cik, ticker, client)
	107	+ except Exception as e: # pragma: no cover
	108	+ r = {"cik": cik, "ticker": ticker, "error": str(e), "discrepancies": []}
	109	+ report["companies"].append(r)
	110	+ n = len(r["discrepancies"])
	111	+ bad += int(bool(n or r.get("error")))
	112	+ log.info("%s cik=%s periods edgar/db=%s/%s discrepancies=%d new_filings=%s %s", ticker, cik, r.get("periods_edgar"),
	113	+ r.get("periods_db"), n, r.get("new_filings"), r.get("error") or "")
	114	+ for d in r["discrepancies"][:5]:
	115	+ log.info(" %s", d)
	116	+ if n and args.fix:
	117	+ res = ingest.ingest_company(client, cik, refresh=False)
	118	+ log.info(" re-ingested %s: rows=%s error=%s", ticker, res.rows, res.error)
	119	+ report["companies_with_discrepancies"] = bad
	120	+ ingest._set_state("reconcile", last_run_at=utcnow(), last_success_at=utcnow(),
	121	+ companies_total=len(sample), companies_done=len(sample) - bad, failures=bad,
	122	+ extra={"companies_with_discrepancies": bad, "sample": [c["ticker"] for c in report["companies"]],
	123	+ "discrepancies": sum(len(c["discrepancies"]) for c in report["companies"])})
	124	+ if args.json:
	125	+ Path(args.json).write_text(json.dumps(report, default=str, indent=1))
	126	+ log.info("reconcile done: %d/%d companies with discrepancies (%s)", bad, len(sample), client.stats)
	127	+ return 1 if bad else 0
	128	+
	129	+
	130	+if __name__ == "__main__":
	131	+ sys.exit(main())
	132