Data Cleaning and Preprocessing for Absolute Beginners

import pandas as pd
import numpy as np

def inspect_dataset(filepath: str) -> pd.DataFrame:
    """First-pass data inspection. Run this before any cleaning.

    Returns the raw DataFrame so you can continue exploring interactively.
    Prints a diagnostic summary covering shape, types, missingness,
    duplicates, numeric ranges, and categorical cardinality.
    """
    df = pd.read_csv(filepath)

    print(f"Shape: {df.shape[0]:,} rows x {df.shape[1]} columns")
    print(f"Memory usage: {df.memory_usage(deep=True).sum() / 1e6:.1f} MB")

    print(f"\n--- Column Types ---")
    print(df.dtypes.value_counts())

    print(f"\n--- Missing Values ---")
    missing = df.isnull().sum()
    missing_pct = (missing / len(df) * 100).round(1)
    missing_report = pd.DataFrame({"count": missing, "percent": missing_pct})
    missing_report = missing_report[missing_report["count"] > 0].sort_values(
        "percent", ascending=False
    )
    if missing_report.empty:
        print("No missing values detected.")
    else:
        print(missing_report)

    print(f"\n--- Duplicate Rows ---")
    dup_count = df.duplicated().sum()
    print(f"Exact duplicates: {dup_count} ({dup_count / len(df) * 100:.1f}%)")

    print(f"\n--- Numeric Summary ---")
    num_desc = df.describe().round(2)
    print(num_desc)

    print(f"\n--- Categorical Columns ---")
    cat_cols = df.select_dtypes(include=["object", "category"]).columns
    for col in cat_cols:
        unique_count = df[col].nunique()
        print(f"  {col}: {unique_count} unique values")
        if unique_count <= 10:
            print(f"    Values: {df[col].unique().tolist()}")
        elif unique_count <= 50:
            print(f"    Top 5: {df[col].value_counts().head(5).to_dict()}")
        else:
            print(f"    High cardinality — consider grouping or encoding")

    return df

import pandas as pd
import numpy as np


class MissingValueHandler:
    """Strategies for handling missing values based on missingness type.

    Usage:
        handler = MissingValueHandler()
        report = handler.diagnose_missingness(df)
        df = handler.impute_numeric(df, 'age', strategy='median')
        df = handler.impute_categorical(df, 'city', strategy='unknown')
    """

    @staticmethod
    def diagnose_missingness(df: pd.DataFrame, target_col: str = None) -> dict:
        """Determine if missingness is random or systematic.

        If target_col is provided, checks whether missingness in each column
        correlates with the target — a signal of potential MNAR.
        """
        report = {}
        for col in df.columns:
            missing_count = df[col].isnull().sum()
            if missing_count == 0:
                continue

            missing_pct = missing_count / len(df) * 100
            entry = {
                "count": missing_count,
                "percent": round(missing_pct, 1),
                "recommendation": MissingValueHandler._recommend(
                    missing_pct, col, df
                ),
            }

            # Check correlation between missingness and target
            if target_col and target_col != col and target_col in df.columns:
                missing_mask = df[col].isnull()
                if df[target_col].dtype in ["float64", "int64"]:
                    mean_when_missing = df.loc[missing_mask, target_col].mean()
                    mean_when_present = df.loc[~missing_mask, target_col].mean()
                    if pd.notna(mean_when_missing) and pd.notna(mean_when_present):
                        diff_pct = (
                            abs(mean_when_missing - mean_when_present)
                            / abs(mean_when_present)
                            * 100
                        )
                        if diff_pct > 10:
                            entry["recommendation"] += (
                                f" WARNING: target mean differs by {diff_pct:.0f}% "
                                f"when {col} is missing — possible MNAR."
                            )

            report[col] = entry
        return report

    @staticmethod
    def _recommend(pct: float, col: str, df: pd.DataFrame) -> str:
        if pct > 50:
            return "DROP COLUMN — more than half the data is missing"
        elif pct > 20:
            return "INVESTIGATE — high missingness may indicate a systematic issue"
        elif df[col].dtype in ["float64", "int64"]:
            return "IMPUTE — use median for skewed, mean for normal distributions"
        else:
            return "IMPUTE — use mode or create an explicit 'Unknown' category"

    @staticmethod
    def impute_numeric(
        df: pd.DataFrame, col: str, strategy: str = "median"
    ) -> pd.DataFrame:
        """Fill missing numeric values. Adds a boolean indicator column."""
        indicator_col = f"{col}_was_missing"
        df[indicator_col] = df[col].isnull().astype(int)

        if strategy == "median":
            fill_value = df[col].median()
        elif strategy == "mean":
            fill_value = df[col].mean()
        elif strategy == "zero":
            fill_value = 0
        else:
            raise ValueError(f"Unknown strategy: {strategy}")

        df[col] = df[col].fillna(fill_value)
        print(f"{col}: imputed {df[indicator_col].sum()} values with {strategy} ({fill_value:.2f})")
        return df

    @staticmethod
    def impute_categorical(
        df: pd.DataFrame, col: str, strategy: str = "mode"
    ) -> pd.DataFrame:
        """Fill missing categorical values."""
        indicator_col = f"{col}_was_missing"
        df[indicator_col] = df[col].isnull().astype(int)

        if strategy == "mode":
            fill_value = df[col].mode()[0]
        elif strategy == "unknown":
            fill_value = "Unknown"
        else:
            raise ValueError(f"Unknown strategy: {strategy}")

        df[col] = df[col].fillna(fill_value)
        print(f"{col}: imputed {df[indicator_col].sum()} values with '{fill_value}'")
        return df

