data-cleaning-pipeline-generator

What this skill does

# Data Cleaning Pipeline Generator

Generates comprehensive data cleaning and preprocessing pipelines using pandas, polars, or PySpark with best practices for handling messy data.

## When to Use

- "Clean my dataset"
- "Generate data cleaning pipeline"
- "Handle missing values"
- "Remove duplicates"
- "Fix data types"
- "Detect and remove outliers"

## Instructions

### 1. Analyze Dataset

```python
import pandas as pd

# Load data
df = pd.read_csv('data.csv')

# Basic info
print(df.info())
print(df.describe())
print(df.head())

# Check for issues
print("\nMissing values:")
print(df.isnull().sum())

print("\nDuplicates:")
print(f"Total duplicates: {df.duplicated().sum()}")

print("\nData types:")
print(df.dtypes)
```

### 2. Generate Pandas Cleaning Pipeline

**Complete Pipeline:**
```python
import pandas as pd
import numpy as np
from datetime import datetime

class DataCleaningPipeline:
    """Data cleaning pipeline for pandas DataFrames."""

    def __init__(self, df: pd.DataFrame):
        self.df = df.copy()
        self.original_shape = df.shape
        self.cleaning_log = []

    def log(self, message: str):
        """Log cleaning steps."""
        self.cleaning_log.append(f"[{datetime.now()}] {message}")
        print(message)

    def remove_duplicates(self, subset=None, keep='first'):
        """Remove duplicate rows."""
        before = len(self.df)
        self.df = self.df.drop_duplicates(subset=subset, keep=keep)
        removed = before - len(self.df)
        self.log(f"Removed {removed} duplicate rows")
        return self

    def handle_missing_values(self, strategy='auto'):
        """Handle missing values based on strategy."""
        missing = self.df.isnull().sum()
        columns_with_missing = missing[missing > 0]

        for col in columns_with_missing.index:
            missing_pct = (missing[col] / len(self.df)) * 100

            if missing_pct > 50:
                self.log(f"Dropping column '{col}' ({missing_pct:.1f}% missing)")
                self.df = self.df.drop(columns=[col])
                continue

            if self.df[col].dtype in ['int64', 'float64']:
                # Numeric columns
                if strategy == 'mean':
                    fill_value = self.df[col].mean()
                elif strategy == 'median':
                    fill_value = self.df[col].median()
                else:  # auto
                    fill_value = self.df[col].median()
                self.df[col] = self.df[col].fillna(fill_value)
                self.log(f"Filled '{col}' with {strategy}: {fill_value:.2f}")
            else:
                # Categorical columns
                if strategy == 'mode':
                    fill_value = self.df[col].mode()[0]
                else:  # auto
                    fill_value = 'Unknown'
                self.df[col] = self.df[col].fillna(fill_value)
                self.log(f"Filled '{col}' with: {fill_value}")

        return self

    def fix_data_types(self, type_mapping=None):
        """Convert columns to appropriate data types."""
        if type_mapping is None:
            type_mapping = {}

        for col in self.df.columns:
            if col in type_mapping:
                try:
                    self.df[col] = self.df[col].astype(type_mapping[col])
                    self.log(f"Converted '{col}' to {type_mapping[col]}")
                except Exception as e:
                    self.log(f"Failed to convert '{col}': {e}")
            else:
                # Auto-detect dates
                if 'date' in col.lower() or 'time' in col.lower():
                    try:
                        self.df[col] = pd.to_datetime(self.df[col])
                        self.log(f"Converted '{col}' to datetime")
                    except:
                        pass

        return self

    def remove_outliers(self, columns=None, method='iqr', threshold=1.5):
        """Remove outliers using IQR or Z-score method."""
        if columns is None:
            columns = self.df.select_dtypes(include=[np.number]).columns

        before = len(self.df)

        for col in columns:
            if method == 'iqr':
                Q1 = self.df[col].quantile(0.25)
                Q3 = self.df[col].quantile(0.75)
                IQR = Q3 - Q1
                lower = Q1 - threshold * IQR
                upper = Q3 + threshold * IQR
                mask = (self.df[col] >= lower) & (self.df[col] <= upper)
            else:  # z-score
                z_scores = np.abs((self.df[col] - self.df[col].mean()) / self.df[col].std())
                mask = z_scores < threshold

            self.df = self.df[mask]

        removed = before - len(self.df)
        self.log(f"Removed {removed} outlier rows using {method} method")
        return self

    def normalize_text(self, columns=None):
        """Normalize text columns (lowercase, strip whitespace)."""
        if columns is None:
            columns = self.df.select_dtypes(include=['object']).columns

        for col in columns:
            self.df[col] = self.df[col].str.strip().str.lower()
            self.log(f"Normalized text in '{col}'")

        return self

    def encode_categorical(self, columns=None, method='label'):
        """Encode categorical variables."""
        if columns is None:
            columns = self.df.select_dtypes(include=['object']).columns

        for col in columns:
            if method == 'label':
                self.df[col] = pd.Categorical(self.df[col]).codes
                self.log(f"Label encoded '{col}'")
            elif method == 'onehot':
                dummies = pd.get_dummies(self.df[col], prefix=col)
                self.df = pd.concat([self.df.drop(columns=[col]), dummies], axis=1)
                self.log(f"One-hot encoded '{col}'")

        return self

    def validate_ranges(self, range_checks):
        """Validate numeric columns are within expected ranges."""
        for col, (min_val, max_val) in range_checks.items():
            invalid = ((self.df[col] < min_val) | (self.df[col] > max_val)).sum()
            if invalid > 0:
                self.log(f"WARNING: {invalid} values in '{col}' outside range [{min_val}, {max_val}]")
                # Remove invalid rows
                self.df = self.df[(self.df[col] >= min_val) & (self.df[col] <= max_val)]

        return self

    def generate_report(self):
        """Generate cleaning report."""
        report = f"""
Data Cleaning Report
====================

Original Shape: {self.original_shape}
Final Shape: {self.df.shape}
Rows Removed: {self.original_shape[0] - self.df.shape[0]}
Columns Removed: {self.original_shape[1] - self.df.shape[1]}

Cleaning Steps:
"""
        for step in self.cleaning_log:
            report += f"  - {step}\n"

        return report

    def get_cleaned_data(self):
        """Return cleaned DataFrame."""
        return self.df


# Usage
pipeline = DataCleaningPipeline(df)
cleaned_df = (
    pipeline
    .remove_duplicates()
    .handle_missing_values(strategy='auto')
    .fix_data_types()
    .remove_outliers(method='iqr', threshold=1.5)
    .normalize_text()
    .validate_ranges({'age': (0, 120), 'price': (0, 1000000)})
    .get_cleaned_data()
)

print(pipeline.generate_report())
cleaned_df.to_csv('cleaned_data.csv', index=False)
```

### 3. Polars Pipeline (Faster for Large Data)

```python
import polars as pl

# Load data
df = pl.read_csv('data.csv')

# Cleaning pipeline
cleaned_df = (
    df
    # Remove duplicates
    .unique()
    # Handle missing values
    .with_columns([
        pl.col('age').fill_null(pl.col('age').median()),
        pl.col('name').fill_null('Unknown'),
    ])
    # Fix data types
    .with_columns([
        pl.col('date').str.strptime(pl.Date, '%Y-%m-%d'),
        pl.col('amount').cast(pl.Float64),
    ])
    # Remove outliers
    .filter(
        (pl.col('age') >= 0) & (pl.col('age') <= 120)
    )
    # Normalize text
    .with_columns([
        pl.col('name').str.to_lowercase().str.str