spb/hfmarketdata

1#!/usr/bin/env python32"""Incremental EDGAR poller: every 2 minutes, detect new filings of tracked CIKs and re-normalise them.34    # 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.py6    # one cycle (cron / debugging)7    venv/bin/python scripts/edgar_incremental.py --once89Sources (see fundamentals.ingest.poll_new_filings): the EDGAR Atom feed `browse-edgar?action=getcurrent` for1010-K / 10-Q / 8-K / 20-F (live, ~4 requests per cycle) + the daily master index of the last two business days11as a safety net (weekend indexes do not exist: 403 from the SEC, skipped silently). For every new12(cik, accession) pair of a tracked form: companyfacts + submissions are refetched (cache bypass), statements13re-normalised (new versions, restatements) and rewritten only when they changed, coverage + screener row14refreshed, and a `filing` event is published on Redis (`filings` channel + `filings:stream` buffer) for the15WebSocket. A Redis SET `edgar:seen_accn` (7 days) guarantees a pair is handled once even when EDGAR misbehaves.16Lag, failures (+ samples) and the mapping failure rate are written to `fund_ingest_state`17(GET /v1/fundamentals/_health).1819Author: Simon-Pierre Boucher <contact@spboucher.ai>20"""21from __future__ import annotations2223import argparse24import logging25import signal26import sys27import time28from pathlib import Path2930HERE = Path(__file__).resolve().parent31sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))3233log = logging.getLogger("edgar_incremental")34_stop = False353637def _sig(*_):38    global _stop39    _stop = True404142def main() -> int:43    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)44    ap.add_argument("--interval", type=float, default=120.0, help="seconds between polls (default 120)")45    ap.add_argument("--once", action="store_true", help="run a single cycle and exit")46    ap.add_argument("--forms", default="10-K,10-Q,8-K,20-F", help="Atom feeds to poll")47    ap.add_argument("--no-daily-index", action="store_true")48    ap.add_argument("--rate", type=float, default=10.0)49    ap.add_argument("-v", "--verbose", action="store_true")50    args = ap.parse_args()51    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")52    if not args.verbose:53        logging.getLogger("httpx").setLevel(logging.WARNING)   # one line per request otherwise (incl. expected 403/404)54    signal.signal(signal.SIGTERM, _sig)55    signal.signal(signal.SIGINT, _sig)5657    from fundamentals import ingest, utcnow58    from fundamentals.edgar_client import EdgarClient59    from fundamentals.models import init_db6061    init_db()62    client = EdgarClient(rate=args.rate)63    forms = tuple(f.strip() for f in args.forms.split(",") if f.strip())64    while not _stop:65        t0 = time.time()66        try:67            summary = ingest.poll_new_filings(client, forms=forms, include_daily_index=not args.no_daily_index)68            log.info("cycle: seen=%s new=%s affected=%s changed=%s events=%s errors=%s index=%s (%.1fs)", summary["seen"],69                     summary["new"], summary["affected_ciks"], summary["rows_changed"], summary["events"], summary["errors"],70                     summary["index_days"], time.time() - t0)71            if summary["rows_changed"]:                       # nothing new stored → the extracts are still fresh72                try:73                    from bulk.build import available_years, build_year74                    for y in available_years()[-2:]:          # keep the two most recent yearly extracts fresh75                        build_year(y)76                except Exception as e:  # pragma: no cover77                    log.warning("bulk rebuild failed: %s", e)78        except Exception as e:79            log.exception("cycle failed: %s", e)80            ingest._set_state("incremental", last_run_at=utcnow(), failures_add=1,81                              failure_samples_add=[f"cycle: {type(e).__name__}: {e}"])82        if args.once:83            break84        for _ in range(int(max(1.0, args.interval - (time.time() - t0)))):85            if _stop:86                break87            time.sleep(1)88    return 0899091if __name__ == "__main__":92    sys.exit(main())93