transformers.py

"""
Data transformation utilities.
"""
import pandas as pd
import numpy as np
from sklearn.preprocessing import LabelEncoder, MinMaxScaler, StandardScaler


class DataTransformer:
    """Provides various data transformation operations."""
    
    @staticmethod
    def drop_missing_rows(df, columns=None):
        """Drop rows with missing values."""
        if columns:
            return df.dropna(subset=columns)
        return df.dropna()
    
    @staticmethod
    def drop_missing_columns(df, threshold=0.5):
        """Drop columns with missing values above threshold."""
        missing_pct = df.isna().sum() / len(df)
        cols_to_keep = missing_pct[missing_pct < threshold].index
        return df[cols_to_keep]
    
    @staticmethod
    def fill_missing_mean(df, columns=None):
        """Fill missing values with column mean."""
        result = df.copy()
        cols = columns if columns else df.select_dtypes(include=[np.number]).columns
        for col in cols:
            if col in result.columns and pd.api.types.is_numeric_dtype(result[col]):
                result[col] = result[col].fillna(result[col].mean())
        return result
    
    @staticmethod
    def fill_missing_median(df, columns=None):
        """Fill missing values with column median."""
        result = df.copy()
        cols = columns if columns else df.select_dtypes(include=[np.number]).columns
        for col in cols:
            if col in result.columns and pd.api.types.is_numeric_dtype(result[col]):
                result[col] = result[col].fillna(result[col].median())
        return result
    
    @staticmethod
    def fill_missing_mode(df, columns=None):
        """Fill missing values with column mode."""
        result = df.copy()
        cols = columns if columns else df.columns
        for col in cols:
            if col in result.columns and not result[col].isna().all():
                mode_val = result[col].mode()
                if len(mode_val) > 0:
                    result[col] = result[col].fillna(mode_val[0])
        return result
    
    @staticmethod
    def fill_missing_forward(df, columns=None):
        """Forward fill missing values."""
        result = df.copy()
        cols = columns if columns else df.columns
        for col in cols:
            if col in result.columns:
                result[col] = result[col].ffill()
        return result
    
    @staticmethod
    def fill_missing_backward(df, columns=None):
        """Backward fill missing values."""
        result = df.copy()
        cols = columns if columns else df.columns
        for col in cols:
            if col in result.columns:
                result[col] = result[col].bfill()
        return result
    
    @staticmethod
    def fill_missing_constant(df, value, columns=None):
        """Fill missing values with a constant."""
        result = df.copy()
        cols = columns if columns else df.columns
        for col in cols:
            if col in result.columns:
                result[col] = result[col].fillna(value)
        return result
    
    @staticmethod
    def normalize_columns(df, columns=None):
        """Normalize numeric columns to 0-1 range."""
        result = df.copy()
        cols = columns if columns else df.select_dtypes(include=[np.number]).columns
        
        for col in cols:
            if col in result.columns and pd.api.types.is_numeric_dtype(result[col]):
                non_null_mask = result[col].notna()
                if non_null_mask.sum() > 0:
                    scaler = MinMaxScaler()
                    scaled_values = scaler.fit_transform(
                        result.loc[non_null_mask, col].values.reshape(-1, 1)
                    ).flatten()
                    # Create a new column with float dtype
                    new_col = result[col].astype(float)
                    new_col.loc[non_null_mask] = scaled_values
                    result[col] = new_col
        return result
    
    @staticmethod
    def standardize_columns(df, columns=None):
        """Standardize numeric columns (z-score normalization)."""
        result = df.copy()
        cols = columns if columns else df.select_dtypes(include=[np.number]).columns
        
        for col in cols:
            if col in result.columns and pd.api.types.is_numeric_dtype(result[col]):
                non_null_mask = result[col].notna()
                if non_null_mask.sum() > 0:
                    scaler = StandardScaler()
                    scaled_values = scaler.fit_transform(
                        result.loc[non_null_mask, col].values.reshape(-1, 1)
                    ).flatten()
                    # Create a new column with float dtype
                    new_col = result[col].astype(float)
                    new_col.loc[non_null_mask] = scaled_values
                    result[col] = new_col
        return result
    
    @staticmethod
    def encode_categorical(df, columns=None):
        """Label encode categorical columns."""
        result = df.copy()
        cols = columns if columns else df.select_dtypes(include=['object']).columns
        
        for col in cols:
            if col in result.columns:
                le = LabelEncoder()
                non_null_mask = result[col].notna()
                if non_null_mask.sum() > 0:
                    encoded_values = le.fit_transform(
                        result.loc[non_null_mask, col].astype(str)
                    )
                    result.loc[non_null_mask, col] = pd.Series(encoded_values, index=result[non_null_mask].index)
        return result
    
    @staticmethod
    def drop_duplicates(df):
        """Remove duplicate rows."""
        return df.drop_duplicates()
    
    @staticmethod
    def drop_columns(df, columns):
        """Drop specified columns."""
        return df.drop(columns=columns, errors='ignore')
    
    @staticmethod
    def rename_column(df, old_name, new_name):
        """Rename a column."""
        return df.rename(columns={old_name: new_name})
    
    @staticmethod
    def reset_index(df):
        """Reset dataframe index."""
        return df.reset_index(drop=True)