spb/hfmarketdata

1"""Record trimmed REAL EDGAR fixtures for the fundamentals tests (run manually, needs network).23    HFMD_DATA_ROOT=/tmp/hfmd-proto .venv/bin/python tests/fixtures/edgar/record.py45Prototype companies: Apple (CIK 320193, FYE late September, 52/53-week), Microsoft (CIK 789019, FYE June 30),6Shake Shack (CIK 1620533, Russell 2000, FYE last Wednesday of December, restaurant cost lines tagged with the7company extension `shak:OperatingMaterialsExpense` → cost_of_revenue legitimately unmapped).89Trimming keeps the fixtures small (< 1 MB gzipped total) while staying real:10* companyfacts: mapped tags + identity tags + a few unmapped us-gaap tags, facts filed since 2022-01-0111  with period end >= 2020-09-01;12* submissions: filings of the tracked forms since 2022-01-01 (older pages dropped);13* MetaLinks of the latest 10-K (SHAK): tag names only.14"""15from __future__ import annotations1617import gzip18import json19import os20import sys21from datetime import date22from pathlib import Path2324HERE = Path(__file__).resolve().parent25ROOT = HERE.parents[2]26sys.path.insert(0, str(ROOT / "hfmarketdata" / "api"))27os.environ.setdefault("HFMD_DATA_ROOT", "/tmp/hfmd-proto")2829from fundamentals import mapping as M  # noqa: E40230from fundamentals.edgar_client import TRACKED_FORMS, EdgarClient  # noqa: E4023132COMPANIES = {"AAPL": 320193, "MSFT": 789019, "SHAK": 1620533}33EXTRA_TICKERS = {"GOOGL": 1652044, "GOOG": 1652044}     # one CIK, two share classes (mapping test only)34MIN_END = "2020-09-01"35MIN_FILED = "2022-01-01"36KEEP_UNMAPPED = ("OperatingExpenses", "LaborAndRelatedExpense", "OccupancyNet", "PreOpeningCosts",37                 "ComprehensiveIncomeNetOfTax", "OtherNonoperatingIncomeExpense", "IncreaseDecreaseInInventories")383940def dump(path: Path, obj) -> None:41    path.parent.mkdir(parents=True, exist_ok=True)42    with gzip.open(path, "wt", encoding="utf-8", compresslevel=9) as fh:43        json.dump(obj, fh, separators=(",", ":"))44    print(f"  {path.name}: {path.stat().st_size / 1024:.0f} KB")454647def trim_companyfacts(cf: dict) -> dict:48    keep = {(t.taxonomy, t.tag) for a in M.ACCOUNTS for t in a.tags} | set(M.IDENTITY_TAGS)49    keep |= {("us-gaap", t) for t in KEEP_UNMAPPED}50    out = {"cik": cf["cik"], "entityName": cf["entityName"], "facts": {}}51    for tax, tags in cf["facts"].items():52        for tag, body in tags.items():53            if (tax, tag) not in keep and tax != "dei":54                continue55            units = {}56            for unit, facts in body["units"].items():57                kept = [f for f in facts if f["end"] >= MIN_END and f["filed"] >= MIN_FILED]58                if kept:59                    units[unit] = kept60            if units:61                out["facts"].setdefault(tax, {})[tag] = {"label": body.get("label"), "description": "", "units": units}62    return out636465def trim_submissions(sub: dict) -> dict:66    rec = sub["filings"]["recent"]67    idx = [i for i, f in enumerate(rec["form"]) if f in TRACKED_FORMS and rec["filingDate"][i] >= MIN_FILED]68    recent = {k: [v[i] for i in idx] for k, v in rec.items()}69    out = {k: v for k, v in sub.items() if k not in ("filings", "addresses", "formerNames", "description")}70    out["filings"] = {"recent": recent, "files": []}71    return out727374def trim_metalinks(ml: dict) -> dict:75    key, inst = next(iter(ml["instance"].items()))76    keep_roles = {r["role"] for r in inst["report"].values() if r.get("menuCat") == "Statements"}77    tags = {}78    for name, info in inst["tag"].items():79        pres = [p for p in (info.get("presentation") or []) if p in keep_roles]80        tags[name] = {"xbrltype": info.get("xbrltype"), "nsuri": info.get("nsuri"), "presentation": pres}81    return {"instance": {key: {"nsprefix": inst["nsprefix"], "nsuri": inst["nsuri"], "baseTaxonomies": inst["baseTaxonomies"],82                               "report": {k: {"role": r.get("role"), "shortName": r.get("shortName"), "menuCat": r.get("menuCat")}83                                          for k, r in inst["report"].items() if r.get("menuCat") == "Statements"},84                               "tag": tags}}}858687def main() -> None:88    c = EdgarClient()89    tickers = c.company_tickers()90    exch = c.company_tickers_exchange()91    wanted = set(COMPANIES.values()) | set(EXTRA_TICKERS.values())92    ct = {str(i): v for i, v in enumerate(v for v in tickers.values() if int(v["cik_str"]) in wanted)}93    dump(HERE / "company_tickers.json.gz", ct)94    dump(HERE / "company_tickers_exchange.json.gz", {"fields": exch["fields"],95                                                     "data": [r for r in exch["data"] if int(r[0]) in wanted]})96    for tk, cik in COMPANIES.items():97        print(tk, cik)98        cf = c.companyfacts(cik)99        dump(HERE / f"companyfacts_CIK{cik:010d}.json.gz", trim_companyfacts(cf))100        sub = c.submissions(cik, include_older=False)101        dump(HERE / f"submissions_CIK{cik:010d}.json.gz", trim_submissions(sub))102        if tk == "SHAK":103            rec = sub["filings"]["recent"]104            accn = next(rec["accessionNumber"][i] for i, f in enumerate(rec["form"]) if f == "10-K")105            dump(HERE / f"metalinks_CIK{cik:010d}_{accn.replace('-', '')}.json.gz", trim_metalinks(c.metalinks(cik, accn)))106    import re107    atom = c.atom_current("10-Q")108    head = atom.split("<entry>")[0]109    entries = re.findall(r"<entry>.*?</entry>", atom, flags=re.S)[:3]110    (HERE / "atom_10-Q.xml").write_text(head + "\n".join(entries) + "\n</feed>\n", encoding="utf-8")111    yesterday = date.fromordinal(date.today().toordinal() - 1)112    for back in range(0, 6):113        d = date.fromordinal(yesterday.toordinal() - back)114        q = (d.month - 1) // 3 + 1115        try:116            txt = c.get_text(f"https://www.sec.gov/Archives/edgar/daily-index/{d.year}/QTR{q}/master.{d:%Y%m%d}.idx")117        except Exception:118            continue119        lines = txt.splitlines()120        head = lines[:11]121        body = [ln for ln in lines[11:] if ln.split("|")[2:3] and ln.split("|")[2] in ("10-K", "10-Q", "8-K", "20-F")][:12]122        (HERE / "master.idx").write_text("\n".join(head + body) + "\n", encoding="utf-8")123        print("  master.idx from", d)124        break125    print(c.stats)126127128if __name__ == "__main__":129    main()130