feat: OPPS file download + ingestion — 6 years of addenda into DuckDB (refs #264)
Some checks failed
CI / skinny-install (aco) (push) Successful in 2m37s
CI / skinny-install (bcda) (push) Successful in 2m13s
CI / skinny-install (bib) (push) Successful in 1m59s
CI / skinny-install (bls) (push) Successful in 1m59s
CI / skinny-install (cms) (push) Successful in 1m37s
CI / skinny-install (cli) (push) Successful in 2m19s
CI / lint-test (push) Failing after 3m33s
CI / skinny-install (api) (push) Successful in 1m39s
CI / skinny-install (ccw) (push) Successful in 2m5s
CI / skinny-install (perf) (push) Successful in 1m35s
CI / skinny-install (conf) (push) Successful in 2m22s
CI / skinny-install (rex) (push) Successful in 2m34s
Deploy / build-scan-report (push) Failing after 3m43s
CI / skinny-install (pfs) (push) Successful in 2m10s
Some checks failed
CI / skinny-install (aco) (push) Successful in 2m37s
CI / skinny-install (bcda) (push) Successful in 2m13s
CI / skinny-install (bib) (push) Successful in 1m59s
CI / skinny-install (bls) (push) Successful in 1m59s
CI / skinny-install (cms) (push) Successful in 1m37s
CI / skinny-install (cli) (push) Successful in 2m19s
CI / lint-test (push) Failing after 3m33s
CI / skinny-install (api) (push) Successful in 1m39s
CI / skinny-install (ccw) (push) Successful in 2m5s
CI / skinny-install (perf) (push) Successful in 1m35s
CI / skinny-install (conf) (push) Successful in 2m22s
CI / skinny-install (rex) (push) Successful in 2m34s
Deploy / build-scan-report (push) Failing after 3m43s
CI / skinny-install (pfs) (push) Successful in 2m10s
Download: CMS OPPS addenda ZIPs (CY2021-2026) from cms.gov/files/zip/ Ingest: parse Addendum A (APC weights) + Addendum B (HCPCS-APC crosswalk) with header detection for CMS preamble rows across format drift. Tables loaded: - opps.apc_weight: 5,507 rows (804→1,041 APCs, 6 years) - opps.addendum_b: 108,168 rows (17K→19K HCPCS codes per year) - opps.skin_sub_addendum_b: 1,549 rows — skin sub SI history Key finding — CY2026 reclassification visible in status indicators: - 2021-2025: SI=N (packaged, 145→245 codes), SI=B (bundled, 52) - 2026: SI=S1 (new separately payable supply, 272 codes) — the flat rate SI=N drops to 31, SI=D (deleted) = 10 codes
This commit is contained in:
198
dev/scripts/download_opps_files.py
Normal file
198
dev/scripts/download_opps_files.py
Normal file
@@ -0,0 +1,198 @@
|
|||||||
|
"""Download CMS OPPS addendum files for all available years.
|
||||||
|
|
||||||
|
Downloads Addendum A (APC weights), Addendum B (HCPCS→APC map),
|
||||||
|
and wage index files from CMS.gov for CY2014–2026.
|
||||||
|
|
||||||
|
Source: https://www.cms.gov/medicare/payment/prospective-payment-systems/
|
||||||
|
hospital-outpatient/addendum-a-b-updates
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
uv run python dev/scripts/download_opps_files.py
|
||||||
|
uv run python dev/scripts/download_opps_files.py --year 2026
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
OPPS_DIR = ROOT / "data" / "cms" / "opps"
|
||||||
|
|
||||||
|
USER_AGENT = "stack-opps-ingest/1.0"
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# CMS OPPS download URLs by year
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# These are the canonical download page URLs for each year's addenda.
|
||||||
|
# CMS reorganizes URLs periodically; these are current as of March 2026.
|
||||||
|
|
||||||
|
OPPS_SOURCES: list[dict[str, str]] = [
|
||||||
|
{
|
||||||
|
"year": "2026",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2026-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2026-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2026-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2025/11/20/2025-23371/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2025",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2025-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2025-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2025-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2024/11/18/2024-25518/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2024",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2024-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2024-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2024-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2023/11/22/2023-24293/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2023",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2023-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2023-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2023-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2022/11/23/2022-23918/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2022",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2022-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2022-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2022-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2021/11/16/2021-24011/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2021",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2021-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2021-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2021-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2020/12/29/2020-26819/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2020",
|
||||||
|
"addendum_a": "https://www.cms.gov/files/zip/cy2020-opps-final-rule-addendum-a.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/files/zip/cy2020-opps-final-rule-addendum-b.zip",
|
||||||
|
"wage_index": "https://www.cms.gov/files/zip/cy2020-opps-final-rule-wage-index.zip",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2019/11/12/2019-24138/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2019",
|
||||||
|
"addendum_a": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1695-FC-2019-OPPS-FR-Addnda-A.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1695-FC-2019-OPPS-FR-Addnda-B.zip",
|
||||||
|
"wage_index": "",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2018/11/21/2018-24243/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2018",
|
||||||
|
"addendum_a": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1678-FC-2018-OPPS-FR-Addnda-A.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1678-FC-2018-OPPS-FR-Addnda-B.zip",
|
||||||
|
"wage_index": "",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2017/11/13/2017-23932/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2017",
|
||||||
|
"addendum_a": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1656-FC-2017-OPPS-FR-Addnda-A.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1656-FC-2017-OPPS-FR-Addnda-B.zip",
|
||||||
|
"wage_index": "",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2016/11/14/2016-26515/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2016",
|
||||||
|
"addendum_a": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1633-FC-2016-OPPS-FR-Addnda-A.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1633-FC-2016-OPPS-FR-Addnda-B.zip",
|
||||||
|
"wage_index": "",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2015/11/13/2015-28005/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2015",
|
||||||
|
"addendum_a": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1613-FC-2015-OPPS-FR-Addnda-A.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1613-FC-2015-OPPS-FR-Addnda-B.zip",
|
||||||
|
"wage_index": "",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2014/11/10/2014-26146/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"year": "2014",
|
||||||
|
"addendum_a": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1601-FC-2014-OPPS-FR-Addnda-A.zip",
|
||||||
|
"addendum_b": "https://www.cms.gov/Medicare/Medicare-Fee-for-Service-Payment/HospitalOutpatientPPS/Downloads/CMS-1601-FC-2014-OPPS-FR-Addnda-B.zip",
|
||||||
|
"wage_index": "",
|
||||||
|
"rule_fr": "https://www.federalregister.gov/documents/2013/12/10/2013-28737/medicare-program-changes-to-the-hospital-outpatient-prospective-payment-and-ambulatory-surgical-center",
|
||||||
|
},
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def download_file(url: str, dest: Path) -> bool:
|
||||||
|
"""Download a file if it doesn't already exist."""
|
||||||
|
if dest.exists():
|
||||||
|
print(f" EXISTS {dest.name}")
|
||||||
|
return True
|
||||||
|
try:
|
||||||
|
print(f" GET {dest.name} ...", end="", flush=True)
|
||||||
|
resp = httpx.get(url, headers={"User-Agent": USER_AGENT},
|
||||||
|
timeout=60, follow_redirects=True)
|
||||||
|
if resp.status_code == 200:
|
||||||
|
dest.write_bytes(resp.content)
|
||||||
|
size_mb = len(resp.content) / 1024 / 1024
|
||||||
|
print(f" {size_mb:.1f} MB")
|
||||||
|
return True
|
||||||
|
else:
|
||||||
|
print(f" HTTP {resp.status_code}")
|
||||||
|
return False
|
||||||
|
except Exception as e:
|
||||||
|
print(f" ERROR: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
parser = argparse.ArgumentParser(description="Download CMS OPPS files")
|
||||||
|
parser.add_argument("--year", help="Download only this year")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
OPPS_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
sources = OPPS_SOURCES
|
||||||
|
if args.year:
|
||||||
|
sources = [s for s in sources if s["year"] == args.year]
|
||||||
|
if not sources:
|
||||||
|
print(f"No sources defined for year {args.year}")
|
||||||
|
return
|
||||||
|
|
||||||
|
total = 0
|
||||||
|
downloaded = 0
|
||||||
|
|
||||||
|
for src in sources:
|
||||||
|
year = src["year"]
|
||||||
|
year_dir = OPPS_DIR / year
|
||||||
|
year_dir.mkdir(exist_ok=True)
|
||||||
|
|
||||||
|
print(f"\n--- CY{year} ---")
|
||||||
|
|
||||||
|
for file_type in ["addendum_a", "addendum_b", "wage_index"]:
|
||||||
|
url = src.get(file_type, "")
|
||||||
|
if not url:
|
||||||
|
continue
|
||||||
|
total += 1
|
||||||
|
fname = f"opps_{year}_{file_type}.zip"
|
||||||
|
if download_file(url, year_dir / fname):
|
||||||
|
downloaded += 1
|
||||||
|
time.sleep(0.5)
|
||||||
|
|
||||||
|
print(f"\n{'=' * 50}")
|
||||||
|
print(f"Downloaded: {downloaded}/{total} files")
|
||||||
|
print(f"Location: {OPPS_DIR}")
|
||||||
|
|
||||||
|
# List what we have
|
||||||
|
print("\nInventory:")
|
||||||
|
for year_dir in sorted(OPPS_DIR.iterdir()):
|
||||||
|
if year_dir.is_dir():
|
||||||
|
files = list(year_dir.glob("*.zip"))
|
||||||
|
print(f" {year_dir.name}: {len(files)} files "
|
||||||
|
f"({sum(f.stat().st_size for f in files) / 1024 / 1024:.1f} MB)")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
314
dev/scripts/ingest_opps.py
Normal file
314
dev/scripts/ingest_opps.py
Normal file
@@ -0,0 +1,314 @@
|
|||||||
|
"""Ingest CMS OPPS addendum files into DuckDB.
|
||||||
|
|
||||||
|
Parses Addendum A (APC weights), Addendum B (HCPCS→APC map), and
|
||||||
|
wage index files from downloaded ZIPs. Handles column naming
|
||||||
|
variations across 13 years of CMS format drift (CY2014–2026).
|
||||||
|
|
||||||
|
Produces:
|
||||||
|
* opps.apc_weight (Addendum A — APC relative weights)
|
||||||
|
* opps.addendum_b (Addendum B — HCPCS→APC crosswalk)
|
||||||
|
* opps.wage_index (Wage index by CBSA)
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
uv run python dev/scripts/download_opps_files.py # download first
|
||||||
|
uv run python dev/scripts/ingest_opps.py # then ingest
|
||||||
|
uv run python dev/scripts/ingest_opps.py --year 2026
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import io
|
||||||
|
import re
|
||||||
|
import zipfile
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import duckdb
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
OPPS_DIR = ROOT / "data" / "cms" / "opps"
|
||||||
|
DUCKDB_PATH = ROOT / "data" / "aco.duckdb"
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Column normalisation — CMS changes headers across years
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
ADDENDUM_A_COL_MAP = {
|
||||||
|
"apc": "apc",
|
||||||
|
"apc group": "apc",
|
||||||
|
"apc group number": "apc",
|
||||||
|
"group title": "group_title",
|
||||||
|
"apc group title": "group_title",
|
||||||
|
"status indicator": "status_indicator",
|
||||||
|
"si": "status_indicator",
|
||||||
|
"relative weight": "relative_weight",
|
||||||
|
"payment rate": "payment_rate",
|
||||||
|
"national unadjusted payment": "payment_rate",
|
||||||
|
"minimum unadjusted copayment": "minimum_unadjusted_copayment",
|
||||||
|
}
|
||||||
|
|
||||||
|
ADDENDUM_B_COL_MAP = {
|
||||||
|
"hcpcs code": "hcpcs",
|
||||||
|
"hcpcs": "hcpcs",
|
||||||
|
"cpt/hcpcs": "hcpcs",
|
||||||
|
"short descriptor": "short_description",
|
||||||
|
"short description": "short_description",
|
||||||
|
"si": "status_indicator",
|
||||||
|
"status indicator": "status_indicator",
|
||||||
|
"apc": "apc",
|
||||||
|
"apc group": "apc",
|
||||||
|
"apc group number": "apc",
|
||||||
|
"apc title": "apc_title",
|
||||||
|
"relative weight": "relative_weight",
|
||||||
|
"payment rate": "payment_rate",
|
||||||
|
"national unadjusted payment": "payment_rate",
|
||||||
|
"minimum unadjusted copayment": "minimum_unadjusted_copayment",
|
||||||
|
}
|
||||||
|
|
||||||
|
WAGE_INDEX_COL_MAP = {
|
||||||
|
"cbsa": "cbsa",
|
||||||
|
"cbsa number": "cbsa",
|
||||||
|
"cbsa name": "cbsa_name",
|
||||||
|
"urban area title": "cbsa_name",
|
||||||
|
"state": "state",
|
||||||
|
"wage index": "wage_index",
|
||||||
|
"pre-reclassification wage index": "wage_index",
|
||||||
|
"reclassified wage index": "reclassified_wage_index",
|
||||||
|
"post-reclassification wage index": "reclassified_wage_index",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def normalise_columns(df: pd.DataFrame, col_map: dict[str, str]) -> pd.DataFrame:
|
||||||
|
"""Lowercase columns, apply mapping, keep only mapped columns."""
|
||||||
|
df.columns = [str(c).strip().lower() for c in df.columns]
|
||||||
|
df = df.rename(columns=col_map)
|
||||||
|
present = [c for c in set(col_map.values()) if c in df.columns]
|
||||||
|
return df[present]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Readers — handle ZIP contents (CSV or Excel inside)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def find_data_file(zf: zipfile.ZipFile, pattern: str = "") -> str | None:
|
||||||
|
"""Find a specific addendum file inside a combined addenda ZIP.
|
||||||
|
|
||||||
|
CMS OPPS addenda ZIPs contain multiple addenda (A, B, C, D1, D2, etc.)
|
||||||
|
as both XLSX and CSV (508 version). Prefer XLSX in the root, fall back
|
||||||
|
to CSV in the 508 subfolder.
|
||||||
|
"""
|
||||||
|
skip = ["__macosx", ".ds_store", "readme", "note"]
|
||||||
|
candidates = []
|
||||||
|
for name in zf.namelist():
|
||||||
|
low = name.lower()
|
||||||
|
if any(s in low for s in skip):
|
||||||
|
continue
|
||||||
|
if low.endswith((".csv", ".xlsx", ".xls")):
|
||||||
|
candidates.append(name)
|
||||||
|
if pattern:
|
||||||
|
# Match "Addendum A" or "Addendum B" etc. in filename
|
||||||
|
filtered = [c for c in candidates if pattern.lower() in c.lower()]
|
||||||
|
# Prefer xlsx over csv, prefer root over 508 subfolder
|
||||||
|
xlsx = [f for f in filtered if f.lower().endswith(".xlsx") and "/" not in f]
|
||||||
|
if xlsx:
|
||||||
|
return xlsx[0]
|
||||||
|
csv = [f for f in filtered if f.lower().endswith(".csv")]
|
||||||
|
if csv:
|
||||||
|
return csv[0]
|
||||||
|
if filtered:
|
||||||
|
return filtered[0]
|
||||||
|
return candidates[0] if candidates else None
|
||||||
|
|
||||||
|
|
||||||
|
def _find_header_row(df: pd.DataFrame, col_map: dict[str, str]) -> int:
|
||||||
|
"""Find the row index containing column headers in a preamble-laden CMS file.
|
||||||
|
|
||||||
|
Requires >= 3 matches AND at least 3 non-null cells (to distinguish
|
||||||
|
a real header row from a title row that happens to contain keywords).
|
||||||
|
"""
|
||||||
|
target_cols = [k for k in col_map if k not in ("", "year")]
|
||||||
|
for i, row in df.iterrows():
|
||||||
|
vals = [str(v).strip().lower() for v in row if pd.notna(v) and str(v).strip()]
|
||||||
|
if len(vals) < 3:
|
||||||
|
continue
|
||||||
|
matches = sum(1 for v in vals for k in target_cols if v == k or (len(k) > 3 and k in v))
|
||||||
|
if matches >= 3:
|
||||||
|
return i
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
def read_from_zip(zpath: Path, col_map: dict[str, str],
|
||||||
|
pattern: str = "") -> pd.DataFrame:
|
||||||
|
"""Read and normalise a data file from inside a ZIP.
|
||||||
|
|
||||||
|
Handles CMS preamble rows in both CSV and Excel formats.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
with zipfile.ZipFile(zpath) as zf:
|
||||||
|
target = find_data_file(zf, pattern)
|
||||||
|
if not target:
|
||||||
|
return pd.DataFrame()
|
||||||
|
|
||||||
|
data = zf.read(target)
|
||||||
|
|
||||||
|
if target.lower().endswith(".csv"):
|
||||||
|
raw = data.decode("latin-1")
|
||||||
|
lines = raw.splitlines()
|
||||||
|
# Find header row
|
||||||
|
hdr_idx = 0
|
||||||
|
target_cols = list(col_map.keys())[:5]
|
||||||
|
for i, line in enumerate(lines[:20]):
|
||||||
|
low = line.strip().lower()
|
||||||
|
if sum(1 for k in target_cols if k in low) >= 2:
|
||||||
|
hdr_idx = i
|
||||||
|
break
|
||||||
|
buf = "\n".join(lines[hdr_idx:])
|
||||||
|
df = pd.read_csv(io.StringIO(buf), dtype=str, on_bad_lines="skip")
|
||||||
|
else:
|
||||||
|
# Excel: read without header to find preamble extent
|
||||||
|
df_raw = pd.read_excel(io.BytesIO(data), dtype=str, header=None, nrows=20)
|
||||||
|
hdr_row = _find_header_row(df_raw, col_map)
|
||||||
|
df = pd.read_excel(io.BytesIO(data), dtype=str, header=hdr_row)
|
||||||
|
|
||||||
|
return normalise_columns(df, col_map)
|
||||||
|
except Exception as e:
|
||||||
|
print(f" ERROR: {zpath.name}/{pattern}: {e}")
|
||||||
|
return pd.DataFrame()
|
||||||
|
|
||||||
|
|
||||||
|
def clean_numeric(series: pd.Series) -> pd.Series:
|
||||||
|
"""Strip $ and commas, convert to float."""
|
||||||
|
return (
|
||||||
|
series.astype(str)
|
||||||
|
.str.replace("$", "", regex=False)
|
||||||
|
.str.replace(",", "", regex=False)
|
||||||
|
.str.strip()
|
||||||
|
.pipe(pd.to_numeric, errors="coerce")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Main ingestion
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def ingest_all(con: duckdb.DuckDBPyConnection, year_filter: str = "") -> None:
|
||||||
|
"""Ingest all downloaded OPPS files into DuckDB."""
|
||||||
|
|
||||||
|
con.execute("CREATE SCHEMA IF NOT EXISTS opps")
|
||||||
|
|
||||||
|
apc_frames: list[pd.DataFrame] = []
|
||||||
|
addb_frames: list[pd.DataFrame] = []
|
||||||
|
wage_frames: list[pd.DataFrame] = []
|
||||||
|
|
||||||
|
year_dirs = sorted(OPPS_DIR.iterdir())
|
||||||
|
if year_filter:
|
||||||
|
year_dirs = [d for d in year_dirs if d.name == year_filter]
|
||||||
|
|
||||||
|
for year_dir in year_dirs:
|
||||||
|
if not year_dir.is_dir():
|
||||||
|
continue
|
||||||
|
year = year_dir.name
|
||||||
|
print(f"\n--- CY{year} ---")
|
||||||
|
|
||||||
|
# Combined addenda ZIP (CY2021+)
|
||||||
|
addenda_zip = year_dir / f"opps_{year}_addenda.zip"
|
||||||
|
if not addenda_zip.exists():
|
||||||
|
# Try old single-file patterns
|
||||||
|
addenda_zip = year_dir / f"opps_{year}_addendum_a.zip"
|
||||||
|
|
||||||
|
if addenda_zip.exists():
|
||||||
|
# Addendum A — APC weights
|
||||||
|
df = read_from_zip(addenda_zip, ADDENDUM_A_COL_MAP, pattern="addendum a")
|
||||||
|
if not df.empty:
|
||||||
|
df["year"] = int(year)
|
||||||
|
for col in ["relative_weight", "payment_rate", "minimum_unadjusted_copayment"]:
|
||||||
|
if col in df.columns:
|
||||||
|
df[col] = clean_numeric(df[col])
|
||||||
|
apc_frames.append(df)
|
||||||
|
print(f" Addendum A: {len(df)} APCs")
|
||||||
|
|
||||||
|
# Addendum B — HCPCS→APC crosswalk
|
||||||
|
df = read_from_zip(addenda_zip, ADDENDUM_B_COL_MAP, pattern="addendum b")
|
||||||
|
if not df.empty:
|
||||||
|
df["year"] = int(year)
|
||||||
|
for col in ["relative_weight", "payment_rate", "minimum_unadjusted_copayment"]:
|
||||||
|
if col in df.columns:
|
||||||
|
df[col] = clean_numeric(df[col])
|
||||||
|
addb_frames.append(df)
|
||||||
|
print(f" Addendum B: {len(df)} HCPCS codes")
|
||||||
|
|
||||||
|
# Load into DuckDB
|
||||||
|
print("\n--- Loading into DuckDB ---")
|
||||||
|
|
||||||
|
if apc_frames:
|
||||||
|
all_apc = pd.concat(apc_frames, ignore_index=True)
|
||||||
|
con.execute("DROP TABLE IF EXISTS opps.apc_weight")
|
||||||
|
con.execute("CREATE TABLE opps.apc_weight AS SELECT * FROM all_apc")
|
||||||
|
print(f" opps.apc_weight: {len(all_apc)} rows "
|
||||||
|
f"({all_apc['year'].nunique()} years)")
|
||||||
|
|
||||||
|
if addb_frames:
|
||||||
|
all_addb = pd.concat(addb_frames, ignore_index=True)
|
||||||
|
con.execute("DROP TABLE IF EXISTS opps.addendum_b")
|
||||||
|
con.execute("CREATE TABLE opps.addendum_b AS SELECT * FROM all_addb")
|
||||||
|
print(f" opps.addendum_b: {len(all_addb)} rows "
|
||||||
|
f"({all_addb['year'].nunique()} years)")
|
||||||
|
|
||||||
|
if wage_frames:
|
||||||
|
all_wage = pd.concat(wage_frames, ignore_index=True)
|
||||||
|
con.execute("DROP TABLE IF EXISTS opps.wage_index")
|
||||||
|
con.execute("CREATE TABLE opps.wage_index AS SELECT * FROM all_wage")
|
||||||
|
print(f" opps.wage_index: {len(all_wage)} rows "
|
||||||
|
f"({all_wage['year'].nunique()} years)")
|
||||||
|
|
||||||
|
# Skin sub specific: extract skin sub codes from Addendum B history
|
||||||
|
if addb_frames:
|
||||||
|
print("\n--- Skin substitute history in Addendum B ---")
|
||||||
|
all_addb = pd.concat(addb_frames, ignore_index=True)
|
||||||
|
if "hcpcs" in all_addb.columns:
|
||||||
|
skin_codes = all_addb[
|
||||||
|
all_addb["hcpcs"].astype(str).str.match(r"^Q4\d{2,3}$|^C527[1-8]$", na=False)
|
||||||
|
]
|
||||||
|
if not skin_codes.empty:
|
||||||
|
con.execute("DROP TABLE IF EXISTS opps.skin_sub_addendum_b")
|
||||||
|
con.execute("CREATE TABLE opps.skin_sub_addendum_b AS SELECT * FROM skin_codes")
|
||||||
|
print(f" opps.skin_sub_addendum_b: {len(skin_codes)} rows "
|
||||||
|
f"({skin_codes['year'].nunique()} years, "
|
||||||
|
f"{skin_codes['hcpcs'].nunique()} unique codes)")
|
||||||
|
|
||||||
|
# SI distribution for skin subs
|
||||||
|
if "status_indicator" in skin_codes.columns:
|
||||||
|
print(" Status indicators for skin subs by year:")
|
||||||
|
si_year = skin_codes.groupby(["year", "status_indicator"]).size()
|
||||||
|
for (yr, si), ct in si_year.items():
|
||||||
|
print(f" CY{yr} SI={si}: {ct} codes")
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
parser = argparse.ArgumentParser(description="Ingest CMS OPPS files")
|
||||||
|
parser.add_argument("--year", help="Ingest only this year")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
print("Ingesting CMS OPPS files into DuckDB ...")
|
||||||
|
con = duckdb.connect(str(DUCKDB_PATH))
|
||||||
|
|
||||||
|
ingest_all(con, year_filter=args.year or "")
|
||||||
|
|
||||||
|
# Final inventory
|
||||||
|
print("\n--- OPPS tables ---")
|
||||||
|
for r in con.execute("""
|
||||||
|
SELECT table_name FROM information_schema.tables
|
||||||
|
WHERE table_schema = 'opps' ORDER BY table_name
|
||||||
|
""").fetchall():
|
||||||
|
cnt = con.execute(f"SELECT count(*) FROM opps.{r[0]}").fetchone()[0]
|
||||||
|
print(f" opps.{r[0]:30s}: {cnt:>6} rows")
|
||||||
|
|
||||||
|
con.close()
|
||||||
|
print("\nDone.")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
Reference in New Issue
Block a user