import pandas as pd
from difflib import SequenceMatcher


class DuplicateHandler:
    """Detect and remove duplicate records — exact and fuzzy.

    Usage:
        handler = DuplicateHandler()
        dups = handler.find_exact_duplicates(df, subset=['email', 'order_date'])
        df_clean = handler.remove_exact_duplicates(df, subset=['email', 'order_date'])
        fuzzy = handler.find_fuzzy_duplicates(df, columns=['name', 'address'])
    """

    @staticmethod
    def find_exact_duplicates(
        df: pd.DataFrame, subset: list = None
    ) -> pd.DataFrame:
        """Identify exact duplicate rows.

        Args:
            df: input DataFrame
            subset: columns to consider for duplication. None = all columns.

        Returns:
            DataFrame containing only the duplicate rows (excluding first occurrence).
        """
        dup_mask = df.duplicated(subset=subset, keep="first")
        dup_count = dup_mask.sum()
        print(f"Exact duplicates found: {dup_count} ({dup_count / len(df) * 100:.1f}%)")
        if subset:
            print(f"  Checked on columns: {subset}")
        return df[dup_mask]

    @staticmethod
    def remove_exact_duplicates(
        df: pd.DataFrame, subset: list = None, keep: str = "first"
    ) -> pd.DataFrame:
        """Remove exact duplicates, keeping one copy."""
        before = len(df)
        df_clean = df.drop_duplicates(subset=subset, keep=keep).reset_index(
            drop=True
        )
        removed = before - len(df_clean)
        print(
            f"Removed {removed} duplicate rows. {len(df_clean):,} rows remaining."
        )
        return df_clean

    @staticmethod
    def find_fuzzy_duplicates(
        df: pd.DataFrame, columns: list, threshold: float = 0.85
    ) -> list:
        """Find near-duplicate records based on string similarity.

        WARNING: O(n^2) complexity. For large datasets (>10k rows), sample
        or use blocking strategies (group by first letter, zip code, etc.)
        before calling this method.

        Args:
            df: input DataFrame
            columns: columns to concatenate for similarity comparison
            threshold: minimum similarity ratio to flag as fuzzy duplicate

        Returns:
            List of (index_i, index_j, similarity_score) tuples.
        """
        fuzzy_pairs = []
        values = df[columns].astype(str).agg(" ".join, axis=1).tolist()

        n = len(values)
        if n > 10_000:
            print(
                f"WARNING: {n:,} rows will produce {n*(n-1)//2:,} comparisons. "
                f"Consider blocking or sampling first."
            )

        for i in range(n):
            for j in range(i + 1, n):
                similarity = SequenceMatcher(None, values[i], values[j]).ratio()
                if similarity >= threshold and similarity < 1.0:
                    fuzzy_pairs.append((i, j, round(similarity, 3)))

        print(
            f"Found {len(fuzzy_pairs)} fuzzy duplicate pairs above "
            f"{threshold} threshold"
        )
        return fuzzy_pairs

import pandas as pd
import numpy as np


