1497 lines
60 KiB
Python
1497 lines
60 KiB
Python
"""Tests for aco.express.readmissions — readmission pipeline functions.
|
|
|
|
Tests verify:
|
|
- Return type is a DataFrame
|
|
- Expected columns are present
|
|
- Filters work correctly (acute inpatient, discharge codes, etc.)
|
|
- Joins produce correct results
|
|
- Edge cases (empty inputs, missing data) are handled
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import polars as pl
|
|
import pytest
|
|
|
|
from aco.express.readmissions import (
|
|
int_encounter,
|
|
int_encounter_data_quality,
|
|
int_encounter_overlap,
|
|
int_encounter_specialty_cohort,
|
|
int_encounter_with_ccs,
|
|
int_exclusion,
|
|
int_index_admission,
|
|
int_index_discharge_requirement,
|
|
int_index_time_requirement,
|
|
int_planned_encounter,
|
|
int_procedure_ccs,
|
|
int_readmission_crude,
|
|
)
|
|
|
|
# ── fixtures ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def core_encounter_df() -> pl.DataFrame:
|
|
"""Encounters with mixed types for readmissions filtering."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": ["ENC001", "ENC002", "ENC003"],
|
|
"person_id": ["P001", "P001", "P002"],
|
|
"patient_id": ["PT001", "PT001", "PT002"],
|
|
"encounter_type": ["acute inpatient", "ed", "acute inpatient"],
|
|
"encounter_start_date": [
|
|
date(2024, 1, 15),
|
|
date(2024, 2, 1),
|
|
date(2024, 3, 10),
|
|
],
|
|
"encounter_end_date": [
|
|
date(2024, 1, 20),
|
|
date(2024, 2, 1),
|
|
date(2024, 3, 15),
|
|
],
|
|
"length_of_stay": [5, 0, 5],
|
|
"admit_source_code": [None, None, None],
|
|
"admit_source_description": [None, None, None],
|
|
"admit_type_code": [None, None, None],
|
|
"admit_type_description": [None, None, None],
|
|
"discharge_disposition_code": ["01", "01", "01"],
|
|
"discharge_disposition_description": ["home", "home", "home"],
|
|
"attending_provider_id": [None, None, None],
|
|
"attending_provider_name": [None, None, None],
|
|
"facility_id": ["FAC001", "FAC001", "FAC002"],
|
|
"facility_name": [None, None, None],
|
|
"primary_diagnosis_code_type": [
|
|
"icd-10-cm",
|
|
"icd-10-cm",
|
|
"icd-10-cm",
|
|
],
|
|
"primary_diagnosis_code": ["I21.0", "R10.9", "J18.9"],
|
|
"primary_diagnosis_description": [None, None, None],
|
|
"drg_code_type": ["ms-drg", None, "ms-drg"],
|
|
"drg_code": ["280", None, "193"],
|
|
"drg_description": [None, None, None],
|
|
"paid_amount": [15000.0, 500.0, 12000.0],
|
|
"allowed_amount": [18000.0, 600.0, 14000.0],
|
|
"charge_amount": [30000.0, 1000.0, 25000.0],
|
|
"encounter_source_type": ["claim", "clinical", "claim"],
|
|
"data_source": ["test", "test", "test"],
|
|
"file_name": ["t.csv", "t.csv", "t.csv"],
|
|
"ingest_datetime": [None, None, None],
|
|
"encounter_group": ["claims", "clinical", "claims"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def readmission_encounter_df() -> pl.DataFrame:
|
|
"""Readmissions int_encounter output shape for downstream tests."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": ["ENC001", "ENC003"],
|
|
"person_id": ["P001", "P002"],
|
|
"admit_date": [date(2024, 1, 15), date(2024, 3, 10)],
|
|
"discharge_date": [date(2024, 1, 20), date(2024, 3, 15)],
|
|
"discharge_disposition_code": ["01", "01"],
|
|
"facility_id": ["FAC001", "FAC002"],
|
|
"drg_code_type": ["ms-drg", "ms-drg"],
|
|
"drg_code": ["280", "193"],
|
|
"paid_amount": [15000.0, 12000.0],
|
|
"primary_diagnosis_code": ["I21.0", "J18.9"],
|
|
"encounter_source_type": ["claim", "claim"],
|
|
"data_source": ["test", "test"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def icd_10_cm_df() -> pl.DataFrame:
|
|
"""ICD-10-CM terminology lookup."""
|
|
return pl.DataFrame(
|
|
{
|
|
"icd_10_cm": ["I21.0", "J18.9", "E11.9"],
|
|
"description": ["AMI", "Pneumonia", "Diabetes"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def icd_10_cm_to_ccs_df() -> pl.DataFrame:
|
|
"""ICD-10-CM to CCS category mapping."""
|
|
return pl.DataFrame(
|
|
{
|
|
"icd_10_cm": ["I21.0", "J18.9", "E11.9"],
|
|
"ccs_diagnosis_category": ["100", "122", "49"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def encounter_with_ccs_df() -> pl.DataFrame:
|
|
"""Encounters enriched with CCS codes."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": ["ENC001", "ENC003"],
|
|
"person_id": ["P001", "P002"],
|
|
"admit_date": [date(2024, 1, 15), date(2024, 3, 10)],
|
|
"discharge_date": [date(2024, 1, 20), date(2024, 3, 15)],
|
|
"discharge_disposition_code": ["01", "01"],
|
|
"facility_id": ["FAC001", "FAC002"],
|
|
"drg_code_type": ["ms-drg", "ms-drg"],
|
|
"drg_code": ["280", "193"],
|
|
"paid_amount": [15000.0, 12000.0],
|
|
"primary_diagnosis_code": ["I21.0", "J18.9"],
|
|
"valid_primary_diagnosis_code_flag": [1, 1],
|
|
"ccs_diagnosis_category": ["100", "122"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def icd_10_pcs_df() -> pl.DataFrame:
|
|
"""ICD-10-PCS terminology lookup."""
|
|
return pl.DataFrame(
|
|
{
|
|
"icd_10_pcs": ["02703ZZ", "0W110J4"],
|
|
"description": ["Dilation coronary artery", "Bypass"],
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def icd_10_pcs_to_ccs_df() -> pl.DataFrame:
|
|
"""ICD-10-PCS to CCS procedure category mapping."""
|
|
return pl.DataFrame(
|
|
{
|
|
"icd_10_pcs": ["02703ZZ", "0W110J4"],
|
|
"ccs_procedure_category": ["47", "99"],
|
|
}
|
|
)
|
|
|
|
|
|
# ── int_encounter ────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntEncounter:
|
|
def test_returns_dataframe(self, core_encounter_df) -> None:
|
|
result = int_encounter(core_encounter_df)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_filters_to_acute_inpatient(self, core_encounter_df) -> None:
|
|
result = int_encounter(core_encounter_df)
|
|
assert len(result) == 2
|
|
assert "ENC002" not in result["encounter_id"].to_list()
|
|
|
|
def test_renames_dates(self, core_encounter_df) -> None:
|
|
result = int_encounter(core_encounter_df)
|
|
assert "admit_date" in result.columns
|
|
assert "discharge_date" in result.columns
|
|
assert "encounter_start_date" not in result.columns
|
|
|
|
def test_expected_columns(self, core_encounter_df) -> None:
|
|
result = int_encounter(core_encounter_df)
|
|
expected = [
|
|
"encounter_id",
|
|
"person_id",
|
|
"admit_date",
|
|
"discharge_date",
|
|
"discharge_disposition_code",
|
|
"facility_id",
|
|
"drg_code_type",
|
|
"drg_code",
|
|
"paid_amount",
|
|
"primary_diagnosis_code",
|
|
"encounter_source_type",
|
|
"data_source",
|
|
]
|
|
for col in expected:
|
|
assert col in result.columns, f"Missing column: {col}"
|
|
|
|
def test_preserves_values(self, core_encounter_df) -> None:
|
|
result = int_encounter(core_encounter_df)
|
|
row = result.filter(pl.col("encounter_id") == "ENC001")
|
|
assert row["person_id"][0] == "P001"
|
|
assert row["paid_amount"][0] == pytest.approx(15000.0)
|
|
|
|
def test_empty_input(self) -> None:
|
|
empty = pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series([], dtype=pl.Utf8),
|
|
"person_id": pl.Series([], dtype=pl.Utf8),
|
|
"encounter_type": pl.Series([], dtype=pl.Utf8),
|
|
"encounter_start_date": pl.Series([], dtype=pl.Date),
|
|
"encounter_end_date": pl.Series([], dtype=pl.Date),
|
|
"discharge_disposition_code": pl.Series([], dtype=pl.Utf8),
|
|
"facility_id": pl.Series([], dtype=pl.Utf8),
|
|
"drg_code_type": pl.Series([], dtype=pl.Utf8),
|
|
"drg_code": pl.Series([], dtype=pl.Utf8),
|
|
"paid_amount": pl.Series([], dtype=pl.Float64),
|
|
"primary_diagnosis_code": pl.Series([], dtype=pl.Utf8),
|
|
"encounter_source_type": pl.Series([], dtype=pl.Utf8),
|
|
"data_source": pl.Series([], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
result = int_encounter(empty)
|
|
assert isinstance(result, pl.DataFrame)
|
|
assert len(result) == 0
|
|
|
|
def test_no_acute_inpatient_returns_empty(self) -> None:
|
|
df = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["ENC001"],
|
|
"person_id": ["P001"],
|
|
"encounter_type": ["ed"],
|
|
"encounter_start_date": [date(2024, 1, 15)],
|
|
"encounter_end_date": [date(2024, 1, 15)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": ["FAC001"],
|
|
"drg_code_type": ["ms-drg"],
|
|
"drg_code": ["280"],
|
|
"paid_amount": [1000.0],
|
|
"primary_diagnosis_code": ["I21.0"],
|
|
"encounter_source_type": ["claim"],
|
|
"data_source": ["test"],
|
|
}
|
|
)
|
|
result = int_encounter(df)
|
|
assert len(result) == 0
|
|
|
|
|
|
# ── int_encounter_with_ccs ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntEncounterWithCcs:
|
|
def test_returns_dataframe(
|
|
self, readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
) -> None:
|
|
result = int_encounter_with_ccs(
|
|
readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
)
|
|
assert isinstance(result, pl.DataFrame)
|
|
|
|
def test_adds_ccs_category(
|
|
self, readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
) -> None:
|
|
result = int_encounter_with_ccs(
|
|
readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
)
|
|
assert "ccs_diagnosis_category" in result.columns
|
|
assert "valid_primary_diagnosis_code_flag" in result.columns
|
|
|
|
def test_valid_diagnosis_flag(
|
|
self, readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
) -> None:
|
|
result = int_encounter_with_ccs(
|
|
readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
)
|
|
assert result["valid_primary_diagnosis_code_flag"].to_list() == [1, 1]
|
|
|
|
def test_invalid_diagnosis_flag(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["ENC001"],
|
|
"person_id": ["P001"],
|
|
"admit_date": [date(2024, 1, 15)],
|
|
"discharge_date": [date(2024, 1, 20)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": ["FAC001"],
|
|
"drg_code_type": ["ms-drg"],
|
|
"drg_code": ["280"],
|
|
"paid_amount": [15000.0],
|
|
"primary_diagnosis_code": ["ZZZZZ"],
|
|
"encounter_source_type": ["claim"],
|
|
"data_source": ["test"],
|
|
}
|
|
)
|
|
icd = pl.DataFrame({"icd_10_cm": ["I21.0"], "description": ["AMI"]})
|
|
ccs = pl.DataFrame({"icd_10_cm": ["I21.0"], "ccs_diagnosis_category": ["100"]})
|
|
result = int_encounter_with_ccs(enc, ccs, icd)
|
|
assert result["valid_primary_diagnosis_code_flag"][0] == 0
|
|
|
|
def test_row_count_preserved(
|
|
self, readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
) -> None:
|
|
result = int_encounter_with_ccs(
|
|
readmission_encounter_df, icd_10_cm_to_ccs_df, icd_10_cm_df
|
|
)
|
|
assert len(result) == len(readmission_encounter_df)
|
|
|
|
|
|
# ── int_index_discharge_requirement ──────────────────────────────────────────
|
|
|
|
|
|
class TestIntIndexDischargeRequirement:
|
|
def test_excludes_invalid_codes(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2", "E3", "E4"],
|
|
"person_id": ["P1"] * 4,
|
|
"admit_date": [date(2024, 1, 1)] * 4,
|
|
"discharge_date": [date(2024, 1, 5)] * 4,
|
|
"discharge_disposition_code": ["01", "02", "07", "20"],
|
|
"facility_id": [None] * 4,
|
|
"drg_code_type": [None] * 4,
|
|
"drg_code": [None] * 4,
|
|
"paid_amount": [None] * 4,
|
|
"primary_diagnosis_code": [None] * 4,
|
|
"encounter_source_type": [None] * 4,
|
|
"data_source": ["test"] * 4,
|
|
}
|
|
)
|
|
result = int_index_discharge_requirement(enc)
|
|
assert isinstance(result, pl.DataFrame)
|
|
assert "E1" in result["encounter_id"].to_list()
|
|
assert "E2" not in result["encounter_id"].to_list()
|
|
assert "E7" not in result["encounter_id"].to_list()
|
|
|
|
def test_keeps_valid_codes(self, readmission_encounter_df) -> None:
|
|
result = int_index_discharge_requirement(readmission_encounter_df)
|
|
assert len(result) == 2
|
|
|
|
def test_empty_input(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series([], dtype=pl.Utf8),
|
|
"person_id": pl.Series([], dtype=pl.Utf8),
|
|
"admit_date": pl.Series([], dtype=pl.Date),
|
|
"discharge_date": pl.Series([], dtype=pl.Date),
|
|
"discharge_disposition_code": pl.Series([], dtype=pl.Utf8),
|
|
"facility_id": pl.Series([], dtype=pl.Utf8),
|
|
"drg_code_type": pl.Series([], dtype=pl.Utf8),
|
|
"drg_code": pl.Series([], dtype=pl.Utf8),
|
|
"paid_amount": pl.Series([], dtype=pl.Float64),
|
|
"primary_diagnosis_code": pl.Series([], dtype=pl.Utf8),
|
|
"encounter_source_type": pl.Series([], dtype=pl.Utf8),
|
|
"data_source": pl.Series([], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
result = int_index_discharge_requirement(enc)
|
|
assert len(result) == 0
|
|
|
|
|
|
# ── int_index_time_requirement ───────────────────────────────────────────────
|
|
|
|
|
|
class TestIntIndexTimeRequirement:
|
|
def test_filters_by_time_window(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2", "E3"],
|
|
"person_id": ["P1", "P1", "P2"],
|
|
"admit_date": [
|
|
date(2024, 1, 1),
|
|
date(2024, 6, 1),
|
|
date(2024, 7, 15),
|
|
],
|
|
"discharge_date": [
|
|
date(2024, 1, 5),
|
|
date(2024, 6, 5),
|
|
date(2024, 7, 20),
|
|
],
|
|
"discharge_disposition_code": ["01", "01", "01"],
|
|
"facility_id": [None] * 3,
|
|
"drg_code_type": [None] * 3,
|
|
"drg_code": [None] * 3,
|
|
"paid_amount": [None] * 3,
|
|
"primary_diagnosis_code": [None] * 3,
|
|
"encounter_source_type": [None] * 3,
|
|
"data_source": ["test"] * 3,
|
|
}
|
|
)
|
|
result = int_index_time_requirement(enc)
|
|
# E3 has latest discharge (Jul 20), so cutoff is Jun 20
|
|
# E1 (Jan 5) and E2 (Jun 5) both before cutoff
|
|
assert len(result) >= 1
|
|
|
|
def test_single_encounter_excluded(self) -> None:
|
|
"""Single encounter can't meet time requirement (no 30-day window)."""
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 7, 1)],
|
|
"discharge_date": [date(2024, 7, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": [None],
|
|
"drg_code_type": [None],
|
|
"drg_code": [None],
|
|
"paid_amount": [None],
|
|
"primary_diagnosis_code": [None],
|
|
"encounter_source_type": [None],
|
|
"data_source": ["test"],
|
|
}
|
|
)
|
|
result = int_index_time_requirement(enc)
|
|
assert len(result) == 0
|
|
|
|
|
|
# ── int_exclusion ────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntExclusion:
|
|
def test_identifies_excluded_encounters(self) -> None:
|
|
enc_ccs = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2", "E3"],
|
|
"person_id": ["P1", "P1", "P2"],
|
|
"admit_date": [date(2024, 1, 1)] * 3,
|
|
"discharge_date": [date(2024, 1, 5)] * 3,
|
|
"discharge_disposition_code": ["01", "01", "01"],
|
|
"facility_id": [None] * 3,
|
|
"drg_code_type": [None] * 3,
|
|
"drg_code": [None] * 3,
|
|
"paid_amount": [None] * 3,
|
|
"primary_diagnosis_code": [None] * 3,
|
|
"valid_primary_diagnosis_code_flag": [1, 1, 1],
|
|
"ccs_diagnosis_category": ["100", "122", "254"],
|
|
}
|
|
)
|
|
excl_ccs = pl.DataFrame({"ccs_diagnosis_category": ["254", "255"]})
|
|
result = int_exclusion(enc_ccs, excl_ccs)
|
|
assert "E3" in result["encounter_id"].to_list()
|
|
assert "E1" not in result["encounter_id"].to_list()
|
|
|
|
def test_no_exclusions(self) -> None:
|
|
enc_ccs = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": [None],
|
|
"drg_code_type": [None],
|
|
"drg_code": [None],
|
|
"paid_amount": [None],
|
|
"primary_diagnosis_code": [None],
|
|
"valid_primary_diagnosis_code_flag": [1],
|
|
"ccs_diagnosis_category": ["100"],
|
|
}
|
|
)
|
|
excl_ccs = pl.DataFrame({"ccs_diagnosis_category": ["999"]})
|
|
result = int_exclusion(enc_ccs, excl_ccs)
|
|
assert len(result) == 0
|
|
|
|
def test_null_ccs_not_excluded(self) -> None:
|
|
enc_ccs = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": [None],
|
|
"drg_code_type": [None],
|
|
"drg_code": [None],
|
|
"paid_amount": [None],
|
|
"primary_diagnosis_code": [None],
|
|
"valid_primary_diagnosis_code_flag": [0],
|
|
"ccs_diagnosis_category": pl.Series([None], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
excl_ccs = pl.DataFrame({"ccs_diagnosis_category": ["100"]})
|
|
result = int_exclusion(enc_ccs, excl_ccs)
|
|
assert len(result) == 0
|
|
|
|
|
|
# ── int_index_admission ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntIndexAdmission:
|
|
def test_combines_requirements_and_exclusion(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2", "E3"],
|
|
"person_id": ["P1", "P1", "P2"],
|
|
"admit_date": [date(2024, 1, 1)] * 3,
|
|
"discharge_date": [date(2024, 1, 5)] * 3,
|
|
"discharge_disposition_code": ["01"] * 3,
|
|
"facility_id": [None] * 3,
|
|
"drg_code_type": [None] * 3,
|
|
"drg_code": [None] * 3,
|
|
"paid_amount": [None] * 3,
|
|
"primary_diagnosis_code": [None] * 3,
|
|
"encounter_source_type": [None] * 3,
|
|
"data_source": ["test"] * 3,
|
|
}
|
|
)
|
|
time_req = pl.DataFrame({"encounter_id": ["E1", "E2"]})
|
|
discharge_req = pl.DataFrame({"encounter_id": ["E1", "E3"]})
|
|
exclusion = pl.DataFrame({"encounter_id": ["E1"]})
|
|
|
|
result = int_index_admission(enc, time_req, discharge_req, exclusion)
|
|
# E1: passes time and discharge but is excluded
|
|
# E2: passes time but not discharge
|
|
# E3: passes discharge but not time
|
|
assert len(result) == 0
|
|
|
|
def test_no_exclusion_keeps_qualifying(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": [None],
|
|
"drg_code_type": [None],
|
|
"drg_code": [None],
|
|
"paid_amount": [None],
|
|
"primary_diagnosis_code": [None],
|
|
"encounter_source_type": [None],
|
|
"data_source": ["test"],
|
|
}
|
|
)
|
|
time_req = pl.DataFrame({"encounter_id": ["E1"]})
|
|
discharge_req = pl.DataFrame({"encounter_id": ["E1"]})
|
|
exclusion = pl.DataFrame({"encounter_id": pl.Series([], dtype=pl.Utf8)})
|
|
|
|
result = int_index_admission(enc, time_req, discharge_req, exclusion)
|
|
assert "E1" in result["encounter_id"].to_list()
|
|
|
|
|
|
# ── int_procedure_ccs ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntProcedureCcs:
|
|
def test_filters_to_icd10_pcs(self) -> None:
|
|
procs = pl.DataFrame(
|
|
{
|
|
"procedure_id": ["PR1", "PR2"],
|
|
"person_id": ["P1", "P1"],
|
|
"encounter_id": ["E1", "E1"],
|
|
"normalized_code_type": ["icd-10-pcs", "hcpcs"],
|
|
"normalized_code": ["02703ZZ", "99213"],
|
|
"data_source": ["test", "test"],
|
|
"member_id": [None, None],
|
|
"patient_id": [None, None],
|
|
"claim_id": [None, None],
|
|
"procedure_date": [None, None],
|
|
"source_code_type": [None, None],
|
|
"source_code": [None, None],
|
|
"source_description": [None, None],
|
|
"normalized_description": [None, None],
|
|
"modifier_1": [None, None],
|
|
"modifier_2": [None, None],
|
|
"modifier_3": [None, None],
|
|
"modifier_4": [None, None],
|
|
"modifier_5": [None, None],
|
|
"practitioner_id": [None, None],
|
|
}
|
|
)
|
|
icd_pcs = pl.DataFrame({"icd_10_pcs": ["02703ZZ"], "description": ["Dilation"]})
|
|
ccs_map = pl.DataFrame(
|
|
{"icd_10_pcs": ["02703ZZ"], "ccs_procedure_category": ["47"]}
|
|
)
|
|
result = int_procedure_ccs(procs, ccs_map, icd_pcs)
|
|
assert len(result) == 1
|
|
assert result["procedure_code"][0] == "02703ZZ"
|
|
assert result["valid_icd_10_pcs_flag"][0] == 1
|
|
|
|
def test_expected_columns(self) -> None:
|
|
procs = pl.DataFrame(
|
|
{
|
|
"procedure_id": ["PR1"],
|
|
"person_id": ["P1"],
|
|
"encounter_id": ["E1"],
|
|
"normalized_code_type": ["icd-10-pcs"],
|
|
"normalized_code": ["02703ZZ"],
|
|
"data_source": ["test"],
|
|
"member_id": [None],
|
|
"patient_id": [None],
|
|
"claim_id": [None],
|
|
"procedure_date": [None],
|
|
"source_code_type": [None],
|
|
"source_code": [None],
|
|
"source_description": [None],
|
|
"normalized_description": [None],
|
|
"modifier_1": [None],
|
|
"modifier_2": [None],
|
|
"modifier_3": [None],
|
|
"modifier_4": [None],
|
|
"modifier_5": [None],
|
|
"practitioner_id": [None],
|
|
}
|
|
)
|
|
icd_pcs = pl.DataFrame({"icd_10_pcs": ["02703ZZ"], "description": ["Dilation"]})
|
|
ccs_map = pl.DataFrame(
|
|
{"icd_10_pcs": ["02703ZZ"], "ccs_procedure_category": ["47"]}
|
|
)
|
|
result = int_procedure_ccs(procs, ccs_map, icd_pcs)
|
|
for col in [
|
|
"encounter_id",
|
|
"procedure_code",
|
|
"valid_icd_10_pcs_flag",
|
|
"ccs_procedure_category",
|
|
]:
|
|
assert col in result.columns
|
|
|
|
|
|
# ── int_encounter_overlap ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntEncounterOverlap:
|
|
def test_non_overlapping_encounters(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2"],
|
|
"person_id": ["P1", "P1"],
|
|
"admit_date": [date(2024, 1, 1), date(2024, 3, 1)],
|
|
"discharge_date": [date(2024, 1, 5), date(2024, 3, 5)],
|
|
"discharge_disposition_code": ["01", "01"],
|
|
"facility_id": [None, None],
|
|
"drg_code_type": ["ms-drg", "ms-drg"],
|
|
"drg_code": ["280", "193"],
|
|
"paid_amount": [1000.0, 2000.0],
|
|
"primary_diagnosis_code": ["I21.0", "J18.9"],
|
|
"encounter_source_type": ["claim", "claim"],
|
|
"data_source": ["test", "test"],
|
|
}
|
|
)
|
|
result = int_encounter_overlap(enc)
|
|
assert isinstance(result, pl.DataFrame)
|
|
assert "is_best_encounter" in result.columns
|
|
# Both should be best since no overlap
|
|
best = result["is_best_encounter"].to_list()
|
|
assert all(b == 1 for b in best)
|
|
|
|
def test_overlapping_encounters_selects_best(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2"],
|
|
"person_id": ["P1", "P1"],
|
|
"admit_date": [date(2024, 1, 1), date(2024, 1, 3)],
|
|
"discharge_date": [date(2024, 1, 10), date(2024, 1, 8)],
|
|
"discharge_disposition_code": ["01", "01"],
|
|
"facility_id": [None, None],
|
|
"drg_code_type": ["ms-drg", "ms-drg"],
|
|
"drg_code": ["280", "193"],
|
|
"paid_amount": [1000.0, 2000.0],
|
|
"primary_diagnosis_code": ["I21.0", "J18.9"],
|
|
"encounter_source_type": ["claim", "claim"],
|
|
"data_source": ["test", "test"],
|
|
}
|
|
)
|
|
result = int_encounter_overlap(enc)
|
|
assert len(result) == 2
|
|
best_count = result["is_best_encounter"].sum()
|
|
assert best_count == 1
|
|
|
|
def test_expected_columns(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": [None],
|
|
"drg_code_type": ["ms-drg"],
|
|
"drg_code": ["280"],
|
|
"paid_amount": [1000.0],
|
|
"primary_diagnosis_code": ["I21.0"],
|
|
"encounter_source_type": ["claim"],
|
|
"data_source": ["test"],
|
|
}
|
|
)
|
|
result = int_encounter_overlap(enc)
|
|
expected = [
|
|
"encounter_id",
|
|
"actual_length_of_stay",
|
|
"source_type_priority",
|
|
"completeness_score",
|
|
"is_best_encounter",
|
|
"selection_reason",
|
|
]
|
|
for col in expected:
|
|
assert col in result.columns
|
|
|
|
|
|
# ── int_readmission_crude ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntReadmissionCrude:
|
|
def test_detects_30_day_readmission(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2"],
|
|
"person_id": ["P1", "P1"],
|
|
"admit_date": [date(2024, 1, 1), date(2024, 1, 20)],
|
|
"discharge_date": [date(2024, 1, 5), date(2024, 1, 25)],
|
|
"discharge_disposition_code": ["01", "01"],
|
|
"facility_id": [None, None],
|
|
"drg_code_type": [None, None],
|
|
"drg_code": [None, None],
|
|
"paid_amount": [None, None],
|
|
"primary_diagnosis_code": [None, None],
|
|
"encounter_source_type": ["claim", "claim"],
|
|
"data_source": ["test", "test"],
|
|
}
|
|
)
|
|
overlap = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2"],
|
|
"person_id": ["P1", "P1"],
|
|
"admit_date": [date(2024, 1, 1), date(2024, 1, 20)],
|
|
"discharge_date": [date(2024, 1, 5), date(2024, 1, 25)],
|
|
"actual_length_of_stay": [4, 5],
|
|
"source_type_priority": [1, 1],
|
|
"completeness_score": [3, 3],
|
|
"overlap_group_id": ["E1", "E2"],
|
|
"has_overlaps": [0, 0],
|
|
"encounter_rank_in_group": [1, 1],
|
|
"is_best_encounter": [1, 1],
|
|
"selection_reason": [
|
|
"No overlapping encounters",
|
|
"No overlapping encounters",
|
|
],
|
|
}
|
|
)
|
|
result = int_readmission_crude(enc, overlap)
|
|
assert isinstance(result, pl.DataFrame)
|
|
e1 = result.filter(pl.col("encounter_id") == "E1")
|
|
if len(e1) > 0:
|
|
assert e1["had_readmission_flag"][0] == 1
|
|
assert e1["readmit_30_flag"][0] == 1
|
|
assert e1["days_to_readmit"][0] == 15
|
|
|
|
def test_no_readmission_beyond_30_days(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2"],
|
|
"person_id": ["P1", "P1"],
|
|
"admit_date": [date(2024, 1, 1), date(2024, 6, 1)],
|
|
"discharge_date": [date(2024, 1, 5), date(2024, 6, 5)],
|
|
"discharge_disposition_code": ["01", "01"],
|
|
"facility_id": [None, None],
|
|
"drg_code_type": [None, None],
|
|
"drg_code": [None, None],
|
|
"paid_amount": [None, None],
|
|
"primary_diagnosis_code": [None, None],
|
|
"encounter_source_type": ["claim", "claim"],
|
|
"data_source": ["test", "test"],
|
|
}
|
|
)
|
|
overlap = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1", "E2"],
|
|
"person_id": ["P1", "P1"],
|
|
"admit_date": [date(2024, 1, 1), date(2024, 6, 1)],
|
|
"discharge_date": [date(2024, 1, 5), date(2024, 6, 5)],
|
|
"actual_length_of_stay": [4, 4],
|
|
"source_type_priority": [1, 1],
|
|
"completeness_score": [3, 3],
|
|
"overlap_group_id": ["E1", "E2"],
|
|
"has_overlaps": [0, 0],
|
|
"encounter_rank_in_group": [1, 1],
|
|
"is_best_encounter": [1, 1],
|
|
"selection_reason": ["No overlapping", "No overlapping"],
|
|
}
|
|
)
|
|
result = int_readmission_crude(enc, overlap)
|
|
e1 = result.filter(pl.col("encounter_id") == "E1")
|
|
if len(e1) > 0:
|
|
assert e1["readmit_30_flag"][0] == 0
|
|
|
|
def test_expected_columns(self) -> None:
|
|
enc = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": [None],
|
|
"drg_code_type": [None],
|
|
"drg_code": [None],
|
|
"paid_amount": [None],
|
|
"primary_diagnosis_code": [None],
|
|
"encounter_source_type": ["claim"],
|
|
"data_source": ["test"],
|
|
}
|
|
)
|
|
overlap = pl.DataFrame(
|
|
{
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"actual_length_of_stay": [4],
|
|
"source_type_priority": [1],
|
|
"completeness_score": [3],
|
|
"overlap_group_id": ["E1"],
|
|
"has_overlaps": [0],
|
|
"encounter_rank_in_group": [1],
|
|
"is_best_encounter": [1],
|
|
"selection_reason": ["No overlapping"],
|
|
}
|
|
)
|
|
result = int_readmission_crude(enc, overlap)
|
|
for col in [
|
|
"encounter_id",
|
|
"had_readmission_flag",
|
|
"days_to_readmit",
|
|
"readmit_30_flag",
|
|
]:
|
|
assert col in result.columns
|
|
|
|
|
|
# ── int_encounter_data_quality ────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntEncounterDataQuality:
|
|
"""Tests for int_encounter_data_quality (lines 56-161).
|
|
|
|
This function joins encounter data with terminology lookups and
|
|
computes 11 data quality flags, then derives disqualified_encounter_flag.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _make_enc_with_ccs(**overrides):
|
|
"""Helper to build a single-row readmissions._int_encounter_with_ccs."""
|
|
defaults = {
|
|
"encounter_id": ["E1"],
|
|
"person_id": ["P1"],
|
|
"admit_date": [date(2024, 1, 1)],
|
|
"discharge_date": [date(2024, 1, 5)],
|
|
"discharge_disposition_code": ["01"],
|
|
"facility_id": ["FAC1"],
|
|
"drg_code_type": ["ms-drg"],
|
|
"drg_code": ["280"],
|
|
"paid_amount": [1000.0],
|
|
"primary_diagnosis_code": ["I21.0"],
|
|
"valid_primary_diagnosis_code_flag": [1],
|
|
"ccs_diagnosis_category": ["100"],
|
|
}
|
|
defaults.update(overrides)
|
|
return pl.DataFrame(defaults)
|
|
|
|
@staticmethod
|
|
def _make_overlap(encounter_ids, is_best):
|
|
"""Helper to build readmissions._int_encounter_overlap."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": encounter_ids,
|
|
"is_best_encounter": is_best,
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _make_dd(codes_and_descs):
|
|
"""Helper for terminology.discharge_disposition."""
|
|
return pl.DataFrame(
|
|
{
|
|
"discharge_disposition_code": [c for c, _ in codes_and_descs],
|
|
"discharge_disposition_description": [d for _, d in codes_and_descs],
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _make_ms_drg(codes):
|
|
return pl.DataFrame({"ms_drg_code": pl.Series(codes, dtype=pl.Utf8)})
|
|
|
|
@staticmethod
|
|
def _make_apr_drg(codes):
|
|
return pl.DataFrame({"apr_drg_code": pl.Series(codes, dtype=pl.Utf8)})
|
|
|
|
def test_clean_encounter_not_disqualified(self):
|
|
"""A fully valid encounter should have all flags = 0."""
|
|
enc = self._make_enc_with_ccs()
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Discharged to home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg(["999"]) # won't match, but ms will
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
assert isinstance(result, pl.DataFrame)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["disqualified_encounter_flag"][0] == 0
|
|
assert row["missing_admit_date_flag"][0] == 0
|
|
assert row["missing_discharge_date_flag"][0] == 0
|
|
assert row["admit_after_discharge_flag"][0] == 0
|
|
assert row["missing_discharge_disposition_code_flag"][0] == 0
|
|
assert row["invalid_discharge_disposition_code_flag"][0] == 0
|
|
assert row["missing_primary_diagnosis_flag"][0] == 0
|
|
assert row["invalid_primary_diagnosis_code_flag"][0] == 0
|
|
assert row["no_diagnosis_ccs_flag"][0] == 0
|
|
assert row["overlaps_with_another_encounter_flag"][0] == 0
|
|
assert row["missing_drg_flag"][0] == 0
|
|
assert row["invalid_drg_flag"][0] == 0
|
|
|
|
def test_missing_admit_date_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
admit_date=pl.Series([None], dtype=pl.Date),
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["missing_admit_date_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_missing_discharge_date_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
discharge_date=pl.Series([None], dtype=pl.Date),
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["missing_discharge_date_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_admit_after_discharge_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
admit_date=[date(2024, 1, 10)],
|
|
discharge_date=[date(2024, 1, 5)],
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["admit_after_discharge_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_missing_discharge_disposition_code_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
discharge_disposition_code=pl.Series([None], dtype=pl.Utf8),
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["missing_discharge_disposition_code_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_invalid_discharge_disposition_code_disqualifies(self):
|
|
"""Code present but not found in terminology lookup."""
|
|
enc = self._make_enc_with_ccs(
|
|
discharge_disposition_code=["ZZ"],
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")]) # ZZ not in lookup
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["invalid_discharge_disposition_code_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_missing_primary_diagnosis_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
primary_diagnosis_code=pl.Series([None], dtype=pl.Utf8),
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["missing_primary_diagnosis_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_invalid_primary_diagnosis_code_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
valid_primary_diagnosis_code_flag=[0],
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["invalid_primary_diagnosis_code_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_no_diagnosis_ccs_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
ccs_diagnosis_category=pl.Series([None], dtype=pl.Utf8),
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["no_diagnosis_ccs_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_overlap_loser_disqualifies(self):
|
|
"""Encounter not selected as best in overlap group."""
|
|
enc = self._make_enc_with_ccs()
|
|
overlap = self._make_overlap(["E1"], [0]) # is_best=0 => loser
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["overlaps_with_another_encounter_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_missing_drg_disqualifies(self):
|
|
enc = self._make_enc_with_ccs(
|
|
drg_code=pl.Series([None], dtype=pl.Utf8),
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg([])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["missing_drg_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_invalid_drg_disqualifies(self):
|
|
"""DRG code present but not found in either ms_drg or apr_drg."""
|
|
enc = self._make_enc_with_ccs(
|
|
drg_code=["999"],
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"]) # 999 not in ms_drg
|
|
apr = self._make_apr_drg(["001"]) # 999 not in apr_drg
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["invalid_drg_flag"][0] == 1
|
|
assert row["disqualified_encounter_flag"][0] == 1
|
|
|
|
def test_valid_apr_drg_not_invalid(self):
|
|
"""DRG code found in apr_drg but not ms_drg is still valid."""
|
|
enc = self._make_enc_with_ccs(
|
|
drg_code=["500"],
|
|
)
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"]) # 500 not in ms_drg
|
|
apr = self._make_apr_drg(["500"]) # 500 IS in apr_drg
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["invalid_drg_flag"][0] == 0
|
|
|
|
def test_output_columns(self):
|
|
enc = self._make_enc_with_ccs()
|
|
overlap = self._make_overlap(["E1"], [1])
|
|
dd = self._make_dd([("01", "Home")])
|
|
ms = self._make_ms_drg(["280"])
|
|
apr = self._make_apr_drg([])
|
|
|
|
result = int_encounter_data_quality(enc, overlap, dd, ms, apr)
|
|
expected_cols = [
|
|
"encounter_id",
|
|
"diagnosis_ccs",
|
|
"disqualified_encounter_flag",
|
|
"missing_admit_date_flag",
|
|
"missing_discharge_date_flag",
|
|
"admit_after_discharge_flag",
|
|
"missing_discharge_disposition_code_flag",
|
|
"invalid_discharge_disposition_code_flag",
|
|
"missing_primary_diagnosis_flag",
|
|
"invalid_primary_diagnosis_code_flag",
|
|
"no_diagnosis_ccs_flag",
|
|
"overlaps_with_another_encounter_flag",
|
|
"missing_drg_flag",
|
|
"invalid_drg_flag",
|
|
]
|
|
for col in expected_cols:
|
|
assert col in result.columns, f"Missing column: {col}"
|
|
|
|
|
|
# ── int_encounter_specialty_cohort ────────────────────────────────────────────
|
|
|
|
|
|
class TestIntEncounterSpecialtyCohort:
|
|
"""Tests for int_encounter_specialty_cohort (lines 364-435).
|
|
|
|
Assigns encounters to specialty cohorts: Surgery/Gynecology (rank 1),
|
|
Cardiorespiratory (rank 2), Cardiovascular (rank 3), Neurology (rank 4),
|
|
Medicine (default).
|
|
"""
|
|
|
|
@staticmethod
|
|
def _enc(ids):
|
|
"""Minimal readmissions._int_encounter."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": ids,
|
|
"person_id": ["P1"] * len(ids),
|
|
"admit_date": [date(2024, 1, 1)] * len(ids),
|
|
"discharge_date": [date(2024, 1, 5)] * len(ids),
|
|
"discharge_disposition_code": ["01"] * len(ids),
|
|
"facility_id": ["F1"] * len(ids),
|
|
"drg_code_type": ["ms-drg"] * len(ids),
|
|
"drg_code": ["280"] * len(ids),
|
|
"paid_amount": [1000.0] * len(ids),
|
|
"primary_diagnosis_code": ["I21.0"] * len(ids),
|
|
"encounter_source_type": ["claim"] * len(ids),
|
|
"data_source": ["test"] * len(ids),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _enc_ccs(ids, ccs_cats):
|
|
"""Minimal readmissions._int_encounter_with_ccs."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series(ids, dtype=pl.Utf8),
|
|
"ccs_diagnosis_category": pl.Series(ccs_cats, dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _proc_ccs(encounter_ids, proc_codes, ccs_cats):
|
|
"""Minimal readmissions._int_procedure_ccs."""
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series(encounter_ids, dtype=pl.Utf8),
|
|
"procedure_code": pl.Series(proc_codes, dtype=pl.Utf8),
|
|
"ccs_procedure_category": pl.Series(ccs_cats, dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _empty_proc():
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series([], dtype=pl.Utf8),
|
|
"procedure_code": pl.Series([], dtype=pl.Utf8),
|
|
"ccs_procedure_category": pl.Series([], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _specialty_vs(rows):
|
|
"""readmissions._value_set_specialty_cohort."""
|
|
return pl.DataFrame(
|
|
{
|
|
"ccs": pl.Series([r[0] for r in rows], dtype=pl.Utf8),
|
|
"specialty_cohort": pl.Series([r[1] for r in rows], dtype=pl.Utf8),
|
|
"procedure_or_diagnosis": pl.Series(
|
|
[r[2] for r in rows], dtype=pl.Utf8
|
|
),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _surg_gyn_vs(icd_codes):
|
|
"""readmissions._value_set_surgery_gynecology_cohort."""
|
|
return pl.DataFrame({"icd_10_pcs": pl.Series(icd_codes, dtype=pl.Utf8)})
|
|
|
|
def test_surgery_gynecology_from_icd_pcs(self):
|
|
"""Surgery/Gynecology via ICD-10-PCS match."""
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["999"]),
|
|
readmissions___int_procedure_ccs=self._proc_ccs(
|
|
["E1"], ["0UT90ZZ"], ["12"]
|
|
),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs([]),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs(
|
|
["0UT90ZZ"]
|
|
),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Surgery/Gynecology"
|
|
|
|
def test_surgery_gynecology_from_ccs_procedure(self):
|
|
"""Surgery/Gynecology via CCS procedure category match."""
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["999"]),
|
|
readmissions___int_procedure_ccs=self._proc_ccs(
|
|
["E1"], ["XXXXX"], ["CCS_SURG"]
|
|
),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs(
|
|
[("CCS_SURG", "Surgery/Gynecology", "Procedure")]
|
|
),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Surgery/Gynecology"
|
|
|
|
def test_cardiorespiratory_from_diagnosis(self):
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["CCS_CR"]),
|
|
readmissions___int_procedure_ccs=self._empty_proc(),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs(
|
|
[("CCS_CR", "Cardiorespiratory", "Diagnosis")]
|
|
),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Cardiorespiratory"
|
|
|
|
def test_cardiovascular_from_diagnosis(self):
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["CCS_CV"]),
|
|
readmissions___int_procedure_ccs=self._empty_proc(),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs(
|
|
[("CCS_CV", "Cardiovascular", "Diagnosis")]
|
|
),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Cardiovascular"
|
|
|
|
def test_neurology_from_diagnosis(self):
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["CCS_N"]),
|
|
readmissions___int_procedure_ccs=self._empty_proc(),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs(
|
|
[("CCS_N", "Neurology", "Diagnosis")]
|
|
),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Neurology"
|
|
|
|
def test_default_medicine_when_no_match(self):
|
|
"""No cohort match defaults to Medicine."""
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["NOMATCH"]),
|
|
readmissions___int_procedure_ccs=self._empty_proc(),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs([]),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Medicine"
|
|
|
|
def test_surgery_takes_priority_over_diagnosis_cohort(self):
|
|
"""Surgery/Gynecology (rank 1) beats Cardiorespiratory (rank 2)."""
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["CCS_CR"]),
|
|
readmissions___int_procedure_ccs=self._proc_ccs(
|
|
["E1"], ["0UT90ZZ"], ["CCS_SURG"]
|
|
),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs(
|
|
[
|
|
("CCS_SURG", "Surgery/Gynecology", "Procedure"),
|
|
("CCS_CR", "Cardiorespiratory", "Diagnosis"),
|
|
]
|
|
),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
row = result.filter(pl.col("encounter_id") == "E1")
|
|
assert row["specialty_cohort"][0] == "Surgery/Gynecology"
|
|
|
|
def test_output_columns(self):
|
|
result = int_encounter_specialty_cohort(
|
|
readmissions___int_encounter=self._enc(["E1"]),
|
|
readmissions___int_encounter_with_ccs=self._enc_ccs(["E1"], ["999"]),
|
|
readmissions___int_procedure_ccs=self._empty_proc(),
|
|
readmissions___value_set_specialty_cohort=self._specialty_vs([]),
|
|
readmissions___value_set_surgery_gynecology_cohort=self._surg_gyn_vs([]),
|
|
)
|
|
assert list(result.columns) == ["encounter_id", "specialty_cohort"]
|
|
|
|
|
|
# ── int_planned_encounter ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIntPlannedEncounter:
|
|
"""Tests for int_planned_encounter (lines 620-681).
|
|
|
|
Identifies planned readmissions: always-planned procedures,
|
|
always-planned diagnoses, and potentially-planned minus acute.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _enc_ccs(ids, diag_codes, ccs_cats):
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series(ids, dtype=pl.Utf8),
|
|
"primary_diagnosis_code": pl.Series(diag_codes, dtype=pl.Utf8),
|
|
"ccs_diagnosis_category": pl.Series(ccs_cats, dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _proc_ccs(encounter_ids, proc_codes, ccs_cats):
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series(encounter_ids, dtype=pl.Utf8),
|
|
"procedure_code": pl.Series(proc_codes, dtype=pl.Utf8),
|
|
"ccs_procedure_category": pl.Series(ccs_cats, dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _empty_proc():
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series([], dtype=pl.Utf8),
|
|
"procedure_code": pl.Series([], dtype=pl.Utf8),
|
|
"ccs_procedure_category": pl.Series([], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _empty_enc_ccs():
|
|
return pl.DataFrame(
|
|
{
|
|
"encounter_id": pl.Series([], dtype=pl.Utf8),
|
|
"primary_diagnosis_code": pl.Series([], dtype=pl.Utf8),
|
|
"ccs_diagnosis_category": pl.Series([], dtype=pl.Utf8),
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _vs_ccs(col_name, vals):
|
|
return pl.DataFrame({col_name: pl.Series(vals, dtype=pl.Utf8)})
|
|
|
|
@staticmethod
|
|
def _vs_icd(col_name, vals):
|
|
return pl.DataFrame({col_name: pl.Series(vals, dtype=pl.Utf8)})
|
|
|
|
def _call(
|
|
self,
|
|
enc_ccs=None,
|
|
proc_ccs=None,
|
|
acute_dx_ccs=None,
|
|
acute_dx_icd=None,
|
|
always_planned_dx=None,
|
|
always_planned_px=None,
|
|
pot_planned_ccs=None,
|
|
pot_planned_icd=None,
|
|
):
|
|
return int_planned_encounter(
|
|
readmissions___int_encounter_with_ccs=(
|
|
enc_ccs if enc_ccs is not None else self._empty_enc_ccs()
|
|
),
|
|
readmissions___int_procedure_ccs=(
|
|
proc_ccs if proc_ccs is not None else self._empty_proc()
|
|
),
|
|
readmissions___value_set_acute_diagnosis_ccs=(
|
|
acute_dx_ccs
|
|
if acute_dx_ccs is not None
|
|
else self._vs_ccs("ccs_diagnosis_category", [])
|
|
),
|
|
readmissions___value_set_acute_diagnosis_icd_10_cm=(
|
|
acute_dx_icd
|
|
if acute_dx_icd is not None
|
|
else self._vs_icd("icd_10_cm", [])
|
|
),
|
|
readmissions___value_set_always_planned_ccs_diagnosis_category=(
|
|
always_planned_dx
|
|
if always_planned_dx is not None
|
|
else self._vs_ccs("ccs_diagnosis_category", [])
|
|
),
|
|
readmissions___value_set_always_planned_ccs_procedure_category=(
|
|
always_planned_px
|
|
if always_planned_px is not None
|
|
else self._vs_ccs("ccs_procedure_category", [])
|
|
),
|
|
readmissions___value_set_potentially_planned_ccs_procedure_category=(
|
|
pot_planned_ccs
|
|
if pot_planned_ccs is not None
|
|
else self._vs_ccs("ccs_procedure_category", [])
|
|
),
|
|
readmissions___value_set_potentially_planned_icd_10_pcs=(
|
|
pot_planned_icd
|
|
if pot_planned_icd is not None
|
|
else self._vs_icd("icd_10_pcs", [])
|
|
),
|
|
)
|
|
|
|
def test_always_planned_procedure(self):
|
|
"""Encounter with always-planned CCS procedure category."""
|
|
result = self._call(
|
|
proc_ccs=self._proc_ccs(["E1"], ["02703ZZ"], ["AP_CCS"]),
|
|
always_planned_px=self._vs_ccs("ccs_procedure_category", ["AP_CCS"]),
|
|
)
|
|
assert "E1" in result["encounter_id"].to_list()
|
|
|
|
def test_always_planned_diagnosis(self):
|
|
"""Encounter with always-planned CCS diagnosis category."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["AP_DX"]),
|
|
always_planned_dx=self._vs_ccs("ccs_diagnosis_category", ["AP_DX"]),
|
|
)
|
|
assert "E1" in result["encounter_id"].to_list()
|
|
|
|
def test_potentially_planned_ccs_without_acute(self):
|
|
"""Potentially planned procedure (CCS) with no acute diagnosis."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["100"]),
|
|
proc_ccs=self._proc_ccs(["E1"], ["PROC1"], ["PP_CCS"]),
|
|
pot_planned_ccs=self._vs_ccs("ccs_procedure_category", ["PP_CCS"]),
|
|
)
|
|
assert "E1" in result["encounter_id"].to_list()
|
|
|
|
def test_potentially_planned_icd_without_acute(self):
|
|
"""Potentially planned procedure (ICD-10-PCS) with no acute dx."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["100"]),
|
|
proc_ccs=self._proc_ccs(["E1"], ["0UT90ZZ"], ["NOMATCH"]),
|
|
pot_planned_icd=self._vs_icd("icd_10_pcs", ["0UT90ZZ"]),
|
|
)
|
|
assert "E1" in result["encounter_id"].to_list()
|
|
|
|
def test_potentially_planned_with_acute_ccs_excluded(self):
|
|
"""Potentially planned but acute CCS diagnosis => not planned."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["ACUTE_CCS"]),
|
|
proc_ccs=self._proc_ccs(["E1"], ["P1"], ["PP_CCS"]),
|
|
pot_planned_ccs=self._vs_ccs("ccs_procedure_category", ["PP_CCS"]),
|
|
acute_dx_ccs=self._vs_ccs("ccs_diagnosis_category", ["ACUTE_CCS"]),
|
|
)
|
|
assert "E1" not in result["encounter_id"].to_list()
|
|
|
|
def test_potentially_planned_with_acute_icd_excluded(self):
|
|
"""Potentially planned but acute ICD-10-CM diagnosis => not planned."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["100"]),
|
|
proc_ccs=self._proc_ccs(["E1"], ["P1"], ["PP_CCS"]),
|
|
pot_planned_ccs=self._vs_ccs("ccs_procedure_category", ["PP_CCS"]),
|
|
acute_dx_icd=self._vs_icd("icd_10_cm", ["I21.0"]),
|
|
)
|
|
assert "E1" not in result["encounter_id"].to_list()
|
|
|
|
def test_no_matches_returns_empty(self):
|
|
"""No match in any planned list => empty result."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["100"]),
|
|
proc_ccs=self._proc_ccs(["E1"], ["P1"], ["NOMATCH"]),
|
|
)
|
|
assert len(result) == 0
|
|
|
|
def test_union_deduplicates(self):
|
|
"""Encounter matching both always-planned px and dx appears once."""
|
|
result = self._call(
|
|
enc_ccs=self._enc_ccs(["E1"], ["I21.0"], ["AP_DX"]),
|
|
proc_ccs=self._proc_ccs(["E1"], ["P1"], ["AP_CCS"]),
|
|
always_planned_px=self._vs_ccs("ccs_procedure_category", ["AP_CCS"]),
|
|
always_planned_dx=self._vs_ccs("ccs_diagnosis_category", ["AP_DX"]),
|
|
)
|
|
assert result["encounter_id"].to_list().count("E1") == 1
|
|
|
|
def test_output_columns(self):
|
|
result = self._call()
|
|
assert "encounter_id" in result.columns
|