feat(pfs): code-family registry + deterministic code detection (refs P48)

src/pfs/families.py

@@ -0,0 +1,99 @@
+"""Code families and deterministic code detection for chat questions.
+
+The registry used to live in a notebook cell
+(``notebooks/palliative_care_rfi.py``); this is its home in ``src`` so
+the chat, notebooks and scripts share one list. Pure: no I/O, no
+DuckDB, no narwhals — it must import inside the ``llm`` container.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+# HCPCS level II (letter + 4 digits) or CPT (5 digits). Validation against
+# the fee schedule happens at lookup time (pfs.valuation), not here.
+CODE_RE = re.compile(r"\b(?:[A-Z]\d{4}|\d{5})\b")
+
+
+def find_codes(text: str) -> tuple[str, ...]:
+    """Sorted unique codes literally present in *text* (case-normalised)."""
+    return tuple(sorted({m.group(0) for m in CODE_RE.finditer(text)}))
+
+
+@dataclass(frozen=True)
+class Family:
+    key: str
+    name: str
+    codes: tuple[str, ...]
+    synonyms: tuple[str, ...]  # lower-case; matched on word boundaries
+
+
+FAMILIES: dict[str, Family] = {
+    "ACP": Family(
+        "ACP",
+        "Advance Care Planning",
+        ("99497", "99498"),
+        ("acp", "advance care planning"),
+    ),
+    "CCM": Family(
+        "CCM",
+        "Chronic Care Management",
+        ("99437", "99439", "99487", "99489", "99490", "99491"),
+        ("ccm", "chronic care management"),
+    ),
+    "PCM": Family(
+        "PCM",
+        "Principal Care Management",
+        ("99424", "99425", "99426", "99427"),
+        ("pcm", "principal care management"),
+    ),
+    "TCM": Family(
+        "TCM",
+        "Transitional Care Management",
+        ("99495", "99496"),
+        ("tcm", "transitional care management"),
+    ),
+    "APCM": Family(
+        "APCM",
+        "Advanced Primary Care Management",
+        ("G0556", "G0557", "G0558"),
+        ("apcm", "advanced primary care management"),
+    ),
+}
+
+_CODE_TO_FAMILY: dict[str, Family] = {c: f for f in FAMILIES.values() for c in f.codes}
+
+
+def family_of(code: str) -> Family | None:
+    return _CODE_TO_FAMILY.get(code.upper())
+
+
+@dataclass(frozen=True)
+class Detection:
+    codes: tuple[str, ...]  # sorted unique: explicit + family-expanded
+    families: tuple[str, ...]  # family keys, sorted
+    explicit: tuple[str, ...]  # codes literally present in the text
+
+
+def detect_codes(text: str) -> Detection:
+    """Codes a question is about: explicit codes plus every code of any
+    family named (by synonym) or touched (by one member code)."""
+    explicit = find_codes(text)
+    lowered = text.lower()
+    families: set[str] = set()
+    for fam in FAMILIES.values():
+        if any(re.search(rf"\b{re.escape(s)}\b", lowered) for s in fam.synonyms):
+            families.add(fam.key)
+    for code in explicit:
+        fam = family_of(code)
+        if fam is not None:
+            families.add(fam.key)
+    codes = set(explicit)
+    for key in families:
+        codes.update(FAMILIES[key].codes)
+    return Detection(
+        codes=tuple(sorted(codes)),
+        families=tuple(sorted(families)),
+        explicit=explicit,
+    )

tests/pfs/test_families.py

@@ -0,0 +1,69 @@
+"""pfs.families — code-family registry + deterministic code detection."""
+
+from __future__ import annotations
+
+from pfs.families import FAMILIES, Detection, detect_codes, family_of, find_codes
+
+
+class TestFindCodes:
+    def test_hcpcs_and_cpt(self):
+        assert find_codes("Codes G0556 and 99490 apply; see g0557.") == (
+            "99490",
+            "G0556",
+        )
+
+    def test_no_false_hits_on_years_and_fr_pages(self):
+        # 2026 (4 digits) and 43842 (5 digits) — the CPT regex will match 43842;
+        # validation against the fee schedule happens at lookup, not here.
+        assert find_codes("91 FR 43842, CY2026") == ("43842",)
+
+    def test_dedupes_and_sorts(self):
+        assert find_codes("G0558 G0556 G0558") == ("G0556", "G0558")
+
+    def test_empty(self):
+        assert find_codes("") == ()
+
+
+class TestRegistry:
+    def test_apcm_family(self):
+        f = FAMILIES["APCM"]
+        assert f.codes == ("G0556", "G0557", "G0558")
+        assert "advanced primary care management" in f.synonyms
+        assert family_of("G0557") is f
+        assert family_of("00000") is None
+
+    def test_all_families_present(self):
+        assert set(FAMILIES) == {"ACP", "CCM", "PCM", "TCM", "APCM"}
+
+
+class TestDetectCodes:
+    def test_family_name_expands_to_codes(self):
+        d = detect_codes("What is APCM and how is it valued?")
+        assert d == Detection(
+            codes=("G0556", "G0557", "G0558"), families=("APCM",), explicit=()
+        )
+
+    def test_synonym_case_insensitive_word_boundary(self):
+        d = detect_codes("Tell me about Advanced Primary Care Management.")
+        assert d.families == ("APCM",)
+        assert detect_codes("the apcmx code").families == ()
+
+    def test_single_member_code_expands_family(self):
+        d = detect_codes("How much does G0557 pay?")
+        assert d.explicit == ("G0557",)
+        assert d.families == ("APCM",)
+        assert d.codes == ("G0556", "G0557", "G0558")
+
+    def test_unregistered_code_stays_alone(self):
+        d = detect_codes("value of 99213")
+        assert d == Detection(codes=("99213",), families=(), explicit=("99213",))
+
+    def test_multiple_families(self):
+        d = detect_codes("compare CCM and TCM")
+        assert d.families == ("CCM", "TCM")
+        assert d.codes == tuple(sorted(FAMILIES["CCM"].codes + FAMILIES["TCM"].codes))
+
+    def test_nothing(self):
+        assert detect_codes("what did commenters say about telehealth?") == Detection(
+            (), (), ()
+        )