class TypeFixer:
    """Fix data type issues in DataFrames.

    Usage:
        fixer = TypeFixer()
        df = fixer.fix_numeric_strings(df, ['revenue', 'price'])
        df = fixer.fix_dates(df, ['order_date', 'ship_date'])
        df = fixer.fix_booleans(df, ['is_active', 'has_subscription'])
        df = fixer.fix_categoricals(df, ['country', 'status'])
    """

    @staticmethod
    def fix_numeric_strings(df: pd.DataFrame, columns: list) -> pd.DataFrame:
        """Convert string columns that should be numeric.

        Handles currency symbols, commas, whitespace, and common
        placeholder strings like 'N/A', '-', 'null', '#REF!'.
        """
        for col in columns:
            original_nulls = df[col].isna().sum()
            cleaned = df[col].astype(str)
            cleaned = cleaned.str.replace(r"[\$\€\£\,]", "", regex=True)
            cleaned = cleaned.str.strip()
            cleaned = cleaned.replace(
                {"": np.nan, "N/A": np.nan, "n/a": np.nan,
                 "-": np.nan, "null": np.nan, "None": np.nan,
                 "nan": np.nan, "#REF!": np.nan, "#N/A": np.nan}
            )
            df[col] = pd.to_numeric(cleaned, errors="coerce")
            new_nulls = df[col].isna().sum() - original_nulls
            print(
                f"{col}: converted to numeric. "
                f"{new_nulls} new NaN values from failed coercion."
            )
        return df

    @staticmethod
    def fix_dates(
        df: pd.DataFrame, columns: list, dayfirst: bool = False,
        fmt: str = None
    ) -> pd.DataFrame:
        """Parse date columns with explicit format handling.

        Args:
            fmt: explicit strftime format string (e.g., '%Y-%m-%d').
                 If None, pandas infers the format — risky in production.
        """
        for col in columns:
            if fmt:
                df[col] = pd.to_datetime(df[col], format=fmt, errors="coerce")
            else:
                df[col] = pd.to_datetime(
                    df[col], dayfirst=dayfirst, errors="coerce"
                )
            failed = df[col].isna().sum()
            print(f"{col}: parsed as datetime. {failed} values are NaT.")
            # Sanity check: flag future dates
            future = (df[col] > pd.Timestamp.now()).sum()
            if future > 0:
                print(f"  WARNING: {future} dates are in the future.")
        return df

    @staticmethod
    def fix_booleans(df: pd.DataFrame, columns: list) -> pd.DataFrame:
        """Convert string booleans to actual booleans."""
        true_values = {"yes", "true", "1", "y", "on", "t"}
        false_values = {"no", "false", "0", "n", "off", "f"}

        for col in columns:
            lower = df[col].astype(str).str.lower().str.strip()
            df[col] = lower.map(
                lambda x: (
                    True if x in true_values
                    else (False if x in false_values else np.nan)
                )
            )
            unmapped = df[col].isna().sum()
            print(f"{col}: converted to boolean. {unmapped} unmapped values.")
        return df

    @staticmethod
    def fix_categoricals(df: pd.DataFrame, columns: list) -> pd.DataFrame:
        """Convert high-memory object columns to categorical dtype."""
        for col in columns:
            before_mem = df[col].memory_usage(deep=True) / 1e6
            df[col] = df[col].astype("category")
            after_mem = df[col].memory_usage(deep=True) / 1e6
            if before_mem > 0:
                reduction = (1 - after_mem / before_mem) * 100
                print(
                    f"{col}: {before_mem:.2f} MB -> {after_mem:.2f} MB "
                    f"({reduction:.0f}% reduction)"
                )
            else:
                print(f"{col}: converted to category dtype.")
        return df

import pandas as pd
import numpy as np


class OutlierHandler:
    """Detect and handle outliers using multiple methods.

    Usage:
        handler = OutlierHandler()
        mask, lower, upper = handler.detect_iqr(df, 'salary')
        mask = handler.detect_zscore(df, 'salary', threshold=3.0)
        df = handler.cap_outliers(df, 'salary', lower=20000, upper=500000)
        df = handler.log_transform(df, 'revenue')
    """

    @staticmethod
    def detect_iqr(
        df: pd.DataFrame, col: str, factor: float = 1.5
    ) -> tuple:
        """Flag outliers using the IQR method.

        Returns:
            Tuple of (boolean mask, lower bound, upper bound).
        """
        q1 = df[col].quantile(0.25)
        q3 = df[col].quantile(0.75)
        iqr = q3 - q1
        lower = q1 - factor * iqr
        upper = q3 + factor * iqr

        outlier_mask = (df[col] < lower) | (df[col] > upper)
        count = outlier_mask.sum()
        print(f"{col}: {count} outliers ({count/len(df)*100:.1f}%) outside [{lower:.2f}, {upper:.2f}]")
        return outlier_mask, lower, upper

    @staticmethod
    def detect_zscore(
        df: pd.DataFrame, col: str, threshold: float = 3.0
    ) -> pd.Series:
        """Flag outliers using z-score method."""
        mean = df[col].mean()
        std = df[col].std()
        if std == 0:
            print(f"{col}: standard deviation is 0 — all values are identical. No outliers.")
            return pd.Series(False, index=df.index)

        z_scores = np.abs((df[col] - mean) / std)
        outlier_mask = z_scores > threshold
        count = outlier_mask.sum()
        print(f"{col}: {count} outliers with |z-score| > {threshold}")
        return outlier_mask

    @staticmethod
    def cap_outliers(
        df: pd.DataFrame, col: str,
        lower: float = None, upper: float = None
    ) -> pd.DataFrame:
        """Cap outlier values to specified bounds (winsorization).

        If bounds are not specified, defaults to 1st and 99th percentiles.
        """
        if lower is None:
            lower = df[col].quantile(0.01)
        if upper is None:
            upper = df[col].quantile(0.99)

        before = df[col].copy()
        df[col] = df[col].clip(lower=lower, upper=upper)
        capped = (before != df[col]).sum()
        print(f"{col}: capped {capped} values to [{lower:.2f}, {upper:.2f}]")
        return df

    @staticmethod
    def log_transform(df: pd.DataFrame, col: str) -> pd.DataFrame:
        """Apply log1p transform to compress right-skewed distributions.

        Handles zeros safely. Does NOT handle negative values — shift first.
        """
        if (df[col] < 0).any():
            min_val = df[col].min()
            print(f"{col}: contains negative values. Shifting by {abs(min_val):.2f} before log transform.")
            df[col] = np.log1p(df[col] - min_val)
        else:
            df[col] = np.log1p(df[col])
        print(f"{col}: log1p transform applied.")
        return df

