4c69d5be9b
One dataset per calendar year on data.cms.gov (2013–2024, UUIDs in pfs.utilization.DATASETS), pulled through the Data API filtered to national rows (~13k a year, one per HCPCS × place of service), cached under data/cms/utilization/, loaded delete-then-insert per year, each year cited in bib as a Source (module:pfs, table:pfs.utilization, source:cms-website, year:N) and logged in cms.ingest_log with the cache file's sha256. Suppressed cells (<11 beneficiaries) load as NULL. read_series() sums places of service and service-weights the average allowed/paid amounts. stack pfs utilization [--year N]... [--write [--offline]] [--code C]... Notebook 7d charts a family's services per year with the exposure dates from 7c as rules; 7e answers the three #694 questions (99490 2015–, G2058→99439 continuity, APCM 2025 vs CCM — no 2025 file yet). CLI docs regenerated (sidebar renumbered around the new page).
300 lines
10 KiB
Python
300 lines
10 KiB
Python
"""Part B utilization per HCPCS code (#694): the CMS "Medicare Physician &
|
||
Other Practitioners — by Geography and Service" public use files, national
|
||
totals only, 2013 on.
|
||
|
||
The direct outcome of a code becoming payable is whether anyone bills it,
|
||
so this is the first outcome series P50 joins to the exposure calendar
|
||
(``pfs.code_exposure``). Each calendar year is its own dataset on
|
||
data.cms.gov (``DATASETS`` maps year → UUID, from the data.json
|
||
catalogue); the Data API is pulled with ``filter[Rndrng_Prvdr_Geo_Lvl]=
|
||
National`` — ~13k rows a year, one per (HCPCS, place of service) — so
|
||
the whole 2013–2024 series is ~160k rows and refetches in minutes.
|
||
|
||
fetch_year(client, year) → the raw API records for one year
|
||
to_rows(year, records, ...) → UtilizationRow per national row
|
||
cite(store, year) → the bib Source item for that year's dataset
|
||
write(con, year, rows) → delete + insert that year (idempotent)
|
||
read_series(con, codes) → per (code, year) totals across places of service
|
||
|
||
Raw pages are cached under ``data/cms/utilization/mup-phy-geo-<year>.json``
|
||
so ``cms.ingest_log`` can carry the file's sha256 and a re-run needs no
|
||
network. Suppressed cells (``Tot_Benes`` < 11) arrive as ``""`` and are
|
||
stored as NULL, never 0.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any, Sequence
|
||
|
||
import httpx
|
||
|
||
from bib.item import Source
|
||
from bib.tag import Tag
|
||
|
||
#: Calendar year → data.cms.gov dataset UUID ("by Geography and Service"),
|
||
#: from https://data.cms.gov/data.json (catalogue read 2026-09-22).
|
||
DATASETS: dict[int, str] = {
|
||
2013: "3c2a4756-0a8c-4e4d-845a-6ad169cb13d3",
|
||
2014: "28181bd2-b377-4003-b73a-4bd92d1db4a9",
|
||
2015: "dbee9609-2c90-43ca-b1b8-161bd9cfcdb2",
|
||
2016: "c7d3f18c-2f00-4553-8cd1-871b727d5cdd",
|
||
2017: "8e96a9f2-ce6e-46fd-b30d-8c695c756bfd",
|
||
2018: "05a85700-052f-4509-af43-7042b9b35868",
|
||
2019: "673030ae-ceed-4561-8fca-b1275395a86a",
|
||
2020: "31eb3018-43e0-4259-a0d8-e7a1112ffd08",
|
||
2021: "f00f462f-bb61-48b1-a321-1c50d78ce175",
|
||
2022: "87304f15-9ed0-41dc-a141-6141a0327453",
|
||
2023: "ddee9e22-7889-4bef-975a-7853e4cd0fbb",
|
||
2024: "0c75b0b3-b40f-4007-a5ac-f9f2fed95862",
|
||
}
|
||
API = "https://data.cms.gov/data-api/v1/dataset/{id}/data"
|
||
LANDING = (
|
||
"https://data.cms.gov/provider-summary-by-type-of-service/"
|
||
"medicare-physician-other-practitioners/"
|
||
"medicare-physician-other-practitioners-by-geography-and-service"
|
||
)
|
||
TITLE = "Medicare Physician & Other Practitioners - by Geography and Service"
|
||
TABLE = "pfs.utilization"
|
||
_PAGE = 5000
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class UtilizationRow:
|
||
year: int
|
||
hcpcs: str
|
||
hcpcs_desc: str
|
||
drug_ind: str
|
||
place_of_service: str
|
||
n_providers: int | None
|
||
n_beneficiaries: int | None
|
||
n_services: float | None
|
||
n_bene_day_services: float | None
|
||
avg_submitted: float | None
|
||
avg_allowed: float | None
|
||
avg_paid: float | None
|
||
avg_standardized: float | None
|
||
dataset_id: str
|
||
item_key: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SeriesRow:
|
||
"""One (code, year) across places of service: counts summed,
|
||
``avg_allowed``/``avg_paid`` weighted by services."""
|
||
|
||
hcpcs: str
|
||
year: int
|
||
n_providers: int | None
|
||
n_beneficiaries: int | None
|
||
n_services: float | None
|
||
avg_allowed: float | None
|
||
avg_paid: float | None
|
||
|
||
|
||
# ── fetch ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def fetch_year(
|
||
client: httpx.Client, year: int, *, page_size: int = _PAGE
|
||
) -> list[dict]:
|
||
"""Every national row of *year*'s dataset, paged ``page_size`` at a
|
||
time until a short (or empty) page. Raises ``KeyError`` for a year
|
||
with no dataset and ``httpx.HTTPStatusError`` on a non-200."""
|
||
url = API.format(id=DATASETS[year])
|
||
out: list[dict] = []
|
||
offset = 0
|
||
while True:
|
||
r = client.get(
|
||
url,
|
||
params={
|
||
"filter[Rndrng_Prvdr_Geo_Lvl]": "National",
|
||
"size": page_size,
|
||
"offset": offset,
|
||
},
|
||
timeout=120,
|
||
)
|
||
r.raise_for_status()
|
||
page = r.json()
|
||
out.extend(page)
|
||
if len(page) < page_size:
|
||
return out
|
||
offset += page_size
|
||
|
||
|
||
def _int(v: Any) -> int | None:
|
||
s = str(v or "").strip()
|
||
if not s:
|
||
return None
|
||
try:
|
||
return int(float(s))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _float(v: Any) -> float | None:
|
||
s = str(v or "").strip()
|
||
if not s:
|
||
return None
|
||
try:
|
||
return float(s)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def to_rows(
|
||
year: int, records: Sequence[dict], *, dataset_id: str, item_key: str
|
||
) -> list[UtilizationRow]:
|
||
"""National rows only (the filter is re-checked here so a cached
|
||
payload from an unfiltered pull still loads cleanly)."""
|
||
out: list[UtilizationRow] = []
|
||
for rec in records:
|
||
if (rec.get("Rndrng_Prvdr_Geo_Lvl") or "") != "National":
|
||
continue
|
||
out.append(
|
||
UtilizationRow(
|
||
year=year,
|
||
hcpcs=(rec.get("HCPCS_Cd") or "").strip().upper(),
|
||
hcpcs_desc=(rec.get("HCPCS_Desc") or "").strip(),
|
||
drug_ind=(rec.get("HCPCS_Drug_Ind") or "").strip(),
|
||
place_of_service=(rec.get("Place_Of_Srvc") or "").strip(),
|
||
n_providers=_int(rec.get("Tot_Rndrng_Prvdrs")),
|
||
n_beneficiaries=_int(rec.get("Tot_Benes")),
|
||
n_services=_float(rec.get("Tot_Srvcs")),
|
||
n_bene_day_services=_float(rec.get("Tot_Bene_Day_Srvcs")),
|
||
avg_submitted=_float(rec.get("Avg_Sbmtd_Chrg")),
|
||
avg_allowed=_float(rec.get("Avg_Mdcr_Alowd_Amt")),
|
||
avg_paid=_float(rec.get("Avg_Mdcr_Pymt_Amt")),
|
||
avg_standardized=_float(rec.get("Avg_Mdcr_Stdzd_Amt")),
|
||
dataset_id=dataset_id,
|
||
item_key=item_key,
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
# ── cache ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def cache_path(year: int, *, root: Path | None = None) -> Path:
|
||
"""``data/cms/utilization/mup-phy-geo-<year>.json`` (next to the
|
||
primary DuckDB, so the sha256 in ``cms.ingest_log`` points at a file
|
||
that lives with the data it produced)."""
|
||
if root is None:
|
||
from conf import path
|
||
|
||
root = path("db.aco").parent / "cms" / "utilization"
|
||
return Path(root) / f"mup-phy-geo-{year}.json"
|
||
|
||
|
||
def save_cache(p: Path, records: Sequence[dict]) -> None:
|
||
p.parent.mkdir(parents=True, exist_ok=True)
|
||
p.write_text(json.dumps(list(records), separators=(",", ":")))
|
||
|
||
|
||
def load_cache(p: Path) -> list[dict]:
|
||
return json.loads(p.read_text())
|
||
|
||
|
||
# ── bib ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def cite(store: Any, year: int) -> str:
|
||
"""The bib Source item for *year*'s dataset — one per year, keyed by
|
||
the dataset's own API URL so re-runs return the same key. Tagged
|
||
``module:pfs``, ``table:pfs.utilization``, ``source:cms-website``,
|
||
``year:<year>``."""
|
||
item = Source(
|
||
title=f"{TITLE}, CY{year} (national totals per HCPCS and place of service)",
|
||
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)
|
||
|
||
|
||
# ── DuckDB ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def ensure_table(con: Any) -> None:
|
||
from pfs.table.utilization import Utilization
|
||
|
||
con.execute("CREATE SCHEMA IF NOT EXISTS pfs")
|
||
con.execute(
|
||
Utilization.to_ddl().replace("CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ", 1)
|
||
)
|
||
|
||
|
||
def write(con: Any, year: int, rows: Sequence[UtilizationRow]) -> int:
|
||
"""Replace *year*'s rows."""
|
||
ensure_table(con)
|
||
con.execute(f"DELETE FROM {TABLE} WHERE year = ?", [year])
|
||
if not rows:
|
||
return 0
|
||
con.executemany(
|
||
f"INSERT INTO {TABLE} VALUES ({','.join('?' * 15)})",
|
||
[
|
||
(
|
||
r.year,
|
||
r.hcpcs,
|
||
r.hcpcs_desc,
|
||
r.drug_ind,
|
||
r.place_of_service,
|
||
r.n_providers,
|
||
r.n_beneficiaries,
|
||
r.n_services,
|
||
r.n_bene_day_services,
|
||
r.avg_submitted,
|
||
r.avg_allowed,
|
||
r.avg_paid,
|
||
r.avg_standardized,
|
||
r.dataset_id,
|
||
r.item_key,
|
||
)
|
||
for r in rows
|
||
],
|
||
)
|
||
return len(rows)
|
||
|
||
|
||
def read_series(
|
||
con: Any, codes: Sequence[str], *, years: Sequence[int] | None = None
|
||
) -> list[SeriesRow]:
|
||
"""Per (code, year) totals across places of service, ordered by code
|
||
then year. Counts are summed (a provider billing both settings is
|
||
counted twice — the file gives no cross-setting distinct count);
|
||
``avg_allowed``/``avg_paid`` are service-weighted means."""
|
||
if not codes:
|
||
return []
|
||
marks = ",".join("?" * len(codes))
|
||
sql = (
|
||
"SELECT hcpcs, year, sum(n_providers), sum(n_beneficiaries), sum(n_services), "
|
||
"sum(avg_allowed * n_services) / nullif(sum(n_services), 0), "
|
||
"sum(avg_paid * n_services) / nullif(sum(n_services), 0) "
|
||
f"FROM {TABLE} WHERE hcpcs IN ({marks})"
|
||
)
|
||
params: list[Any] = [c.upper() for c in codes]
|
||
if years:
|
||
sql += f" AND year IN ({','.join('?' * len(years))})"
|
||
params.extend(int(y) for y in years)
|
||
sql += " GROUP BY hcpcs, year ORDER BY hcpcs, year"
|
||
return [
|
||
SeriesRow(
|
||
hcpcs=h,
|
||
year=int(y),
|
||
n_providers=int(p) if p is not None else None,
|
||
n_beneficiaries=int(b) if b is not None else None,
|
||
n_services=float(s) if s is not None else None,
|
||
avg_allowed=float(a) if a is not None else None,
|
||
avg_paid=float(pd) if pd is not None else None,
|
||
)
|
||
for h, y, p, b, s, a, pd in con.execute(sql, params).fetchall()
|
||
]
|