#!/usr/bin/env python3 """Reconciliation: re-check a random sample of companies against EDGAR and report discrepancies. venv/bin/python scripts/edgar_reconcile.py # 20 random tracked companies venv/bin/python scripts/edgar_reconcile.py --sample 50 --json /tmp/reconcile.json venv/bin/python scripts/edgar_reconcile.py --tickers AAPL,MSFT For each sampled CIK the companyfacts document is refetched (cache bypassed), normalised in memory with the current mapping and compared — latest version of every period, every public account — with what the database serves. Differences (value changed, period missing on either side) are listed per company; the summary is stored in `fund_ingest_state` (key `reconcile`) and exposed on GET /v1/fundamentals/_health. Exit code 1 when any discrepancy is found (so it can be wired to an alert). Use `--fix` to re-ingest the companies that differ. Author: Simon-Pierre Boucher """ from __future__ import annotations import argparse import json import logging import math import random import sys from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api")) log = logging.getLogger("edgar_reconcile") def compare(cik: int, ticker: str, client) -> dict: from sqlalchemy import select from core.db import session from fundamentals import mapping as M from fundamentals import normalize as N from fundamentals.models import EdgarCompany, fund_statements cf = client.companyfacts(cik, refresh=True) sub = client.submissions(cik, refresh=True) with session() as s: c = s.get(EdgarCompany, cik) db_rows = [dict(r._mapping) for r in s.execute(select(fund_statements).where(fund_statements.c.cik == cik))] if cf is None: return {"cik": cik, "ticker": ticker, "error": "companyfacts_404", "discrepancies": []} fresh = N.normalize_company(cik, ticker, N.facts_frame(cf), N.filings_from_submissions(sub), (sub or {}).get("fiscalYearEnd") or (c.fiscal_year_end if c else None)) a = {(r["statement"], r["fiscal_year"], r["fiscal_quarter"]): r for r in N.select_as_of(fresh.rows, None)} b = {(r["statement"], r["fiscal_year"], r["fiscal_quarter"]): r for r in N.select_as_of(db_rows, None)} diffs = [] for key in sorted(set(a) | set(b)): if key not in a: diffs.append({"period": key, "issue": "period_missing_on_edgar_side"}) continue if key not in b: diffs.append({"period": key, "issue": "period_missing_in_db"}) continue for acc in M.PUBLIC_ACCOUNTS: if M.ACCOUNT_BY_NAME[acc].statement != key[0]: continue x, y = a[key].get(acc), b[key].get(acc) if x is None and y is None: continue if x is None or y is None or not math.isclose(x, y, rel_tol=1e-6, abs_tol=0.5): diffs.append({"period": key, "account": acc, "edgar": x, "db": y, "issue": "value_differs"}) from fundamentals.edgar_client import TRACKED_FORMS known = {r["accn"] for r in db_rows} 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 [] return {"cik": cik, "ticker": ticker, "periods_edgar": len(a), "periods_db": len(b), "discrepancies": diffs, "new_filings": new[:5]} def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--sample", type=int, default=20) ap.add_argument("--tickers", help="explicit tickers instead of a random sample") ap.add_argument("--seed", type=int) ap.add_argument("--json", help="write the full report to this path") ap.add_argument("--fix", action="store_true", help="re-ingest companies with discrepancies") 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 %(name)s: %(message)s") from sqlalchemy import select from core.db import session from fundamentals import ingest, utcnow from fundamentals.edgar_client import EdgarClient from fundamentals.models import EdgarCompany, init_db init_db() client = EdgarClient() with session() as s: companies = [(c.cik, c.ticker) for c in s.scalars(select(EdgarCompany).where(EdgarCompany.normalized_at.isnot(None)))] if args.tickers: want = {t.strip().upper() for t in args.tickers.split(",")} sample = [x for x in companies if x[1] in want] else: rng = random.Random(args.seed) sample = rng.sample(companies, min(args.sample, len(companies))) report = {"run_at": utcnow().isoformat(), "sample": len(sample), "companies": []} bad = 0 for cik, ticker in sample: try: r = compare(cik, ticker, client) except Exception as e: # pragma: no cover r = {"cik": cik, "ticker": ticker, "error": str(e), "discrepancies": []} report["companies"].append(r) n = len(r["discrepancies"]) bad += int(bool(n or r.get("error"))) log.info("%s cik=%s periods edgar/db=%s/%s discrepancies=%d new_filings=%s %s", ticker, cik, r.get("periods_edgar"), r.get("periods_db"), n, r.get("new_filings"), r.get("error") or "") for d in r["discrepancies"][:5]: log.info(" %s", d) if n and args.fix: res = ingest.ingest_company(client, cik, refresh=False) log.info(" re-ingested %s: rows=%s error=%s", ticker, res.rows, res.error) report["companies_with_discrepancies"] = bad ingest._set_state("reconcile", last_run_at=utcnow(), last_success_at=utcnow(), companies_total=len(sample), companies_done=len(sample) - bad, failures=bad, extra={"companies_with_discrepancies": bad, "sample": [c["ticker"] for c in report["companies"]], "discrepancies": sum(len(c["discrepancies"]) for c in report["companies"])}) if args.json: Path(args.json).write_text(json.dumps(report, default=str, indent=1)) try: # housekeeping: hand free pages back to the OS when the database runs auto_vacuum=INCREMENTAL (no-op otherwise) from fundamentals.migrations import incremental_vacuum log.info("sqlite incremental_vacuum done, freelist pages left: %d", incremental_vacuum()) except Exception as e: # pragma: no cover log.warning("incremental_vacuum failed: %s", e) log.info("reconcile done: %d/%d companies with discrepancies (%s)", bad, len(sample), client.stats) return 1 if bad else 0 if __name__ == "__main__": sys.exit(main())