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.*`, runs the connector
       pipeline (`bcda.pipe.cclf`: v3 routing on EOB-TYPE, Final-Action filter,
       SharedSystem→NCH dedupe) into `input_layer.*`, then republishes the
       read-only replica.
    5. **Proof** — identity, coverage shape, claim provenance, line amounts,
       referential integrity, and how many dictionary-mapped columns actually
       carry data, each marked pass / fail.
    6. **input_layer** — the connector's medical_claim / pharmacy_claim /
       eligibility reconciled against the `bcda.*` tables they came from.

    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(
            run_pipeline=True
        )  # 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 _(mo, q):
    # ── 6. input_layer — the connector pipeline reconciled against bcda.* ──
    _il = q(
        "select table_name from information_schema.tables where table_schema = 'input_layer'"
    )["table_name"].to_list()
    mo.stop(
        not {"medical_claim", "pharmacy_claim", "eligibility"} <= set(_il),
        mo.md(
            "## 6 · input_layer\n\n⚠️ `input_layer.*` not in the replica yet — run the "
            "load above (it runs `bcda.pipe.cclf`) or `stack load bcda --skip-flatten --run-pipeline`."
        ),
    )
    il_checks: list[tuple[bool, str, str]] = []

    def il(ok: bool, name: str, detail: str) -> None:
        il_checks.append((bool(ok), name, detail))

    _src = q("""
        select
            count(*) filter (where final_action = 'FinalAction' and eob_type <> 'PDE') final_medical,
            count(*) filter (where final_action = 'FinalAction' and eob_type = 'PDE') final_pde,
            count(*) filter (where final_action = 'NotFinalAction') not_final,
            count(*) filter (where final_action = 'FinalAction' and eob_type in ('INPATIENT','OUTPATIENT','SNF','HHA','HOSPICE')) final_inst,
            count(*) filter (where final_action = 'FinalAction' and eob_type in ('CARRIER','DME')) final_prof
        from bcda.explanation_of_benefit
    """).to_dicts()[0]
    _mc = q("""
        select claim_type, count(*) lines, count(distinct claim_id) claims,
               count(person_id) with_person, count(paid_amount) with_paid, count(charge_amount) with_charge,
               count(deductible_amount) with_ded, count(coinsurance_amount) with_coins,
               count(hcpcs_code) with_hcpcs, count(rendering_npi) with_rendering, count(billing_npi) with_billing,
               count(bill_type_code) with_tob, count(drg_code) with_drg, count(admission_date) with_adm,
               count(place_of_service_code) with_pos, count(diagnosis_code_1) with_dx, count(paid_date) with_paid_dt
        from input_layer.medical_claim group by 1 order by 1
    """)
    _by = {r["claim_type"]: r for r in _mc.to_dicts()}
    _inst, _prof = _by.get("institutional", {}), _by.get("professional", {})
    _claims = sum(r["claims"] for r in _mc.to_dicts())
    il(
        _claims == _src["final_medical"],
        "Every final-action medical EOB is in medical_claim, once",
        f"{_claims:,} claims = {_src['final_medical']:,} FinalAction non-PDE EOBs; {_src['not_final']:,} NotFinalAction excluded",
    )
    il(
        _inst.get("claims") == _src["final_inst"]
        and _prof.get("claims") == _src["final_prof"],
        "Routed by EOB-TYPE: institutional vs professional",
        f"institutional {_inst.get('claims', 0):,} / {_src['final_inst']:,} · professional {_prof.get('claims', 0):,} / {_src['final_prof']:,}",
    )
    _no_person = q(
        "select count(*) n from input_layer.medical_claim where person_id is null"
    ).to_dicts()[0]["n"]
    il(
        _no_person == 0,
        "Every line resolves to an MBI (person_id)",
        f"{_no_person:,} lines without MBI",
    )
    il(
        _inst.get("with_adm") == _inst.get("lines")
        and _inst.get("with_tob", 0) >= 0.99 * _inst.get("lines", 1),
        "Institutional lines carry admission date and NUBC type of bill",
        f"TOB {_inst.get('with_tob', 0):,}, admission {_inst.get('with_adm', 0):,} of {_inst.get('lines', 0):,}",
    )
    _tob_len = q(
        "select count(*) n from input_layer.medical_claim where claim_type='institutional' and length(bill_type_code) <> 4"
    ).to_dicts()[0]["n"]
    il(
        _tob_len == 0,
        "Type of bill is the 4-character NUBC code (not fac+clsfctn+freq)",
        f"{_tob_len:,} malformed",
    )
    il(
        _inst.get("with_paid") == _inst.get("lines")
        and _inst.get("with_charge") == _inst.get("lines")
        and _inst.get("with_ded") == _inst.get("lines")
        and _inst.get("with_coins") == _inst.get("lines"),
        "Institutional paid / charge / deductible / coinsurance from v3 Adjudication",
        f"paid {_inst.get('with_paid', 0):,} · charge {_inst.get('with_charge', 0):,} · ded {_inst.get('with_ded', 0):,} · coins {_inst.get('with_coins', 0):,}",
    )
    il(
        _prof.get("with_paid") == _prof.get("lines")
        and _prof.get("with_pos", 0) >= 0.95 * _prof.get("lines", 1)
        and _prof.get("with_rendering") == _prof.get("lines"),
        "Professional lines: paid, place of service, rendering NPI",
        f"paid {_prof.get('with_paid', 0):,} · POS {_prof.get('with_pos', 0):,} · rendering {_prof.get('with_rendering', 0):,} of {_prof.get('lines', 0):,}",
    )
    il(
        _inst.get("with_drg", 0) > 0 and _inst.get("with_dx", 0) == _inst.get("lines"),
        "Institutional DRG and header diagnoses present",
        f"DRG {_inst.get('with_drg', 0):,} · dx1 {_inst.get('with_dx', 0):,}",
    )
    il(
        (_inst.get("with_paid_dt", 0) + _prof.get("with_paid_dt", 0))
        == (_inst.get("lines", 0) + _prof.get("lines", 0)),
        "Paid date resolved (clm_pd_dt → NCH weekly → CMS process)",
        f"{_inst.get('with_paid_dt', 0) + _prof.get('with_paid_dt', 0):,} lines",
    )

    _rx = q("""
        select count(*) n, count(distinct claim_id) claims, count(prescribing_provider_npi) prs, count(dispensing_provider_npi) dsp,
               count(ndc_code) ndc, count(quantity) qty, count(days_supply) ds, count(refills) rf, count(charge_amount) chg,
               count(copayment_amount) cop, count(paid_amount) paid, count(paid_date) pd
        from input_layer.pharmacy_claim
    """).to_dicts()[0]
    il(
        _rx["claims"] == _src["final_pde"],
        "Every final-action PDE is in pharmacy_claim, once",
        f"{_rx['claims']:,} = {_src['final_pde']:,}",
    )
    il(
        _rx["prs"] == _rx["n"]
        and _rx["dsp"] == _rx["n"]
        and _rx["ndc"] == _rx["n"]
        and _rx["qty"] == _rx["n"]
        and _rx["ds"] == _rx["n"],
        "Part D: prescriber, pharmacy, NDC, quantity, days supply populated",
        f"{_rx['n']:,} rows",
    )
    il(
        _rx["chg"] == _rx["n"] and _rx["cop"] == _rx["n"],
        "Part D: total drug cost and paid-by-patient populated",
        f"charge {_rx['chg']:,} · copay {_rx['cop']:,} · plan-paid {_rx['paid']:,} (sandbox emits none)",
    )

    _el = q("""
        select count(*) n, count(distinct person_id) people, count(gender) g, count(enrollment_start_date) es,
               count(medicare_status_code) ms, count(original_reason_entitlement_code) orec, count(dual_status_code) dual,
               count(enrollment_status) st, count(*) filter (where gender = 'unknown') g_unk, count(*) filter (where race = 'unknown') r_unk
        from input_layer.eligibility
    """).to_dicts()[0]
    _pat = q("select count(*) n from bcda.patient").to_dicts()[0]["n"]
    _dual_rows = q(
        "select count(distinct subscriber_id) n from bcda.coverage where class_plan = 'Dual Medicare/Medicaid'"
    ).to_dicts()[0]["n"]
    _ab_rows = q(
        "select count(distinct subscriber_id) n from bcda.coverage where class_plan in ('Part A', 'Part B')"
    ).to_dicts()[0]["n"]
    il(
        _el["n"] == _pat and _el["people"] == _pat,
        "One eligibility row per Patient",
        f"{_el['n']:,} = {_pat:,}",
    )
    il(
        _el["g_unk"] == 0,
        "Gender resolved for every row",
        f"{_el['g_unk']:,} unknown; race unknown {_el['r_unk']:,} (OMB UNK)",
    )
    il(
        _el["es"] == _el["ms"] == _el["st"] == _ab_rows,
        "Enrollment period, Medicare status and status from the Part A/B Coverage row",
        f"{_el['es']:,} rows = {_ab_rows:,} beneficiaries with a Part A/B row; {_el['n'] - _el['es']:,} patients have none in this pull",
    )
    il(
        _el["dual"] == _dual_rows,
        "Dual status from the Dual Medicare/Medicaid Coverage row",
        f"{_el['dual']:,} = {_dual_rows:,} beneficiaries with a Dual row",
    )

    _passed = sum(1 for ok, _, _ in il_checks if ok)
    mo.vstack(
        [
            mo.md(f"## 6 · input_layer — {_passed} / {len(il_checks)} checks pass"),
            mo.md(
                "\n".join(
                    f"- {'✅' if ok else '❌'} **{name}** — {detail}"
                    for ok, name, detail in il_checks
                )
            ),
            mo.ui.table(
                _mc,
                label="medical_claim by claim_type: lines, claims and populated fields",
            ),
        ]
    )
    return (il_checks,)


@app.cell(hide_code=True)
def _(ROOT, con, mo, pl, q):
    # ── 7. 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"""
            ## 7 · 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


@app.cell(hide_code=True)
def _(STORE_PATH, mo, pl):
    """## 8 · Incremental pulls — watermarks, checks, ledger."""
    from bcda import sync as _sync
    from bcda.store import Store as _Store

    _store = _Store(path=str(STORE_PATH))
    _status = _sync.status(_store)
    _next = _sync.next_since(_store)
    _blocks = []
    for _rtype, _since in _next.items():
        if _since is None:
            _blocks.append(
                {
                    "type": _rtype,
                    "_since": "— no baseline —",
                    "level": "block",
                    "code": "no_baseline",
                    "message": "run an unfiltered pull first",
                }
            )
            continue
        for _c in _sync.check_since(_store, _since, types=[_rtype]):
            _blocks.append({"type": _rtype, "_since": _since, **_c.__dict__})
    _ledger_rows = [
        {
            "seq": _r["applied_seq"],
            "job": _r["job_key"],
            "kind": _r["kind"],
            "ok": _r["ok"],
            "since": _r["since"],
            "transaction_time": _r["transaction_time"],
            "table": _t,
            **_c,
        }
        for _r in _status["ledger"]
        for _t, _c in (_r["counts"] or {}).items()
    ]
    mo.vstack(
        [
            mo.md(f"""
            ## 8 · Incremental pulls (`_since`)

            BCDA filters `_since` on `meta.lastUpdated`, returns **all** resources for newly attributed
            enrollees on `/Group/all`, and recommends the previous job's `transactionTime` as the next `_since`.
            Each claim *version* has a new FHIR id, so claims are keyed by control number; the newest
            `lastUpdated` wins, then NCH/DDPS over SharedSystem, then the later job. Only a baseline retires rows.
            `bcda.sync` keeps a per-job ledger and re-derives the current tables from the applied set on every
            apply, so the result is independent of arrival order (tested in `tests/bcda/test_sync.py`).

            Store: `{STORE_PATH}` · {len(_status["jobs"])} complete job(s), {len(_status["pending"])} not yet reconciled.
            Next `_since`: {", ".join(f"`{t}` → `{v or '—'}`" for t, v in _next.items())}
            """),
            mo.ui.table(
                pl.DataFrame(_status["watermarks"])
                if _status["watermarks"]
                else pl.DataFrame({"note": ["no complete jobs"]}),
                label="Watermarks",
            ),
            mo.ui.table(
                pl.DataFrame(_blocks).select(
                    "type", "_since", "level", "code", "message"
                )
                if _blocks
                else pl.DataFrame({"note": ["—"]}),
                label="Checks for the watermark _since",
            ),
            mo.ui.table(
                pl.DataFrame(_ledger_rows)
                if _ledger_rows
                else pl.DataFrame(
                    {"note": ["nothing applied yet — `stack bcda sync apply`"]}
                ),
                label="Sync ledger (per job × table)",
            ),
        ]
    )
    return


if __name__ == "__main__":
    app.run()