spb/hfmarketdata

1#!/usr/bin/env python32"""Reconciliation: re-check a random sample of companies against EDGAR and report discrepancies.34    venv/bin/python scripts/edgar_reconcile.py                 # 20 random tracked companies5    venv/bin/python scripts/edgar_reconcile.py --sample 50 --json /tmp/reconcile.json6    venv/bin/python scripts/edgar_reconcile.py --tickers AAPL,MSFT78For each sampled CIK the companyfacts document is refetched (cache bypassed), normalised in memory with the9current mapping and compared — latest version of every period, every public account — with what the database10serves. Differences (value changed, period missing on either side) are listed per company; the summary is11stored in `fund_ingest_state` (key `reconcile`) and exposed on GET /v1/fundamentals/_health. Exit code 1 when12any discrepancy is found (so it can be wired to an alert). Use `--fix` to re-ingest the companies that differ.1314Author: Simon-Pierre Boucher <contact@spboucher.ai>15"""16from __future__ import annotations1718import argparse19import json20import logging21import math22import random23import sys24from pathlib import Path2526HERE = Path(__file__).resolve().parent27sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))2829log = logging.getLogger("edgar_reconcile")303132def compare(cik: int, ticker: str, client) -> dict:33    from sqlalchemy import select3435    from core.db import session36    from fundamentals import mapping as M37    from fundamentals import normalize as N38    from fundamentals.models import EdgarCompany, fund_statements3940    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            continue56        if key not in b:57            diffs.append({"period": key, "issue": "period_missing_in_db"})58            continue59        for acc in M.PUBLIC_ACCOUNTS:60            if M.ACCOUNT_BY_NAME[acc].statement != key[0]:61                continue62            x, y = a[key].get(acc), b[key].get(acc)63            if x is None and y is None:64                continue65            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_FORMS68    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]}727374def 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")8485    from sqlalchemy import select8687    from core.db import session88    from fundamentals import ingest, utcnow89    from fundamentals.edgar_client import EdgarClient90    from fundamentals.models import EdgarCompany, init_db9192    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 = 0104    for cik, ticker in sample:105        try:106            r = compare(cik, ticker, client)107        except Exception as e:  # pragma: no cover108            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"] = bad120    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    try:  # housekeeping: hand free pages back to the OS when the database runs auto_vacuum=INCREMENTAL (no-op otherwise)127        from fundamentals.migrations import incremental_vacuum128        log.info("sqlite incremental_vacuum done, freelist pages left: %d", incremental_vacuum())129    except Exception as e:  # pragma: no cover130        log.warning("incremental_vacuum failed: %s", e)131    log.info("reconcile done: %d/%d companies with discrepancies (%s)", bad, len(sample), client.stats)132    return 1 if bad else 0133134135if __name__ == "__main__":136    sys.exit(main())137