"""Engineer features used across the analysis layers. Adds census_region, urban_rural, denominator-specific charge ratios, drg_mdc / apc_family, and produces provider_combined.parquet (one row per provider joined across inpatient + outpatient). """ import pandas as pd import numpy as np from .clean import clean_inpatient, clean_outpatient from .config import ( DATA_PROC, FIPS_TO_REGION, ruca_to_urban_rural, COMPATIBILITY_RATIO_ALIAS, MAPPING_METADATA, MAPPING_STATUS, MAPPING_VALIDATION_EXAMPLES, drg_mapping_quarantine_reason, drg_to_mdc, apc_to_family, ) # --------------------------------------------------------------------------- # Per-row feature engineering # --------------------------------------------------------------------------- def _safe_ratio(numerator: pd.Series, denominator: pd.Series) -> pd.Series: return (numerator / denominator).replace([np.inf, -np.inf], np.nan) def add_features_inpatient(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df["census_region"] = df["state_fips"].map(FIPS_TO_REGION).fillna("Unknown") df["urban_rural"] = df["ruca_code"].map(ruca_to_urban_rural) df["drg_mdc"] = df.apply(lambda r: drg_to_mdc(r["drg_code"], r["drg_desc"]), axis=1) df["mapping_quarantined"] = df.apply( lambda r: drg_mapping_quarantine_reason(r["drg_code"], r["drg_desc"]) is not None, axis=1, ) df["is_maryland_exception"] = df["state"].eq("MD") df["charge_to_total_payment_ratio"] = _safe_ratio( df["avg_submitted_charge"], df["avg_total_payment"] ) # Compatibility alias required by the unchanged analysis modules. The # canonical field above names the actual inpatient denominator. df[COMPATIBILITY_RATIO_ALIAS] = df["charge_to_total_payment_ratio"] return df def add_features_outpatient(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df["census_region"] = df["state_fips"].map(FIPS_TO_REGION).fillna("Unknown") df["urban_rural"] = df["ruca_code"].map(ruca_to_urban_rural) df["apc_family"] = df["apc_desc"].map(apc_to_family) df["mapping_quarantined"] = False df["is_maryland_exception"] = df["state"].eq("MD") # The outpatient denominator is the allowed amount, not Medicare-only # payment. Suppressed allowed amounts therefore produce missing ratios. df["charge_to_allowed_amount_ratio"] = _safe_ratio( df["avg_submitted_charge"], df["avg_allowed_amount"] ) # Compatibility alias required by the unchanged analysis modules. The # canonical field above names the actual outpatient denominator. df[COMPATIBILITY_RATIO_ALIAS] = df["charge_to_allowed_amount_ratio"] return df # --------------------------------------------------------------------------- # Provider-level aggregation + cross-dataset join # --------------------------------------------------------------------------- def provider_summary_inpatient(df: pd.DataFrame) -> pd.DataFrame: """One row per provider summarizing inpatient activity.""" g = df.groupby("provider_ccn", as_index=False).agg( provider_name=("provider_name", "first"), state=("state", "first"), state_fips=("state_fips", "first"), census_region=("census_region", "first"), urban_rural=("urban_rural", "first"), is_maryland_exception=("is_maryland_exception", "first"), inp_drgs_seen=("drg_code", "nunique"), inp_total_discharges=("total_discharges", "sum"), inp_avg_payment=("avg_total_payment", "mean"), inp_median_payment=("avg_total_payment", "median"), inp_avg_charge_to_total_payment_ratio=("charge_to_total_payment_ratio", "mean"), ) return g def provider_summary_outpatient(df: pd.DataFrame) -> pd.DataFrame: """One row per provider summarizing outpatient activity (cost rows only).""" g = df.groupby("provider_ccn", as_index=False).agg( provider_name=("provider_name", "first"), state=("state", "first"), state_fips=("state_fips", "first"), census_region=("census_region", "first"), urban_rural=("urban_rural", "first"), is_maryland_exception=("is_maryland_exception", "first"), out_apcs_seen=("apc_code", "nunique"), out_total_services=("apc_services", "sum"), out_avg_allowed=("avg_allowed_amount", "mean"), out_median_allowed=("avg_allowed_amount", "median"), out_avg_charge_to_allowed_amount_ratio=("charge_to_allowed_amount_ratio", "mean"), ) return g def _outpatient_presence_summary(df: pd.DataFrame) -> pd.DataFrame: """Summarize full-frame outpatient presence independently of suppression.""" aggregations = { "provider_name": ("provider_name", "first"), "state": ("state", "first"), "state_fips": ("state_fips", "first"), } for column in ("census_region", "urban_rural"): if column in df.columns: aggregations[column] = (column, "first") result = df.groupby("provider_ccn", as_index=False).agg(**aggregations) result["out_full_rows"] = df.groupby("provider_ccn").size().reindex( result["provider_ccn"] ).to_numpy() result["out_full_apcs_seen"] = df.groupby("provider_ccn")["apc_code"].nunique().reindex( result["provider_ccn"] ).to_numpy() return result def build_provider_combined( inp_feat: pd.DataFrame, out_feat: pd.DataFrame, out_full_feat: pd.DataFrame | None = None, ) -> pd.DataFrame: """Join inpatient metrics to cost-observed metrics and full outpatient presence.""" inp_s = provider_summary_inpatient(inp_feat) out_s = provider_summary_outpatient(out_feat) merged = inp_s.merge(out_s, on="provider_ccn", how="outer", suffixes=("_inp", "_out")) if out_full_feat is not None: merged = merged.merge( _outpatient_presence_summary(out_full_feat), on="provider_ccn", how="outer", ) # Reconcile geography explicitly: inpatient wins, then full outpatient # presence, then the cost-observed outpatient summary. for column in ("provider_name", "state", "state_fips", "census_region", "urban_rural"): candidates = [f"{column}_inp", column, f"{column}_out"] available = [merged[candidate] for candidate in candidates if candidate in merged] if available: value = available[0] for candidate in available[1:]: value = value.fillna(candidate) merged[column] = value maryland_candidates = [ merged[column] for column in ( "is_maryland_exception_inp", "is_maryland_exception_out" ) if column in merged ] merged["is_maryland_exception"] = ( maryland_candidates[0].fillna(maryland_candidates[1]) if len(maryland_candidates) == 2 else maryland_candidates[0] if maryland_candidates else False ).fillna(False).astype(bool) merged = merged.drop(columns=[ c for c in merged.columns if c.endswith(("_inp", "_out")) and c not in ( "provider_name", "state", "state_fips", "census_region", "urban_rural", "is_maryland_exception", ) ]) # Boolean presence flags — useful for outlier profiling merged["has_inpatient"] = merged["inp_total_discharges"].notna() merged["has_outpatient_cost_observed"] = merged["out_total_services"].notna() merged["has_outpatient_full"] = merged.get("out_full_rows", pd.Series(index=merged.index)).notna() merged["has_outpatient"] = merged["has_outpatient_full"] merged["outpatient_presence_basis"] = "full outpatient frame" return merged # --------------------------------------------------------------------------- # Mapping validation # --------------------------------------------------------------------------- _UNMAPPED_LABELS = {"Unknown", "Other", "Other / Unclassified"} def _normalise_mapping_code(value, width: int) -> str | None: if pd.isna(value): return None text = str(value).strip() if text.endswith(".0") and text[:-2].isdigit(): text = text[:-2] return text.zfill(width) if text.isdigit() else text def _validate_mapping( df: pd.DataFrame, code_col: str, desc_col: str, mapper, examples: list[dict], code_width: int, feature_field: str, quarantine_fn=None, ) -> dict: observed = df[[code_col, desc_col]].copy() observed["_mapping_code"] = observed[code_col].map( lambda value: _normalise_mapping_code(value, code_width) ) observed = observed[observed["_mapping_code"].notna()] observed_rows = len(observed) observed_pairs = observed.drop_duplicates(["_mapping_code", desc_col]) labels_by_code = {} quarantine_by_code = {} for code, group in observed_pairs.groupby("_mapping_code", sort=True): descriptions = group[desc_col].where(group[desc_col].notna(), "") labels_by_code[code] = sorted({mapper(code, desc) for desc in descriptions}) if quarantine_fn: reasons = { quarantine_fn(code, desc) for desc in descriptions if quarantine_fn(code, desc) } if reasons: quarantine_by_code[code] = sorted(reasons) unmapped_codes = sorted( code for code, labels in labels_by_code.items() if not labels or all(label in _UNMAPPED_LABELS for label in labels) ) ambiguous_codes = sorted( code for code, labels in labels_by_code.items() if len(labels) > 1 ) validation_examples = [] for example in examples: code = _normalise_mapping_code(example["code"], code_width) matches = observed_pairs[observed_pairs["_mapping_code"] == code] description = "" if not matches.empty: value = matches.iloc[0][desc_col] description = "" if pd.isna(value) else str(value) actual = mapper(code, description) marker = example["description_contains"] description_check = marker.upper() in description.upper() validation_examples.append({ "code": example["code"], "observed": not matches.empty, "observed_description": description, "expected_mapping": example["expected_mapping"], "actual_mapping": actual, "description_contains": marker, "description_check_passed": description_check, "passed": bool( not matches.empty and description_check and actual == example["expected_mapping"] ), }) observed_code_count = len(labels_by_code) return { "mapping_status": MAPPING_STATUS, "feature_field": feature_field, "code_column": code_col, "description_column": desc_col, "observed_rows": int(observed_rows), "observed_code_count": int(observed_code_count), "observed_codes": sorted(labels_by_code), "mapped_code_count": int(observed_code_count - len(unmapped_codes)), "unmapped_code_count": int(len(unmapped_codes)), "unmapped_count": int(len(unmapped_codes)), "unmapped_codes": unmapped_codes, "ambiguity_rule": "A code is ambiguous when its observed descriptions produce more than one distinct exploratory label.", "ambiguous_code_count": int(len(ambiguous_codes)), "ambiguous_count": int(len(ambiguous_codes)), "ambiguous_codes": ambiguous_codes, "validation_examples": validation_examples, "quarantined_code_count": int(len(quarantine_by_code)), "quarantined_codes": [ {"code": code, "reasons": reasons} for code, reasons in sorted(quarantine_by_code.items()) ], "quarantined_row_count": int( observed["_mapping_code"].isin(quarantine_by_code).sum() ), "modeling_policy": ( "Rows carrying a quarantined DRG mapping are excluded from predictor modeling." if quarantine_fn else "No APC mapping quarantine rules are defined." ), } def validate_observed_mappings( inpatient_df: pd.DataFrame, outpatient_df: pd.DataFrame ) -> dict: """Validate observed code coverage without implying an official crosswalk.""" result = dict(MAPPING_METADATA) result["mapping_status"] = MAPPING_STATUS result["drg"] = _validate_mapping( inpatient_df, "drg_code", "drg_desc", drg_to_mdc, MAPPING_VALIDATION_EXAMPLES["drg"], code_width=3, feature_field="drg_mdc", quarantine_fn=drg_mapping_quarantine_reason, ) result["apc"] = _validate_mapping( outpatient_df, "apc_code", "apc_desc", lambda _code, desc: apc_to_family(desc), MAPPING_VALIDATION_EXAMPLES["apc"], code_width=4, feature_field="apc_family", ) return result # --------------------------------------------------------------------------- # Orchestration # --------------------------------------------------------------------------- def run() -> dict: """Run the full feature engineering step. Returns dict of artifacts.""" inp_clean = clean_inpatient() out_cost = clean_outpatient(drop_suppressed=True) out_full = clean_outpatient(drop_suppressed=False) inp_feat = add_features_inpatient(inp_clean) out_feat = add_features_outpatient(out_cost) out_full_feat = add_features_outpatient(out_full) provider_combined = build_provider_combined(inp_feat, out_feat, out_full_feat) inp_feat.to_parquet(DATA_PROC / "inpatient_features.parquet", index=False) out_feat.to_parquet(DATA_PROC / "outpatient_features.parquet", index=False) out_full_feat.to_parquet(DATA_PROC / "outpatient_full_features.parquet", index=False) provider_combined.to_parquet(DATA_PROC / "provider_combined.parquet", index=False) return { "inpatient_features": inp_feat, "outpatient_features": out_feat, "provider_combined": provider_combined, } if __name__ == "__main__": # pragma: no cover res = run() for k, df in res.items(): print(f"{k}: {df.shape}") print(df.head(2).to_string()) print()