import pandas as pd
import numpy as np
from sklearn.preprocessing import LabelEncoder, OneHotEncoder


class CategoricalEncoder:
    """Encode categorical variables for model consumption.

    Usage:
        encoder = CategoricalEncoder()
        df = encoder.encode_onehot(df, ['color', 'size'])
        df = encoder.encode_ordinal(df, 'education', ['high_school', 'bachelors', 'masters', 'phd'])
        df = encoder.encode_frequency(df, ['city', 'product_id'])
    """

    @staticmethod
    def encode_onehot(
        df: pd.DataFrame, columns: list, max_cardinality: int = 10
    ) -> pd.DataFrame:
        """One-hot encode low-cardinality columns.

        Skips columns that exceed max_cardinality and prints a warning.
        Uses drop_first=True to avoid multicollinearity in linear models.
        """
        for col in columns:
            cardinality = df[col].nunique()
            if cardinality > max_cardinality:
                print(
                    f"SKIPPED: {col} has {cardinality} unique values. "
                    f"Use frequency or target encoding instead."
                )
                continue

            dummies = pd.get_dummies(df[col], prefix=col, drop_first=True)
            df = pd.concat([df.drop(col, axis=1), dummies], axis=1)
            print(f"{col}: one-hot encoded into {len(dummies.columns)} columns")
        return df

    @staticmethod
    def encode_ordinal(
        df: pd.DataFrame, column: str, order: list
    ) -> pd.DataFrame:
        """Encode ordered categories (e.g., Low < Medium < High).

        Unmapped values are set to NaN so they surface as missing rather
        than silently receiving an arbitrary integer.
        """
        mapping = {val: idx for idx, val in enumerate(order)}
        unmapped = set(df[column].dropna().unique()) - set(mapping.keys())
        if unmapped:
            print(f"WARNING: {column} has unmapped values: {unmapped}")
        df[column] = df[column].map(mapping)
        print(f"{column}: ordinally encoded with order {order}")
        return df

    @staticmethod
    def encode_frequency(
        df: pd.DataFrame, columns: list
    ) -> pd.DataFrame:
        """Replace categories with their frequency in the training data.

        Safe for high-cardinality features. Does not create new columns.
        """
        for col in columns:
            freq_map = df[col].value_counts(normalize=True).to_dict()
            df[col] = df[col].map(freq_map)
            print(f"{col}: frequency encoded ({len(freq_map)} categories)")
        return df

    @staticmethod
    def encode_label(
        df: pd.DataFrame, columns: list
    ) -> tuple:
        """Label encode — safe for tree-based models only.

        WARNING: creates false ordinal relationships for linear models.
        Returns the DataFrame and a dict of fitted LabelEncoders for
        inverse transformation and production reuse.
        """
        encoders = {}
        for col in columns:
            le = LabelEncoder()
            df[col] = le.fit_transform(df[col].astype(str))
            encoders[col] = le
            print(f"{col}: label encoded ({len(le.classes_)} classes)")
        return df, encoders

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler


