diff --git a/src/pfs/lineage.py b/src/pfs/lineage.py new file mode 100644 index 0000000..503c500 --- /dev/null +++ b/src/pfs/lineage.py @@ -0,0 +1,73 @@ +"""Lineage events for a code: what happened, when, and where the record +says so. + +Two independent sources: the RVU files (``pfs.rvu`` base rows year over +year — appeared / disappeared / status_change / descriptor_change / +revalued) and the Federal Register (``fr_anchors`` paragraphs that name +the code next to an event verb — created / adopted_cpt / replaces / +replaced_by / deleted / crosswalk / bundled / telehealth_list). +``lineage`` merges them and marks each RVU event ``anchored`` when an FR +event for the same code lies within ±1 rule year; the rest are +``unanchored`` for review, never silently accepted. +""" + +from __future__ import annotations + +from typing import Any + +from pfs.codetables import EventRow + +RVU_KINDS = ( + "appeared", + "disappeared", + "status_change", + "descriptor_change", + "revalued", +) + + +def rvu_year_span(con: Any) -> tuple[int, int]: + lo, hi = con.execute("SELECT min(year), max(year) FROM pfs.rvu").fetchone() + return int(lo), int(hi) + + +def _base_rows(con: Any, code: str) -> list[tuple[int, str, str, float | None]]: + return [ + (int(y), s or "", d or "", t) + for y, s, d, t in con.execute( + "SELECT year, status_code, description, non_fac_total FROM pfs.rvu " + "WHERE hcpcs = ? AND (mod IS NULL OR mod = '') " + "QUALIFY row_number() OVER (PARTITION BY year ORDER BY mod NULLS FIRST) = 1 " + "ORDER BY year", + [code.upper()], + ).fetchall() + ] + + +def _ev(code: str, year: int, kind: str, note: str = "") -> EventRow: + return EventRow(code.upper(), year, kind, "", "", "", 0, 0, "rvu", False, note) + + +def rvu_events( + con: Any, code: str, *, revalue_threshold: float = 0.10 +) -> list[EventRow]: + rows = _base_rows(con, code) + if not rows: + return [] + lo, hi = rvu_year_span(con) + out: list[EventRow] = [] + first, last = rows[0][0], rows[-1][0] + if first > lo: + out.append(_ev(code, first, "appeared")) + if last < hi: + out.append(_ev(code, last + 1, "disappeared")) + for (py, ps, pd, pt), (y, s, d, t) in zip(rows, rows[1:]): + if s != ps: + out.append(_ev(code, y, "status_change", f"{ps}→{s}")) + if d != pd: + out.append(_ev(code, y, "descriptor_change", f"{pd} → {d}")) + if pt and t is not None and pt > 0: + delta = (t - pt) / pt + if abs(delta) > revalue_threshold: + out.append(_ev(code, y, "revalued", f"{delta:+.1%}")) + return sorted(out, key=lambda e: (e.year, RVU_KINDS.index(e.kind))) diff --git a/tests/pfs/test_lineage.py b/tests/pfs/test_lineage.py new file mode 100644 index 0000000..c6673ff --- /dev/null +++ b/tests/pfs/test_lineage.py @@ -0,0 +1,83 @@ +"""pfs.lineage — dated events from RVU diffs and FR paragraphs, cross-checked.""" + +from __future__ import annotations + +import duckdb +import pytest + +from pfs.lineage import rvu_events, rvu_year_span + +RVU_COLS = "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER" + + +@pytest.fixture +def con(): + c = duckdb.connect(":memory:") + c.execute("CREATE SCHEMA pfs") + c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})") + rows = [ + # table spans 2015..2026 (filler code keeps the span honest) + *[("00000", None, "filler", "A", 1.0, y) for y in range(2015, 2027)], + # 99487: B in 2015-2016, A from 2017; description rename 2021; +25% revalue 2022 + ("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2015), + ("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2016), + ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.00, 2017), + ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.02, 2018), + ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.05, 2019), + ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.10, 2020), + ("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.12, 2021), + ("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, 2022), + *[ + ("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, y) + for y in range(2023, 2027) + ], + ("99487", "26", "modifier row ignored", "A", 99.0, 2022), + # G2058: 2020 only + ("G2058", None, "Ccm add 20min", "A", 1.0, 2020), + # 99439: 2021 onward + *[ + ("99439", "", "Chrnc care mgmt svc ea addl", "A", 1.0, y) + for y in range(2021, 2027) + ], + ] + c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", rows) + yield c + c.close() + + +class TestSpan: + def test_span(self, con): + assert rvu_year_span(con) == (2015, 2026) + + +class TestRvuEvents: + def test_status_descriptor_and_revalue(self, con): + ev = rvu_events(con, "99487") + kinds = [(e.year, e.kind, e.note) for e in ev] + assert (2017, "status_change", "B→A") in kinds + assert ( + 2021, + "descriptor_change", + "Cmplx chron care w/o pt vsit → Cplx chrnc care 1st 60 min", + ) in kinds + assert (2022, "revalued", "+25.0%") in kinds + assert not any( + k == "appeared" for _, k, _ in kinds + ) # present from the first table year + assert all( + e.source == "rvu" and e.anchored is False and e.code == "99487" for e in ev + ) + + def test_appeared_and_disappeared(self, con): + ev = rvu_events(con, "G2058") + assert [(e.year, e.kind) for e in ev] == [ + (2020, "appeared"), + (2021, "disappeared"), + ] + + def test_appeared_only(self, con): + ev = rvu_events(con, "99439") + assert [(e.year, e.kind) for e in ev] == [(2021, "appeared")] + + def test_unknown_code(self, con): + assert rvu_events(con, "99999") == []