#!/usr/bin/env python3 """Build a reproducible Medicare Part D GLP-1 spending-by-state dataset. Source: CMS Medicare Part D Prescribers - by Geography and Drug, 2023 and 2024. Scope: standalone GLP-1 receptor agonists plus tirzepatide (dual GIP/GLP-1), excluding fixed-ratio insulin/GLP-1 combination products from headline totals. This script uses only the Python standard library. """ from __future__ import annotations import csv import hashlib import json from collections import defaultdict from dataclasses import dataclass from datetime import date from decimal import Decimal, ROUND_HALF_UP, getcontext from pathlib import Path from typing import Dict, Iterable, List, Mapping, MutableMapping, Optional, Sequence, Tuple getcontext().prec = 34 BASE_DIR = Path(__file__).resolve().parent SOURCE_2023 = BASE_DIR / "MUP_DPR_RY25_P04_V10_DY23_Geo.csv" SOURCE_2024 = BASE_DIR / "MUP_DPR_RY26_P04_V10_DY24_Geo.csv" VERIFIED_DATE = "2026-08-01" DATASET_VERSION = "1.0.0" CMS_2023_URL = ( "https://data.cms.gov/sites/default/files/2025-04/" "9fe6b8a6-0cb9-4b7c-9760-87800da010a8/" "MUP_DPR_RY25_P04_V10_DY23_Geo.csv" ) CMS_2024_URL = ( "https://data.cms.gov/sites/default/files/2026-05/" "3e80b44e-5a87-4414-959a-b6a7c8c18720/" "MUP_DPR_RY26_P04_V10_DY24_Geo.csv" ) CMS_DATASET_PAGE = ( "https://data.cms.gov/provider-summary-by-type-of-service/" "medicare-part-d-prescribers/medicare-part-d-prescribers-by-geography-and-drug" ) CMS_METHOD_PAGE = "https://data.cms.gov/resources/medicare-part-d-prescribers-methodology" CMS_DICTIONARY_PAGE = ( "https://data.cms.gov/resources/" "medicare-part-d-prescribers-by-geography-and-drug-data-dictionary" ) FDA_GLP1_LIST_URL = "https://www.fda.gov/media/175358/download" # Year-specific CMS rows that implement the same conceptual rule: # all standalone GLP-1 products and tirzepatide rows present in that year's PUF. # Fixed-ratio insulin combinations are excluded from the headline dataset. INCLUDED_BRANDS_BY_YEAR: Mapping[int, Sequence[str]] = { 2023: ( "Trulicity", "Byetta", "Bydureon Bcise", "Bydureon Pen", "Saxenda", "Victoza 2-Pak", "Victoza 3-Pak", "Ozempic", "Rybelsus", "Wegovy", "Mounjaro", ), 2024: ( "Trulicity", "Byetta", "Bydureon Bcise", "Liraglutide", "Victoza 2-Pak", "Victoza 3-Pak", "Ozempic", "Rybelsus", "Wegovy", "Mounjaro", "Zepbound", ), } EXCLUDED_FIXED_RATIO_COMBINATIONS: Sequence[str] = ( "Xultophy 100-3.6", "Soliqua 100-33", ) STATES: Mapping[str, str] = { "Alabama": "AL", "Alaska": "AK", "Arizona": "AZ", "Arkansas": "AR", "California": "CA", "Colorado": "CO", "Connecticut": "CT", "Delaware": "DE", "District of Columbia": "DC", "Florida": "FL", "Georgia": "GA", "Hawaii": "HI", "Idaho": "ID", "Illinois": "IL", "Indiana": "IN", "Iowa": "IA", "Kansas": "KS", "Kentucky": "KY", "Louisiana": "LA", "Maine": "ME", "Maryland": "MD", "Massachusetts": "MA", "Michigan": "MI", "Minnesota": "MN", "Mississippi": "MS", "Missouri": "MO", "Montana": "MT", "Nebraska": "NE", "Nevada": "NV", "New Hampshire": "NH", "New Jersey": "NJ", "New Mexico": "NM", "New York": "NY", "North Carolina": "NC", "North Dakota": "ND", "Ohio": "OH", "Oklahoma": "OK", "Oregon": "OR", "Pennsylvania": "PA", "Rhode Island": "RI", "South Carolina": "SC", "South Dakota": "SD", "Tennessee": "TN", "Texas": "TX", "Utah": "UT", "Vermont": "VT", "Virginia": "VA", "Washington": "WA", "West Virginia": "WV", "Wisconsin": "WI", "Wyoming": "WY", } NUMERIC_FIELDS = ( "Tot_Prscrbrs", "Tot_Clms", "Tot_30day_Fills", "Tot_Drug_Cst", "Tot_Benes", "GE65_Tot_Clms", "GE65_Tot_30day_Fills", "GE65_Tot_Drug_Cst", "GE65_Tot_Benes", "LIS_Bene_Cst_Shr", "NonLIS_Bene_Cst_Shr", ) def dec(value: Optional[str]) -> Decimal: if value is None or value == "": return Decimal(0) return Decimal(value) def blank_or_decimal(value: Optional[str]) -> Optional[Decimal]: if value is None or value == "": return None return Decimal(value) def sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def read_selected_rows(path: Path, year: int) -> Tuple[List[dict], List[dict], List[dict]]: included = set(INCLUDED_BRANDS_BY_YEAR[year]) excluded = set(EXCLUDED_FIXED_RATIO_COMBINATIONS) included_rows: List[dict] = [] combo_rows: List[dict] = [] all_rows: List[dict] = [] with path.open(newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) required = { "Prscrbr_Geo_Lvl", "Prscrbr_Geo_Desc", "Brnd_Name", "Gnrc_Name", "Tot_Clms", "Tot_30day_Fills", "Tot_Drug_Cst", } missing = required.difference(reader.fieldnames or []) if missing: raise ValueError(f"Missing required CMS columns in {path.name}: {sorted(missing)}") for row in reader: brand = row["Brnd_Name"] if brand in included: included_rows.append(row) all_rows.append(row) elif brand in excluded: combo_rows.append(row) all_rows.append(row) return included_rows, combo_rows, all_rows def sum_field(rows: Iterable[Mapping[str, str]], field: str) -> Decimal: return sum((dec(row.get(field)) for row in rows), Decimal(0)) def q_money(value: Decimal) -> str: return str(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) def q_count(value: Decimal) -> str: return str(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP)) def q_fill(value: Decimal) -> str: return str(value.quantize(Decimal("0.1"), rounding=ROUND_HALF_UP)) def q_pct(value: Optional[Decimal]) -> str: if value is None: return "" return str((value * 100).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)) def q_rate(value: Optional[Decimal]) -> str: if value is None: return "" return str(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) def aggregate_state(rows: Sequence[dict], year: int) -> Tuple[Dict[str, dict], Dict[Tuple[str, str], dict]]: state_agg: Dict[str, dict] = {} state_product: Dict[Tuple[str, str], dict] = {} for state in STATES: state_agg[state] = { "Tot_Drug_Cst": Decimal(0), "Tot_Clms": Decimal(0), "Tot_30day_Fills": Decimal(0), "GE65_Tot_Drug_Cst": Decimal(0), "LIS_Bene_Cst_Shr": Decimal(0), "NonLIS_Bene_Cst_Shr": Decimal(0), "reported_brands": set(), "suppressed_age_65_plus_utilization_product_cells": 0, "suppressed_age_65_plus_beneficiary_product_cells": 0, } for row in rows: if row["Prscrbr_Geo_Lvl"] != "State": continue state = row["Prscrbr_Geo_Desc"] if state not in STATES: continue brand = row["Brnd_Name"] a = state_agg[state] for field in ( "Tot_Drug_Cst", "Tot_Clms", "Tot_30day_Fills", "GE65_Tot_Drug_Cst", "LIS_Bene_Cst_Shr", "NonLIS_Bene_Cst_Shr", ): a[field] += dec(row.get(field)) a["reported_brands"].add(brand) if row.get("GE65_Tot_Drug_Cst", "") == "": a["suppressed_age_65_plus_utilization_product_cells"] += 1 if row.get("GE65_Tot_Benes", "") == "": a["suppressed_age_65_plus_beneficiary_product_cells"] += 1 state_product[(state, brand)] = row return state_agg, state_product def national_rows(rows: Sequence[dict]) -> Dict[str, dict]: return { row["Brnd_Name"]: row for row in rows if row["Prscrbr_Geo_Lvl"] == "National" } def build_outputs() -> None: if not SOURCE_2023.exists() or not SOURCE_2024.exists(): raise FileNotFoundError( "Place the CMS 2023 and 2024 geography-and-drug CSVs beside this script. " f"Missing: {[str(p) for p in (SOURCE_2023, SOURCE_2024) if not p.exists()]}" ) rows23, combo23, _ = read_selected_rows(SOURCE_2023, 2023) rows24, combo24, _ = read_selected_rows(SOURCE_2024, 2024) n23 = national_rows(rows23) n24 = national_rows(rows24) expected23 = set(INCLUDED_BRANDS_BY_YEAR[2023]) expected24 = set(INCLUDED_BRANDS_BY_YEAR[2024]) if set(n23) != expected23: raise AssertionError(f"Unexpected 2023 national product rows: {sorted(set(n23) ^ expected23)}") if set(n24) != expected24: raise AssertionError(f"Unexpected 2024 national product rows: {sorted(set(n24) ^ expected24)}") nat23_cost = sum_field(n23.values(), "Tot_Drug_Cst") nat24_cost = sum_field(n24.values(), "Tot_Drug_Cst") nat23_claims = sum_field(n23.values(), "Tot_Clms") nat24_claims = sum_field(n24.values(), "Tot_Clms") nat23_fills = sum_field(n23.values(), "Tot_30day_Fills") nat24_fills = sum_field(n24.values(), "Tot_30day_Fills") nat24_age65 = sum_field(n24.values(), "GE65_Tot_Drug_Cst") # Integrity checks against the published build's expected totals. assert nat24_cost == Decimal("27503414515.35") assert nat24_claims == Decimal("21825833") assert nat24_fills == Decimal("29286812.4") assert nat23_cost == Decimal("22204068493.62") assert nat23_claims == Decimal("16229271") assert nat23_fills == Decimal("22753714.3") state23, product23 = aggregate_state(rows23, 2023) state24, product24 = aggregate_state(rows24, 2024) state_records: List[dict] = [] for state, abbr in STATES.items(): a23 = state23[state] a24 = state24[state] cost23 = a23["Tot_Drug_Cst"] cost24 = a24["Tot_Drug_Cst"] claims23 = a23["Tot_Clms"] claims24 = a24["Tot_Clms"] fills23 = a23["Tot_30day_Fills"] fills24 = a24["Tot_30day_Fills"] product_costs24 = { brand: dec(product24[(state, brand)]["Tot_Drug_Cst"]) for brand in INCLUDED_BRANDS_BY_YEAR[2024] if (state, brand) in product24 } top_brand, top_cost = max(product_costs24.items(), key=lambda x: x[1]) state_records.append({ "state": state, "state_abbreviation": abbr, "gross_drug_cost_2024_usd": cost24, "gross_drug_cost_2023_usd": cost23, "gross_drug_cost_change_2023_to_2024_usd": cost24 - cost23, "gross_drug_cost_change_2023_to_2024_pct": (cost24 / cost23 - 1) if cost23 else None, "claims_2024": claims24, "claims_2023": claims23, "claims_change_2023_to_2024_pct": (claims24 / claims23 - 1) if claims23 else None, "standardized_30_day_fills_2024": fills24, "standardized_30_day_fills_2023": fills23, "fills_change_2023_to_2024_pct": (fills24 / fills23 - 1) if fills23 else None, "gross_cost_per_claim_2024_usd": cost24 / claims24 if claims24 else None, "gross_cost_per_standardized_30_day_fill_2024_usd": cost24 / fills24 if fills24 else None, "share_of_national_gross_cost_2024_pct": cost24 / nat24_cost if nat24_cost else None, "reported_age_65_plus_gross_drug_cost_2024_usd": a24["GE65_Tot_Drug_Cst"], "reported_age_65_plus_share_of_state_cost_2024_pct": ( a24["GE65_Tot_Drug_Cst"] / cost24 if cost24 else None ), "age_65_plus_share_is_lower_bound": ( a24["suppressed_age_65_plus_utilization_product_cells"] > 0 ), "suppressed_age_65_plus_utilization_product_cells_2024": ( a24["suppressed_age_65_plus_utilization_product_cells"] ), "suppressed_age_65_plus_beneficiary_product_cells_2024": ( a24["suppressed_age_65_plus_beneficiary_product_cells"] ), "top_product_2024": top_brand, "top_product_gross_cost_2024_usd": top_cost, "top_product_share_of_state_cost_2024_pct": top_cost / cost24 if cost24 else None, "reported_product_cells_2024": len(a24["reported_brands"]), "product_cells_without_reportable_cms_row_2024": ( len(INCLUDED_BRANDS_BY_YEAR[2024]) - len(a24["reported_brands"]) ), }) state_records.sort(key=lambda r: r["gross_drug_cost_2024_usd"], reverse=True) for rank, record in enumerate(state_records, 1): record["rank_2024"] = rank state_total_2024 = sum((r["gross_drug_cost_2024_usd"] for r in state_records), Decimal(0)) assert state_total_2024 == Decimal("27365405390.12") # State summary output. state_path = BASE_DIR / "medicare_glp1_spending_by_state_2024.csv" state_fields = [ "rank_2024", "state", "state_abbreviation", "gross_drug_cost_2024_usd", "gross_drug_cost_2023_usd", "gross_drug_cost_change_2023_to_2024_usd", "gross_drug_cost_change_2023_to_2024_pct", "claims_2024", "claims_2023", "claims_change_2023_to_2024_pct", "standardized_30_day_fills_2024", "standardized_30_day_fills_2023", "fills_change_2023_to_2024_pct", "gross_cost_per_claim_2024_usd", "gross_cost_per_standardized_30_day_fill_2024_usd", "share_of_national_gross_cost_2024_pct", "reported_age_65_plus_gross_drug_cost_2024_usd", "reported_age_65_plus_share_of_state_cost_2024_pct", "age_65_plus_share_is_lower_bound", "suppressed_age_65_plus_utilization_product_cells_2024", "suppressed_age_65_plus_beneficiary_product_cells_2024", "top_product_2024", "top_product_gross_cost_2024_usd", "top_product_share_of_state_cost_2024_pct", "reported_product_cells_2024", "product_cells_without_reportable_cms_row_2024", "geography_basis", "cost_basis", "source_data_year", "verified_date", "dataset_version", ] with state_path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=state_fields) writer.writeheader() for r in state_records: writer.writerow({ **{k: r.get(k, "") for k in state_fields}, "gross_drug_cost_2024_usd": q_money(r["gross_drug_cost_2024_usd"]), "gross_drug_cost_2023_usd": q_money(r["gross_drug_cost_2023_usd"]), "gross_drug_cost_change_2023_to_2024_usd": q_money(r["gross_drug_cost_change_2023_to_2024_usd"]), "gross_drug_cost_change_2023_to_2024_pct": q_pct(r["gross_drug_cost_change_2023_to_2024_pct"]), "claims_2024": q_count(r["claims_2024"]), "claims_2023": q_count(r["claims_2023"]), "claims_change_2023_to_2024_pct": q_pct(r["claims_change_2023_to_2024_pct"]), "standardized_30_day_fills_2024": q_fill(r["standardized_30_day_fills_2024"]), "standardized_30_day_fills_2023": q_fill(r["standardized_30_day_fills_2023"]), "fills_change_2023_to_2024_pct": q_pct(r["fills_change_2023_to_2024_pct"]), "gross_cost_per_claim_2024_usd": q_rate(r["gross_cost_per_claim_2024_usd"]), "gross_cost_per_standardized_30_day_fill_2024_usd": q_rate(r["gross_cost_per_standardized_30_day_fill_2024_usd"]), "share_of_national_gross_cost_2024_pct": q_pct(r["share_of_national_gross_cost_2024_pct"]), "reported_age_65_plus_gross_drug_cost_2024_usd": q_money(r["reported_age_65_plus_gross_drug_cost_2024_usd"]), "reported_age_65_plus_share_of_state_cost_2024_pct": q_pct(r["reported_age_65_plus_share_of_state_cost_2024_pct"]), "age_65_plus_share_is_lower_bound": str(r["age_65_plus_share_is_lower_bound"]).lower(), "top_product_gross_cost_2024_usd": q_money(r["top_product_gross_cost_2024_usd"]), "top_product_share_of_state_cost_2024_pct": q_pct(r["top_product_share_of_state_cost_2024_pct"]), "geography_basis": "Prescriber state (provider location in NPPES), not beneficiary residence", "cost_basis": "CMS total drug cost: gross Part D drug cost before manufacturer rebates; includes amounts paid by plans, beneficiaries, subsidies, and third parties", "source_data_year": "2024 (with 2023 comparison)", "verified_date": VERIFIED_DATE, "dataset_version": DATASET_VERSION, }) # Complete state-product matrix. Missing CMS rows remain blank, not zero. product_path = BASE_DIR / "medicare_glp1_spending_state_product_2024.csv" product_fields = [ "state", "state_abbreviation", "product_brand", "generic_name", "cms_row_reported", "total_prescribers", "claims", "standardized_30_day_fills", "gross_drug_cost_usd", "beneficiaries_product_specific", "age_65_plus_claims", "age_65_plus_standardized_30_day_fills", "age_65_plus_gross_drug_cost_usd", "age_65_plus_beneficiaries_product_specific", "age_65_plus_utilization_fields_suppressed", "age_65_plus_utilization_suppression_flag", "age_65_plus_beneficiary_field_suppressed", "age_65_plus_beneficiary_suppression_flag", "low_income_subsidy_beneficiary_cost_share_usd", "non_low_income_subsidy_beneficiary_cost_share_usd", "note", "source_data_year", "verified_date", "dataset_version", ] with product_path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=product_fields) writer.writeheader() for state, abbr in STATES.items(): for brand in INCLUDED_BRANDS_BY_YEAR[2024]: row = product24.get((state, brand)) if row is None: writer.writerow({ "state": state, "state_abbreviation": abbr, "product_brand": brand, "generic_name": "", "cms_row_reported": "false", "note": "No reportable state-product row in the CMS 2024 geography file; do not interpret as a confirmed zero.", "source_data_year": 2024, "verified_date": VERIFIED_DATE, "dataset_version": DATASET_VERSION, }) continue age_utilization_suppressed = row.get("GE65_Tot_Drug_Cst", "") == "" age_beneficiary_suppressed = row.get("GE65_Tot_Benes", "") == "" note_parts = [] if age_utilization_suppressed: note_parts.append( "CMS suppressed the age-65-plus claims, standardized-fill, and gross-cost fields; blanks do not mean zero." ) if age_beneficiary_suppressed: note_parts.append( "CMS suppressed the age-65-plus beneficiary count; a blank does not mean zero." ) note_parts.append( "Beneficiary counts are unique within this state-product row only and must not be summed across products as unique class users." ) writer.writerow({ "state": state, "state_abbreviation": abbr, "product_brand": brand, "generic_name": row["Gnrc_Name"], "cms_row_reported": "true", "total_prescribers": row["Tot_Prscrbrs"], "claims": row["Tot_Clms"], "standardized_30_day_fills": row["Tot_30day_Fills"], "gross_drug_cost_usd": row["Tot_Drug_Cst"], "beneficiaries_product_specific": row["Tot_Benes"], "age_65_plus_claims": row["GE65_Tot_Clms"], "age_65_plus_standardized_30_day_fills": row["GE65_Tot_30day_Fills"], "age_65_plus_gross_drug_cost_usd": row["GE65_Tot_Drug_Cst"], "age_65_plus_beneficiaries_product_specific": row["GE65_Tot_Benes"], "age_65_plus_utilization_fields_suppressed": str(age_utilization_suppressed).lower(), "age_65_plus_utilization_suppression_flag": row.get("GE65_Sprsn_Flag", ""), "age_65_plus_beneficiary_field_suppressed": str(age_beneficiary_suppressed).lower(), "age_65_plus_beneficiary_suppression_flag": row.get("GE65_Bene_Sprsn_Flag", ""), "low_income_subsidy_beneficiary_cost_share_usd": row["LIS_Bene_Cst_Shr"], "non_low_income_subsidy_beneficiary_cost_share_usd": row["NonLIS_Bene_Cst_Shr"], "note": " ".join(note_parts), "source_data_year": 2024, "verified_date": VERIFIED_DATE, "dataset_version": DATASET_VERSION, }) # National product table. national_path = BASE_DIR / "medicare_glp1_spending_national_product_2024.csv" national_fields = [ "rank_2024", "product_brand", "generic_name", "gross_drug_cost_2024_usd", "share_of_included_national_gross_cost_2024_pct", "claims_2024", "standardized_30_day_fills_2024", "gross_cost_per_claim_2024_usd", "gross_cost_per_standardized_30_day_fill_2024_usd", "beneficiaries_product_specific_2024", "age_65_plus_gross_drug_cost_2024_usd", "source_data_year", "verified_date", "dataset_version", ] national_records = [] for brand, row in n24.items(): cost = dec(row["Tot_Drug_Cst"]) claims = dec(row["Tot_Clms"]) fills = dec(row["Tot_30day_Fills"]) national_records.append({ "product_brand": brand, "generic_name": row["Gnrc_Name"], "gross_drug_cost_2024_usd": cost, "share_of_included_national_gross_cost_2024_pct": cost / nat24_cost, "claims_2024": claims, "standardized_30_day_fills_2024": fills, "gross_cost_per_claim_2024_usd": cost / claims if claims else None, "gross_cost_per_standardized_30_day_fill_2024_usd": cost / fills if fills else None, "beneficiaries_product_specific_2024": blank_or_decimal(row["Tot_Benes"]), "age_65_plus_gross_drug_cost_2024_usd": dec(row["GE65_Tot_Drug_Cst"]), }) national_records.sort(key=lambda r: r["gross_drug_cost_2024_usd"], reverse=True) with national_path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=national_fields) writer.writeheader() for rank, r in enumerate(national_records, 1): writer.writerow({ "rank_2024": rank, "product_brand": r["product_brand"], "generic_name": r["generic_name"], "gross_drug_cost_2024_usd": q_money(r["gross_drug_cost_2024_usd"]), "share_of_included_national_gross_cost_2024_pct": q_pct(r["share_of_included_national_gross_cost_2024_pct"]), "claims_2024": q_count(r["claims_2024"]), "standardized_30_day_fills_2024": q_fill(r["standardized_30_day_fills_2024"]), "gross_cost_per_claim_2024_usd": q_rate(r["gross_cost_per_claim_2024_usd"]), "gross_cost_per_standardized_30_day_fill_2024_usd": q_rate(r["gross_cost_per_standardized_30_day_fill_2024_usd"]), "beneficiaries_product_specific_2024": ( q_count(r["beneficiaries_product_specific_2024"]) if r["beneficiaries_product_specific_2024"] is not None else "" ), "age_65_plus_gross_drug_cost_2024_usd": q_money(r["age_65_plus_gross_drug_cost_2024_usd"]), "source_data_year": 2024, "verified_date": VERIFIED_DATE, "dataset_version": DATASET_VERSION, }) # Territories / other prescriber geographies in the 2024 source. other_path = BASE_DIR / "medicare_glp1_spending_other_geographies_2024.csv" other_agg: MutableMapping[str, dict] = defaultdict(lambda: { "gross_drug_cost": Decimal(0), "claims": Decimal(0), "fills": Decimal(0), "reported_cells": 0, }) for row in rows24: if row["Prscrbr_Geo_Lvl"] != "State": continue geo = row["Prscrbr_Geo_Desc"] if geo in STATES: continue a = other_agg[geo] a["gross_drug_cost"] += dec(row["Tot_Drug_Cst"]) a["claims"] += dec(row["Tot_Clms"]) a["fills"] += dec(row["Tot_30day_Fills"]) a["reported_cells"] += 1 with other_path.open("w", newline="", encoding="utf-8") as f: fields = [ "prescriber_geography", "gross_drug_cost_2024_usd", "claims_2024", "standardized_30_day_fills_2024", "reported_product_cells_2024", "source_data_year", "verified_date", "dataset_version", ] writer = csv.DictWriter(f, fieldnames=fields) writer.writeheader() for geo, a in sorted(other_agg.items(), key=lambda x: x[1]["gross_drug_cost"], reverse=True): writer.writerow({ "prescriber_geography": geo, "gross_drug_cost_2024_usd": q_money(a["gross_drug_cost"]), "claims_2024": q_count(a["claims"]), "standardized_30_day_fills_2024": q_fill(a["fills"]), "reported_product_cells_2024": a["reported_cells"], "source_data_year": 2024, "verified_date": VERIFIED_DATE, "dataset_version": DATASET_VERSION, }) # Product inclusion crosswalk. inclusion_path = BASE_DIR / "medicare_glp1_product_scope_2023_2024.csv" all_brands = sorted(set(INCLUDED_BRANDS_BY_YEAR[2023]) | set(INCLUDED_BRANDS_BY_YEAR[2024]) | set(EXCLUDED_FIXED_RATIO_COMBINATIONS)) with inclusion_path.open("w", newline="", encoding="utf-8") as f: fields = ["product_brand", "included_2023", "included_2024", "headline_scope", "reason"] writer = csv.DictWriter(f, fieldnames=fields) writer.writeheader() for brand in all_brands: excluded = brand in EXCLUDED_FIXED_RATIO_COMBINATIONS writer.writerow({ "product_brand": brand, "included_2023": str(brand in INCLUDED_BRANDS_BY_YEAR[2023]).lower(), "included_2024": str(brand in INCLUDED_BRANDS_BY_YEAR[2024]).lower(), "headline_scope": "excluded" if excluded else "included when present in that year's CMS file", "reason": ( "Fixed-ratio insulin/GLP-1 combination; reported separately in sensitivity analysis" if excluded else "Standalone GLP-1 product or tirzepatide (dual GIP/GLP-1)" ), }) combo24_national = [r for r in combo24 if r["Prscrbr_Geo_Lvl"] == "National"] combo24_cost = sum_field(combo24_national, "Tot_Drug_Cst") affected_age_states = [ r["state"] for r in state_records if r["age_65_plus_share_is_lower_bound"] ] suppressed_age_utilization_cells = sum( r["suppressed_age_65_plus_utilization_product_cells_2024"] for r in state_records ) suppressed_age_beneficiary_cells = sum( r["suppressed_age_65_plus_beneficiary_product_cells_2024"] for r in state_records ) reported_state_product_cells = sum(r["reported_product_cells_2024"] for r in state_records) absent_state_product_cells = sum(r["product_cells_without_reportable_cms_row_2024"] for r in state_records) summary = { "dataset_name": "Medicare GLP-1 Spending by Prescriber State", "dataset_version": DATASET_VERSION, "verified_date": VERIFIED_DATE, "national_2024": { "gross_drug_cost_usd": q_money(nat24_cost), "claims": q_count(nat24_claims), "standardized_30_day_fills": q_fill(nat24_fills), "gross_cost_per_claim_usd": q_rate(nat24_cost / nat24_claims), "gross_cost_per_standardized_30_day_fill_usd": q_rate(nat24_cost / nat24_fills), "age_65_plus_gross_drug_cost_usd": q_money(nat24_age65), "age_65_plus_share_pct": q_pct(nat24_age65 / nat24_cost), }, "national_2023": { "gross_drug_cost_usd": q_money(nat23_cost), "claims": q_count(nat23_claims), "standardized_30_day_fills": q_fill(nat23_fills), }, "change_2023_to_2024": { "gross_drug_cost_pct": q_pct(nat24_cost / nat23_cost - 1), "claims_pct": q_pct(nat24_claims / nat23_claims - 1), "standardized_30_day_fills_pct": q_pct(nat24_fills / nat23_fills - 1), }, "state_and_dc_2024": { "gross_drug_cost_usd": q_money(state_total_2024), "share_of_national_pct": q_pct(state_total_2024 / nat24_cost), "national_less_states_dc_usd": q_money(nat24_cost - state_total_2024), }, "scope_sensitivity": { "excluded_fixed_ratio_combo_gross_cost_2024_usd": q_money(combo24_cost), "headline_plus_fixed_ratio_combo_gross_cost_2024_usd": q_money(nat24_cost + combo24_cost), "fixed_ratio_combo_increment_over_headline_pct": q_pct(combo24_cost / nat24_cost), }, "state_product_reporting_2024": { "possible_state_product_cells": len(STATES) * len(INCLUDED_BRANDS_BY_YEAR[2024]), "reported_state_product_cells": reported_state_product_cells, "cells_without_a_reportable_cms_row": absent_state_product_cells, "missing_row_semantics": "No reportable CMS state-product row; do not interpret as a confirmed zero.", }, "state_age_65_plus_suppression_2024": { "suppressed_age_65_plus_utilization_product_cells": suppressed_age_utilization_cells, "suppressed_age_65_plus_beneficiary_product_cells": suppressed_age_beneficiary_cells, "affected_jurisdiction_count": len(affected_age_states), "affected_jurisdictions": sorted(affected_age_states), "state_share_semantics": "For affected jurisdictions, the reported age-65-plus gross-cost total and share are lower bounds because one or more product-level age fields are suppressed.", }, "sources": { "cms_dataset_page": CMS_DATASET_PAGE, "cms_2023_csv": CMS_2023_URL, "cms_2024_csv": CMS_2024_URL, "cms_methodology": CMS_METHOD_PAGE, "cms_data_dictionary": CMS_DICTIONARY_PAGE, "fda_glp1_product_list": FDA_GLP1_LIST_URL, }, "source_sha256": { SOURCE_2023.name: sha256(SOURCE_2023), SOURCE_2024.name: sha256(SOURCE_2024), }, } summary_path = BASE_DIR / "medicare_glp1_spending_summary_2024.json" summary_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") output_paths = [state_path, product_path, national_path, other_path, inclusion_path, summary_path] manifest = { "dataset_version": DATASET_VERSION, "verified_date": VERIFIED_DATE, "files": [ { "file": p.name, "bytes": p.stat().st_size, "sha256": sha256(p), } for p in output_paths ], } manifest_path = BASE_DIR / "medicare_glp1_spending_manifest_2024.json" manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") print(json.dumps({ "national_2024_gross_cost": q_money(nat24_cost), "national_2024_claims": q_count(nat24_claims), "national_2023_gross_cost": q_money(nat23_cost), "cost_growth_pct": q_pct(nat24_cost / nat23_cost - 1), "states_dc_2024_gross_cost": q_money(state_total_2024), "outputs": [p.name for p in output_paths] + [manifest_path.name], }, indent=2)) if __name__ == "__main__": build_outputs()