Spaces:
Running
Running
File size: 3,657 Bytes
dc3d345 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """Leakage-safe preprocessing: a scikit-learn transformer that imputes, encodes,
and engineers clinical features using statistics learned on the training split
only. Reused identically by training and by the live API so there is no
train/serve skew.
"""
import numpy as np
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from .config import CAT_COLUMNS, MODEL_COLUMNS, NUMERIC_COLUMNS
def clean_text_series(series: pd.Series) -> pd.Series:
"""Normalize the raw UCI whitespace/tab artifacts (e.g. ' yes', '\\tyes')."""
cleaned = (
series.astype(str)
.str.replace("\t", "", regex=False)
.str.replace(r"\s+", "", regex=True)
.str.lower()
.str.strip()
)
return cleaned.replace({"": np.nan, "?": np.nan, "nan": np.nan, "none": np.nan})
class CKDPreprocessor(BaseEstimator, TransformerMixin):
"""Train-only imputation, categorical encoding, and clinical feature engineering."""
def fit(self, X, y=None):
df = X.copy()
self.numeric_medians_ = df[NUMERIC_COLUMNS].median(numeric_only=True)
self.category_modes_ = {}
self.category_maps_ = {}
for column in CAT_COLUMNS:
mode = df[column].dropna().mode()
self.category_modes_[column] = mode.iloc[0] if not mode.empty else "unknown"
values = (
df[column]
.fillna(self.category_modes_[column])
.astype(str)
.sort_values()
.unique()
)
self.category_maps_[column] = {v: code for code, v in enumerate(values)}
# Learn the age_group fill value on the training split only (no leakage).
age_group_fit = pd.cut(
df["age"], bins=[0, 30, 45, 60, 120], labels=[0, 1, 2, 3]
).astype(float)
self.age_group_median_ = age_group_fit.median()
return self
def transform(self, X):
df = X.copy()
# Coerce to numeric first: single-row API payloads with missing fields
# arrive as object dtype, which makes fillna emit a downcast warning.
df[NUMERIC_COLUMNS] = df[NUMERIC_COLUMNS].apply(pd.to_numeric, errors="coerce")
df[NUMERIC_COLUMNS] = df[NUMERIC_COLUMNS].fillna(self.numeric_medians_)
for column in CAT_COLUMNS:
df[column] = (
df[column]
.fillna(self.category_modes_[column])
.astype(str)
.map(self.category_maps_[column])
.fillna(-1)
.astype(int)
)
df["kidney_stress_index"] = (df["sc"] * df["bu"]) / (df["hemo"] + 1e-6)
df["anemia_risk"] = df["hemo"] / (df["pcv"] + 1e-6)
df["age_bp_risk"] = (df["age"] * df["bp"]) / 1000.0
df["age_group"] = pd.cut(
df["age"], bins=[0, 30, 45, 60, 120], labels=[0, 1, 2, 3]
).astype(float)
df["age_group"] = df["age_group"].fillna(self.age_group_median_)
return df[MODEL_COLUMNS]
def load_dataset(path) -> pd.DataFrame:
"""Load the UCI CKD CSV and normalize known raw-value artifacts."""
df = pd.read_csv(path, na_values=["?", "nan", "\t?", " ?"])
df = df.rename(columns={"wc": "wbcc", "rc": "rbcc", "classification": "class"})
if "id" in df.columns:
df = df.drop(columns=["id"])
for column in df.columns:
if df[column].dtype == object:
df[column] = clean_text_series(df[column])
for column in NUMERIC_COLUMNS:
df[column] = pd.to_numeric(df[column], errors="coerce")
df["target"] = (df["class"] == "ckd").astype(int)
return df.drop(columns=["class"])
|