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

    con = connect.duckdb()

    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 locality, locality_name
        FROM pfs.gpci
        ORDER BY locality
        """
    ).fetchall()

    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={f"{loc} - {name}": loc for loc, name in _localities},
        value=f"{_localities[0][0]} - {_localities[0][1]}",
        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
    _locality = locality_picker.value
    _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):
    carrier_result = q(f"""
        SELECT year, 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 locality = '{locality_picker.value}'
          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)
    _locality = locality_picker.value
    _hcpcs = hcpcs_input.value.strip()
    _cf = RULES[_year].conversion_factor

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

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

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

    calc_f = payment(rvu_df, gpci_df, 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)
    _cf = _RULES[_year].conversion_factor

    sql_result = q(f"""
        SELECT
            r.hcpcs,
            r.mod,
            r.year,
            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.locality = '{locality_picker.value}'
        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)
        _comparison = f"""
        ## Comparison

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

        Small differences are normal — CMS rounds intermediate values
        and may apply adjustments not captured in the base formula.
        """
    else:
        _comparison = "*Select a valid year/locality/code combination to see comparison.*"

    mo.md(_comparison)
    return


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