92ae58a2ec
pfs.event_study: pre/post windows around an exposure year (the effective date is 1 January, so the exposure year is the first treated year), log ratio of window means, difference in log changes against the never-treated control set from pfs.exposure.control_codes (same RVU status and decile band, no exposure within ±2 years), 95% percentile-bootstrap interval over the controls with a fixed seed. A code with no pre-period of its own (a new code or a successor) is measured through its family's summed series (scope="family"), so G2058 → 99439 is one continuous line. pre_post() gives single-series window means for outcomes without a code-level control set (occupation wages, employment). utilization_panel() ties an exposure row, its pfs.utilization series, its controls and the estimate. notebooks/event_study.py: pick one of nine fixture exposures, see the treated services series with the exposure rule, the indexed control band (10th–90th pct), the contrast table with its interval and FR anchor, clinical-staff wages and health-care employment around the same year, and a table of every fixture contrast with anchors. A 2025 exposure has no post-period until the 2025 PUF ships — stated, never filled. Descriptive contrasts only; the notebook says so and the chat only links here. Market/industry series are deferred: no keyless quote source is reachable from the rack.
264 lines
8.7 KiB
Python
264 lines
8.7 KiB
Python
"""pfs.event_study — pre/post windows around exposures with never-treated controls (#696)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import math
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from pfs.codetables import ExposureRow, ensure_tables, write_exposures
|
|
from pfs.event_study import (
|
|
Effect,
|
|
PrePost,
|
|
did_effect,
|
|
log_change,
|
|
pre_post,
|
|
utilization_panel,
|
|
windows,
|
|
)
|
|
from pfs.utilization import UtilizationRow
|
|
from pfs.utilization import write as write_util
|
|
|
|
|
|
class TestWindows:
|
|
def test_pre_before_post_from_exposure_year(self):
|
|
pre, post = windows(2017, available=range(2013, 2025))
|
|
assert pre == (2015, 2016) and post == (2017, 2018)
|
|
|
|
def test_truncated_at_the_data_edge(self):
|
|
pre, post = windows(2024, available=range(2013, 2025), pre=3, post=3)
|
|
assert pre == (2021, 2022, 2023) and post == (2024,)
|
|
assert windows(2013, available=range(2013, 2025)) == ((), (2013, 2014))
|
|
|
|
|
|
class TestLogChange:
|
|
def test_ratio_of_means(self):
|
|
s = {2015: 100.0, 2016: 100.0, 2017: 200.0, 2018: 200.0}
|
|
assert log_change(s, (2015, 2016), (2017, 2018)) == pytest.approx(math.log(2))
|
|
|
|
def test_missing_side_is_none(self):
|
|
assert log_change({2017: 1.0}, (2015, 2016), (2017,)) is None
|
|
assert log_change({2015: 1.0}, (2015,), (2017,)) is None
|
|
assert (
|
|
log_change({2015: 0.0, 2017: 1.0}, (2015,), (2017,)) is None
|
|
) # zero pre → undefined
|
|
|
|
|
|
class TestDidEffect:
|
|
def _controls(self, n=20, growth=0.1):
|
|
return {
|
|
f"C{i:04d}": {
|
|
2015: 100.0,
|
|
2016: 100.0,
|
|
2017: 100.0 * (1 + growth + 0.01 * (i % 5)),
|
|
2018: 100.0 * (1 + growth + 0.01 * (i % 5)),
|
|
}
|
|
for i in range(n)
|
|
}
|
|
|
|
def test_effect_is_treated_change_minus_control_change(self):
|
|
treated = {2015: 100.0, 2016: 100.0, 2017: 300.0, 2018: 300.0}
|
|
e = did_effect(
|
|
treated,
|
|
self._controls(),
|
|
2017,
|
|
code="X",
|
|
kind="becomes-payable",
|
|
outcome="services",
|
|
)
|
|
assert isinstance(e, Effect)
|
|
assert e.pre_years == (2015, 2016) and e.post_years == (2017, 2018)
|
|
assert e.treated_change == pytest.approx(math.log(3))
|
|
assert 0.09 < e.control_change < 0.14
|
|
assert e.effect == pytest.approx(e.treated_change - e.control_change)
|
|
assert e.ci_low is not None and e.ci_low < e.effect < e.ci_high
|
|
assert e.n_controls == 20
|
|
assert e.effect_pct == pytest.approx(math.exp(e.effect) - 1)
|
|
|
|
def test_ci_is_deterministic_and_shrinks_with_more_controls(self):
|
|
treated = {2015: 100.0, 2016: 100.0, 2017: 150.0, 2018: 150.0}
|
|
a = did_effect(
|
|
treated, self._controls(10), 2017, code="X", kind="k", outcome="o"
|
|
)
|
|
b = did_effect(
|
|
treated, self._controls(10), 2017, code="X", kind="k", outcome="o"
|
|
)
|
|
c = did_effect(
|
|
treated, self._controls(200), 2017, code="X", kind="k", outcome="o"
|
|
)
|
|
assert (a.ci_low, a.ci_high) == (b.ci_low, b.ci_high)
|
|
assert (c.ci_high - c.ci_low) < (a.ci_high - a.ci_low)
|
|
|
|
def test_too_few_controls_gives_no_ci(self):
|
|
treated = {2015: 1.0, 2016: 1.0, 2017: 2.0}
|
|
e = did_effect(
|
|
treated,
|
|
{"C1": {2015: 1.0, 2017: 1.1}},
|
|
2017,
|
|
code="X",
|
|
kind="k",
|
|
outcome="o",
|
|
)
|
|
assert e.n_controls == 1 and e.ci_low is None and e.effect is not None
|
|
e0 = did_effect(treated, {}, 2017, code="X", kind="k", outcome="o")
|
|
assert e0.n_controls == 0 and e0.control_change is None and e0.effect is None
|
|
|
|
def test_treated_without_post_data(self):
|
|
e = did_effect(
|
|
{2015: 1.0, 2016: 1.0},
|
|
self._controls(),
|
|
2017,
|
|
code="X",
|
|
kind="k",
|
|
outcome="o",
|
|
)
|
|
assert e.treated_change is None and e.effect is None and e.n_controls == 20
|
|
|
|
|
|
class TestPrePost:
|
|
def test_means_and_change(self):
|
|
p = pre_post({2015: 10.0, 2016: 12.0, 2017: 20.0, 2018: 24.0}, 2017)
|
|
assert isinstance(p, PrePost)
|
|
assert p.pre_mean == 11.0 and p.post_mean == 22.0
|
|
assert p.change == pytest.approx(math.log(2))
|
|
|
|
def test_empty(self):
|
|
p = pre_post({}, 2017)
|
|
assert p.pre_mean is None and p.change is None
|
|
|
|
|
|
RVU_COLS = "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER"
|
|
|
|
|
|
def _util(code, year, services):
|
|
return UtilizationRow(
|
|
year,
|
|
code,
|
|
"d",
|
|
"N",
|
|
"O",
|
|
10,
|
|
100,
|
|
float(services),
|
|
float(services),
|
|
1.0,
|
|
2.0,
|
|
3.0,
|
|
4.0,
|
|
"ds",
|
|
"K",
|
|
)
|
|
|
|
|
|
def _exp(code, year, kind, status="A", band=1, item_key="ITEM", p_id=5):
|
|
return ExposureRow(
|
|
code,
|
|
"FAM",
|
|
dt.date(year, 1, 1),
|
|
year,
|
|
kind,
|
|
status,
|
|
"",
|
|
band,
|
|
1.0,
|
|
0.0,
|
|
item_key,
|
|
p_id,
|
|
9,
|
|
"created",
|
|
False,
|
|
"",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def con():
|
|
c = duckdb.connect(":memory:")
|
|
ensure_tables(c)
|
|
c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
|
|
rows = []
|
|
# treated code T appears payable 2017; controls C1..C6 payable across the span in the same band
|
|
for y in range(2015, 2021):
|
|
for i in range(1, 7):
|
|
rows.append((f"C000{i}", None, "c", "A", 1.0, y))
|
|
if y >= 2017:
|
|
rows.append(("T0001", None, "t", "A", 1.0, y))
|
|
c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", rows)
|
|
write_exposures(c, [_exp("T0001", 2017, "becomes-payable")])
|
|
util = []
|
|
for y in range(2015, 2021):
|
|
for i in range(1, 7):
|
|
util.append(_util(f"C000{i}", y, 100 * (1 + 0.05 * (y - 2015))))
|
|
if y >= 2017:
|
|
util.append(_util("T0001", y, 500 + 100 * (y - 2017)))
|
|
for y in range(2015, 2021):
|
|
write_util(c, y, [r for r in util if r.year == y])
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
class TestUtilizationPanel:
|
|
def test_panel_uses_control_set_and_carries_anchor(self, con):
|
|
panel = utilization_panel(con, "T0001", 2017, "becomes-payable")
|
|
assert panel.effect.code == "T0001" and panel.effect.outcome == "n_services"
|
|
assert panel.effect.n_controls == 6
|
|
assert (
|
|
panel.exposure is not None
|
|
and panel.exposure.item_key == "ITEM"
|
|
and panel.exposure.p_id == 5
|
|
)
|
|
# treated has no pre-period utilization (it did not exist) → change undefined, stated not faked
|
|
assert panel.effect.treated_change is None and panel.effect.scope == "code"
|
|
assert set(panel.controls) == {f"C000{i}" for i in range(1, 7)}
|
|
assert panel.treated == {2017: 500.0, 2018: 600.0, 2019: 700.0, 2020: 800.0}
|
|
|
|
def test_panel_with_pre_period(self, con):
|
|
write_exposures(
|
|
con,
|
|
[_exp("T0001", 2017, "becomes-payable"), _exp("C0001", 2018, "revalued")],
|
|
)
|
|
panel = utilization_panel(con, "C0001", 2018, "revalued")
|
|
assert panel.effect.treated_change is not None
|
|
assert panel.effect.n_controls == 5 # the other controls, not itself
|
|
assert panel.effect.effect == pytest.approx(
|
|
0.0, abs=1e-9
|
|
) # it grows like its controls
|
|
|
|
def test_new_code_falls_back_to_the_family_series(self, con):
|
|
# T0001 has no pre-period of its own; its family (FAM) also holds P0001,
|
|
# billed before 2017 — the family total stands in as the treated series.
|
|
con.executemany(
|
|
"INSERT INTO pfs.code_family VALUES (?,?,?,?,?,?,?,?,?)",
|
|
[
|
|
("FAM", "Fam", "T0001", "successor", 2017, None, "", 0, ""),
|
|
("FAM", "Fam", "P0001", "predecessor", 2015, 2016, "", 0, ""),
|
|
],
|
|
)
|
|
for y in (2015, 2016):
|
|
write_util(
|
|
con,
|
|
y,
|
|
[
|
|
_util(f"C000{i}", y, 100 * (1 + 0.05 * (y - 2015)))
|
|
for i in range(1, 7)
|
|
]
|
|
+ [_util("P0001", y, 400.0)],
|
|
)
|
|
panel = utilization_panel(con, "T0001", 2017, "becomes-payable")
|
|
assert panel.effect.scope == "family"
|
|
assert panel.treated[2015] == 400.0 and panel.treated[2017] == 500.0
|
|
assert (
|
|
panel.effect.treated_change is not None and panel.effect.effect is not None
|
|
)
|
|
assert panel.effect.n_controls == 6
|
|
|
|
def test_unknown_exposure_still_returns_a_panel(self, con):
|
|
panel = utilization_panel(con, "ZZZZZ", 2018, "ends")
|
|
assert (
|
|
panel.exposure is None
|
|
and panel.effect.n_controls == 0
|
|
and panel.treated == {}
|
|
)
|