diff --git a/src/pfs/valuation.py b/src/pfs/valuation.py new file mode 100644 index 0000000..00de133 --- /dev/null +++ b/src/pfs/valuation.py @@ -0,0 +1,166 @@ +"""RVUs, conversion factor and national unadjusted payment per code and +vintage, read from the DuckDB replica (``pfs.rvu`` final years, +``pfs.rvu_proposed`` for the newest NPRM). Conversion factors come from +``pfs.rules`` — never from ``pfs.rvu.conv_factor``. Pure DuckDB SQL; no +narwhals (this module runs inside the ``llm`` container). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Sequence + +from pfs.nprm import NPRM_SOURCES +from pfs.rules import RULES, proposed_for + + +@dataclass(frozen=True) +class ValuationRow: + code: str + description: str + vintage: str # "CY2026 final" | "CY2027 proposed" + year: int + proposed: bool + status: str + work: float | None + pe_nf: float | None + pe_f: float | None + mp: float | None + total_nf: float | None + total_f: float | None + cf: float + pay_nf: float | None + pay_f: float | None + label: str # "[PFS CY2026 Addendum B]" + citation: str # "90 FR 49266" + url: str + + +def vintage_label(year: int, proposed: bool) -> str: + return f"[CY{year} NPRM Addendum B]" if proposed else f"[PFS CY{year} Addendum B]" + + +def fr_citation_url(cite: str) -> str: + parts = cite.split() + if len(parts) != 3 or parts[1].upper() != "FR": + return "" + return f"https://www.federalregister.gov/citation/{parts[0]}-FR-{parts[2]}" + + +def _pay(total: float | None, cf: float) -> float | None: + return None if total is None else round(total * cf, 2) + + +_FINAL_SQL = """ +WITH base AS ( + SELECT hcpcs, year, description, status_code, work_rvu, non_fac_pe_rvu, + fac_pe_rvu, mp_rvu, non_fac_total, fac_total + FROM pfs.rvu + WHERE hcpcs IN (SELECT UNNEST(?::VARCHAR[])) + QUALIFY row_number() OVER ( + PARTITION BY hcpcs, year ORDER BY (mod IS NULL OR mod = '') DESC, mod + ) = 1 +) +SELECT * FROM base +QUALIFY dense_rank() OVER (PARTITION BY hcpcs ORDER BY year DESC) <= ? +ORDER BY hcpcs, year +""" + +_PROPOSED_SQL = """ +SELECT hcpcs, description, status_code, work_rvu, non_fac_pe_rvu, fac_pe_rvu, mp_rvu +FROM pfs.rvu_proposed +WHERE cms_rule_id = ? AND hcpcs IN (SELECT UNNEST(?::VARCHAR[])) +QUALIFY row_number() OVER ( + PARTITION BY hcpcs ORDER BY (mod IS NULL OR mod = '') DESC, mod +) = 1 +ORDER BY hcpcs +""" + + +def _sum(*vals: float | None) -> float | None: + if any(v is None for v in vals): + return None + return round(sum(vals), 4) # type: ignore[arg-type] + + +def valuation( + con: Any, codes: Sequence[str], *, years: int = 4 +) -> tuple[list[ValuationRow], list[str]]: + """Rows for *codes* (last *years* final vintages + the newest NPRM), + ordered by code then year with the proposed row last per code, and the + codes that produced no rows at all.""" + wanted = sorted({c.upper() for c in codes}) + if not wanted: + return [], [] + by_code: dict[str, list[ValuationRow]] = {c: [] for c in wanted} + + for hcpcs, year, desc, status, work, pe_nf, pe_f, mp, tot_nf, tot_f in con.execute( + _FINAL_SQL, [wanted, years] + ).fetchall(): + rule = RULES.get(int(year)) + if rule is None: + continue + cf = rule.conversion_factor + cite = rule.federal_register_citation + by_code[hcpcs].append( + ValuationRow( + code=hcpcs, + description=desc or "", + vintage=f"CY{year} final", + year=int(year), + proposed=False, + status=status or "", + work=work, + pe_nf=pe_nf, + pe_f=pe_f, + mp=mp, + total_nf=tot_nf, + total_f=tot_f, + cf=cf, + pay_nf=_pay(tot_nf, cf), + pay_f=_pay(tot_f, cf), + label=vintage_label(int(year), False), + citation=cite, + url=fr_citation_url(cite), + ) + ) + + nprm_year, _tag, rule_id, _pin = max(NPRM_SOURCES, key=lambda s: s[0]) + prop = proposed_for(nprm_year) + if prop is not None: + cf = prop.conversion_factor + cite = prop.federal_register_citation + for hcpcs, desc, status, work, pe_nf, pe_f, mp in con.execute( + _PROPOSED_SQL, [rule_id, wanted] + ).fetchall(): + tot_nf, tot_f = _sum(work, pe_nf, mp), _sum(work, pe_f, mp) + by_code[hcpcs].append( + ValuationRow( + code=hcpcs, + description=desc or "", + vintage=f"CY{nprm_year} proposed", + year=nprm_year, + proposed=True, + status=status or "", + work=work, + pe_nf=pe_nf, + pe_f=pe_f, + mp=mp, + total_nf=tot_nf, + total_f=tot_f, + cf=cf, + pay_nf=_pay(tot_nf, cf), + pay_f=_pay(tot_f, cf), + label=vintage_label(nprm_year, True), + citation=cite, + url=fr_citation_url(cite), + ) + ) + + rows = [ + r + for c in wanted + for r in sorted(by_code[c], key=lambda r: (r.year, r.proposed)) + ] + unpriced = [c for c in wanted if not by_code[c]] + return rows, unpriced diff --git a/tests/pfs/test_valuation.py b/tests/pfs/test_valuation.py new file mode 100644 index 0000000..5aac543 --- /dev/null +++ b/tests/pfs/test_valuation.py @@ -0,0 +1,215 @@ +"""pfs.valuation — RVUs + national payment by vintage from a DuckDB replica.""" + +from __future__ import annotations + +import duckdb +import pytest + +from pfs.rules import RULES, proposed_for +from pfs.valuation import ValuationRow, fr_citation_url, valuation, vintage_label + +RVU_COLS = ( + "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, " + "work_rvu DOUBLE, non_fac_pe_rvu DOUBLE, fac_pe_rvu DOUBLE, mp_rvu DOUBLE, " + "non_fac_total DOUBLE, fac_total DOUBLE, conv_factor DOUBLE, year INTEGER" +) +PROPOSED_COLS = ( + "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, " + "work_rvu DOUBLE, non_fac_pe_rvu DOUBLE, fac_pe_rvu DOUBLE, mp_rvu DOUBLE, cms_rule_id VARCHAR" +) + + +@pytest.fixture +def con(): + c = duckdb.connect(":memory:") + c.execute("CREATE SCHEMA pfs") + c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})") + c.execute(f"CREATE TABLE pfs.rvu_proposed ({PROPOSED_COLS})") + rows = [ + # G0556: 2025 + 2026 final, with a modifier row that must be ignored + ( + "G0556", + None, + "Adv prim care mgmt lvl 1", + "A", + 0.25, + 0.20, + 0.10, + 0.02, + 0.47, + 0.37, + 32.3465, + 2025, + ), + ( + "G0556", + None, + "Adv prim care mgmt lvl 1", + "A", + 0.25, + 0.22, + 0.06, + 0.02, + 0.49, + 0.33, + 33.4009, + 2026, + ), + ( + "G0556", + "26", + "Adv prim care mgmt lvl 1", + "A", + 9.0, + 9.0, + 9.0, + 9.0, + 27.0, + 27.0, + 33.4009, + 2026, + ), + # 99490: six years so the last-N cut applies (2021..2026) + *[ + ( + "99490", + "", + "Chrnc care mgmt srvc 20 min", + "A", + 1.0, + 1.0, + 0.5, + 0.05, + 2.05, + 1.55, + 30.0, + y, + ) + for y in range(2021, 2027) + ], + # I-status code with NULL components + ( + "G9999", + None, + "Not priced", + "I", + None, + None, + None, + None, + None, + None, + 33.4009, + 2026, + ), + ] + c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", rows) + c.executemany( + "INSERT INTO pfs.rvu_proposed VALUES (?,?,?,?,?,?,?,?,?)", + [ + ( + "G0556", + None, + "Adv prim care mgmt lvl 1", + "A", + 0.25, + 0.23, + 0.07, + 0.02, + "CMS-1848-P", + ), + ( + "G0556", + "26", + "Adv prim care mgmt lvl 1", + "A", + 9.0, + 9.0, + 9.0, + 9.0, + "CMS-1848-P", + ), + ("G0556", None, "older nprm", "A", 0.1, 0.1, 0.1, 0.1, "CMS-1832-P"), + ], + ) + yield c + c.close() + + +class TestHelpers: + def test_vintage_label(self): + assert vintage_label(2026, False) == "[PFS CY2026 Addendum B]" + assert vintage_label(2027, True) == "[CY2027 NPRM Addendum B]" + + def test_fr_citation_url(self): + assert ( + fr_citation_url("90 FR 49266") + == "https://www.federalregister.gov/citation/90-FR-49266" + ) + assert fr_citation_url("") == "" + + +class TestValuation: + def test_final_and_proposed_rows_with_cf_and_payment(self, con): + rows, unpriced = valuation(con, ["G0556"], years=4) + assert unpriced == [] + assert [(r.year, r.proposed) for r in rows] == [ + (2025, False), + (2026, False), + (2027, True), + ] + r26 = rows[1] + assert isinstance(r26, ValuationRow) + assert (r26.code, r26.status, r26.work, r26.pe_nf, r26.pe_f, r26.mp) == ( + "G0556", + "A", + 0.25, + 0.22, + 0.06, + 0.02, + ) + assert (r26.total_nf, r26.total_f) == (0.49, 0.33) + assert r26.cf == RULES[2026].conversion_factor + assert r26.pay_nf == round(0.49 * RULES[2026].conversion_factor, 2) + assert r26.pay_f == round(0.33 * RULES[2026].conversion_factor, 2) + assert r26.label == "[PFS CY2026 Addendum B]" + assert r26.citation == RULES[2026].federal_register_citation + assert r26.url == fr_citation_url(RULES[2026].federal_register_citation) + assert r26.vintage == "CY2026 final" + p = rows[2] + assert p.label == "[CY2027 NPRM Addendum B]" and p.vintage == "CY2027 proposed" + assert p.cf == proposed_for(2027).conversion_factor + assert p.total_nf == round(0.25 + 0.23 + 0.02, 4) and p.total_f == round( + 0.25 + 0.07 + 0.02, 4 + ) + assert p.pay_nf == round(p.total_nf * p.cf, 2) + assert p.citation == proposed_for(2027).federal_register_citation + + def test_modifier_rows_are_ignored(self, con): + rows, _ = valuation(con, ["G0556"]) + assert all(r.work == 0.25 for r in rows) + + def test_last_n_years_only(self, con): + rows, _ = valuation(con, ["99490"], years=4) + assert [r.year for r in rows if not r.proposed] == [2023, 2024, 2025, 2026] + + def test_unpriced_codes_reported(self, con): + rows, unpriced = valuation(con, ["G0556", "Z9999"]) + assert unpriced == ["Z9999"] + assert {r.code for r in rows} == {"G0556"} + + def test_null_components_pass_through(self, con): + rows, unpriced = valuation(con, ["G9999"]) + assert unpriced == [] + (r,) = rows + assert r.status == "I" and r.work is None and r.pay_nf is None + + def test_order_is_code_then_year(self, con): + rows, _ = valuation(con, ["G0556", "99490"], years=2) + assert [(r.code, r.year) for r in rows] == [ + ("99490", 2025), + ("99490", 2026), + ("G0556", 2025), + ("G0556", 2026), + ("G0556", 2027), + ]