From 5ab1aad74be21f0793f994fb2be55d9266d4f3b8 Mon Sep 17 00:00:00 2001 From: kert Date: Thu, 12 Nov 2026 18:08:10 -0500 Subject: [PATCH] =?UTF-8?q?feat(pfs):=20money=20stakes=20=E2=80=94=20PSPS?= =?UTF-8?q?=20utilization=20by=20specialty=20and=20CMS=20specialty=20impac?= =?UTF-8?q?t=20tables?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/cli/pfs.py | 218 +++++++++++++ src/pfs/codetables.py | 107 +++++++ src/pfs/psps.py | 234 ++++++++++++++ src/pfs/stakes.py | 382 +++++++++++++++++++++++ tests/cli/test_pfs_cli.py | 285 +++++++++++++++++ tests/pfs/fixtures/impact_cy2017_fr.txt | 14 + tests/pfs/fixtures/impact_cy2027_508.txt | 15 + tests/pfs/test_psps.py | 245 +++++++++++++++ tests/pfs/test_stakes.py | 382 +++++++++++++++++++++++ 9 files changed, 1882 insertions(+) create mode 100644 src/pfs/psps.py create mode 100644 src/pfs/stakes.py create mode 100644 tests/pfs/fixtures/impact_cy2017_fr.txt create mode 100644 tests/pfs/fixtures/impact_cy2027_508.txt create mode 100644 tests/pfs/test_psps.py create mode 100644 tests/pfs/test_stakes.py diff --git a/src/cli/pfs.py b/src/cli/pfs.py index 5e5b24c..57d5776 100644 --- a/src/cli/pfs.py +++ b/src/cli/pfs.py @@ -78,9 +78,25 @@ from pfs.guidance import build as build_guidance from pfs.guideline_facts import guideline_facts from pfs.lineage import lineage, lineage_all, seen_codes from pfs.origin import DEFAULT_BOOK, family_origin, origin_inventory, resolve_book +from pfs.psps import DATASETS as PSPS_YEARS +from pfs.psps import cache_path as psps_cache_path +from pfs.psps import cite as cite_psps +from pfs.psps import fetch_code_year as fetch_psps +from pfs.psps import load_cache as load_psps_cache +from pfs.psps import read_by_specialty as read_psps_rows +from pfs.psps import save_cache as save_psps_cache +from pfs.psps import to_rows as psps_rows +from pfs.psps import write as write_psps_rows from pfs.reaction import series as reaction_series from pfs.rounds import family_rounds, rounds_inventory from pfs.runlog import RunLog, ledger_path +from pfs.stakes import ( + discover_508, + discover_fr, + parse_impact_508, + parse_impact_fr_text, + write_impacts, +) from pfs.stance import ( DEFAULT_GENERATION, DEFAULT_METHOD, @@ -1068,6 +1084,208 @@ def utilization( con.close() +def _psps_names(con: Any) -> dict[str, str]: + """Medicare provider specialty code -> description.""" + try: + rows = con.execute( + "SELECT specialty_code, specialty_description " + "FROM terminology.cms_acceptable_provider_specialty_codes" + ).fetchall() + except Exception as exc: + if _missing_table(exc): + return {} + raise + return {str(c): str(d or "") for c, d in rows} + + +def _family_codes(con: Any, key: str) -> list[str]: + refresh_from(con) + fam = FAMILIES.get(normalize_key(key)) + if fam is None: + raise typer.BadParameter( + f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}" + ) + return sorted(fam.codes) + + +@app.command() +def psps( + family: str = typer.Option(..., "--family", help="Family key, e.g. HOME-VISITS."), + year: list[int] = typer.Option( + [], "--year", help="Calendar year(s); default every PSPS year." + ), + write: bool = typer.Option( + False, "--write", help="Fetch data.cms.gov, load, cite, republish." + ), + offline: bool = typer.Option( + False, "--offline", help="With --write: use the cached JSON, no network." + ), +) -> None: + """Medicare utilization by provider specialty for a family's codes + (pfs.utilization_specialty): the PSPS national totals per code, year + and specialty. --write pulls each (year, code) through the Data API, + caches it under data/cms/psps/ (never overwriting a cached file) and + republishes; without --write it prints what the replica holds.""" + import time + + years = sorted(year) if year else sorted(PSPS_YEARS) + unknown = [y for y in years if y not in PSPS_YEARS] + if unknown: + raise typer.BadParameter( + f"no PSPS dataset for {unknown}; known {min(PSPS_YEARS)}-{max(PSPS_YEARS)}" + ) + con = _read() + try: + codes = _family_codes(con, family) + names = _psps_names(con) + if not write: + try: + rows = read_psps_rows(con, codes, years=years) + except Exception as exc: + if _missing_table(exc): + typer.echo("no PSPS table yet - run with --write") + return + raise + finally: + con.close() + if not write: + for r in rows: + svc = f"{r.services:,.0f}" if r.services is not None else "-" + typer.echo( + f"{r.year} {r.hcpcs} {r.specialty_cd} {r.specialty:<32} {svc:>12}" + ) + typer.echo(f"psps: {len(rows)} rows") + return + store = _store() + staged: list[tuple[int, str, list[Any], str, str]] = [] + client = None if offline else _http_dossier() + try: + for y in years: + key = cite_psps(store, y) + for code in codes: + cache = psps_cache_path(y, code) + if offline: + if not cache.is_file(): + typer.echo(f"{y} {code}: no cached file, skipped") + continue + records = load_psps_cache(cache) + else: + records = fetch_psps(client, y, code) + if records: + cache = save_psps_cache(cache, records) + time.sleep(0.2) + rows = psps_rows( + y, + code, + records, + dataset_id=PSPS_YEARS[y], + item_key=key, + names=names, + ) + staged.append((y, code, rows, key, str(cache))) + typer.echo( + f"{y} {code}: {len(records)} lines -> {len(rows)} specialties" + ) + finally: + if client is not None: + client.close() + from cms.ingest_log import log_ingest + + with _batch() as con: + for y, code, rows, key, cache in staged: + n = write_psps_rows(con, y, code, rows) + log_ingest( + con, + module="pfs.psps", + table_name="pfs.utilization_specialty", + rows=n, + source_file=cache, + pincite_key=key, + ) + _publish() + typer.echo(f"psps: {sum(len(r) for _, _, r, _, _ in staged)} rows written") + + +def _http_dossier() -> Any: + import httpx + + return httpx.Client(headers={"User-Agent": "stack-pfs-dossier/1.0"}) + + +@app.command() +def impacts( + scan: bool = typer.Option( + False, + "--scan", + help="Every CMS 508 impact file in bib / data/cms/pfs_final and every " + "PFS rule whose FR text prints the table.", + ), + path: Path = typer.Option(None, "--path", help="One impact file or FR text."), + year: int = typer.Option(0, "--year", help="Rule year, with --path."), + vintage: str = typer.Option("final", "--vintage", help="final | proposed"), + write: bool = typer.Option(False, "--write"), +) -> None: + """CMS's estimated impact on total allowed charges by specialty + (pfs.specialty_impact): the 508 text files CMS posts with a rule and + the table printed in the 2012-2019 Federal Register texts.""" + from conf import ROOT + + if scan == (path is not None): + raise typer.BadParameter("pass exactly one of --scan or --path") + if vintage not in ("final", "proposed"): + raise typer.BadParameter("--vintage is final or proposed") + groups: list[tuple[int, str, str, list[Any], str, str]] = [] + if scan: + store = _store() + sources = discover_508(store, ROOT / "data/cms/pfs_final") + sources += discover_fr(store) + else: + if not year: + raise typer.BadParameter("--path needs --year") + kind = "508" if path.name.endswith("_508.txt") else "fr" + from pfs.stakes import ImpactSource + + sources = [ImpactSource(path, year, vintage, "", kind)] + for src in sources: + if src.kind == "508": + rows = parse_impact_508( + src.path, + rule_year=src.rule_year, + vintage=src.vintage, + item_key=src.item_key, + ) + else: + rows = parse_impact_fr_text( + src.path.read_text(errors="replace"), + rule_year=src.rule_year, + vintage=src.vintage, + item_key=src.item_key, + ) + groups.append( + (src.rule_year, src.vintage, src.kind, rows, src.item_key, str(src.path)) + ) + typer.echo( + f"CY{src.rule_year} {src.vintage} ({src.kind}): {len(rows)} rows " + f"<- {src.path.name}" + ) + if not write: + return + from cms.ingest_log import log_ingest + + with _batch() as con: + for ry, vin, kind, rows, key, file in groups: + n = write_impacts(con, ry, vin, rows, source=kind) + log_ingest( + con, + module="pfs.stakes", + table_name="pfs.specialty_impact", + rows=n, + source_file=file, + pincite_key=key, + ) + _publish() + + @app.command() def review(code: str = typer.Option("", "--code")) -> None: """Element lines the classifier could not place.""" diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py index f8b5256..10780be 100644 --- a/src/pfs/codetables.py +++ b/src/pfs/codetables.py @@ -498,8 +498,61 @@ class GuidelineFactRow: extractor: str +@dataclass(frozen=True) +class PspsRow: + """One (year, code, provider specialty) national total from the CMS + Physician/Supplier Procedure Summary (``pfs.utilization_specialty``). + The counts and dollars are sums over every carrier, locality, place + of service and modifier line of the file; ``None`` when any summed + cell was suppressed (``"*"``), never 0.""" + + year: int + hcpcs: str + specialty_cd: str + specialty: str + services: float | None + allowed: float | None + paid: float | None + dataset_id: str + item_key: str + + +@dataclass(frozen=True) +class ImpactRow: + """One row of CMS's "Estimated Impact on Total Allowed Charges by + Specialty" (``pfs.specialty_impact``): a specialty's allowed charges + (millions of dollars) and the estimated change from the work, PE, MP + and combined RVU changes, as fractions (``-0.02`` = -2%). ``vintage`` is + ``"final"`` or ``"proposed"``; ``setting`` is ``TOTAL``, + ``Non-Facility`` or ``Facility`` (older Federal Register tables carry + ``TOTAL`` only). ``source`` is ``"508"`` (the CMS accessible file) or + ``"fr"`` (the table printed in the rule); ``page`` is the FR page of an + ``fr`` row, 0 otherwise.""" + + rule_year: int + vintage: str + specialty: str + setting: str + allowed_mil: float | None + impact_work: float | None + impact_pe: float | None + impact_mp: float | None + impact_combined: float | None + source: str + item_key: str + page: int + + _DDL = """ CREATE SCHEMA IF NOT EXISTS pfs; +CREATE TABLE IF NOT EXISTS pfs.utilization_specialty ( + year INTEGER, hcpcs VARCHAR, specialty_cd VARCHAR, specialty VARCHAR, + services DOUBLE, allowed DOUBLE, paid DOUBLE, dataset_id VARCHAR, + item_key VARCHAR); +CREATE TABLE IF NOT EXISTS pfs.specialty_impact ( + rule_year INTEGER, vintage VARCHAR, specialty VARCHAR, setting VARCHAR, + allowed_mil DOUBLE, impact_work DOUBLE, impact_pe DOUBLE, impact_mp DOUBLE, + impact_combined DOUBLE, source VARCHAR, item_key VARCHAR, page INTEGER); CREATE TABLE IF NOT EXISTS pfs.code_guideline_fact ( family VARCHAR, code VARCHAR, edition_year INTEGER, fact VARCHAR, value VARCHAR, cite_item VARCHAR, cite_locator VARCHAR, extractor VARCHAR); @@ -1126,3 +1179,57 @@ def read_stances(con: Any, family: str, docket: str | None = None) -> list[Stanc rows = con.execute(sql + " ORDER BY docket, item_key, event_key", params).fetchall() # a replica published before ``reviewed`` existed has shorter rows return [StanceRow(*r[:8], bool(r[8]) if len(r) > 8 else False) for r in rows] + + +def write_psps(con: Any, year: int, code: str, rows: Sequence[PspsRow]) -> int: + """Replace the (year, code) rows.""" + ensure_tables(con) + con.execute( + "DELETE FROM pfs.utilization_specialty WHERE year = ? AND hcpcs = ?", + [year, code], + ) + return _insert(con, "pfs.utilization_specialty", rows) + + +def read_psps( + con: Any, codes: Sequence[str], *, years: Sequence[int] | None = None +) -> list[PspsRow]: + if not codes: + return [] + sql = ( + "SELECT * FROM pfs.utilization_specialty " + f"WHERE hcpcs IN ({','.join('?' * len(codes))})" + ) + params: list[Any] = [c.upper() for c in codes] + if years: + sql += f" AND year IN ({','.join('?' * len(years))})" + params += [int(y) for y in years] + rows = con.execute(sql + " ORDER BY year, hcpcs, specialty_cd", params).fetchall() + return [PspsRow(*r) for r in rows] + + +def write_specialty_impacts( + con: Any, rule_year: int, vintage: str, source: str, rows: Sequence[ImpactRow] +) -> int: + """Replace the (rule_year, vintage, source) rows.""" + ensure_tables(con) + con.execute( + "DELETE FROM pfs.specialty_impact " + "WHERE rule_year = ? AND vintage = ? AND source = ?", + [rule_year, vintage, source], + ) + return _insert(con, "pfs.specialty_impact", rows) + + +def read_specialty_impacts( + con: Any, rule_years: Sequence[int] | None = None +) -> list[ImpactRow]: + sql = "SELECT * FROM pfs.specialty_impact" + params: list[Any] = [] + if rule_years: + sql += f" WHERE rule_year IN ({','.join('?' * len(rule_years))})" + params = [int(y) for y in rule_years] + rows = con.execute( + sql + " ORDER BY rule_year, vintage, source, specialty, setting", params + ).fetchall() + return [ImpactRow(*r) for r in rows] diff --git a/src/pfs/psps.py b/src/pfs/psps.py new file mode 100644 index 0000000..416d508 --- /dev/null +++ b/src/pfs/psps.py @@ -0,0 +1,234 @@ +"""Medicare Part B utilization by provider specialty (the money stakes, T10): +the CMS "Physician/Supplier Procedure Summary" (PSPS) Data API, summed to +one national row per (year, HCPCS code, provider specialty). + +The PSPS file carries one line per carrier x locality x specialty x place of +service x modifier, so a code's national total by specialty is the sum of +those lines. Cells under the privacy threshold arrive as ``"*"``; a sum with +any suppressed part is ``None``, never a smaller number that looks complete. +Lines flagged with an error indicator (``PSPS_ERROR_IND_CD`` other than +``"0"``) are CMS's own invalid lines and are excluded. + + fetch_code_year(client, year, code) the raw API records for one code/year + to_rows(year, code, records, ...) PspsRow per specialty + cite(store, year) the bib Source item for that year + write(con, year, code, rows) delete + insert (idempotent) + read_by_specialty(con, codes, ...) PspsRow list from the replica + +Raw pages are cached under ``data/cms/psps//.json``. The cache is +hoard-safe: ``save_cache`` never overwrites a cached file; a different payload +lands beside it as ``..json``. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any, Mapping, Sequence + +import httpx + +from bib.item import Source +from bib.tag import Tag +from pfs.codetables import PspsRow, read_psps, write_psps + +__all__ = [ + "API", + "DATASETS", + "PspsRow", + "cache_path", + "cite", + "fetch_code_year", + "load_cache", + "read_by_specialty", + "save_cache", + "to_rows", + "write", +] + +#: Calendar year -> data.cms.gov "Physician/Supplier Procedure Summary" +#: Data API id, from https://data.cms.gov/data.json (read 2026-11-11). 2025 +#: lists two distributions; it stays out until one is confirmed as the full +#: year. +DATASETS: dict[int, str] = { + 2010: "f4344bd9-ac34-43f0-b144-0186f6fa099f", + 2011: "1e9f7a5a-5550-4525-bbbf-ef66115184b8", + 2012: "b597d146-81bc-43f7-80b1-c2e85168170d", + 2013: "53027e7f-1b1d-4628-a286-03b0bc0b5f2f", + 2014: "3d28f6b0-e9df-48ea-b2eb-5cac10eaaf45", + 2015: "f8392bbd-981b-4ea6-b9f9-0891825105d7", + 2016: "b8a1bee7-62b2-44eb-8585-0c257a3d8c3f", + 2017: "3060244e-269e-4442-a983-8c190adfe7f1", + 2018: "6056deb7-17a7-49bc-a5b3-c799b4f91568", + 2019: "0ce67ad8-524b-4d4f-8ac8-55f4a68f0711", + 2020: "5eb3eebe-20cb-405c-864f-444873134f24", + 2021: "72200613-6706-4ad2-8219-e200cc4391cc", + 2022: "e4786716-10e9-40f2-b936-7444177474d5", + 2023: "b72adfb6-22cf-4241-9c64-050ad9061e03", + 2024: "647c8fa8-5dd6-460d-a2ec-18faf15b3fb2", +} +API = "https://data.cms.gov/data-api/v1/dataset/{id}/data" +TITLE = "Medicare Physician/Supplier Procedure Summary" +TABLE = "pfs.utilization_specialty" +USER_AGENT = "stack-pfs-dossier/1.0" +_PAGE = 5000 +_RETRIES = 3 + + +def fetch_code_year( + client: httpx.Client, + year: int, + code: str, + *, + page_size: int = _PAGE, + sleep: Any = None, +) -> list[dict]: + """Every PSPS line of *code* in *year*, paged until a short page. A 5xx + is retried with backoff; a 4xx (including 403/429) raises + ``httpx.HTTPStatusError`` and the caller stops that source. ``KeyError`` + for a year with no dataset.""" + import time + + nap = sleep or time.sleep + url = API.format(id=DATASETS[year]) + out: list[dict] = [] + offset = 0 + while True: + for attempt in range(_RETRIES + 1): + r = client.get( + url, + params={ + "filter[HCPCS_CD]": code, + "size": page_size, + "offset": offset, + }, + timeout=120, + ) + if r.status_code >= 500 and attempt < _RETRIES: + nap(2.0 * 2**attempt) + continue + break + r.raise_for_status() + page = r.json() + out.extend(page) + if len(page) < page_size: + return out + offset += page_size + + +def cache_path(year: int, code: str, *, root: Path | None = None) -> Path: + """``data/cms/psps//.json``, next to the primary DuckDB.""" + if root is None: + from conf import path + + root = path("db.aco").parent / "cms" / "psps" + return Path(root) / str(year) / f"{code.upper()}.json" + + +def save_cache(p: Path, records: Sequence[dict]) -> Path: + """Write *records* to *p* unless *p* exists. Identical bytes: no-op, + returns *p*. Different bytes: the old file is kept untouched and the new + payload is saved as ``.`` beside it (hoard rule); + that path is returned.""" + body = json.dumps(list(records), separators=(",", ":")).encode() + p.parent.mkdir(parents=True, exist_ok=True) + if p.exists(): + if p.read_bytes() == body: + return p + p = p.with_name(f"{p.stem}.{hashlib.sha256(body).hexdigest()[:8]}{p.suffix}") + if p.exists(): + return p + p.write_bytes(body) + return p + + +def load_cache(p: Path) -> list[dict]: + return json.loads(p.read_text()) + + +def _num(v: Any) -> float | None: + s = str(v if v is not None else "").strip() + if not s or s == "*": + return None + try: + return float(s) + except ValueError: + return None + + +def _sum(vals: list[float | None]) -> float | None: + if any(v is None for v in vals): + return None + return round(sum(vals), 2) # type: ignore[arg-type] + + +def to_rows( + year: int, + code: str, + records: Sequence[dict], + *, + dataset_id: str, + item_key: str, + names: Mapping[str, str], +) -> list[PspsRow]: + """National totals by ``PROVIDER_SPEC_CD`` for *code*: submitted services, + allowed charges and NCH payment summed over every line (carrier, + locality, place of service, modifier). Error-flagged lines are skipped; + a suppressed (``"*"``) or blank cell makes that sum ``None``.""" + code = code.upper() + buckets: dict[str, list[dict]] = {} + for rec in records: + if (rec.get("HCPCS_CD") or "").strip().upper() != code: + continue + if str(rec.get("PSPS_ERROR_IND_CD") or "0").strip() != "0": + continue + buckets.setdefault((rec.get("PROVIDER_SPEC_CD") or "").strip(), []).append(rec) + out: list[PspsRow] = [] + for cd in sorted(buckets): + recs = buckets[cd] + out.append( + PspsRow( + year=year, + hcpcs=code, + specialty_cd=cd, + specialty=names.get(cd, ""), + services=_sum( + [_num(r.get("PSPS_SUBMITTED_SERVICE_CNT")) for r in recs] + ), + allowed=_sum([_num(r.get("PSPS_ALLOWED_CHARGE_AMT")) for r in recs]), + paid=_sum([_num(r.get("PSPS_NCH_PAYMENT_AMT")) for r in recs]), + dataset_id=dataset_id, + item_key=item_key, + ) + ) + return out + + +def cite(store: Any, year: int) -> str: + """The bib Source item for *year*'s dataset, keyed by its API URL so + re-runs return the same key. Tagged ``module:pfs``, + ``table:pfs.utilization_specialty``, ``source:cms-website``, + ``year:``.""" + item = Source( + title=f"{TITLE}, CY{year} (national totals per HCPCS and provider specialty)", + url=API.format(id=DATASETS[year]), + doc_type="dataset", + ) + tags = [ + Tag.module("pfs"), + Tag.table(TABLE), + Tag.source("cms-website"), + Tag.year(year), + ] + return store.upsert(item, tags=tags) + + +def write(con: Any, year: int, code: str, rows: Sequence[PspsRow]) -> int: + return write_psps(con, year, code, rows) + + +def read_by_specialty( + con: Any, codes: Sequence[str], *, years: Sequence[int] | None = None +) -> list[PspsRow]: + return read_psps(con, codes, years=years) diff --git a/src/pfs/stakes.py b/src/pfs/stakes.py new file mode 100644 index 0000000..ab691ae --- /dev/null +++ b/src/pfs/stakes.py @@ -0,0 +1,382 @@ +"""Money stakes (T10): who stood to gain or lose when a code family was +revalued, in two public series. + +* CMS's "Estimated Impact on Total Allowed Charges by Specialty" for each + rule: the accessible ``_508.txt`` file CMS posts with a final rule, and + the same table printed in the Federal Register text of the 2012-2019 + rules. ``parse_impact_508`` / ``parse_impact_fr_text`` read them into + ``ImpactRow``; ``write_impacts`` / ``read_impacts`` store them in + ``pfs.specialty_impact``. +* ``money_by_specialty`` joins a family's payment change per code (from + ``pfs.valuation``) with each specialty's billing of those codes (PSPS, + ``pfs.psps``) into a dollar delta per specialty. + +Percents are stored as fractions (``-2%`` -> ``-0.02``). A bare integer in +the Federal Register tables is a percent point, so ``1`` -> ``0.01``. +""" + +from __future__ import annotations + +import csv +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Sequence + +from pfs.codetables import ( + ImpactRow, + PspsRow, + read_specialty_impacts, + write_specialty_impacts, +) +from pfs.commenters import specialty_key +from pfs.valuation import ValuationRow + +__all__ = [ + "ImpactRow", + "SpecialtyMoney", + "billing_share", + "money_by_specialty", + "parse_impact_508", + "parse_impact_fr_text", + "read_impacts", + "write_impacts", +] + +SETTINGS = ("TOTAL", "Non-Facility", "Facility") +_TITLE_RE = re.compile( + r"Impact on Total Allowed Charges by Specialty\s*\**\s*$", re.IGNORECASE +) +_PAGE_RE = re.compile(r"^\[\[Page (\d+)\]\]") +_NUM = r"-?\d+(?:\.\d+)?" +_ROW_RE = re.compile( + r"^\s*(?P[A-Za-z0-9][^\n]*?)\s*(?:\.{2,}|\s{2,})\s*" + rf"(?P\$?\s*[\d,]+(?:\.\d+)?)(?P(?:\s+{_NUM}%?){{3,4}})\s*$" +) +_CODE_PREFIX_RE = re.compile(r"^\d{1,3}-+") + + +def _money(s: str) -> float | None: + t = (s or "").strip().replace("$", "").replace(",", "") + try: + return float(t) + except ValueError: + return None + + +def _pct(s: str, *, bare_points: bool = False) -> float | None: + """``"-2%"`` -> -0.02. A bare number is read as percent points when + *bare_points* (the FR tables), otherwise it is not a percent.""" + t = (s or "").strip() + if not t: + return None + if t.endswith("%"): + t = t[:-1].strip() + elif not bare_points: + return None + try: + return round(float(t) / 100.0, 6) + except ValueError: + return None + + +def parse_impact_508( + path: str | Path, *, rule_year: int, vintage: str, item_key: str = "" +) -> list[ImpactRow]: + """Rows of a CMS ``..._508.txt`` impact file: tab-separated, columns + specialty, setting, allowed (mil), work, PE, MP, combined. The header and + footnote lines (whose setting cell is not TOTAL / Non-Facility / + Facility) are skipped.""" + out: list[ImpactRow] = [] + with open(path, newline="", encoding="utf-8-sig") as fh: + for rec in csv.reader(fh, delimiter="\t"): + if len(rec) < 7 or rec[1].strip() not in SETTINGS: + continue + out.append( + ImpactRow( + rule_year=rule_year, + vintage=vintage, + specialty=rec[0].strip(), + setting=rec[1].strip(), + allowed_mil=_money(rec[2]), + impact_work=_pct(rec[3]), + impact_pe=_pct(rec[4]), + impact_mp=_pct(rec[5]), + impact_combined=_pct(rec[6]), + source="508", + item_key=item_key, + page=0, + ) + ) + return out + + +def parse_impact_fr_text( + text: str, *, rule_year: int, vintage: str, item_key: str +) -> list[ImpactRow]: + """Rows of every ``... Impact on Total Allowed Charges by Specialty`` + table in a Federal Register text file (the by-selected-proposal tables + have a longer title and are not read). Setting is ``TOTAL``; ``page`` is + the nearest preceding ``[[Page N]]`` marker.""" + out: list[ImpactRow] = [] + page = 0 + in_table = False + seen_rows = False + for line in text.splitlines(): + m = _PAGE_RE.match(line) + if m: + page = int(m.group(1)) + continue + s = line.strip() + if not in_table: + if s.startswith("Table") and _TITLE_RE.search(s): + in_table, seen_rows = True, False + continue + row = _ROW_RE.match(line) + if row: + seen_rows = True + nums = [_pct(t, bare_points=True) for t in row["nums"].split()] + # CY2013-14 tables have one "work and MP" column (three numbers): + # it is kept as impact_work and impact_mp stays None. + work, pe, mp, comb = ( + nums if len(nums) == 4 else [nums[0], nums[1], None, nums[2]] + ) + name = _CODE_PREFIX_RE.sub("", row["name"].strip()).rstrip("* ") + out.append( + ImpactRow( + rule_year=rule_year, + vintage=vintage, + specialty=name, + setting="TOTAL", + allowed_mil=_money(row["allowed"]), + impact_work=work, + impact_pe=pe, + impact_mp=mp, + impact_combined=comb, + source="fr", + item_key=item_key, + page=page, + ) + ) + elif seen_rows and s and not set(s) <= {"-"}: + in_table = False # the footnote after the rows ends the table + return out + + +def write_impacts( + con: Any, rule_year: int, vintage: str, rows: Sequence[ImpactRow], *, source: str +) -> int: + """Replace the (rule_year, vintage, source) rows.""" + return write_specialty_impacts(con, rule_year, vintage, source, rows) + + +def read_impacts(con: Any, rule_years: Sequence[int] | None = None) -> list[ImpactRow]: + return read_specialty_impacts(con, rule_years) + + +# ── discovery ──────────────────────────────────────────────────────── + +_FILE_RE = re.compile(r"CY\s*(\d{4}) PFS (Proposed|Final) Rule", re.IGNORECASE) +_FR_YEAR_RE = re.compile(r"Table[^\n]*?CY\s*(\d{4})[^\n]*Impact on Total Allowed") + + +@dataclass(frozen=True) +class ImpactSource: + path: Path + rule_year: int + vintage: str + item_key: str + kind: str # "508" | "fr" + + +def fr_table_year(text: str) -> int | None: + """The rule year in the first by-specialty impact table title of an FR + text (``Table 52--CY 2017 PFS Estimated Impact ...``).""" + for line in text.splitlines(): + s = line.strip() + if s.startswith("Table") and _TITLE_RE.search(s): + m = _FR_YEAR_RE.search(s) + return int(m.group(1)) if m else None + return None + + +def discover_508(store: Any, final_root: Path) -> list[ImpactSource]: + """Every CMS ``Impact on Total Allowed Charges by Specialty..._508.txt`` + file: bib attachments first, then the same files under *final_root* + (``data/cms/pfs_final``) that bib does not hold. Rule year and vintage + come from the file name (``CY 2027 PFS Final Rule ...``).""" + pat = re.compile(r"Impact on Total Allowed Charges by Specialty.*_508\.txt$") + out: dict[tuple[int, str], ImpactSource] = {} + + def add(path: Path, item_key: str) -> None: + m = _FILE_RE.search(path.name) + if m and path.is_file(): + key = (int(m.group(1)), m.group(2).lower()) + out.setdefault(key, ImpactSource(path, key[0], key[1], item_key, "508")) + + con = store._con() # noqa: SLF001 + for fn, sp, ik in con.execute( + "SELECT a.filename, a.storage_path, i.key FROM attachments a " + "JOIN items i ON i.id = a.item_id ORDER BY a.id" + ).fetchall(): + if pat.search(str(fn)) and sp: + add(Path(sp), ik) + for p in sorted(Path(final_root).glob("*/files/**/*_508.txt")): + if pat.search(p.name): + add(p, "") + return list(out.values()) + + +def discover_fr(store: Any) -> list[ImpactSource]: + """Every proposed or final PFS rule in bib whose FR text prints a + by-specialty impact table (2012-2019 rules).""" + from pfs.rounds import fr_text_path, rule_items + + out: list[ImpactSource] = [] + for it in rule_items(store, []): + if it["kind"] not in ("proposed", "final"): + continue + p = fr_text_path(store, it["item_key"]) + if p is None: + continue + year = fr_table_year(p.read_text(errors="replace")) + if year is not None: + out.append(ImpactSource(p, year, it["kind"], it["item_key"], "fr")) + return out + + +# ── money by specialty ─────────────────────────────────────────────── + + +@dataclass(frozen=True) +class SpecialtyMoney: + """What a family's payment change meant for one specialty in *year*. + + ``services`` is the specialty's billing of the family's codes in each + code's latest PSPS year before *year*. ``delta_per_service`` / + ``delta_total`` use the final national non-facility payment, year vs + year-1, over the codes that have both (``None`` when none do); + ``delta_pct`` is ``delta_total`` over the same codes' year-1 payments, + as a fraction. ``proposed_*`` are the same figures from the NPRM's + proposed payment (``rvu_proposed``, CY2026-27) against the year-1 final, + kept apart from the final figures. ``new_payment`` is True when a code + the specialty bills first became payable in *year*.""" + + specialty: str + services: float + delta_per_service: float | None + delta_total: float | None + share: float + delta_pct: float | None = None + new_payment: bool = False + proposed_delta_per_service: float | None = None + proposed_delta_total: float | None = None + proposed_delta_pct: float | None = None + + +def _latest_services(psps: Sequence[PspsRow], year: int) -> dict[str, dict[str, float]]: + """code -> {specialty key -> services} for the code's latest PSPS year + <= year-1 (suppressed cells skipped).""" + latest: dict[str, int] = {} + for r in psps: + if r.year <= year - 1: + latest[r.hcpcs] = max(latest.get(r.hcpcs, 0), r.year) + out: dict[str, dict[str, float]] = {} + for r in psps: + if latest.get(r.hcpcs) != r.year or r.services is None: + continue + key = specialty_key(r.specialty) or f"CODE {r.specialty_cd}" + d = out.setdefault(r.hcpcs, {}) + d[key] = d.get(key, 0.0) + r.services + return out + + +def billing_share(psps: Sequence[PspsRow], *, year: int) -> dict[str, float]: + """Each specialty's share (by ``specialty_key``) of the family's + services, latest PSPS year <= year-1 per code.""" + totals: dict[str, float] = {} + for per in _latest_services(psps, year).values(): + for k, v in per.items(): + totals[k] = totals.get(k, 0.0) + v + whole = sum(totals.values()) + if whole <= 0: + return {} + return {k: v / whole for k, v in totals.items()} + + +def money_by_specialty( + payment: Sequence[ValuationRow], psps: Sequence[PspsRow], *, year: int +) -> list[SpecialtyMoney]: + """Per specialty (``specialty_key``), the family's payment change + in *year*, largest billers first. A code missing either year's final + payment contributes ``None`` deltas but its services still count in + ``services`` and ``share``.""" + final: dict[tuple[str, int], float | None] = {} + proposed: dict[str, float | None] = {} + for v in payment: + if v.proposed: + if v.year == year: + proposed[v.code] = v.pay_nf + else: + final[(v.code, v.year)] = v.pay_nf + services = _latest_services(psps, year) + acc: dict[str, dict[str, Any]] = {} + for code, per in services.items(): + now, before = final.get((code, year)), final.get((code, year - 1)) + prop = proposed.get(code) + # first payable this year: paid now, and never paid in any earlier year + is_new = now is not None and not any( + pay is not None for (c, y), pay in final.items() if c == code and y < year + ) + for spec, n in per.items(): + a = acc.setdefault( + spec, + dict(svc=0.0, d=None, base=None, pd=None, pbase=None, new=False), + ) + a["svc"] += n + a["new"] = a["new"] or is_new + if now is not None and before is not None: + a["d"] = (a["d"] or 0.0) + (now - before) * n + a["base"] = (a["base"] or 0.0) + before * n + if prop is not None and before is not None: + a["pd"] = (a["pd"] or 0.0) + (prop - before) * n + a["pbase"] = (a["pbase"] or 0.0) + before * n + whole = sum(a["svc"] for a in acc.values()) + + def ratio(num: float | None, den: float | None) -> float | None: + return num / den if num is not None and den else None + + out: list[SpecialtyMoney] = [] + for spec, a in acc.items(): + # services of only the codes that carry a delta, for the per-service mean + d_svc = sum( + n + for code, per in services.items() + for s, n in per.items() + if s == spec + and final.get((code, year)) is not None + and final.get((code, year - 1)) is not None + ) + p_svc = sum( + n + for code, per in services.items() + for s, n in per.items() + if s == spec + and proposed.get(code) is not None + and final.get((code, year - 1)) is not None + ) + out.append( + SpecialtyMoney( + specialty=spec, + services=a["svc"], + delta_per_service=ratio(a["d"], d_svc), + delta_total=a["d"], + share=a["svc"] / whole if whole else 0.0, + delta_pct=ratio(a["d"], a["base"]), + new_payment=a["new"], + proposed_delta_per_service=ratio(a["pd"], p_svc), + proposed_delta_total=a["pd"], + proposed_delta_pct=ratio(a["pd"], a["pbase"]), + ) + ) + return sorted(out, key=lambda m: (-m.services, m.specialty)) diff --git a/tests/cli/test_pfs_cli.py b/tests/cli/test_pfs_cli.py index 07bfc55..e1ede6e 100644 --- a/tests/cli/test_pfs_cli.py +++ b/tests/cli/test_pfs_cli.py @@ -2774,3 +2774,288 @@ class TestStanceGeneration: + (["--family", "HOME-VISITS"] if cmd == "stance" else []), ) assert res.exit_code != 0 and "--generation" in res.output + + +class TestPsps: + _REC = { + "HCPCS_CD": "99350", + "PROVIDER_SPEC_CD": "50", + "PSPS_SUBMITTED_SERVICE_CNT": "10", + "PSPS_ALLOWED_CHARGE_AMT": "100", + "PSPS_NCH_PAYMENT_AMT": "80", + "PSPS_ERROR_IND_CD": "0", + } + + @pytest.fixture + def env(self, con, monkeypatch, tmp_path): + from bib.store import Store + + store = Store(":memory:", storage_dir=tmp_path / "storage") + monkeypatch.setattr(pfs_cli, "_store", lambda: store) + monkeypatch.setattr( + pfs_cli, + "psps_cache_path", + lambda y, code, root=None: tmp_path / str(y) / f"{code}.json", + ) + con.execute("CREATE SCHEMA IF NOT EXISTS terminology") + con.execute( + "CREATE TABLE terminology.cms_acceptable_provider_specialty_codes " + "(specialty_code VARCHAR, specialty_description VARCHAR, " + "pecos_specialty_description VARCHAR)" + ) + con.execute( + "INSERT INTO terminology.cms_acceptable_provider_specialty_codes " + "VALUES ('50', 'Nurse Practitioner', 'NP')" + ) + fetched: list[tuple[int, str]] = [] + + def fake_fetch(client, y, code, **kw): + fetched.append((y, code)) + return [dict(self._REC, HCPCS_CD=code)] if code == "99350" else [] + + monkeypatch.setattr(pfs_cli, "fetch_psps", fake_fetch) + monkeypatch.setattr(pfs_cli, "_http_dossier", lambda: MagicMock()) + import time + + monkeypatch.setattr(time, "sleep", lambda s: None) + try: + yield store, fetched, tmp_path + finally: + store.close() + + def test_write_fetches_every_family_code_loads_cites_logs(self, con, env): + store, fetched, tmp_path = env + res = runner.invoke( + app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"] + ) + assert res.exit_code == 0, res.output + assert len(fetched) == 20 and {y for y, _ in fetched} == {2023} + rows = con.execute( + "SELECT hcpcs, specialty_cd, specialty, services, allowed, paid, item_key " + "FROM pfs.utilization_specialty" + ).fetchall() + assert rows[0][:6] == ("99350", "50", "Nurse Practitioner", 10.0, 100.0, 80.0) + assert "year:2023" in store.get(rows[0][6]).tags + log = con.execute( + "SELECT module, table_name, rows, source_file FROM cms.ingest_log " + "WHERE source_file LIKE '%99350.json'" + ).fetchone() + assert log[:3] == ("pfs.psps", "pfs.utilization_specialty", 1) + assert (tmp_path / "2023" / "99350.json").exists() + assert con.published == [True] + assert "2023 99350: 1 lines -> 1 specialties" in res.output + assert "2023 99349: 0 lines -> 0 specialties" in res.output + + def test_offline_uses_cache_and_skips_missing(self, con, env): + store, fetched, tmp_path = env + runner.invoke( + app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"] + ) + n = len(fetched) + res = runner.invoke( + app, + [ + "pfs", + "psps", + "--family", + "HOME-VISITS", + "--write", + "--offline", + "--year", + "2023", + ], + ) + assert res.exit_code == 0, res.output + assert len(fetched) == n # no new fetch + assert "2023 99349: no cached file, skipped" in res.output + assert con.execute( + "SELECT count(*) FROM pfs.utilization_specialty" + ).fetchone() == (1,) + + def test_psps_cli_offline_uses_cache(self, con, env): + # the brief's named test: a cached file alone is enough offline + store, fetched, tmp_path = env + from pfs.psps import save_cache + + save_cache(tmp_path / "2022" / "99350.json", [self._REC]) + res = runner.invoke( + app, + [ + "pfs", + "psps", + "--family", + "HOME-VISITS", + "--write", + "--offline", + "--year", + "2022", + ], + ) + assert res.exit_code == 0, res.output + assert fetched == [] + assert con.execute( + "SELECT year, hcpcs, services FROM pfs.utilization_specialty" + ).fetchall() == [(2022, "99350", 10.0)] + + def test_names_missing_when_terminology_table_absent(self, con, env): + con.execute("DROP TABLE terminology.cms_acceptable_provider_specialty_codes") + res = runner.invoke( + app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"] + ) + assert res.exit_code == 0, res.output + assert con.execute( + "SELECT specialty FROM pfs.utilization_specialty" + ).fetchone() == ("",) + + def test_terminology_error_other_than_missing_table_propagates(self, monkeypatch): + class Bad: + def execute(self, *a): + raise RuntimeError("boom") + + with pytest.raises(RuntimeError): + pfs_cli._psps_names(Bad()) + + def test_unknown_year_and_family_rejected_before_any_fetch(self, con, env): + store, fetched, _ = env + res = runner.invoke( + app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "1999"] + ) + assert res.exit_code != 0 + res = runner.invoke(app, ["pfs", "psps", "--family", "NOPE", "--write"]) + assert res.exit_code != 0 and "unknown family" in res.output + assert fetched == [] + + def test_read_prints_stored_rows_without_the_write_lock( + self, con, env, monkeypatch + ): + runner.invoke( + app, ["pfs", "psps", "--family", "HOME-VISITS", "--write", "--year", "2023"] + ) + + def fail_batch(): + raise AssertionError("must not open a RW connection") + + monkeypatch.setattr(pfs_cli, "_batch", fail_batch) + res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"]) + assert res.exit_code == 0, res.output + assert "Nurse Practitioner" in res.output and "psps: 1 rows" in res.output + + def test_read_prints_dash_for_suppressed_and_reports_missing_table(self, con, env): + con.execute( + "INSERT INTO pfs.utilization_specialty VALUES " + "(2023, '99350', '50', 'NP', NULL, NULL, NULL, 'd', 'K')" + ) + res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"]) + assert res.exit_code == 0 and " -" in res.output + con.execute("DROP TABLE pfs.utilization_specialty") + res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"]) + assert "no PSPS table yet" in res.output + + def test_read_reraises_other_errors(self, con, env, monkeypatch): + def boom(*a, **k): + raise RuntimeError("boom") + + monkeypatch.setattr(pfs_cli, "read_psps_rows", boom) + res = runner.invoke(app, ["pfs", "psps", "--family", "HOME-VISITS"]) + assert res.exit_code != 0 and isinstance(res.exception, RuntimeError) + + def test_http_client_sends_the_honest_agent(self): + c = pfs_cli._http_dossier() + try: + assert c.headers["User-Agent"] == "stack-pfs-dossier/1.0" + finally: + c.close() + + +class TestImpacts: + _NAME = ( + "CY 2027 PFS Final Rule Estimated Impact on Total Allowed Charges " + "by Specialty_508.txt" + ) + + @pytest.fixture + def sources(self, con, monkeypatch, tmp_path): + from pfs.stakes import ImpactSource + + fix = Path(__file__).parents[1] / "pfs" / "fixtures" + p508 = tmp_path / self._NAME + p508.write_text((fix / "impact_cy2027_508.txt").read_text()) + pfr = tmp_path / "2016-26668.txt" + pfr.write_text((fix / "impact_cy2017_fr.txt").read_text()) + monkeypatch.setattr(pfs_cli, "_store", lambda: object()) + monkeypatch.setattr( + pfs_cli, + "discover_508", + lambda store, root: [ImpactSource(p508, 2027, "final", "BJG", "508")], + ) + monkeypatch.setattr( + pfs_cli, + "discover_fr", + lambda store: [ImpactSource(pfr, 2017, "final", "FRK", "fr")], + ) + return p508, pfr + + def test_scan_write_loads_both_kinds_and_logs(self, con, sources): + res = runner.invoke(app, ["pfs", "impacts", "--scan", "--write"]) + assert res.exit_code == 0, res.output + got = con.execute( + "SELECT rule_year, vintage, source, count(*) FROM pfs.specialty_impact " + "GROUP BY ALL ORDER BY 1" + ).fetchall() + assert got == [(2017, "final", "fr", 4), (2027, "final", "508", 6)] + logs = con.execute( + "SELECT module, table_name, rows, pincite_key FROM cms.ingest_log ORDER BY rows" + ).fetchall() + assert logs == [ + ("pfs.stakes", "pfs.specialty_impact", 4, "FRK"), + ("pfs.stakes", "pfs.specialty_impact", 6, "BJG"), + ] + assert con.published == [True] + assert "CY2027 final (508): 6 rows" in res.output + + def test_scan_without_write_only_reports(self, con, sources): + res = runner.invoke(app, ["pfs", "impacts", "--scan"]) + assert res.exit_code == 0, res.output + assert "CY2017 final (fr): 4 rows" in res.output + assert con.published == [] + assert con.execute("SELECT count(*) FROM pfs.specialty_impact").fetchone() == ( + 0, + ) + + def test_path_form_reads_one_file(self, con, sources): + p508, pfr = sources + res = runner.invoke( + app, + ["pfs", "impacts", "--path", str(p508), "--year", "2027", "--write"], + ) + assert res.exit_code == 0, res.output + res = runner.invoke( + app, + [ + "pfs", + "impacts", + "--path", + str(pfr), + "--year", + "2016", + "--vintage", + "proposed", + "--write", + ], + ) + assert res.exit_code == 0, res.output + assert con.execute( + "SELECT rule_year, vintage, source, count(*) FROM pfs.specialty_impact " + "GROUP BY ALL ORDER BY 1" + ).fetchall() == [(2016, "proposed", "fr", 4), (2027, "final", "508", 6)] + + def test_argument_errors(self, con, sources): + p508, _ = sources + for args in ( + [], + ["--scan", "--path", str(p508)], + ["--path", str(p508)], + ["--scan", "--vintage", "draft"], + ): + res = runner.invoke(app, ["pfs", "impacts", *args]) + assert res.exit_code != 0, args diff --git a/tests/pfs/fixtures/impact_cy2017_fr.txt b/tests/pfs/fixtures/impact_cy2017_fr.txt new file mode 100644 index 0000000..a2f7be1 --- /dev/null +++ b/tests/pfs/fixtures/impact_cy2017_fr.txt @@ -0,0 +1,14 @@ +[[Page 80544]] + + Table 52--CY 2017 PFS Estimated Impact on Total Allowed Charges by Specialty * +---------------------------------------------------------------------------------------------------------------- + (C) Impact of (D) Impact of (E) Impact of + (A) Specialty (B) Allowed work RVU PE RVU changes MP RVU changes (F) Combined + charges (mil) changes (%) (%) (%) impact ** (%) +---------------------------------------------------------------------------------------------------------------- +TOTAL........................... $89,866 0 0 0 0 +ALLERGY/IMMUNOLOGY.............. 231 0 1 0 1 +ANESTHESIOLOGY.................. 1,982 0 0 0 0 +AUDIOLOGIST..................... 61 0 0 0 0 + +** Column F may not equal the sum of columns C, D, and E due to rounding. diff --git a/tests/pfs/fixtures/impact_cy2027_508.txt b/tests/pfs/fixtures/impact_cy2027_508.txt new file mode 100644 index 0000000..1a0e576 --- /dev/null +++ b/tests/pfs/fixtures/impact_cy2027_508.txt @@ -0,0 +1,15 @@ +CY 2027 PFS Final Rule Estimated Impact on Total Allowed Charges by Specialty +"(A) + Specialty" "(B) +Facility" "(C) + Allowed Charges (mil)" "(D) +Impact of Work RVU Changes" "(E) +Impact of PE RVU Changes" "(F) +Impact of MP RVU Changes" "(G) +Combined Impact" +ALLERGY/IMMUNOLOGY TOTAL $240 0% -2% 0% -2% +ALLERGY/IMMUNOLOGY Non-Facility $233 0% -2% 0% -2% +ALLERGY/IMMUNOLOGY Facility $7 0% -1% 0% 0% +ANESTHESIOLOGY TOTAL "$1,672" 0% 0% 0% 0% +ANESTHESIOLOGY Non-Facility $343 0% -1% 0% -1% +ANESTHESIOLOGY Facility "$1,329" 0% 0% 0% 0% diff --git a/tests/pfs/test_psps.py b/tests/pfs/test_psps.py new file mode 100644 index 0000000..6386f57 --- /dev/null +++ b/tests/pfs/test_psps.py @@ -0,0 +1,245 @@ +"""pfs.psps - Physician/Supplier Procedure Summary by specialty (T10).""" + +from __future__ import annotations + +import json + +import duckdb +import httpx +import pytest + +from bib.store import Store +from pfs.codetables import PspsRow, ensure_tables +from pfs.psps import ( + API, + DATASETS, + cache_path, + cite, + fetch_code_year, + load_cache, + read_by_specialty, + save_cache, + to_rows, + write, +) + + +def rec(**kw): + base = { + "HCPCS_CD": "99350", + "HCPCS_INITIAL_MODIFIER_CD": "", + "PROVIDER_SPEC_CD": "50", + "PSPS_SUBMITTED_SERVICE_CNT": "10", + "PSPS_ALLOWED_CHARGE_AMT": "100.5", + "PSPS_NCH_PAYMENT_AMT": "80.25", + "PSPS_ERROR_IND_CD": "0", + } + base.update(kw) + return base + + +NAMES = {"50": "Nurse Practitioner", "08": "Family Practice"} + + +class TestDatasets: + def test_2010_to_2024_and_no_2025(self): + assert sorted(DATASETS) == list(range(2010, 2025)) + assert DATASETS[2023] == "b72adfb6-22cf-4241-9c64-050ad9061e03" + + +class TestToRows: + def test_psps_aggregates_by_specialty(self): + records = [ + rec(), + rec(HCPCS_INITIAL_MODIFIER_CD="95", PSPS_SUBMITTED_SERVICE_CNT="5"), + rec(PROVIDER_SPEC_CD="08", PSPS_ALLOWED_CHARGE_AMT="*"), + rec(PROVIDER_SPEC_CD="08", PSPS_ALLOWED_CHARGE_AMT="7"), + rec(PSPS_SUBMITTED_SERVICE_CNT="999", PSPS_ERROR_IND_CD="1"), + rec(HCPCS_CD="99349"), + ] + rows = to_rows( + 2023, "99350", records, dataset_id="d", item_key="K", names=NAMES + ) + by = {r.specialty_cd: r for r in rows} + assert set(by) == {"08", "50"} + np_ = by["50"] + assert (np_.services, np_.allowed, np_.paid) == (15.0, 201.0, 160.5) + assert np_.specialty == "Nurse Practitioner" + assert (np_.year, np_.hcpcs, np_.dataset_id, np_.item_key) == ( + 2023, + "99350", + "d", + "K", + ) + fp = by["08"] + assert fp.services == 20.0 + assert fp.allowed is None # a suppressed part makes the sum None + + def test_unknown_specialty_name_blank_and_bad_numbers_none(self): + rows = to_rows( + 2023, + "99350", + [ + rec( + PROVIDER_SPEC_CD="ZZ", + PSPS_SUBMITTED_SERVICE_CNT="", + PSPS_NCH_PAYMENT_AMT="x", + ) + ], + dataset_id="d", + item_key="K", + names=NAMES, + ) + assert rows[0].specialty == "" + assert rows[0].services is None and rows[0].paid is None + + def test_missing_error_indicator_counts_as_clean(self): + r = rec() + del r["PSPS_ERROR_IND_CD"] + assert to_rows(2023, "99350", [r], dataset_id="d", item_key="K", names={}) + + +def _paged(pages, seen, statuses=None): + it = iter(pages) + st = iter(statuses or []) + + def handler(request): + seen.append(str(request.url)) + code = next(st, 200) + if code != 200: + return httpx.Response(code) + return httpx.Response(200, json=next(it)) + + return httpx.Client(transport=httpx.MockTransport(handler)) + + +class TestFetch: + def test_filters_by_code_and_pages_until_short(self): + seen: list[str] = [] + client = _paged([[rec()] * 2, [rec()]], seen) + out = fetch_code_year(client, 2023, "99350", page_size=2) + assert len(out) == 3 + assert "filter%5BHCPCS_CD%5D=99350" in seen[0] + assert "offset=2" in seen[1] + assert seen[0].startswith(API.format(id=DATASETS[2023])) + + def test_5xx_is_retried_with_backoff(self): + seen: list[str] = [] + naps: list[float] = [] + client = _paged([[rec()]], seen, statuses=[503, 502]) + out = fetch_code_year(client, 2023, "99350", sleep=naps.append) + assert len(out) == 1 and len(seen) == 3 + assert naps == [2.0, 4.0] + + def test_persistent_5xx_raises(self): + client = httpx.Client( + transport=httpx.MockTransport(lambda r: httpx.Response(503)) + ) + with pytest.raises(httpx.HTTPStatusError): + fetch_code_year(client, 2023, "99350", sleep=lambda s: None) + + def test_4xx_is_not_retried(self): + seen: list[str] = [] + client = _paged([], seen, statuses=[429]) + with pytest.raises(httpx.HTTPStatusError): + fetch_code_year(client, 2023, "99350", sleep=lambda s: None) + assert len(seen) == 1 + + def test_unknown_year(self): + with pytest.raises(KeyError): + fetch_code_year(httpx.Client(), 1999, "99350") + + def test_default_sleep_is_time_sleep(self, monkeypatch): + import time + + naps: list[float] = [] + monkeypatch.setattr(time, "sleep", naps.append) + client = _paged([[rec()]], [], statuses=[500]) + fetch_code_year(client, 2023, "99350") + assert naps == [2.0] + + +class TestCache: + def test_path_layout(self, tmp_path): + assert ( + cache_path(2023, "99350", root=tmp_path) == tmp_path / "2023" / "99350.json" + ) + + def test_default_root_is_next_to_the_primary_db(self): + p = cache_path(2023, "99350") + assert ( + p.parts[-4:] == ("psps", "2023", "99350.json")[-4:] + or p.name == "99350.json" + ) + assert "psps" in p.parts + + def test_roundtrip(self, tmp_path): + p = tmp_path / "2023" / "99350.json" + assert save_cache(p, [rec()]) == p + assert load_cache(p) == [rec()] + + def test_hoard_never_overwrites_a_cached_file(self, tmp_path): + p = tmp_path / "2023" / "99350.json" + save_cache(p, [rec()]) + before = p.read_bytes() + # same bytes: a no-op on the same path + assert save_cache(p, [rec()]) == p + # different bytes: the old file is untouched, new one beside it + q = save_cache(p, [rec(PSPS_SUBMITTED_SERVICE_CNT="11")]) + assert q != p and p.read_bytes() == before + assert ( + q.name.startswith("99350.") and q.suffix == ".json" and len(q.stem) == 6 + 8 + ) + assert json.loads(q.read_text())[0]["PSPS_SUBMITTED_SERVICE_CNT"] == "11" + # the same different payload again does not rewrite or add a file + assert save_cache(p, [rec(PSPS_SUBMITTED_SERVICE_CNT="11")]) == q + assert len(list(p.parent.iterdir())) == 2 + + +class TestCite: + def test_one_bib_source_per_year(self): + store = Store(":memory:") + try: + k = cite(store, 2023) + assert cite(store, 2023) == k + assert cite(store, 2022) != k + item = store.get(k) + assert "CY2023" in item.title + for t in ("module:pfs", "table:pfs.utilization_specialty", "year:2023"): + assert t in item.tags + finally: + store.close() + + +@pytest.fixture +def con(): + c = duckdb.connect(":memory:") + ensure_tables(c) + yield c + c.close() + + +class TestWriteRead: + def _row(self, y, code, cd="50", n=1.0): + return PspsRow(y, code, cd, "NP", n, 2.0, 3.0, "d", "K") + + def test_write_replaces_year_and_code_only(self, con): + assert write(con, 2023, "99350", [self._row(2023, "99350")]) == 1 + write(con, 2023, "99349", [self._row(2023, "99349")]) + write(con, 2022, "99350", [self._row(2022, "99350")]) + assert write(con, 2023, "99350", [self._row(2023, "99350", n=9.0)]) == 1 + rows = read_by_specialty(con, ["99350", "99349"]) + assert [(r.year, r.hcpcs, r.services) for r in rows] == [ + (2022, "99350", 1.0), + (2023, "99349", 1.0), + (2023, "99350", 9.0), + ] + + def test_read_filters_years_and_handles_empty(self, con): + write(con, 2023, "99350", [self._row(2023, "99350")]) + write(con, 2022, "99350", [self._row(2022, "99350")]) + assert [r.year for r in read_by_specialty(con, ["99350"], years=[2022])] == [ + 2022 + ] + assert read_by_specialty(con, []) == [] + assert write(con, 2021, "99350", []) == 0 diff --git a/tests/pfs/test_stakes.py b/tests/pfs/test_stakes.py new file mode 100644 index 0000000..f7315ae --- /dev/null +++ b/tests/pfs/test_stakes.py @@ -0,0 +1,382 @@ +"""pfs.stakes - CMS specialty impact tables and money by specialty (T10).""" + +from __future__ import annotations + +from pathlib import Path + +import duckdb +import pytest + +from bib.item import Source +from bib.store import Store +from pfs import stakes +from pfs.codetables import ImpactRow, PspsRow, ensure_tables +from pfs.stakes import ( + billing_share, + discover_508, + discover_fr, + fr_table_year, + money_by_specialty, + parse_impact_508, + parse_impact_fr_text, + read_impacts, + write_impacts, +) +from pfs.valuation import ValuationRow + +FIX = Path(__file__).parent / "fixtures" + + +def test_parse_impact_508(): + rows = parse_impact_508( + FIX / "impact_cy2027_508.txt", rule_year=2027, vintage="final", item_key="K" + ) + assert len(rows) == 6 + a = rows[0] + assert (a.specialty, a.setting, a.allowed_mil) == ( + "ALLERGY/IMMUNOLOGY", + "TOTAL", + 240.0, + ) + assert (a.impact_work, a.impact_pe, a.impact_mp, a.impact_combined) == ( + 0.0, + -0.02, + 0.0, + -0.02, + ) + assert (a.rule_year, a.vintage, a.source, a.item_key, a.page) == ( + 2027, + "final", + "508", + "K", + 0, + ) + assert [r.setting for r in rows[:3]] == ["TOTAL", "Non-Facility", "Facility"] + anes = rows[3] + assert (anes.specialty, anes.setting, anes.allowed_mil) == ( + "ANESTHESIOLOGY", + "TOTAL", + 1672.0, + ) + + +def test_parse_impact_fr_text(): + rows = parse_impact_fr_text( + (FIX / "impact_cy2017_fr.txt").read_text(), + rule_year=2017, + vintage="final", + item_key="FR", + ) + assert [r.specialty for r in rows] == [ + "TOTAL", + "ALLERGY/IMMUNOLOGY", + "ANESTHESIOLOGY", + "AUDIOLOGIST", + ] + tot, allergy = rows[0], rows[1] + assert tot.allowed_mil == 89866.0 + assert (allergy.allowed_mil, allergy.impact_pe, allergy.impact_combined) == ( + 231.0, + 0.01, + 0.01, + ) + assert (allergy.setting, allergy.source, allergy.page) == ("TOTAL", "fr", 80544) + + +def test_fr_text_other_table_titles_and_wrapped_names_and_page_changes(): + text = "\n".join( + [ + "[[Page 10]]", + "Table 72--CY 2014 PFS Estimated Impact on Total Allowed Charges by Specialty by Selected Proposal*", + "-----", + "ALLERGY..... 1 0 0 0 0", + "Table 7--CY 2014 PFS Estimated Impact on Total Allowed Charges by Specialty *", + "-----", + "(A) Specialty (B)", + "MULTISPECIALTY CLINIC/OTHER PHYS 129 -1.5 0 0 1", + "[[Page 11]]", + "", + "OTHER........... 27 0 0 0 -1", + "A footnote line", + "NOT A ROW..... 1 0 0 0 0", + ] + ) + rows = parse_impact_fr_text(text, rule_year=2014, vintage="proposed", item_key="X") + assert [(r.specialty, r.page) for r in rows] == [ + ("MULTISPECIALTY CLINIC/OTHER PHYS", 10), + ("OTHER", 11), + ] + assert rows[0].impact_work == -0.015 + assert rows[1].impact_combined == -0.01 + + +def test_fr_text_older_layouts(): + text = "\n".join( + [ + "[[Page 5]]", + "Table 134--CY 2013 PFS Final Rule With Comment Period Estimated Impact on Total Allowed Charges by Specialty *", + "-----", + "TOTAL........................................... $ 86,588 0 0 0", + "02-ANESTHESIOLOGY **............................ 1,923 0 1 1", + " TOTAL................ $89,020 0% 0% 0% -1%", + "ANESTHESIOLOGY....... $1,970 0% 1% -2% 0%", + ] + ) + rows = parse_impact_fr_text(text, rule_year=2013, vintage="final", item_key="X") + assert [r.specialty for r in rows] == [ + "TOTAL", + "ANESTHESIOLOGY", + "TOTAL", + "ANESTHESIOLOGY", + ] + tot, anes, tot16, anes16 = rows + assert tot.allowed_mil == 86588.0 + assert (anes.impact_work, anes.impact_pe, anes.impact_mp, anes.impact_combined) == ( + 0.0, + 0.01, + None, + 0.01, + ) + assert (tot16.allowed_mil, tot16.impact_combined) == (89020.0, -0.01) + assert (anes16.impact_pe, anes16.impact_mp) == (0.01, -0.02) + + +def test_percent_and_money_helpers(): + assert stakes._pct("-2%") == -0.02 + assert stakes._pct("3 %") == 0.03 + assert stakes._pct("") is None + assert stakes._pct("1") is None # a bare number is not a 508 percent + assert stakes._pct("1", bare_points=True) == 0.01 + assert stakes._pct("x%") is None + assert stakes._pct("x", bare_points=True) is None + assert stakes._money("$1,672") == 1672.0 + assert stakes._money("n/a") is None + + +def test_508_skips_short_and_unknown_setting_lines(tmp_path): + p = tmp_path / "x_508.txt" + p.write_text('title\nA\tB\tC\n"* note"\tx\nFOO\tTOTAL\t$5\t0%\t0%\t0%\t1%\t\n') + rows = parse_impact_508(p, rule_year=2027, vintage="final") + assert [ + (r.specialty, r.allowed_mil, r.impact_combined, r.item_key) for r in rows + ] == [("FOO", 5.0, 0.01, "")] + + +@pytest.fixture +def con(): + c = duckdb.connect(":memory:") + ensure_tables(c) + yield c + c.close() + + +def test_write_and_read_impacts(con): + rows = parse_impact_508( + FIX / "impact_cy2027_508.txt", rule_year=2027, vintage="final" + ) + fr = parse_impact_fr_text( + (FIX / "impact_cy2017_fr.txt").read_text(), + rule_year=2017, + vintage="final", + item_key="FR", + ) + assert write_impacts(con, 2027, "final", rows, source="508") == 6 + assert write_impacts(con, 2017, "final", fr, source="fr") == 4 + assert write_impacts(con, 2027, "final", rows[:3], source="508") == 3 # replaces + got = read_impacts(con) + assert [(r.rule_year, r.source) for r in got].count((2027, "508")) == 3 + assert [r.rule_year for r in read_impacts(con, [2017])] == [2017] * 4 + assert isinstance(got[0], ImpactRow) + assert write_impacts(con, 2020, "final", [], source="508") == 0 + + +def test_fr_table_year(): + t = "x\nTable 52--CY 2017 PFS Estimated Impact on Total Allowed Charges by Specialty *\n" + assert fr_table_year(t) == 2017 + assert ( + fr_table_year("Table 5 Estimated Impact on Total Allowed Charges by Specialty") + is None + ) + assert fr_table_year("nothing here") is None + + +def test_impacts_scan_finds_508(tmp_path): + name = "CY 2027 PFS Final Rule Estimated Impact on Total Allowed Charges by Specialty_508.txt" + src = tmp_path / "src" + src.mkdir() + f = src / name + f.write_text((FIX / "impact_cy2027_508.txt").read_text()) + other = src / "other_508.txt" + other.write_text("x") + store = Store(":memory:", storage_dir=tmp_path / "storage") + try: + key = store.upsert(Source(title="Impacts", url="https://x/y")) + store.attach_file(key, f) + store.attach_file(key, other) + # an older year that lives only under the pfs_final directory tree + root = tmp_path / "pfs_final" + d = root / "2026" / "files" / "d" + d.mkdir(parents=True) + (d / name.replace("2027", "2026")).write_text("x") + (d / "unrelated_508.txt").write_text("x") + (d / "CY 2026 Impact on Total Allowed Charges by Specialty_508.txt").write_text( + "x" + ) + # the 2027 tree copy must not duplicate the bib one + d27 = root / "2027" / "files" / "d" + d27.mkdir(parents=True) + (d27 / name).write_text("x") + found = discover_508(store, root) + finally: + store.close() + got = {(s.rule_year, s.vintage): s for s in found} + assert set(got) == {(2027, "final"), (2026, "final")} + assert got[(2027, "final")].item_key == key and got[(2027, "final")].kind == "508" + assert got[(2026, "final")].item_key == "" + + +def test_discover_fr(tmp_path, monkeypatch): + import pfs.rounds as rounds + + t = tmp_path / "a.txt" + t.write_text((FIX / "impact_cy2017_fr.txt").read_text()) + n = tmp_path / "n.txt" + n.write_text("no table") + items = [ + {"item_key": "A", "kind": "final"}, + {"item_key": "B", "kind": "rule"}, + {"item_key": "C", "kind": "proposed"}, # no text on disk + {"item_key": "D", "kind": "final"}, # text without the table + ] + paths = {"A": t, "B": t, "C": None, "D": n} + monkeypatch.setattr(rounds, "rule_items", lambda store, anchors: items) + monkeypatch.setattr(rounds, "fr_text_path", lambda store, k: paths[k]) + found = discover_fr(object()) + assert [(s.rule_year, s.vintage, s.item_key, s.kind) for s in found] == [ + (2017, "final", "A", "fr") + ] + + +# ── money by specialty ─────────────────────────────────────────────── + + +def val(code, year, pay, *, proposed=False, status="A"): + return ValuationRow( + code=code, + description="", + vintage=f"CY{year} {'proposed' if proposed else 'final'}", + year=year, + proposed=proposed, + status=status, + work=None, + pe_nf=None, + pe_f=None, + mp=None, + total_nf=None, + total_f=None, + cf=1.0, + pay_nf=pay, + pay_f=None, + label="", + citation="", + url="", + ) + + +def ps(year, code, spec, n): + return PspsRow(year, code, "00", spec, n, None, None, "d", "K") + + +def test_money_by_specialty_exact(): + payment = [val("99350", 2022, 100.0), val("99350", 2023, 90.0)] + psps = [ + ps(2022, "99350", "Family Practice", 10.0), + ps(2022, "99350", "Nurse Practitioner", 30.0), + ] + out = money_by_specialty(payment, psps, year=2023) + by = {m.specialty: m for m in out} + fp, np_ = by["FAMILY PRACTICE"], by["NURSE PRACTITIONER"] + assert (fp.delta_per_service, fp.delta_total, fp.share) == (-10.0, -100.0, 0.25) + assert (np_.delta_per_service, np_.delta_total, np_.share) == (-10.0, -300.0, 0.75) + assert fp.delta_pct == pytest.approx(-0.10) + assert fp.new_payment is False + assert fp.proposed_delta_total is None and fp.proposed_delta_pct is None + assert [m.specialty for m in out] == ["NURSE PRACTITIONER", "FAMILY PRACTICE"] + + +def test_missing_year_gives_none_delta_but_counts_in_share(): + payment = [ + val("A", 2022, 100.0), + val("A", 2023, 50.0), + val("B", 2023, 10.0), # no prior-year payment + val("C", 2022, 5.0), + val("C", 2023, None, status="B"), # no longer paid + ] + psps = [ + ps(2021, "A", "Family Practice", 5.0), # superseded by the 2022 row + ps(2022, "A", "Family Practice", 10.0), + ps(2022, "B", "Family Practice", 10.0), + ps(2022, "B", "Cardiology", 20.0), + ps(2022, "C", "Cardiology", 5.0), + ps(2022, "A", "Cardiology", None), # suppressed, skipped + ps(2024, "A", "Family Practice", 999.0), # after year-1, ignored + ] + by = {m.specialty: m for m in money_by_specialty(payment, psps, year=2023)} + fp, ca = by["FAMILY PRACTICE"], by["CARDIOLOGY"] + assert fp.services == 20.0 and fp.share == pytest.approx(20 / 45) + assert fp.delta_total == -500.0 # only code A carries a delta + assert fp.delta_per_service == -50.0 # per service of A, not of B + assert fp.delta_pct == pytest.approx(-0.5) + assert fp.new_payment is True # B is first payable in 2023 + assert ca.services == 25.0 + assert ( + ca.delta_total is None and ca.delta_per_service is None and ca.delta_pct is None + ) + assert ca.new_payment is True + + +def test_proposed_vs_final_delta_kept_apart(): + payment = [ + val("A", 2026, 100.0), + val("A", 2027, 90.0), + val("A", 2027, 80.0, proposed=True), + val("A", 2028, 1.0, proposed=True), # another year's NPRM, ignored + val("B", 2027, 40.0, proposed=True), # proposed but no prior final + ] + psps = [ + ps(2026, "A", "Family Practice", 10.0), + ps(2026, "B", "Family Practice", 1.0), + ] + (m,) = money_by_specialty(payment, psps, year=2027) + assert m.delta_total == -100.0 + assert m.proposed_delta_total == -200.0 + assert m.proposed_delta_per_service == -20.0 + assert m.proposed_delta_pct == pytest.approx(-0.2) + + +def test_unnamed_specialty_falls_back_to_code_and_empty_inputs(): + (m,) = money_by_specialty( + [val("A", 2022, 1.0), val("A", 2023, 2.0)], + [PspsRow(2022, "A", "99", "", 4.0, None, None, "d", "K")], + year=2023, + ) + assert m.specialty == "CODE 99" and m.delta_total == 4.0 + assert money_by_specialty([], [], year=2023) == [] + # zero services: share is 0, never a division error + (z,) = money_by_specialty([], [ps(2022, "A", "Family Practice", 0.0)], year=2023) + assert z.share == 0.0 and z.delta_total is None + + +def test_billing_share(): + psps = [ + ps(2022, "A", "Family Practice", 10.0), + ps(2022, "A", "Cardiology", 30.0), + ps(2023, "B", "Family Practice", 60.0), + ] + got = billing_share(psps, year=2024) + assert got == {"FAMILY PRACTICE": 0.7, "CARDIOLOGY": 0.3} + assert billing_share(psps, year=2023) == { + "FAMILY PRACTICE": 0.25, + "CARDIOLOGY": 0.75, + } + assert billing_share([], year=2023) == {}