Spaces:
Sleeping
Sleeping
File size: 5,843 Bytes
d81f51d | 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 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """Schema-aware imputation, encoding, and scaling."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from .adapters.base import CANONICAL_ID_COL, DatasetSchema
@dataclass
class PreprocessResult:
ids: np.ndarray
X: np.ndarray
transformer: ColumnTransformer
scaler: ColumnTransformer
feature_names: list[str]
clean_features: pd.DataFrame
keep_mask: np.ndarray = None # type: ignore[assignment]
"""Boolean mask over the transformer's raw output, marking finite + non-constant
columns. Live-predict path applies this to ``transformer.transform(new_row)``
so the new feature vector lines up with the training X exactly."""
def __iter__(self) -> Iterator[object]:
"""Preserve the old six-value unpacking API used by tests/scripts."""
yield self.ids
yield self.X
yield self.transformer
yield self.scaler
yield self.feature_names
yield self.clean_features
def _legacy_schema(feature_matrix: pd.DataFrame) -> DatasetSchema:
if CANONICAL_ID_COL not in feature_matrix.columns:
raise ValueError("feature_matrix must contain an id_student column")
numeric_cols = [
col for col in feature_matrix.columns
if col != CANONICAL_ID_COL
]
return DatasetSchema(
dataset_name="legacy",
adapter_name="legacy",
source_id_col=CANONICAL_ID_COL,
feature_cols=numeric_cols,
numeric_feature_cols=numeric_cols,
)
def _feature_name(name: str) -> str:
for prefix in ["num__", "cat__"]:
if name.startswith(prefix):
return name[len(prefix):]
return name
def preprocess(
feature_matrix: pd.DataFrame,
schema: DatasetSchema | None = None,
family_weights: dict[str, float] | None = None,
) -> PreprocessResult:
"""Return IDs, scaled feature matrix, fitted transformer, and clean features."""
schema = schema or _legacy_schema(feature_matrix)
if schema.id_col not in feature_matrix.columns:
raise ValueError(f"feature_matrix must contain a {schema.id_col!r} column")
ids = feature_matrix[schema.id_col].to_numpy()
numeric_cols = [col for col in schema.numeric_feature_cols if col in feature_matrix.columns]
categorical_cols = [col for col in schema.categorical_feature_cols if col in feature_matrix.columns]
selected_cols = list(dict.fromkeys(numeric_cols + categorical_cols))
if not selected_cols:
raise ValueError("No clustering features are available after applying dataset schema")
work = feature_matrix[selected_cols].copy()
for col in numeric_cols:
work[col] = pd.to_numeric(work[col], errors="coerce")
transformers: list[tuple[str, Pipeline, list[str]]] = []
if numeric_cols:
transformers.append(
(
"num",
Pipeline(
steps=[
("imputer", SimpleImputer(strategy="median")),
("scaler", StandardScaler()),
]
),
numeric_cols,
)
)
if categorical_cols:
transformers.append(
(
"cat",
Pipeline(
steps=[
("imputer", SimpleImputer(strategy="most_frequent")),
("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
# Z-scoring one-hot columns is intentional: all encoded
# dimensions then contribute comparably to distance-based clustering.
("scaler", StandardScaler()),
]
),
categorical_cols,
)
)
transformer = ColumnTransformer(transformers=transformers, remainder="drop", verbose_feature_names_out=True)
X = transformer.fit_transform(work)
X = np.asarray(X, dtype=float)
feature_names = [_feature_name(name) for name in transformer.get_feature_names_out()]
finite_mask = np.isfinite(X).all(axis=0)
nonzero_mask = np.nanvar(X, axis=0) > 0
keep = finite_mask & nonzero_mask
if not keep.any():
raise ValueError("No non-constant features remain after preprocessing")
X = X[:, keep]
feature_names = [name for name, keep_col in zip(feature_names, keep) if keep_col]
encoded = pd.DataFrame(X, columns=feature_names, index=feature_matrix.index)
if family_weights:
for prefix, weight in family_weights.items():
cols = [col for col in encoded.columns if col.startswith(prefix)]
if cols:
encoded.loc[:, cols] *= float(weight)
X = encoded.to_numpy()
clean_features = work.copy()
if numeric_cols:
num_imputer = SimpleImputer(strategy="median")
clean_features[numeric_cols] = num_imputer.fit_transform(work[numeric_cols])
for col in categorical_cols:
mode = work[col].mode(dropna=True)
fill = mode.iloc[0] if not mode.empty else "missing"
clean_features[col] = work[col].fillna(fill).astype(str)
kept_source_cols = [
col for col in selected_cols
if col in feature_names or any(name.startswith(f"{col}_") for name in feature_names)
]
clean_features = clean_features[kept_source_cols]
return PreprocessResult(
ids=ids,
X=X,
transformer=transformer,
scaler=transformer,
feature_names=feature_names,
clean_features=clean_features.reset_index(drop=True),
keep_mask=keep,
)
|