"""Build filings ground-truth JSONL from downloaded EDGAR data. Reads the sidecars + XBRL companyfacts written by `download_edgar.py` and produces `evaluation/smoke_filings_sample.jsonl` — the file the eval CLI reads in `--mode live --doc-type filing` runs. Each JSONL row has: id, source, text (full 10-K plaintext), ground_truth (Filing schema dict) Cover fields come from the sidecar (auto). Financials come from XBRL companyfacts using the most-common US-GAAP concepts (see FIN_MAP below). Risk factors are left EMPTY in the auto-generated ground truth — they're free text and there's no canonical source. Backfill by hand for a stricter qualitative eval. Usage ----- python scripts/build_filings_gt.py python scripts/build_filings_gt.py --raw-dir data/raw/10k --out evaluation/smoke_filings_sample.jsonl """ from __future__ import annotations import argparse import json from pathlib import Path ROOT = Path(__file__).resolve().parents[1] # US-GAAP concept name -> our schema field. # Values are lists of candidate concept names in order of preference — real # filers use different names for the same line ("Revenues", "SalesRevenueNet", # "RevenueFromContractWithCustomerExcludingAssessedTax"...). # US-GAAP concept mapping. Values are ordered lists of candidate concept # names — the first one that has an FP='FY' entry for the target fiscal-year # end date wins. `total_debt` is special: we SUM all matching concepts because # real filers split debt across many line items (short-term borrowings + # long-term debt current portion + long-term debt noncurrent + notes payable). # # v2.2 expansions: # - Added Deposits + LongTermBorrowings so banks (JPM) get real total_debt. # - Added StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest # as first choice for total_equity (matches how large filers actually report). # - Added several revenue variants (retail filers use SalesRevenueGoodsNet, etc.). # - Added short-term debt concepts (ShortTermBorrowings, etc.) into total_debt sum. FIN_MAP = { "revenue": [ "Revenues", "RevenueFromContractWithCustomerExcludingAssessedTax", "RevenueFromContractWithCustomerIncludingAssessedTax", "SalesRevenueNet", "SalesRevenueGoodsNet", ], "cost_of_revenue": [ "CostOfRevenue", "CostOfGoodsAndServicesSold", "CostOfGoodsSold", ], "gross_profit": ["GrossProfit"], "operating_income": ["OperatingIncomeLoss", "IncomeLossFromContinuingOperations"], "net_income": [ "NetIncomeLoss", "ProfitLoss", "NetIncomeLossAttributableToParent", ], "eps_basic": ["EarningsPerShareBasic", "IncomeLossFromContinuingOperationsPerBasicShare"], "eps_diluted": ["EarningsPerShareDiluted", "IncomeLossFromContinuingOperationsPerDilutedShare"], "cash_and_equivalents": [ "CashAndCashEquivalentsAtCarryingValue", "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents", "CashCashEquivalentsAndShortTermInvestments", ], "total_assets": ["Assets"], "total_equity": [ "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest", "StockholdersEquity", ], "total_debt": [ # SUM of every matching concept — filers split debt many ways "LongTermDebt", "LongTermDebtNoncurrent", "LongTermDebtCurrent", "LongTermBorrowings", # banks (JPM, Citi) use this instead of LongTermDebt "ShortTermBorrowings", "CommercialPaper", "DebtCurrent", "DebtLongtermAndShorttermCombinedAmount", "Deposits", # bank funding — treat as debt for a bank ], "operating_cash_flow": [ "NetCashProvidedByUsedInOperatingActivities", "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations", ], } def _pick_fy_value(concept_data: dict, fy_end: str) -> float | None: """Return the concept value for the fiscal year ending `fy_end` (ISO date). XBRL companyfacts nests values by unit (USD, USD/shares, pure). We take the first unit that produces a match. Match rule: same `end` date + form starting with '10-K' + `fp == 'FY'`. """ units = concept_data.get("units", {}) for _unit, entries in units.items(): for e in entries: if e.get("end") == fy_end and e.get("form", "").startswith("10-K") and e.get("fp") == "FY": try: return float(e["val"]) except (KeyError, ValueError, TypeError): continue return None def build_financials(facts: dict, fy_end: str) -> dict: """Populate as many FilingFinancials fields as possible from XBRL.""" us_gaap = (facts.get("facts", {}) or {}).get("us-gaap", {}) or {} fin: dict = {"currency": "USD", "fiscal_year": int(fy_end[:4])} for field, candidates in FIN_MAP.items(): val = None collected: list[float] = [] for concept in candidates: data = us_gaap.get(concept) if not data: continue v = _pick_fy_value(data, fy_end) if v is None: continue if field == "total_debt": # Sum across all matching debt concepts (short-term + long-term). collected.append(v) else: val = v break if field == "total_debt" and collected: val = sum(collected) if val is not None: fin[field] = val return fin def build_row(txt_path: Path) -> dict | None: """Produce one ground-truth row for the JSONL, or None if metadata is missing.""" stem = txt_path.name[:-len(".txt")] meta_path = txt_path.parent / f"{stem}.meta.json" facts_path = txt_path.parent / f"{stem}.facts.json" if not meta_path.exists(): print(f"[!] no sidecar for {txt_path.name} — skipping") return None meta = json.loads(meta_path.read_text(encoding="utf-8")) facts = json.loads(facts_path.read_text(encoding="utf-8")) if facts_path.exists() else None fy_end = meta["reporting_period_end"] # Backfill exchange + state_of_incorporation from the SEC submissions feed # (both were added to the sidecar in download_edgar.py v2.2). If they're # missing (older sidecar without the field), fall back to None — the # schema field is Optional so this doesn't fail extraction. exchange = meta.get("exchange") # Prefer the two-letter code (matches the schema validator that uppercases + strips). state = meta.get("state_of_incorporation") or meta.get("state_of_incorporation_desc") cover = { "company_name": meta["company_name"], "cik": meta["cik10"], "ticker": meta["ticker"], "form_type": meta["form"], "fiscal_year_end": fy_end, "filing_date": meta["filing_date"], } if exchange: cover["exchange"] = exchange if state: cover["state_of_incorporation"] = state financials = build_financials(facts, fy_end) if facts else {"currency": "USD", "fiscal_year": int(fy_end[:4])} return { "id": f"filing_{meta['ticker']}_{fy_end}", "source": "sec_edgar", "text": txt_path.read_text(encoding="utf-8"), "ground_truth": { "cover": cover, "financials": financials, "top_risk_factors": [], # backfill by hand for qualitative eval }, } def main(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--raw-dir", default="data/raw/10k") ap.add_argument("--out", default="evaluation/smoke_filings_sample.jsonl") args = ap.parse_args(argv) raw = ROOT / args.raw_dir out = ROOT / args.out out.parent.mkdir(parents=True, exist_ok=True) txts = sorted(raw.glob("*_10k.txt")) if not txts: print(f"[!] no *_10k.txt files under {raw} — run download_edgar.py first.") return 2 n_written = 0 with out.open("w", encoding="utf-8") as f: for p in txts: row = build_row(p) if not row: continue f.write(json.dumps(row) + "\n") n_written += 1 fin = row["ground_truth"]["financials"] print(f" {row['id']:36s} revenue={fin.get('revenue')} net_income={fin.get('net_income')}") print(f"\n-> {out} ({n_written} rows)") return 0 if __name__ == "__main__": raise SystemExit(main())