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

feat(pfs): money stakes — PSPS utilization by specialty and CMS specialty impact tables

Browse Source
This commit is contained in:
kert
2026-11-12 18:08:10 -05:00
parent 045b99daed
commit 5ab1aad74b
9 changed files with 1882 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
src/cli/pfs.py
View File

@@ -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."""

107
src/pfs/codetables.py
View File

@@ -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]

234
src/pfs/psps.py Normal file
View File

@@ -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/<year>/<code>.json``. The cache is
hoard-safe: ``save_cache`` never overwrites a cached file; a different payload
lands beside it as ``<code>.<sha8>.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/<year>/<code>.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 ``<stem>.<sha8><suffix>`` 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:<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)

382
src/pfs/stakes.py Normal file
View File

@@ -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<name>[A-Za-z0-9][^\n]*?)\s*(?:\.{2,}|\s{2,})\s*"
rf"(?P<allowed>\$?\s*[\d,]+(?:\.\d+)?)(?P<nums>(?:\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))

285
tests/cli/test_pfs_cli.py
View File

@@ -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

14
tests/pfs/fixtures/impact_cy2017_fr.txt Normal file
View File

@@ -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.

15
tests/pfs/fixtures/impact_cy2027_508.txt Normal file
View File

@@ -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%

245
tests/pfs/test_psps.py Normal file
View File

@@ -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

382
tests/pfs/test_stakes.py Normal file
View File

@@ -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) == {}