| """Data preprocessing pipeline.""" |
|
|
| import pandas as pd |
| import numpy as np |
| from sklearn.preprocessing import StandardScaler, LabelEncoder |
| from typing import Tuple, List, Dict, Optional |
| from sklearn.model_selection import train_test_split |
| from config.settings import RANDOM_SEED, TEST_SIZE, VAL_SIZE, NUMERIC_FEATURES_DTYPE, CATEGORICAL_FEATURES_DTYPE |
|
|
|
|
| class Preprocessor: |
| """Handles data preprocessing including encoding, scaling, and splitting.""" |
|
|
| def __init__(self, df: pd.DataFrame, target_col: str): |
| """Initialize preprocessor.""" |
| self.df = df.copy() |
| self.target_col = target_col |
| self.numeric_cols = df.select_dtypes(include=NUMERIC_FEATURES_DTYPE).columns.tolist() |
| self.categorical_cols = df.select_dtypes(include=CATEGORICAL_FEATURES_DTYPE).columns.tolist() |
| |
| |
| if self.target_col in self.numeric_cols: |
| self.numeric_cols.remove(self.target_col) |
| if self.target_col in self.categorical_cols: |
| self.categorical_cols.remove(self.target_col) |
| |
| self.scaler = StandardScaler() |
| self.label_encoders = {} |
| self.target_encoder = None |
| self.feature_names = [] |
|
|
| def encode_categorical(self) -> None: |
| """Encode categorical features using LabelEncoder.""" |
| df_encoded = self.df.copy() |
| |
| |
| for col in self.categorical_cols: |
| le = LabelEncoder() |
| df_encoded[col] = le.fit_transform(df_encoded[col].astype(str)) |
| self.label_encoders[col] = le |
| |
| |
| self.is_classification = self.df[self.target_col].dtype in CATEGORICAL_FEATURES_DTYPE or \ |
| self.df[self.target_col].nunique() < 20 |
| |
| if self.is_classification and self.target_col in self.categorical_cols: |
| le = LabelEncoder() |
| df_encoded[self.target_col] = le.fit_transform(df_encoded[self.target_col].astype(str)) |
| self.target_encoder = le |
| |
| self.df = df_encoded |
|
|
| def scale_features(self, X: pd.DataFrame) -> np.ndarray: |
| """Scale numeric features using StandardScaler.""" |
| X_scaled = X.copy() |
| |
| if self.numeric_cols: |
| X_scaled[self.numeric_cols] = self.scaler.fit_transform(X[self.numeric_cols]) |
| |
| return X_scaled |
|
|
| def prepare_data(self, test_size: Optional[float] = None) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: |
| """Prepare data for training. |
| |
| Returns: |
| Tuple of (X_train, X_test, y_train, y_test) |
| """ |
| if test_size is None: |
| test_size = TEST_SIZE |
| |
| |
| self.encode_categorical() |
| |
| |
| X = self.df.drop(columns=[self.target_col]) |
| y = self.df[self.target_col] |
| |
| |
| self.feature_names = X.columns.tolist() |
| |
| |
| X_scaled = self.scale_features(X) |
| |
| |
| X_train, X_test, y_train, y_test = train_test_split( |
| X_scaled, y, test_size=test_size, random_state=RANDOM_SEED, stratify=y if self.is_classification else None |
| ) |
| |
| return X_train, X_test, y_train, y_test |
|
|
| def get_feature_names(self) -> List[str]: |
| """Return list of feature names.""" |
| return self.feature_names |
|
|
| def get_task_type(self) -> str: |
| """Return task type: 'classification' or 'regression'.""" |
| return "classification" if self.is_classification else "regression" |
|
|
| def rank_target_columns(df: pd.DataFrame) -> List[Tuple[str, int, str, int]]: |
| """ |
| Rank DataFrame columns by suitability as target columns. |
| |
| Ranking system: |
| 1 (Best): Binary columns (exactly 2 unique values) |
| 2: Low cardinality (2-20 unique values) |
| 3: Medium cardinality (21-50 unique values) |
| 4: High cardinality (50+ unique values) - likely features, not targets |
| |
| Args: |
| df: Input DataFrame |
| |
| Returns: |
| List of tuples: (column_name, rank, reason, unique_count) |
| Sorted by rank (best first) |
| """ |
| ranked = [] |
| |
| for col in df.columns: |
| unique_count = df[col].nunique() |
| |
| |
| if unique_count < 2: |
| continue |
| |
| |
| if unique_count == 2: |
| rank = 1 |
| reason = f"✅ Binary ({unique_count} classes)" |
| elif unique_count <= 20: |
| rank = 2 |
| reason = f"✅ Low cardinality ({unique_count} classes)" |
| elif unique_count <= 50: |
| rank = 3 |
| reason = f"⚠️ Medium cardinality ({unique_count} classes)" |
| else: |
| rank = 4 |
| reason = f"❌ High cardinality ({unique_count} classes) - likely a feature" |
| |
| ranked.append((col, rank, reason, unique_count)) |
| |
| |
| ranked.sort(key=lambda x: (x[1], x[0])) |
| |
| return ranked |
