homelab/stack
1
0
Fork 0
Code Issues 247 Pull Requests 2 Actions Packages Projects 1 Releases Wiki Activity

fix(pfs): pay only A/R/T statuses; window over RULES years; labelled CFs (refs P48)

Browse Source 
Multiplying RVUs by the CF for a bundled (B) or carrier-priced (C) code
put a payment on screen that Medicare never makes: pay_nf/pay_f are now
None outside statuses A/R/T and every row carries a status_note saying
why. The last-N window counted vintages pfs.rules has no CF for, which
could empty the table; it now counts only RULES years and warns about
the rest (and about a missing NPRM ProposedRule). cf_note labels which
of the year's several conversion factors is shown (non-APM standard,
the QP CF, a mid-year split).
This commit is contained in:
kert
2026-09-08 23:21:00 -04:00
parent 29fd22c1b1
commit 8d8d6a30f7
2 changed files with 199 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
src/pfs/valuation.py
View File

@@ -7,11 +7,14 @@ narwhals (this module runs inside the ``llm`` container).
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any, Sequence
from pfs.nprm import NPRM_SOURCES
from pfs.rules import RULES, proposed_for
from pfs.rules import RULES, ProposedRule, RuleYear, proposed_for
log = logging.getLogger(__name__)
@dataclass(frozen=True)
@@ -34,6 +37,8 @@ class ValuationRow:
label: str # "[PFS CY2026 Addendum B]"
citation: str # "90 FR 49266"
url: str
status_note: str = "" # why a non-A/R/T status is not paid ("" when paid)
cf_note: str = "" # which of the year's several CFs this one is
def vintage_label(year: int, proposed: bool) -> str:
@@ -47,8 +52,80 @@ def fr_citation_url(cite: str) -> str:
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)
#: Payment status codes paid as total RVUs × CF. Every other status is
#: priced some other way (or not at all), so multiplying its RVUs by the
#: CF would put a number on screen that Medicare never pays.
PAID_STATUSES = frozenset({"A", "R", "T"})
_STATUS_NOTES = {
"B": "bundled — no separate PFS payment",
"C": "carrier priced by the MAC",
"J": "anesthesia — paid on base units × the anesthesia CF",
"I": "not valid for Medicare",
"N": "non-covered",
"E": "excluded by regulation",
"M": "measurement code — not paid",
"X": "statutory exclusion",
"P": "bundled/excluded",
}
def status_note(status: str) -> str:
"""Why *status* is not paid from RVUs × CF — ``""`` when it is."""
code = (status or "").strip().upper()
if code in PAID_STATUSES:
return ""
if code in _STATUS_NOTES:
return _STATUS_NOTES[code]
if not code:
return "status not stated — no RVU-based payment"
return f"status {code} — no RVU-based payment"
def final_cf_note(rule: RuleYear) -> str:
"""Which of a final year's several conversion factors ``cf`` is.
Since CY2026 CMS publishes separate QP (APM) and non-QP standard
CFs, and a mid-year legislative update can split a year in two — an
unlabelled dollar figure would hide both.
"""
parts = ["non-APM standard CF"]
if rule.cf_qp is not None and rule.cf_qp != rule.conversion_factor:
parts.append(f"QP CF ${rule.cf_qp:.4f}")
segs = sorted(rule.cf_segments, key=lambda s: s.start)
if len(segs) > 1:
earlier = next(
(
(seg, nxt)
for seg, nxt in zip(segs, segs[1:])
if seg.cf != rule.conversion_factor
),
None,
)
if earlier is None:
parts.append("mid-year split — dominant CF shown")
else:
seg, nxt = earlier
parts.append(
f"mid-year split — dominant CF shown (${seg.cf:.4f} before {nxt.start})"
)
return "; ".join(parts)
def proposed_cf_note(prop: ProposedRule) -> str:
"""Which CF a proposed row shows (the NPRM's non-QP standard one)."""
parts = ["proposed; non-QP CF"]
if prop.cf_qp is not None and prop.cf_qp != prop.conversion_factor:
parts.append(f"QP CF ${prop.cf_qp:.4f}")
return "; ".join(parts)
def _pay(total: float | None, cf: float, status: str) -> float | None:
"""National unadjusted payment, or ``None`` when the status is not
paid from RVUs (see :data:`PAID_STATUSES`)."""
if total is None or (status or "").strip().upper() not in PAID_STATUSES:
return None
return round(total * cf, 2)
_FINAL_SQL = """
@@ -57,6 +134,7 @@ WITH base AS (
fac_pe_rvu, mp_rvu, non_fac_total, fac_total
FROM pfs.rvu
WHERE hcpcs IN (SELECT UNNEST(?::VARCHAR[]))
AND year IN (SELECT UNNEST(?::INTEGER[]))
QUALIFY row_number() OVER (
PARTITION BY hcpcs, year ORDER BY (mod IS NULL OR mod = '') DESC, mod
) = 1
@@ -77,6 +155,14 @@ ORDER BY hcpcs
"""
_UNKNOWN_YEARS_SQL = """
SELECT DISTINCT year FROM pfs.rvu
WHERE hcpcs IN (SELECT UNNEST(?::VARCHAR[]))
AND year NOT IN (SELECT UNNEST(?::INTEGER[]))
ORDER BY year
"""
def _sum(*vals: float | None) -> float | None:
if any(v is None for v in vals):
return None
@@ -93,13 +179,24 @@ def valuation(
if not wanted:
return [], []
by_code: dict[str, list[ValuationRow]] = {c: [] for c in wanted}
rule_years = sorted(RULES)
skipped = [
int(y)
for (y,) in con.execute(_UNKNOWN_YEARS_SQL, [wanted, rule_years]).fetchall()
]
if skipped:
log.warning(
"valuation: %s priced in pfs.rvu year(s) %s with no pfs.rules entry "
"— skipped (no conversion factor)",
", ".join(wanted),
", ".join(str(y) for y in skipped),
)
for hcpcs, year, desc, status, work, pe_nf, pe_f, mp, tot_nf, tot_f in con.execute(
_FINAL_SQL, [wanted, years]
_FINAL_SQL, [wanted, rule_years, years]
).fetchall():
rule = RULES.get(int(year))
if rule is None:
continue
rule = RULES[int(year)]
cf = rule.conversion_factor
cite = rule.federal_register_citation
by_code[hcpcs].append(
@@ -110,6 +207,7 @@ def valuation(
year=int(year),
proposed=False,
status=status or "",
status_note=status_note(status or ""),
work=work,
pe_nf=pe_nf,
pe_f=pe_f,
@@ -117,8 +215,9 @@ def valuation(
total_nf=tot_nf,
total_f=tot_f,
cf=cf,
pay_nf=_pay(tot_nf, cf),
pay_f=_pay(tot_f, cf),
pay_nf=_pay(tot_nf, cf, status or ""),
pay_f=_pay(tot_f, cf, status or ""),
cf_note=final_cf_note(rule),
label=vintage_label(int(year), False),
citation=cite,
url=fr_citation_url(cite),
@@ -127,7 +226,13 @@ def valuation(
nprm_year, _tag, rule_id, _pin = max(NPRM_SOURCES, key=lambda s: s[0])
prop = proposed_for(nprm_year)
if prop is not None:
if prop is None:
log.warning(
"valuation: no ProposedRule for CY%s — the newest NPRM vintage is "
"omitted from the table",
nprm_year,
)
else:
cf = prop.conversion_factor
cite = prop.federal_register_citation
for hcpcs, desc, status, work, pe_nf, pe_f, mp in con.execute(
@@ -142,6 +247,7 @@ def valuation(
year=nprm_year,
proposed=True,
status=status or "",
status_note=status_note(status or ""),
work=work,
pe_nf=pe_nf,
pe_f=pe_f,
@@ -149,8 +255,9 @@ def valuation(
total_nf=tot_nf,
total_f=tot_f,
cf=cf,
pay_nf=_pay(tot_nf, cf),
pay_f=_pay(tot_f, cf),
pay_nf=_pay(tot_nf, cf, status or ""),
pay_f=_pay(tot_f, cf, status or ""),
cf_note=proposed_cf_note(prop),
label=vintage_label(nprm_year, True),
citation=cite,
url=fr_citation_url(cite),

80
tests/pfs/test_valuation.py
View File

@@ -87,6 +87,37 @@ def con():
)
for y in range(2021, 2027)
],
# A vintage older than the first year pfs.rules knows (RULES starts
# at 2014): it has no CF, so it must not consume a window slot.
(
"G0556",
None,
"Adv prim care mgmt lvl 1",
"A",
0.25,
0.20,
0.10,
0.02,
0.47,
0.37,
30.0,
2013,
),
# B-status code with full RVUs — bundled, so never paid
(
"G0559",
None,
"Bundled service",
"B",
0.30,
0.20,
0.10,
0.02,
0.52,
0.42,
33.4009,
2026,
),
# I-status code with NULL components
(
"G9999",
@@ -204,6 +235,55 @@ class TestValuation:
(r,) = rows
assert r.status == "I" and r.work is None and r.pay_nf is None
def test_bundled_status_is_not_paid_and_is_annotated(self, con):
rows, _ = valuation(con, ["G0559"])
(r,) = rows
assert (r.status, r.total_nf, r.total_f) == ("B", 0.52, 0.42)
assert r.pay_nf is None and r.pay_f is None
assert r.status_note == "bundled — no separate PFS payment"
def test_paid_statuses_have_no_note(self, con):
rows, _ = valuation(con, ["G0556"], years=1)
r = rows[0]
assert r.status == "A" and r.status_note == ""
assert r.pay_nf == round(r.total_nf * r.cf, 2)
def test_unknown_status_note_names_the_code(self, con):
con.execute(
"INSERT INTO pfs.rvu VALUES "
"('G0560',NULL,'Odd','Q',0.1,0.1,0.1,0.0,0.2,0.2,33.4009,2026)"
)
(r,) = valuation(con, ["G0560"])[0]
assert r.pay_nf is None
assert r.status_note == "status Q — no RVU-based payment"
def test_window_counts_only_years_pfs_rules_knows(self, con, caplog):
"""A vintage with no RuleYear must not eat a slot in the last-N
window — otherwise asking for one year returns nothing at all."""
with caplog.at_level("WARNING"):
rows, _ = valuation(con, ["G0556"], years=1)
assert [(r.year, r.proposed) for r in rows] == [(2026, False), (2027, True)]
assert "2013" in caplog.text and "G0556" in caplog.text
def test_cf_note_final_year(self, con):
rows, _ = valuation(con, ["G0556"], years=1)
r = rows[0]
assert r.cf_note == (f"non-APM standard CF; QP CF ${RULES[2026].cf_qp:.4f}")
def test_cf_note_flags_a_mid_year_split(self, con):
rows, _ = valuation(con, ["99490"], years=3)
r24 = next(r for r in rows if r.year == 2024)
assert r24.cf_note.startswith("non-APM standard CF; mid-year split")
assert "$32.7442 before 2024-03-09" in r24.cf_note
def test_cf_note_proposed(self, con):
rows, _ = valuation(con, ["G0556"], years=1)
p = rows[-1]
assert p.proposed
assert p.cf_note == (
f"proposed; non-QP CF; QP CF ${proposed_for(2027).cf_qp:.4f}"
)
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] == [