class FeatureScaler:
    """Scale numeric features for model consumption.

    Usage:
        scaler_util = FeatureScaler()
        df_train, scaler = scaler_util.scale_standard(df_train, num_cols)
        df_test[num_cols] = scaler.transform(df_test[num_cols])  # same scaler!
    """

    @staticmethod
    def scale_standard(
        df: pd.DataFrame, columns: list
    ) -> tuple:
        """Z-score normalization: mean=0, std=1.

        Use for normally distributed features with linear models,
        logistic regression, SVMs, and PCA.
        """
        scaler = StandardScaler()
        df[columns] = scaler.fit_transform(df[columns])
        print(f"StandardScaler applied to {len(columns)} columns")
        return df, scaler

    @staticmethod
    def scale_minmax(
        df: pd.DataFrame, columns: list,
        feature_range: tuple = (0, 1)
    ) -> tuple:
        """Min-max scaling to a fixed range.

        Use for neural networks (which expect bounded inputs) and
        features with known theoretical bounds.
        """
        scaler = MinMaxScaler(feature_range=feature_range)
        df[columns] = scaler.fit_transform(df[columns])
        print(
            f"MinMaxScaler applied to {len(columns)} columns, "
            f"range={feature_range}"
        )
        return df, scaler

    @staticmethod
    def scale_robust(df: pd.DataFrame, columns: list) -> tuple:
        """Scale using median and IQR instead of mean and std.

        Use when data has outliers that would distort StandardScaler.
        Outliers still exist after scaling but no longer dominate the range.
        """
        scaler = RobustScaler()
        df[columns] = scaler.fit_transform(df[columns])
        print(
            f"RobustScaler applied to {len(columns)} columns "
            f"(outlier-resistant)"
        )
        return df, scaler

import pandas as pd
import numpy as np
import joblib
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer


class DataCleaningPipeline:
    """Reproducible data cleaning pipeline for production use.

    Encapsulates imputation, encoding, and scaling into a single
    serializable object. Fit once on training data, transform everything
    else with the same fitted parameters.

    Usage:
        pipeline = DataCleaningPipeline(
            numeric_cols=['age', 'income', 'score'],
            categorical_cols=['city', 'status', 'tier']
        )
        X_train_clean = pipeline.fit_transform(X_train)
        X_test_clean = pipeline.transform(X_test)
        pipeline.save('preprocessing_pipeline.pkl')

        # At inference time:
        loaded = DataCleaningPipeline.load('preprocessing_pipeline.pkl')
        X_new_clean = loaded.transform(X_new)
    """

    def __init__(self, numeric_cols: list, categorical_cols: list):
        self.numeric_cols = numeric_cols
        self.categorical_cols = categorical_cols
        self.pipeline = self._build_pipeline()
        self._is_fitted = False

    def _build_pipeline(self) -> ColumnTransformer:
        numeric_transformer = Pipeline(steps=[
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler()),
        ])

        categorical_transformer = Pipeline(steps=[
            ("imputer", SimpleImputer(strategy="constant", fill_value="Unknown")),
            ("encoder", OneHotEncoder(
                handle_unknown="ignore", sparse_output=False
            )),
        ])

        preprocessor = ColumnTransformer(transformers=[
            ("numeric", numeric_transformer, self.numeric_cols),
            ("categorical", categorical_transformer, self.categorical_cols),
        ])

        return preprocessor

    def fit(self, X_train: pd.DataFrame) -> "DataCleaningPipeline":
        """Fit all transformers on training data only."""
        self.pipeline.fit(X_train)
        self._is_fitted = True
        print(f"Pipeline fitted on {len(X_train):,} rows, "
              f"{len(self.numeric_cols)} numeric + "
              f"{len(self.categorical_cols)} categorical columns.")
        return self

    def transform(self, X: pd.DataFrame) -> np.ndarray:
        """Transform new data using the fitted pipeline."""
        if not self._is_fitted:
            raise RuntimeError(
                "Pipeline is not fitted. Call .fit(X_train) first."
            )
        return self.pipeline.transform(X)

    def fit_transform(self, X_train: pd.DataFrame) -> np.ndarray:
        """Fit and transform in one step. Use on training data."""
        self.fit(X_train)
        return self.pipeline.transform(X_train)

    def save(self, filepath: str) -> None:
        """Serialize the fitted pipeline to disk."""
        if not self._is_fitted:
            raise RuntimeError(
                "Cannot save an unfitted pipeline. Call .fit() first."
            )
        joblib.dump(self, filepath)
        print(f"Pipeline saved to {filepath}")

    @staticmethod
    def load(filepath: str) -> "DataCleaningPipeline":
        """Load a previously saved pipeline."""
        pipeline = joblib.load(filepath)
        print(f"Pipeline loaded from {filepath}")
        return pipeline