stack/notebooks/bcda_v3_sandbox.py

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()