Spaces:
Running
Running
| """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"]) | |