diff --git a/docs/superpowers/specs/2026-10-25-bcda-v3-design.md b/docs/superpowers/specs/2026-10-25-bcda-v3-design.md new file mode 100644 index 0000000..6716a31 --- /dev/null +++ b/docs/superpowers/specs/2026-10-25-bcda-v3-design.md @@ -0,0 +1,54 @@ +# P52 — BCDA v3: client, crosswalk-driven flatteners, vendored upstream, proving notebook + +Tracker: milestone P52, #1139. Builds on the v2 `bcda` module (client/store/flatten, #6, #8) and the `bcda.express.cclf` input-layer pipeline. + +## Why now + +BCDA v3 went GA on 2026-07-01 and CMS discontinues v1/v2 on 2027-07-30. The module targets v2 end to end: `Client` defaults to `/api/v2`, and `bcda.express.flatten` matches BlueButton extensions by the retired `https://bluebutton.cms.gov/resources/variables/` URL suffix. On a v3 payload that match never fires, so almost every coded column comes out NULL without an error. The live spike against the public Large Model Entity sandbox (2026-10-25) confirmed the transport still works on v3 and that the payload is where the migration lives. + +## Sources of truth (in priority order) + +1. **BCDA v3 Data Dictionary** (`BCDA_v3_Data_Dictionary.xlsx`, published 2026-07-01). Each v3 field has a `fhirPath`, the IDR `sourceColumn`, and two crosswalk columns: `ccwMapping` (the CCW/v2 variable names the current table models are named after) and `cclfMapping` (CCLF file.element). The `CCLF-BCDA v3 Mapping` sheet is the CCLF-side view of the same crosswalk; `CCLF fields absent from BCDA v3` lists what cannot be reproduced (HICNs, FIPS county, address lines, BETOS, several CCLFA/B amounts). +2. **Migration guide** (`/api-documentation/how-to-migrate-v3.html`) and the **July 9 2026 office-hours transition deck**: endpoint path change only; `/auth` and `/data` unchanged; Claim/ClaimResponse retired into EOB; `_typeFilter` with the `System-Type` tag (`SharedSystem`, `NationalClaimsHistory`, `DDPS`; default omits SharedSystem); extension/CodeSystem URLs move to `fhir/StructureDefinition/` and `fhir/CodeSystem/`; Patient/Coverage/EOB ids are new (match on MBI and claim control number); Coverage drops reference year/month for `Coverage.period` and gains a Dual Medicare/Medicaid row; BETOS removed; v3 deduplicates claim versions within a source. +3. **Official v3 sample files** (`v3_Patient/Coverage/ExplanationOfBenefit.ndjson` from bcda.cms.gov) — the fixtures the flatteners are tested against. +4. **CMSgov/bcda-app** — the server. Vendored sparsely: the v3 handler and its tests, request validation (`_typeFilter`, `_since`, Prefer/Accept rules), the v3 OperationOutcome writer, the v3 Postman collection, gzip download fixtures, the sandbox model config, and LICENSE. + +## Decisions + +1. **One set of tables, two payload dialects.** `bcda.patient`, `bcda.coverage`, `bcda.explanation_of_benefit`, `bcda.eob_item` keep their CCW-named columns (the `ccwMapping` column is exactly the crosswalk from v3 fields onto them). v3-native data with no v2 home gets new nullable columns rather than being crammed into monthly slots. The flattener detects the dialect per resource (`meta.profile` carries `C4BB-…|2.2.0`, or any `bluebutton.cms.gov/fhir/` URL) and dispatches; v2 seeds and tests keep passing. +2. **v3 extraction is table-driven.** `bcda.express.v3` has a handful of primitives that mirror the dictionary's fhirPath idioms (identifier by system, `type` coding by system, supportingInfo by category code → code/timing/value, header `adjudication`/`total` by Adjudication code, item adjudication by Adjudication code, item extension by StructureDefinition code, careTeam by role, contained provider by `provider.reference`) and declarative maps from v3 codes to columns. Every map entry cites the dictionary `sourceColumn`, so a reader can check it against the spreadsheet. +3. **Identity follows the migration guide.** `bene_id` is the v3 `Patient.id` (BENE_SK; the EOB `patient.reference` uses it). `bene_mbi_id` is the current MBI (the `us-mbi` identifier without a `period.end`, else the latest). `clm_id` is the C4BB unique claim id; `clm_cntl_num` is the claim control number and the key to pair a SharedSystem EOB with its NationalClaimsHistory successor. `system_type` and `final_action` come from `meta.tag`; `eob_type` from the `EOB-TYPE` coding, `clm_type_cd` from `CLM-TYPE-CD`. +4. **Coverage is one row per plan per period**, as v3 delivers it: `class_plan` ∈ {Part A, Part B, Part C, Part D, Dual Medicare/Medicaid}, `period_start`/`period_end` plus the plan's extensions in v3-named columns. The legacy part-A/B/D begin/end columns are filled from the matching plan's period so `bcda.express.cclf` keeps working; `dual_01` is left NULL and the pipeline reads `coalesce(dual_01, dual_stus_cd)` after a self-join to the Dual row — the only pipeline change in scope. +5. **Race.** v3 moves race to the US Core extension (OMB codes). `us_core_race` holds the OMB code; `bene_race_cd` is derived through the published CCW crosswalk (2106-3→1, 2054-5→2, 2028-9/2076-8→4, 1002-5→6, Hispanic ethnicity→5, unknown→0) so the eligibility race labels do not go blank. +6. **Client**: default `v3`; `type_filter=` on `start_export`/`export`; `jobs()` sends `Prefer: respond-async` + `Accept: application/fhir+json` (required on v2 and v3) and returns the Task resources from the Bundle; every non-2xx surfaces the OperationOutcome diagnostics (`BcdaHttpError`, still an `httpx.HTTPStatusError`); a poll that hits `404 Job has been cancelled` raises `JobCancelledError` (the public sandbox credentials are shared, so this happens). `/_version` and `/_health` are exposed for the notebook's preflight. +7. **Crosswalk is data, not prose.** The dictionary and both CCLF sheets are flattened to `dev/seeds/bcda_v3_data_dictionary.csv`, `dev/seeds/bcda_v3_cclf_crosswalk.csv`, `dev/seeds/bcda_v3_cclf_absent.csv` and land in `reference_data` via `stack load seed`, so the notebook (and later the chat) can show, per table column, where it came from and which CCLF element it replaces. +8. **Vendoring is a manifest, not a submodule.** `dev/vendor/MANIFEST.toml` pins the bcda-app commit and lists each vendored path with its git blob sha; bcda.cms.gov downloads are pinned by sha256. `dev/scripts/vendor_bcda_app.py --check` compares against upstream (GitHub tree API + content hashes) and exits non-zero on drift; `--sync` refreshes files and the manifest. A weekly workflow runs `--check` and files an issue on drift (no auto-commit: a payload change needs a human to look at the flattener). `tests/bcda/test_vendor.py` verifies the tree matches the manifest offline, and the flattener tests run on the vendored official samples. +9. **The notebook proves, it does not analyse.** `notebooks/bcda_v3_sandbox.py` walks preflight → auth → attribution → export (reusing a completed job younger than 24 h when one exists, since a fresh Large-entity export takes 20–40 min) → `Store` download → flatten → DuckDB load (`duckdb_batch` + `publish_replica`, never a long-lived writer) → checks. Each check is a sentence with a number and a pass/fail mark: every Patient has an MBI, Coverage rows per plan, EOB counts by System-Type × EOB-TYPE, claim-control-number uniqueness within a source, item amounts populated, dictionary field population per table. Pulls are gated behind `BCDA_PULL=1` or the button so the nightly notebooks-integration run only re-validates data already on disk. + +## Data flow + +``` +sandbox /api/v3 ──Client──▶ Store (data/bcda/exports//*.ndjson, jobs/files parquet) + └─ flatten_export ──dialect?──▶ v2 flatteners (unchanged) + └──────────▶ bcda.express.v3 ──▶ flat/*.parquet + └─ load_bcda ──▶ aco.duckdb bcda.* ──▶ publish_replica +BCDA_v3_Data_Dictionary.xlsx ──gen_bcda_v3_crosswalk.py──▶ dev/seeds/bcda_v3_*.csv ──stack load seed──▶ reference_data.* +CMSgov/bcda-app + bcda.cms.gov ──vendor_bcda_app.py──▶ dev/vendor/{bcda-app,bcda-site}/ (+ MANIFEST.toml) ──▶ tests, weekly drift check +``` + +## Out of scope + +Reworking `bcda.express.cclf` semantics for v3 (claim routing by `nch_near_line_rec_ident_cd` still works because v3 carries `CLM_NRLN_RIC_CD`; a v3-native routing on `eob_type` is a follow-up). Partially adjudicated exports end to end (the client supports `_typeFilter`; the sandbox Large entity is adjudicated-only). Databricks/Iceberg landing of the new columns. + +## Results (2026-10-25, Large Model Entity sandbox) + +Pulled as three single-type jobs: the EOB job (89596, 6 min server-side, 192 files / 2.2 GB) was adopted in 44 s; Patient (11 MB) and Coverage (44 MB) exports took 24 and 32 min, most of it queued as *Pending*. Flatten + load of the whole pull: ~2 min. + +| table | rows | +|---|---| +| `bcda.patient` | 9,582 | +| `bcda.coverage` | 19,396 (Part A 5,643 · Part B 5,649 · Part C 3,236 · Dual 4,868) | +| `bcda.explanation_of_benefit` | 44,361 (PDE 14,395 · CARRIER 6,298 · DME 6,242 · SNF 5,813 · HOSPICE 2,942 · INPATIENT 2,910 · OUTPATIENT 2,883 · HHA 2,878) | +| `bcda.eob_item` | 248,528 | + +All 20 notebook checks pass: every Patient has a unique current MBI and `bene_id == Patient.id`; every Coverage row has `period.start`, every Part A row `ms_cd`, every Dual row `dual_stus_cd`, no orphan beneficiaries; every EOB has a claim id, a claim control number, System-Type and Final-Action tags, EOB-TYPE and CLM-TYPE-CD; control numbers are unique within a source (v3 deduplication holds); lines carry HCPCS or NDC on 100% and paid/submitted amounts on 94% (Part D lines carry drug-cost amounts instead); no orphan lines. The entity is adjudicated-only (no SharedSystem claims), so the control-number pairing check is exercised but finds zero pairs — partially adjudicated data needs an entity with PAC access and `_typeFilter`. diff --git a/notebooks/bcda_v3_sandbox.py b/notebooks/bcda_v3_sandbox.py new file mode 100644 index 0000000..7b83c8b --- /dev/null +++ b/notebooks/bcda_v3_sandbox.py @@ -0,0 +1,761 @@ +import marimo + +__generated_with = "0.23.13" +app = marimo.App(width="full") + + +@app.cell(hide_code=True) +def _(): + import marimo as mo + + return (mo,) + + +@app.cell(hide_code=True) +def _(mo): + mo.md(""" + # BCDA v3 — sandbox pull, flatten, load, prove + + The Beneficiary Claims Data API moved to **v3** on 2026-07-01 (v1/v2 end on + 2027-07-30). This notebook walks the whole path against CMS's public sandbox + for the **Large Model Entity** (10,000 synthetic enrollees) and checks each + step with a number rather than a feeling: + + 1. **Preflight** — `/_version`, `/_health`, `/api/v3/metadata` (no token). + 2. **Auth + attribution** — client-credentials token, `attribution_status`, + the entity's job history. + 3. **Pull** — one export job per resource type (a combined job takes 20–40 + minutes and the shared sandbox credentials mean anyone can cancel it; the + EOB-only job finishes in ~6 minutes and 190 files / 2 GB). A completed job + younger than 24 h is **adopted** instead of re-run. + 4. **Flatten + load** — `bcda.express.v3` (table-driven from the v3 Data + Dictionary's `fhirPath` / `sourceColumn` / `ccwMapping` columns) streams + the ndjson into Parquet and `aco.duckdb` `bcda.*`, then the read-only + replica is republished. + 5. **Proof** — identity, coverage shape, claim provenance, line amounts, + referential integrity, and how many dictionary-mapped columns actually + carry data, each marked pass / fail. + + Network steps run only when **`BCDA_PULL=1`** is set or the buttons below are + pressed (`BCDA_LOAD=1` flattens and loads what is already in the store without + touching the network); otherwise the notebook validates whatever is already in + the replica, so the nightly notebooks-integration run never starts an export. + + Sources: BCDA v3 Data Dictionary (`dev/vendor/bcda-site`), the migration + guide and July-2026 office-hours transition deck, `dev/seeds/bcda_v3_*.csv` + (CCLF ↔ BCDA v3 crosswalk), CMSgov/bcda-app (sparse vendor, `dev/vendor/bcda-app`). + """) + return + + +@app.cell(hide_code=True) +def _(): + # ── 0. Setup ── + import os + import time + from datetime import datetime, timedelta, timezone + + import altair as alt + import polars as pl + + from bcda.client import SANDBOX, BcdaError, Client + from bcda.store import Store + from conf import ROOT, connect, path + + connect.theme() + + # Public sandbox credentials from https://bcda.cms.gov/api-documentation/get-a-bearer-token.html + # ("Large Model Entity", 10,000 synthetic enrollees). Real credentials come in + # through the environment and win. + LARGE_ENTITY = { + "name": "Large Model Entity (sandbox, synthetic)", + "client_id": "f5b6a686-0d7b-4080-873a-e60534fd2995", + "client_secret": "96ac1483449c652f363396737937f23cdae47a56a116c8a4405dbd54b31d2686f005df502c5c86ab", + "enrollees": 10_000, + } + CLIENT_ID = os.environ.get("BCDA_CLIENT_ID") or LARGE_ENTITY["client_id"] + CLIENT_SECRET = ( + os.environ.get("BCDA_CLIENT_SECRET") or LARGE_ENTITY["client_secret"] + ) + BASE_URL = os.environ.get("BCDA_BASE_URL") or SANDBOX + PULL = os.environ.get("BCDA_PULL") == "1" + LOAD = ( + os.environ.get("BCDA_LOAD") == "1" + ) # flatten + load what is on disk, no network + TYPES = ("Patient", "Coverage", "ExplanationOfBenefit") + STORE_PATH = path("storage.bcda") + + def client() -> Client: + return Client( + CLIENT_ID, CLIENT_SECRET, base_url=BASE_URL, output_dir=STORE_PATH + ) + + NOTES: dict[str, str] = {} + return ( + BASE_URL, + BcdaError, + LARGE_ENTITY, + LOAD, + NOTES, + PULL, + ROOT, + STORE_PATH, + Store, + TYPES, + alt, + client, + connect, + datetime, + pl, + time, + timedelta, + timezone, + ) + + +@app.cell(hide_code=True) +def _(BASE_URL, LARGE_ENTITY, PULL, mo): + pull_btn = mo.ui.run_button(label="Pull from sandbox now") + load_btn = mo.ui.run_button(label="Flatten + load into DuckDB") + mo.vstack( + [ + mo.md(f""" + **Target** `{BASE_URL}` · entity *{LARGE_ENTITY["name"]}* · + `BCDA_PULL` {"set — network steps run headless" if PULL else "unset — press a button to pull or load"} + """), + mo.hstack([pull_btn, load_btn], justify="start"), + ] + ) + return load_btn, pull_btn + + +@app.cell(hide_code=True) +def _(BcdaError, NOTES, client, mo): + # ── 1. Preflight (no token) ── + preflight = {} + try: + with client() as _c: + _md = _c.metadata() + preflight["fhirVersion"] = _md.get("fhirVersion") + preflight["software"] = (_md.get("software") or {}).get("version") + preflight["releaseDate"] = (_md.get("software") or {}).get("releaseDate") + preflight["resources"] = ", ".join( + r.get("type", "?") + for r in (_md.get("rest") or [{}])[0].get("resource", []) + ) + try: + preflight["_version"] = _c.server_version().get("version") + except BcdaError as e: # noqa: BLE001 — informational endpoint + preflight["_version"] = ( + f"unavailable ({e.status_code if hasattr(e, 'status_code') else e})" + ) + try: + preflight["_health"] = _c.health().get("database") + except BcdaError as e: # noqa: BLE001 + preflight["_health"] = f"unavailable ({e})" + ONLINE = True + except Exception as e: # noqa: BLE001 — no network: validate disk only + ONLINE = False + NOTES["preflight"] = f"sandbox unreachable: {e}" + + def _tick(ok: bool) -> str: + return "✅" if ok else "❌" + + if ONLINE: + _rows = [ + ( + _tick(preflight.get("fhirVersion") == "4.0.1"), + "FHIR R4 metadata", + f"fhirVersion {preflight.get('fhirVersion')}", + ), + ( + _tick(bool(preflight.get("software"))), + "bcda-app release", + f"{preflight.get('software')} ({preflight.get('releaseDate')})", + ), + ( + _tick("ExplanationOfBenefit" in (preflight.get("resources") or "")), + "v3 resources advertised", + preflight.get("resources"), + ), + ( + _tick(preflight.get("_health") == "ok"), + "`/_health` database", + str(preflight.get("_health")), + ), + ("ℹ️", "`/_version`", str(preflight.get("_version"))), + ] + mo.output.replace( + mo.md( + "## 1 · Preflight\n\n" + + "\n".join(f"- {t} **{n}** — {v}" for t, n, v in _rows) + ) + ) + else: + mo.output.replace( + mo.md( + f"## 1 · Preflight\n\n⚠️ {NOTES['preflight']} — continuing with data already on disk." + ) + ) + return (ONLINE,) + + +@app.cell(hide_code=True) +def _(ONLINE, TYPES, client, datetime, mo, pl, timedelta, timezone): + # ── 2. Auth, attribution, job history ── + attribution = {} + completed: list[dict] = [] + reusable: dict[str, dict] = {} + if ONLINE: + with client() as _c: + _tok = _c.authenticate() + attribution = _c.attribution_status() + completed = _c.completed_jobs() + _cutoff = datetime.now(timezone.utc) - timedelta(hours=23) + for _j in completed: # newest first; keep the newest single-type job per type + if ( + len(_j["types"]) == 1 + and _j["types"][0] in TYPES + and _j["types"][0] not in reusable + ): + try: + _end = datetime.fromisoformat(_j["end"].replace("Z", "+00:00")) + except ValueError: + continue + if _end >= _cutoff: + reusable[_j["types"][0]] = _j + _ing = { + d["type"]: d["timestamp"] for d in attribution.get("ingestion_dates", []) + } + _exp = { + d["type"]: d["timestamp"] for d in attribution.get("expiration_dates", []) + } + _hist = ( + pl.DataFrame( + [ + { + "job_id": j["job_id"], + "types": ",".join(j["types"]) or "(all)", + "completed": j["end"], + "reusable_24h": any(r is j for r in reusable.values()), + } + for j in completed[:15] + ] + ) + if completed + else pl.DataFrame({"job_id": []}) + ) + mo.output.replace( + mo.vstack( + [ + mo.md(f""" + ## 2 · Auth + attribution + + - ✅ **token** issued ({len(_tok)} chars, 20-minute lifetime) + - ✅ **attribution** last updated `{_ing.get("last_attribution_update")}`, runout `{_ing.get("last_runout_update")}`; access expires `{_exp.get("attribution_access_expiration")}` + - {"✅" if completed else "ℹ️"} **{len(completed)} completed v3 job(s)** on this entity; reusable single-type jobs (<24 h): {", ".join(sorted(reusable)) or "none"} + """), + mo.ui.table( + _hist, + label="Most recent completed v3 jobs (shared entity — other sandbox users' jobs appear here too)", + ), + ] + ) + ) + else: + mo.output.replace(mo.md("## 2 · Auth + attribution\n\n⚠️ skipped — offline.")) + return (reusable,) + + +@app.cell(hide_code=True) +def _(ONLINE, PULL, STORE_PATH, Store, TYPES, client, mo, pl, pull_btn, reusable, time): + # ── 3. Pull (one job per type; adopt a fresh completed job when one exists) ── + store = Store(path=str(STORE_PATH)) + pulled: dict[str, str] = {} + _do_pull = ONLINE and (PULL or pull_btn.value) + if _do_pull: + _log = [] + with client() as _c: + for _t in TYPES: + _t0 = time.time() + if _t in reusable: + _key = store.adopt( + _c, reusable[_t]["job_url"], types=[_t], poll_interval=10 + ) + _how = f"adopted job {reusable[_t]['job_id']}" + else: + _key = store.export( + _c, types=[_t], poll_interval=15, cancel_retries=8 + ) + _how = "exported" + pulled[_t] = _key + _log.append( + f"- ✅ **{_t}** {_how} → `{_key}`, {len(store.get_files(_key))} file(s), {time.time() - _t0:.0f}s" + ) + mo.output.replace(mo.md("## 3 · Pull\n\n" + "\n".join(_log))) + else: + _jobs = store.list_jobs() + _df = ( + pl.DataFrame( + [ + { + "key": j["key"], + "status": j["status"], + "types": j["types"], + "ts": j["ts"], + "files": len(store.get_files(j["key"])), + } + for j in _jobs + ] + ).sort("ts", descending=True) + if _jobs + else pl.DataFrame({"key": []}) + ) + mo.output.replace( + mo.vstack( + [ + mo.md( + f"## 3 · Pull\n\nNo pull requested ({'offline' if not ONLINE else 'set `BCDA_PULL=1` or press *Pull from sandbox now*'}). Jobs already in the store at `{STORE_PATH}`:" + ), + mo.ui.table( + _df, + label="Store job history (append-only; last row per key is current)", + ), + ] + ) + ) + return pulled, store + + +@app.cell(hide_code=True) +def _(LOAD, PULL, STORE_PATH, connect, load_btn, mo, pulled, store, time): + # ── 4. Flatten + load ── + from aco.load.bcda import _find_latest_export, load_bcda + + _complete = [j for j in store.list_jobs(status="complete")] + _do_load = bool(_complete) and (PULL or LOAD or load_btn.value or bool(pulled)) + load_stats: dict[str, int] = {} + if _do_load: + _dirs = _find_latest_export(STORE_PATH) + _t0 = time.time() + load_stats = ( + load_bcda() + ) # newest export per resource type → bcda.* in aco.duckdb + _dt = time.time() - _t0 + _replica = connect.publish_replica("aco") + mo.output.replace( + mo.md( + "## 4 · Flatten + load\n\n" + + f"- ✅ flattened from {len(_dirs)} export dir(s): {', '.join(d.name for d in _dirs)}\n" + + "\n".join( + f"- ✅ `{t}` — {n:,} rows" for t, n in sorted(load_stats.items()) + ) + + f"\n- ✅ {_dt:.0f}s end to end; replica republished at `{_replica}`" + ) + ) + else: + mo.output.replace( + mo.md( + "## 4 · Flatten + load\n\n" + + ( + "No complete export in the store yet." + if not _complete + else "Not requested — reading the tables already in the replica." + ) + ) + ) + return + + +@app.cell(hide_code=True) +def _(connect, mo, pl): + # ── 5. Proof — against the read-only replica ── + con = connect.duckdb("aco", read_only=True) + + def q(sql: str) -> pl.DataFrame: + return con.execute(sql).pl() + + _tables = q( + "select table_name from information_schema.tables where table_schema = 'bcda'" + )["table_name"].to_list() + HAVE = {"patient", "coverage", "explanation_of_benefit", "eob_item"} <= set(_tables) + mo.stop( + not HAVE, + mo.md( + f"## 5 · Proof\n\n⚠️ `bcda.*` tables not in the replica yet (found: {_tables or 'none'}). Run the pull and load above first." + ), + ) + return HAVE, con, q + + +@app.cell(hide_code=True) +def _(mo, q): + checks: list[tuple[bool, str, str]] = [] + + def check(ok: bool, name: str, detail: str) -> None: + checks.append((bool(ok), name, detail)) + + counts = { + r["t"]: r["n"] + for r in q(""" + select 'patient' t, count(*) n from bcda.patient + union all select 'coverage', count(*) from bcda.coverage + union all select 'explanation_of_benefit', count(*) from bcda.explanation_of_benefit + union all select 'eob_item', count(*) from bcda.eob_item + """).to_dicts() + } + + # Identity + _p = q(""" + select count(*) n, count(bene_mbi_id) with_mbi, count(distinct bene_mbi_id) distinct_mbi, + count(distinct bene_id) distinct_sk, sum(case when bene_id = id then 1 else 0 end) sk_is_id, + count(cntct_lang_cd) with_lang, count(bene_death_dt) deceased, + sum(case when bene_race_cd is not null then 1 else 0 end) with_race + from bcda.patient + """).to_dicts()[0] + check( + _p["n"] > 0 and _p["with_mbi"] == _p["n"], + "Every Patient carries a current MBI", + f"{_p['with_mbi']:,} / {_p['n']:,}", + ) + check( + _p["distinct_mbi"] == _p["n"], + "MBIs are unique", + f"{_p['distinct_mbi']:,} distinct", + ) + check( + _p["sk_is_id"] == _p["n"], + "bene_id is the v3 Patient.id (BENE_SK)", + f"{_p['sk_is_id']:,} rows", + ) + check( + _p["with_race"] == _p["n"], + "Race crosswalked from US Core OMB codes", + f"{_p['with_race']:,} rows; {_p['deceased']:,} deceased; {_p['with_lang']:,} with contact language", + ) + + # Coverage shape + plans = q( + "select class_plan, count(*) n, count(period_start) with_start, count(ms_cd) with_ms, count(dual_stus_cd) with_dual from bcda.coverage group by 1 order by 1" + ) + _plan_names = set(plans["class_plan"].to_list()) + check( + {"Part A", "Part B"} <= _plan_names, + "Coverage has one row per plan (Part A/B present)", + ", ".join(f"{r['class_plan']}={r['n']:,}" for r in plans.to_dicts()), + ) + check( + plans["with_start"].sum() == plans["n"].sum(), + "Every Coverage row has Coverage.period.start", + f"{plans['with_start'].sum():,} / {plans['n'].sum():,}", + ) + _a = plans.filter(plans["class_plan"] == "Part A") + check( + len(_a) and _a["with_ms"][0] == _a["n"][0], + "Part A rows carry Medicare status (ms_cd)", + f"{_a['with_ms'][0] if len(_a) else 0:,} / {_a['n'][0] if len(_a) else 0:,}", + ) + _d = plans.filter(plans["class_plan"] == "Dual Medicare/Medicaid") + check( + len(_d) == 0 or _d["with_dual"][0] == _d["n"][0], + "Dual rows carry dual_stus_cd", + f"{_d['with_dual'][0] if len(_d) else 0:,} dual rows", + ) + _cov_ref = q( + "select count(*) n, sum(case when p.bene_id is null then 1 else 0 end) orphans from bcda.coverage c left join bcda.patient p on p.bene_id = c.beneficiary_reference" + ).to_dicts()[0] + check( + _cov_ref["orphans"] == 0, + "Coverage.beneficiary resolves to a Patient", + f"{_cov_ref['orphans']:,} orphans of {_cov_ref['n']:,}", + ) + + # Claims provenance + prov = q( + "select system_type, eob_type, count(*) n, sum(clm_pmt_amt) paid from bcda.explanation_of_benefit group by 1, 2 order by 1, 2" + ) + _e = q(""" + select count(*) n, count(clm_id) with_id, count(clm_cntl_num) with_ccn, count(system_type) with_sys, count(eob_type) with_eob_type, + count(clm_type_cd) with_clm_type, count(final_action) with_fa, count(clm_from_dt) with_from, + min(clm_from_dt) first_dt, max(clm_thru_dt) last_dt + from bcda.explanation_of_benefit + """).to_dicts()[0] + check( + _e["with_id"] == _e["n"] and _e["with_ccn"] == _e["n"], + "Every EOB has a unique claim id and a claim control number", + f"{_e['n']:,} EOBs", + ) + check( + _e["with_sys"] == _e["n"] and _e["with_fa"] == _e["n"], + "Every EOB is tagged System-Type and Final-Action", + f"{', '.join(sorted(set(prov['system_type'].to_list())))}", + ) + check( + _e["with_eob_type"] == _e["n"] and _e["with_clm_type"] == _e["n"], + "EOB-TYPE and CLM-TYPE-CD present on every claim", + f"{', '.join(sorted(set(prov['eob_type'].to_list())))}", + ) + check( + _e["with_from"] == _e["n"], + "Billable period present", + f"{_e['first_dt']} → {_e['last_dt']}", + ) + _dupe = q(""" + select count(*) dupes from ( + select system_type, clm_cntl_num, count(*) c from bcda.explanation_of_benefit + group by 1, 2 having count(*) > 1) + """).to_dicts()[0]["dupes"] + check( + _dupe == 0, + "Claim control number unique within a source system (v3 dedupes claim versions)", + f"{_dupe:,} duplicated control numbers", + ) + _pairs = q(""" + select count(*) pairs from ( + select clm_cntl_num from bcda.explanation_of_benefit + group by 1 having count(distinct system_type) > 1) + """).to_dicts()[0]["pairs"] + check( + True, + "SharedSystem ↔ NationalClaimsHistory pairs (same control number)", + f"{_pairs:,} claims seen in two sources", + ) + _eob_ref = q( + "select count(*) n, sum(case when p.bene_id is null then 1 else 0 end) orphans from bcda.explanation_of_benefit e left join bcda.patient p on p.bene_id = e.bene_id" + ).to_dicts()[0] + check( + _eob_ref["orphans"] == 0, + "EOB.patient resolves to a Patient", + f"{_eob_ref['orphans']:,} orphans of {_eob_ref['n']:,}", + ) + + # Lines + _i = q(""" + select count(*) n, count(distinct clm_id) claims, count(clm_line_hcpcs_cd) hcpcs, count(clm_line_ndc_cd) ndc, + count(clm_line_cvrd_pd_amt) paid, count(clm_line_sbmtd_chrg_amt) submitted, count(clm_line_from_dt) dated + from bcda.eob_item + """).to_dicts()[0] + check( + _i["claims"] == _e["n"] or _i["claims"] > 0, + "Every claim has line items", + f"{_i['n']:,} lines on {_i['claims']:,} claims", + ) + check( + _i["hcpcs"] + _i["ndc"] >= _i["n"] * 0.95, + "Lines carry a HCPCS or NDC code", + f"HCPCS {_i['hcpcs']:,}, NDC {_i['ndc']:,} of {_i['n']:,}", + ) + check( + _i["paid"] >= _i["n"] * 0.9 and _i["dated"] == _i["n"], + "Line paid amount and service date populated", + f"paid {_i['paid']:,}, submitted {_i['submitted']:,}, dated {_i['dated']:,}", + ) + _orph = q( + "select count(*) n from bcda.eob_item i left join bcda.explanation_of_benefit e on e.clm_id = i.clm_id where e.clm_id is null" + ).to_dicts()[0]["n"] + check( + _orph == 0, + "Every line item belongs to a loaded claim", + f"{_orph:,} orphan lines", + ) + + _passed = sum(1 for ok, _, _ in checks if ok) + mo.vstack( + [ + mo.md( + f"## 5 · Proof — {_passed} / {len(checks)} checks pass\n\n" + + " · ".join(f"`{t}` {n:,}" for t, n in counts.items()) + ), + mo.md( + "\n".join( + f"- {'✅' if ok else '❌'} **{name}** — {detail}" + for ok, name, detail in checks + ) + ), + ] + ) + return checks, counts, plans, prov + + +@app.cell(hide_code=True) +def _(alt, mo, plans, prov, q): + # ── Charts: one measure each, single hue from the fhirworx theme ── + _by_type = q( + "select eob_type, count(*) n, sum(clm_pmt_amt) paid from bcda.explanation_of_benefit group by 1 order by 2 desc" + ) + claims_chart = ( + alt.Chart(_by_type) + .mark_bar(cornerRadiusEnd=4, size=18) + .encode( + y=alt.Y("eob_type:N", sort="-x", title=None), + x=alt.X("n:Q", title="claims", axis=alt.Axis(grid=True)), + tooltip=[ + alt.Tooltip("eob_type:N", title="EOB type"), + alt.Tooltip("n:Q", title="claims", format=","), + alt.Tooltip("paid:Q", title="paid", format="$,.0f"), + ], + ) + .properties(title="Claims by EOB type", height=28 * len(_by_type), width=420) + ) + _monthly = q(""" + select date_trunc('month', clm_from_dt) as svc_month, count(*) n + from bcda.explanation_of_benefit where clm_from_dt >= date '2015-01-01' + group by 1 order by 1 + """) + monthly_chart = ( + alt.Chart(_monthly) + .mark_line(strokeWidth=2, point=alt.OverlayMarkDef(size=36)) + .encode( + x=alt.X("svc_month:T", title=None), + y=alt.Y("n:Q", title="claims per month"), + tooltip=[ + alt.Tooltip("svc_month:T", title="month", format="%b %Y"), + alt.Tooltip("n:Q", title="claims", format=","), + ], + ) + .properties(title="Claims by service month", height=220, width=620) + ) + plan_chart = ( + alt.Chart(plans) + .mark_bar(cornerRadiusEnd=4, size=18) + .encode( + y=alt.Y("class_plan:N", sort="-x", title=None), + x=alt.X("n:Q", title="coverage rows"), + tooltip=[ + alt.Tooltip("class_plan:N", title="plan"), + alt.Tooltip("n:Q", title="rows", format=","), + ], + ) + .properties(title="Coverage rows by plan", height=28 * len(plans), width=420) + ) + mo.vstack( + [ + mo.hstack( + [mo.ui.altair_chart(claims_chart), mo.ui.altair_chart(plan_chart)], + justify="start", + ), + mo.ui.altair_chart(monthly_chart), + mo.ui.table(prov, label="Claims and payments by System-Type × EOB-TYPE"), + ] + ) + return + + +@app.cell(hide_code=True) +def _(ROOT, con, mo, pl, q): + # ── 6. Dictionary field population + CCLF crosswalk ── + _cols = q( + "select table_name, column_name from information_schema.columns where table_schema = 'bcda' order by 1, 2" + ) + _pop_rows = [] + for _t in ("patient", "coverage", "explanation_of_benefit", "eob_item"): + _names = _cols.filter(pl.col("table_name") == _t)["column_name"].to_list() + _expr = ", ".join(f'count("{c}") as "{c}"' for c in _names) + _r = con.execute(f"select count(*) as _n, {_expr} from bcda.{_t}").fetchone() + _n = _r[0] + for _c, _v in zip(_names, _r[1:], strict=True): + _pop_rows.append( + { + "table": _t, + "column": _c, + "non_null": _v, + "pct": round(100 * _v / _n, 1) if _n else 0.0, + } + ) + population = pl.DataFrame(_pop_rows) + _summary = ( + population.group_by("table") + .agg( + pl.len().alias("columns"), + (pl.col("non_null") > 0).sum().alias("populated"), + ) + .with_columns( + (100 * pl.col("populated") / pl.col("columns")) + .round(1) + .alias("pct_populated") + ) + .sort("table") + ) + + # Crosswalk: the dictionary row behind each populated column, by IDR source column + _seeds = ROOT / "dev" / "seeds" + _has_ref = ( + con.execute( + "select count(*) from information_schema.tables where table_schema='reference_data' and table_name='bcda_v3_data_dictionary'" + ).fetchone()[0] + > 0 + ) + if _has_ref: + dictionary = q( + "select fhir_resource, field_name, fhir_path, source_column, ccw_mapping, cclf_mapping from reference_data.bcda_v3_data_dictionary" + ) + crosswalk = q( + "select cclf_file, element, field_name, field_label, bcda_v3_field, fhir_path from reference_data.bcda_v3_cclf_crosswalk" + ) + absent = q( + "select cclf_file, element, field_name, field_description from reference_data.bcda_v3_cclf_absent" + ) + _src = "reference_data (stack load seed)" + elif (_seeds / "bcda_v3_data_dictionary.csv").exists(): + dictionary = pl.read_csv( + _seeds / "bcda_v3_data_dictionary.csv", infer_schema_length=10000 + ) + crosswalk = pl.read_csv( + _seeds / "bcda_v3_cclf_crosswalk.csv", infer_schema_length=10000 + ) + absent = pl.read_csv( + _seeds / "bcda_v3_cclf_absent.csv", infer_schema_length=10000 + ) + _src = "dev/seeds CSVs" + else: + dictionary = crosswalk = absent = None + _src = "unavailable" + + _mapped = ( + crosswalk.filter( + pl.col("bcda_v3_field").is_not_null() & (pl.col("bcda_v3_field") != "") + ) + if crosswalk is not None + else None + ) + mo.vstack( + [ + mo.md(f""" + ## 6 · Dictionary field population and the CCLF crosswalk + + Columns are named after the v3 Data Dictionary's `ccwMapping` (legacy CCW names) or its IDR + `sourceColumn` (v3-only). A populated column means the v3 payload actually carried that element + for this entity; empty ones are either v2-only (monthly `dual_*`, `buyin*`, `rfrnc_yr`), Part D-only + when no Part D coverage exists, or not emitted by the sandbox generator. Crosswalk source: *{_src}*. + """), + mo.ui.table(_summary, label="Populated columns per table"), + mo.ui.table( + population.sort(["table", "pct"], descending=[False, True]), + label="Non-null count per column", + ), + ] + + ( + [ + mo.md( + f"**CCLF → BCDA v3**: {len(_mapped):,} of {len(crosswalk):,} CCLF elements have a v3 field; {len(absent):,} CCLF fields have no v3 equivalent (HICNs, FIPS county, address lines, BETOS, several CCLFA/B amounts)." + ), + mo.ui.table( + dictionary, + label="BCDA v3 Data Dictionary (fhirPath · IDR sourceColumn · CCW · CCLF)", + ), + mo.ui.table(_mapped, label="CCLF element → BCDA v3 field"), + mo.ui.table(absent, label="CCLF fields absent from BCDA v3"), + ] + if dictionary is not None + else [ + mo.md( + "Crosswalk seeds not found — run `uv run python dev/scripts/gen_bcda_v3_crosswalk.py` and `stack load seed`." + ) + ] + ) + ) + return + + +if __name__ == "__main__": + app.run()