stack/notebooks/pfs_calcs.py

import marimo

__generated_with = "0.20.2"
app = marimo.App(width="medium")


@app.cell(hide_code=True)
def _():
    import marimo as mo

    return (mo,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    # PFS Payment Calculator

    Three ways to get a Medicare Physician Fee Schedule payment rate
    for a given **year**, **locality**, and **HCPCS code**.
    """)
    return


@app.cell(hide_code=True)
def _():
    import polars as pl

    from conf import connect

    # PFS reference data lives in the DuckLake lakehouse (M5, #514);
    # queries are unchanged — the lake is the default database.
    con = connect.ducklake()

    def q(sql):
        return con.execute(sql).pl()

    return con, pl, q


@app.cell(hide_code=True)
def _(con, mo):
    _years = sorted(
        r[0]
        for r in con.execute(
            "SELECT DISTINCT year FROM pfs.rvu ORDER BY year"
        ).fetchall()
    )
    _localities = con.execute(
        """
        SELECT DISTINCT mac, locality, locality_name
        FROM pfs.gpci
        ORDER BY locality_name
        """
    ).fetchall()

    _loc_options = {f"{name} ({loc})": f"{mac}|{loc}" for mac, loc, name in _localities}
    _first_key = next(iter(_loc_options))

    year_picker = mo.ui.dropdown(
        options={str(y): y for y in _years},
        value=str(_years[-1]),
        label="Year",
    )
    locality_picker = mo.ui.dropdown(
        options=_loc_options,
        value=_first_key,
        label="Locality",
    )
    hcpcs_input = mo.ui.text(value="99213", label="HCPCS Code")

    mo.hstack([year_picker, locality_picker, hcpcs_input], justify="start", gap=1)
    return hcpcs_input, locality_picker, year_picker


@app.cell(hide_code=True)
def _(hcpcs_input, locality_picker, year_picker):
    _year = year_picker.value
    _mac, _locality = locality_picker.value.split("|")
    _hcpcs = hcpcs_input.value.strip()
    return


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## 1 - Carrier Locality Lookup

    Pre-computed by CMS. The `pfs.carrier_locality` table has the
    published fee for every year/locality/code combination.
    This is the ground truth — it accounts for rounding, GPCI floors,
    and carrier-priced edge cases.
    """)
    return


@app.cell
def _(hcpcs_input, locality_picker, q, year_picker):
    _mac, _loc = locality_picker.value.split("|")
    carrier_result = q(f"""
        SELECT year, mac, locality, hcpcs, mod,
               non_fac_fee, fac_fee,
               non_fac_limiting_charge, fac_limiting_charge
        FROM pfs.carrier_locality
        WHERE year = {year_picker.value}
          AND mac = '{_mac}'
          AND locality = '{_loc}'
          AND hcpcs = '{hcpcs_input.value.strip()}'
        ORDER BY mod
    """)
    carrier_result
    return (carrier_result,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## 2 - Calculated from RVUs + GPCIs

    Uses `pfs.calcs.payment` which implements:

    ```
    Payment = (Work_RVU x Work_GPCI
             + PE_RVU x PE_GPCI
             + MP_RVU x MP_GPCI) x CF
    ```
    """)
    return


@app.cell
def _(hcpcs_input, locality_picker, pl, q, year_picker):
    from pfs.calcs import payment
    from pfs.rules import RULES

    _year = int(year_picker.value)
    _mac, _locality = locality_picker.value.split("|")
    _hcpcs = hcpcs_input.value.strip()
    _cf = RULES[_year].conversion_factor

    rvu_df = q(f"""
        SELECT *, '{_locality}' as locality, '{_mac}' as mac
        FROM pfs.rvu
        WHERE year = {_year} AND hcpcs = '{_hcpcs}'
    """)

    gpci_df = q(f"""
        SELECT * FROM pfs.gpci
        WHERE year = {_year} AND mac = '{_mac}' AND locality = '{_locality}'
    """)

    calc_nf = payment(rvu_df, gpci_df, cf=_cf, facility=False).select(
        "hcpcs",
        "mod",
        "work_rvu",
        "non_fac_pe_rvu",
        "mp_rvu",
        "work_gpci",
        "pe_gpci",
        "mp_gpci",
        pl.col("payment_amount").round(2).alias("non_fac_payment"),
    )

    calc_f = payment(rvu_df, gpci_df, cf=_cf, facility=True).select(
        "hcpcs",
        "mod",
        pl.col("payment_amount").round(2).alias("fac_payment"),
    )

    calc_result = calc_nf.join(calc_f, on=["hcpcs", "mod"], how="left")
    calc_result
    return (calc_result,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## 3 - Pure SQL

    Same formula, single query. Good for ad-hoc analysis.
    """)
    return


@app.cell
def _(hcpcs_input, locality_picker, q, year_picker):
    from pfs.rules import RULES as _RULES

    _year = int(year_picker.value)
    _mac, _loc = locality_picker.value.split("|")
    _cf = _RULES[_year].conversion_factor

    sql_result = q(f"""
        SELECT
            r.hcpcs,
            r.mod,
            r.year,
            g.mac,
            g.locality,
            g.locality_name,
            r.work_rvu,
            r.non_fac_pe_rvu,
            r.fac_pe_rvu,
            r.mp_rvu,
            g.work_gpci,
            g.pe_gpci,
            g.mp_gpci,
            round(
                (r.work_rvu * g.work_gpci
                 + r.non_fac_pe_rvu * g.pe_gpci
                 + r.mp_rvu * g.mp_gpci)
                * {_cf}, 2
            ) as non_fac_payment,
            round(
                (r.work_rvu * g.work_gpci
                 + r.fac_pe_rvu * g.pe_gpci
                 + r.mp_rvu * g.mp_gpci)
                * {_cf}, 2
            ) as fac_payment
        FROM pfs.rvu r
        JOIN pfs.gpci g ON r.year = g.year
        WHERE r.year = {_year}
          AND r.hcpcs = '{hcpcs_input.value.strip()}'
          AND g.mac = '{_mac}'
          AND g.locality = '{_loc}'
        ORDER BY r.mod
    """)
    sql_result
    return (sql_result,)


@app.cell(hide_code=True)
def _(carrier_result, mo, pl, calc_result):
    _has_carrier = carrier_result.height > 0
    _has_calc = calc_result.height > 0

    _comparison = ""
    if _has_carrier and _has_calc:
        _carrier_base = carrier_result.filter(
            (pl.col("mod").is_null()) | (pl.col("mod") == "")
        )
        _calc_base = calc_result.filter(
            (pl.col("mod").is_null()) | (pl.col("mod") == "")
        )
        _carrier_nf = _carrier_base.head(1).select("non_fac_fee").item()
        _calc_nf = _calc_base.head(1).select("non_fac_payment").item()
        _diff = round(abs(_carrier_nf - _calc_nf), 2)
        _match = "EXACT MATCH" if _diff == 0 else f"${_diff:.2f} difference"
        _comparison = f"""
        ## Comparison

        | Source | Non-Facility Fee |
        |--------|-----------------|
        | CMS Carrier File | **${_carrier_nf:.2f}** |
        | Calculated (RVU x GPCI x CF) | **${_calc_nf:.2f}** |
        | Difference | **{_match}** |

        The MAC-aware join (RVU x GPCI by MAC + locality) produces
        exact parity with CMS carrier file rates.  Any difference
        indicates a carrier-priced or status-indicator edge case.
        """
    else:
        _comparison = (
            "*Select a valid year/locality/code combination to see comparison.*"
        )

    mo.md(_comparison)
    return


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