feat(pfs): valuation rows by vintage from the DuckDB replica with pfs.rules conversion factors (refs P48)

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

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),
+        ]