ASP pricing trajectories (skin_subs.asp_trajectories): - 1,974 rows: QoQ % change, cumulative change, entry quarter - 39 products with anomalous >50% QoQ spikes - Jan 2026 flat-rate impact: 105 losers, 47 gainers - XWRAP Dual: $5,893/cm² payment vs $127.28 flat rate Market segmentation (skin_subs.products_enriched, skin_subs.manufacturers): - 286 products enriched with ASP stats + claims utilization - 68 manufacturers; Organogenesis, MiMedx, Smith+Nephew top 3 - Amniotic membrane category: 214 products, avg ASP $1,392/cm² Utilization metrics (skin_subs.providers, beneficiaries, utilization_summary): - 200 providers, 1,255 beneficiaries from synthetic claims - Podiatry: 60% of claims ($1.02M) — consistent with DOJ cases - Office setting: 68% of spend — lowest oversight environment - Top provider: $55K paid, 13 patients, $4,247/patient (FL dermatology)
This commit is contained in:
kert
386
dev/scripts/build_skin_subs_analytics.py
Normal file
386
dev/scripts/build_skin_subs_analytics.py
Normal file
@@ -0,0 +1,386 @@
|
||||
"""Build skin substitutes analytical tables in DuckDB.
|
||||
|
||||
Creates derived tables for ASP pricing trajectories (#239),
|
||||
manufacturer/product market segmentation (#238), and utilization
|
||||
metrics from synthetic claims (#234).
|
||||
|
||||
Usage:
|
||||
uv run python dev/scripts/build_skin_subs_analytics.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import duckdb
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
DUCKDB_PATH = ROOT / "data" / "aco.duckdb"
|
||||
|
||||
FLAT_RATE_2026 = 127.28 # CMS Jan 2026 reclassification: $/cm²
|
||||
|
||||
|
||||
def build_asp_trajectories(con: duckdb.DuckDBPyConnection) -> None:
|
||||
"""#239 — Product-level ASP pricing time series with analytics."""
|
||||
print("\n=== ASP Pricing Trajectories (#239) ===")
|
||||
|
||||
con.execute("DROP TABLE IF EXISTS skin_subs.asp_trajectories")
|
||||
con.execute(f"""
|
||||
CREATE TABLE skin_subs.asp_trajectories AS
|
||||
WITH quarterly AS (
|
||||
SELECT
|
||||
a.quarter,
|
||||
a.hcpcs_code,
|
||||
a.short_description,
|
||||
h.product_name,
|
||||
h.manufacturer,
|
||||
h.category,
|
||||
a.asp_per_unit,
|
||||
a.payment_limit,
|
||||
-- Quarter as sortable integer for window functions
|
||||
(CAST(split_part(a.quarter, '-Q', 1) AS INTEGER) * 4
|
||||
+ CAST(split_part(a.quarter, '-Q', 2) AS INTEGER)) AS q_ord
|
||||
FROM skin_subs.asp_quarterly a
|
||||
LEFT JOIN skin_subs.hcpcs_universe h
|
||||
ON a.hcpcs_code = h.hcpcs_code
|
||||
),
|
||||
with_lag AS (
|
||||
SELECT *,
|
||||
LAG(asp_per_unit) OVER (
|
||||
PARTITION BY hcpcs_code ORDER BY q_ord
|
||||
) AS prev_asp,
|
||||
FIRST_VALUE(asp_per_unit) OVER (
|
||||
PARTITION BY hcpcs_code ORDER BY q_ord
|
||||
) AS first_asp,
|
||||
FIRST_VALUE(quarter) OVER (
|
||||
PARTITION BY hcpcs_code ORDER BY q_ord
|
||||
) AS entry_quarter,
|
||||
COUNT(*) OVER (
|
||||
PARTITION BY hcpcs_code
|
||||
) AS total_quarters
|
||||
FROM quarterly
|
||||
)
|
||||
SELECT
|
||||
quarter,
|
||||
hcpcs_code,
|
||||
short_description,
|
||||
product_name,
|
||||
manufacturer,
|
||||
category,
|
||||
asp_per_unit,
|
||||
payment_limit,
|
||||
entry_quarter,
|
||||
total_quarters,
|
||||
-- Quarter-over-quarter change
|
||||
CASE WHEN prev_asp > 0
|
||||
THEN round((asp_per_unit - prev_asp) / prev_asp * 100, 2)
|
||||
ELSE NULL END AS qoq_pct_change,
|
||||
-- Cumulative change from entry
|
||||
CASE WHEN first_asp > 0
|
||||
THEN round((asp_per_unit - first_asp) / first_asp * 100, 2)
|
||||
ELSE NULL END AS cumulative_pct_change,
|
||||
-- Jan 2026 flat-rate impact: difference from $127.28/cm²
|
||||
round(payment_limit - {FLAT_RATE_2026}, 2) AS flat_rate_delta,
|
||||
-- Flag: would this product lose or gain under flat rate?
|
||||
CASE WHEN payment_limit > {FLAT_RATE_2026} THEN 'loses'
|
||||
WHEN payment_limit < {FLAT_RATE_2026} THEN 'gains'
|
||||
ELSE 'neutral' END AS flat_rate_impact,
|
||||
-- Anomaly flag: >50% QoQ spike
|
||||
CASE WHEN prev_asp > 0
|
||||
AND abs(asp_per_unit - prev_asp) / prev_asp > 0.5
|
||||
THEN true ELSE false END AS anomalous_spike
|
||||
FROM with_lag
|
||||
ORDER BY hcpcs_code, quarter
|
||||
""")
|
||||
|
||||
count = con.execute(
|
||||
"SELECT count(*) FROM skin_subs.asp_trajectories"
|
||||
).fetchone()[0]
|
||||
print(f" Rows: {count}")
|
||||
|
||||
# Summary stats
|
||||
for label, q in [
|
||||
("Products with anomalous spikes",
|
||||
"SELECT count(DISTINCT hcpcs_code) FROM skin_subs.asp_trajectories WHERE anomalous_spike"),
|
||||
("Products that lose under flat rate (latest quarter)",
|
||||
"""SELECT count(DISTINCT hcpcs_code) FROM skin_subs.asp_trajectories
|
||||
WHERE flat_rate_impact = 'loses'
|
||||
AND quarter = (SELECT max(quarter) FROM skin_subs.asp_trajectories)"""),
|
||||
("Products that gain under flat rate (latest quarter)",
|
||||
"""SELECT count(DISTINCT hcpcs_code) FROM skin_subs.asp_trajectories
|
||||
WHERE flat_rate_impact = 'gains'
|
||||
AND quarter = (SELECT max(quarter) FROM skin_subs.asp_trajectories)"""),
|
||||
]:
|
||||
val = con.execute(q).fetchone()[0]
|
||||
print(f" {label}: {val}")
|
||||
|
||||
# Top losers under flat rate
|
||||
print("\n Top 10 losers under Jan 2026 flat rate (latest quarter):")
|
||||
for r in con.execute("""
|
||||
SELECT hcpcs_code, product_name, manufacturer,
|
||||
round(payment_limit, 2) as pay, round(flat_rate_delta, 2) as delta
|
||||
FROM skin_subs.asp_trajectories
|
||||
WHERE quarter = (SELECT max(quarter) FROM skin_subs.asp_trajectories)
|
||||
ORDER BY flat_rate_delta DESC LIMIT 10
|
||||
""").fetchall():
|
||||
print(f" {r[0]} {(r[1] or ''):30s} {(r[2] or ''):20s} pay=${r[3]:>8} delta=${r[4]:>+8}")
|
||||
|
||||
|
||||
def build_market_segmentation(con: duckdb.DuckDBPyConnection) -> None:
|
||||
"""#238 — Manufacturer/product market segmentation."""
|
||||
print("\n=== Market Segmentation (#238) ===")
|
||||
|
||||
# Products enriched with market data
|
||||
con.execute("DROP TABLE IF EXISTS skin_subs.products_enriched")
|
||||
con.execute(f"""
|
||||
CREATE TABLE skin_subs.products_enriched AS
|
||||
WITH latest_asp AS (
|
||||
SELECT DISTINCT ON (hcpcs_code)
|
||||
hcpcs_code, asp_per_unit, payment_limit, quarter AS latest_quarter
|
||||
FROM skin_subs.asp_quarterly
|
||||
ORDER BY hcpcs_code, quarter DESC
|
||||
),
|
||||
asp_stats AS (
|
||||
SELECT
|
||||
hcpcs_code,
|
||||
count(DISTINCT quarter) AS quarters_on_market,
|
||||
min(quarter) AS first_quarter,
|
||||
max(quarter) AS last_quarter,
|
||||
round(avg(asp_per_unit), 2) AS avg_asp,
|
||||
round(min(asp_per_unit), 2) AS min_asp,
|
||||
round(max(asp_per_unit), 2) AS max_asp,
|
||||
round(max(asp_per_unit) - min(asp_per_unit), 2) AS asp_range
|
||||
FROM skin_subs.asp_quarterly
|
||||
GROUP BY hcpcs_code
|
||||
),
|
||||
claims_stats AS (
|
||||
SELECT
|
||||
hcpcs_code,
|
||||
count(*) AS claim_lines,
|
||||
round(sum(paid_amount), 2) AS total_paid,
|
||||
round(avg(paid_amount), 2) AS avg_paid_per_line,
|
||||
count(DISTINCT rendering_npi) AS unique_providers,
|
||||
count(DISTINCT person_id) AS unique_benes
|
||||
FROM skin_subs.claims_synthetic
|
||||
WHERE claim_type = 'product'
|
||||
GROUP BY hcpcs_code
|
||||
)
|
||||
SELECT
|
||||
h.hcpcs_code,
|
||||
h.product_name,
|
||||
h.manufacturer,
|
||||
h.category,
|
||||
h.fda_pathway,
|
||||
h.effective_date,
|
||||
h.termination_date,
|
||||
h.status,
|
||||
-- ASP pricing
|
||||
la.asp_per_unit AS latest_asp,
|
||||
la.payment_limit AS latest_payment_limit,
|
||||
la.latest_quarter,
|
||||
s.quarters_on_market,
|
||||
s.first_quarter,
|
||||
s.last_quarter,
|
||||
s.avg_asp,
|
||||
s.min_asp,
|
||||
s.max_asp,
|
||||
s.asp_range,
|
||||
-- Flat rate impact
|
||||
CASE WHEN la.payment_limit IS NOT NULL
|
||||
THEN round(la.payment_limit - {FLAT_RATE_2026}, 2)
|
||||
ELSE NULL END AS flat_rate_delta,
|
||||
-- Claims utilization
|
||||
cs.claim_lines,
|
||||
cs.total_paid,
|
||||
cs.avg_paid_per_line,
|
||||
cs.unique_providers,
|
||||
cs.unique_benes
|
||||
FROM skin_subs.hcpcs_universe h
|
||||
LEFT JOIN latest_asp la ON h.hcpcs_code = la.hcpcs_code
|
||||
LEFT JOIN asp_stats s ON h.hcpcs_code = s.hcpcs_code
|
||||
LEFT JOIN claims_stats cs ON h.hcpcs_code = cs.hcpcs_code
|
||||
ORDER BY cs.total_paid DESC NULLS LAST
|
||||
""")
|
||||
|
||||
count = con.execute(
|
||||
"SELECT count(*) FROM skin_subs.products_enriched"
|
||||
).fetchone()[0]
|
||||
print(f" Products: {count}")
|
||||
|
||||
# Manufacturer summary
|
||||
con.execute("DROP TABLE IF EXISTS skin_subs.manufacturers")
|
||||
con.execute("""
|
||||
CREATE TABLE skin_subs.manufacturers AS
|
||||
SELECT
|
||||
manufacturer,
|
||||
count(*) AS product_count,
|
||||
count(*) FILTER (WHERE status = 'Active') AS active_products,
|
||||
count(*) FILTER (WHERE claim_lines IS NOT NULL) AS products_with_claims,
|
||||
round(sum(total_paid), 2) AS total_revenue,
|
||||
round(avg(latest_asp), 2) AS avg_asp,
|
||||
string_agg(DISTINCT category, '; ' ORDER BY category) AS categories,
|
||||
min(first_quarter) AS earliest_entry,
|
||||
max(latest_quarter) AS latest_data
|
||||
FROM skin_subs.products_enriched
|
||||
WHERE manufacturer IS NOT NULL AND manufacturer != ''
|
||||
GROUP BY manufacturer
|
||||
ORDER BY total_revenue DESC NULLS LAST
|
||||
""")
|
||||
|
||||
print("\n Top manufacturers by revenue (synthetic claims):")
|
||||
for r in con.execute("""
|
||||
SELECT manufacturer, product_count, active_products,
|
||||
total_revenue, avg_asp
|
||||
FROM skin_subs.manufacturers
|
||||
WHERE total_revenue IS NOT NULL
|
||||
ORDER BY total_revenue DESC LIMIT 10
|
||||
""").fetchall():
|
||||
rev = f"${r[3]:,.0f}" if r[3] else "n/a"
|
||||
print(f" {(r[0] or ''):30s} prods={r[1]:3d} active={r[2]:3d} rev={rev:>12} avg_asp=${r[4] or 0:>8}")
|
||||
|
||||
print("\n Products by category:")
|
||||
for r in con.execute("""
|
||||
SELECT category, count(*) as n,
|
||||
count(*) FILTER (WHERE claim_lines IS NOT NULL) as with_claims,
|
||||
round(avg(latest_asp), 2) as avg_asp
|
||||
FROM skin_subs.products_enriched
|
||||
WHERE category IS NOT NULL
|
||||
GROUP BY category ORDER BY n DESC
|
||||
""").fetchall():
|
||||
print(f" {(r[0] or ''):40s} n={r[1]:3d} claims={r[2]:3d} avg_asp=${r[3] or 0:>8}")
|
||||
|
||||
|
||||
def build_utilization_metrics(con: duckdb.DuckDBPyConnection) -> None:
|
||||
"""#234 — Utilization metrics from synthetic claims."""
|
||||
print("\n=== Utilization Metrics (#234) ===")
|
||||
|
||||
# Provider-level aggregation
|
||||
con.execute("DROP TABLE IF EXISTS skin_subs.providers")
|
||||
con.execute("""
|
||||
CREATE TABLE skin_subs.providers AS
|
||||
SELECT
|
||||
rendering_npi,
|
||||
provider_specialty,
|
||||
state,
|
||||
count(*) AS total_claim_lines,
|
||||
count(*) FILTER (WHERE claim_type = 'product') AS product_lines,
|
||||
count(*) FILTER (WHERE claim_type = 'application') AS application_lines,
|
||||
count(DISTINCT person_id) AS unique_patients,
|
||||
count(DISTINCT service_date) AS service_days,
|
||||
round(sum(paid_amount), 2) AS total_paid,
|
||||
round(avg(paid_amount), 2) AS avg_paid_per_line,
|
||||
round(sum(paid_amount) / nullif(count(DISTINCT person_id), 0), 2)
|
||||
AS paid_per_patient,
|
||||
count(DISTINCT hcpcs_code) AS unique_products,
|
||||
min(service_date) AS first_service,
|
||||
max(service_date) AS last_service
|
||||
FROM skin_subs.claims_synthetic
|
||||
GROUP BY rendering_npi, provider_specialty, state
|
||||
ORDER BY total_paid DESC
|
||||
""")
|
||||
|
||||
# Beneficiary-level aggregation
|
||||
con.execute("DROP TABLE IF EXISTS skin_subs.beneficiaries")
|
||||
con.execute("""
|
||||
CREATE TABLE skin_subs.beneficiaries AS
|
||||
SELECT
|
||||
person_id,
|
||||
patient_age,
|
||||
patient_gender,
|
||||
state,
|
||||
count(*) AS total_claim_lines,
|
||||
count(DISTINCT rendering_npi) AS unique_providers,
|
||||
count(DISTINCT service_date) AS service_dates,
|
||||
count(DISTINCT hcpcs_code) AS unique_products,
|
||||
round(sum(paid_amount), 2) AS total_paid,
|
||||
min(service_date) AS first_service,
|
||||
max(service_date) AS last_service,
|
||||
-- Days between first and last service
|
||||
julian(max(service_date)) - julian(min(service_date))
|
||||
AS treatment_span_days
|
||||
FROM skin_subs.claims_synthetic
|
||||
GROUP BY person_id, patient_age, patient_gender, state
|
||||
ORDER BY total_paid DESC
|
||||
""")
|
||||
|
||||
# Utilization summary by dimensions
|
||||
con.execute("DROP TABLE IF EXISTS skin_subs.utilization_summary")
|
||||
con.execute("""
|
||||
CREATE TABLE skin_subs.utilization_summary AS
|
||||
SELECT
|
||||
provider_specialty,
|
||||
place_of_service_description AS setting,
|
||||
state,
|
||||
count(*) AS claim_lines,
|
||||
count(DISTINCT person_id) AS unique_benes,
|
||||
count(DISTINCT rendering_npi) AS unique_providers,
|
||||
round(sum(paid_amount), 2) AS total_paid,
|
||||
round(avg(paid_amount), 2) AS avg_paid,
|
||||
round(avg(units), 1) AS avg_units,
|
||||
round(sum(paid_amount) / nullif(count(DISTINCT person_id), 0), 2)
|
||||
AS paid_per_bene
|
||||
FROM skin_subs.claims_synthetic
|
||||
WHERE claim_type = 'product'
|
||||
GROUP BY provider_specialty, place_of_service_description, state
|
||||
ORDER BY total_paid DESC
|
||||
""")
|
||||
|
||||
# Print summaries
|
||||
prov_count = con.execute("SELECT count(*) FROM skin_subs.providers").fetchone()[0]
|
||||
bene_count = con.execute("SELECT count(*) FROM skin_subs.beneficiaries").fetchone()[0]
|
||||
util_count = con.execute("SELECT count(*) FROM skin_subs.utilization_summary").fetchone()[0]
|
||||
print(f" Providers: {prov_count}")
|
||||
print(f" Beneficiaries: {bene_count}")
|
||||
print(f" Utilization summary rows: {util_count}")
|
||||
|
||||
print("\n Top providers by total paid:")
|
||||
for r in con.execute("""
|
||||
SELECT rendering_npi, provider_specialty, state,
|
||||
unique_patients, total_paid, paid_per_patient
|
||||
FROM skin_subs.providers ORDER BY total_paid DESC LIMIT 10
|
||||
""").fetchall():
|
||||
print(f" NPI {r[0]} {r[1]:15s} {r[2]} pts={r[3]:3d} paid=${r[4]:>10,.2f} per_pt=${r[5]:>8,.2f}")
|
||||
|
||||
print("\n Utilization by setting:")
|
||||
for r in con.execute("""
|
||||
SELECT setting, sum(claim_lines) as lines, sum(total_paid) as paid,
|
||||
round(avg(paid_per_bene), 2) as avg_per_bene
|
||||
FROM skin_subs.utilization_summary
|
||||
GROUP BY setting ORDER BY paid DESC
|
||||
""").fetchall():
|
||||
print(f" {r[0]:15s} lines={r[1]:5d} paid=${r[2]:>12,.2f} per_bene=${r[3]:>8,.2f}")
|
||||
|
||||
print("\n Utilization by specialty:")
|
||||
for r in con.execute("""
|
||||
SELECT provider_specialty, sum(claim_lines) as lines,
|
||||
sum(total_paid) as paid, sum(unique_benes) as benes
|
||||
FROM skin_subs.utilization_summary
|
||||
GROUP BY provider_specialty ORDER BY paid DESC
|
||||
""").fetchall():
|
||||
print(f" {r[0]:20s} lines={r[1]:5d} paid=${r[2]:>12,.2f} benes={r[3]:5d}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
print("Building skin substitutes analytical tables ...")
|
||||
con = duckdb.connect(str(DUCKDB_PATH))
|
||||
con.execute("CREATE SCHEMA IF NOT EXISTS skin_subs")
|
||||
|
||||
build_asp_trajectories(con)
|
||||
build_market_segmentation(con)
|
||||
build_utilization_metrics(con)
|
||||
|
||||
# Final table inventory
|
||||
print("\n=== All skin_subs tables ===")
|
||||
for r in con.execute("""
|
||||
SELECT table_name FROM information_schema.tables
|
||||
WHERE table_schema = 'skin_subs' ORDER BY table_name
|
||||
""").fetchall():
|
||||
cnt = con.execute(f"SELECT count(*) FROM skin_subs.{r[0]}").fetchone()[0]
|
||||
print(f" skin_subs.{r[0]:30s}: {cnt:>6} rows")
|
||||
|
||||
con.close()
|
||||
print("\nDone.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user