Feature Scaling: Standardization, Normalization, and Robust Scaling

import numpy as np
import pandas as pd

# Create sample data with different scales and outliers
np.random.seed(42)
data = {
    'income': [50000, 60000, 55000, 70000, 45000, 58000, 62000, 48000, 75000, 52000, 1000000],  # Outlier at end
    'age': [25, 30, 28, 35, 22, 29, 32, 26, 38, 27, 40],
    'score': [7.5, 8.2, 7.8, 9.1, 6.9, 8.0, 8.5, 7.2, 9.5, 7.6, 8.8]
}
df = pd.DataFrame(data)

print("Original Dataset:")
print(df)
print("
Summary Statistics:")
print(df.describe())

# 1. STANDARDIZATION (Z-score) - Manual
def standardize_manual(series):
    """
    Manual Z-score standardization: (x - mean) / std
    Result: mean=0, std=1
    """
    mean_val = np.sum(series) / len(series)
    variance = np.sum((series - mean_val) ** 2) / (len(series) - 1)  # Sample variance
    std_val = np.sqrt(variance)
    
    standardized = (series - mean_val) / std_val
    
    return standardized, mean_val, std_val

print("
=== 1. STANDARDIZATION (Z-score) - Manual ===")
df_std = df.copy()
for col in df.columns:
    standardized, mean_val, std_val = standardize_manual(df[col].values)
    df_std[f'{col}_std'] = standardized
    print(f"
{col}:")
    print(f"  Mean: {mean_val:.2f}, Std: {std_val:.2f}")
    print(f"  Standardized mean: {np.mean(standardized):.6f}")
    print(f"  Standardized std: {np.std(standardized, ddof=1):.6f}")

print("
Standardized data:")
print(df_std[[c for c in df_std.columns if '_std' in c]])

# 2. MIN-MAX NORMALIZATION - Manual
def min_max_scale_manual(series, feature_range=(0, 1)):
    """
    Manual Min-Max scaling: (x - min) / (max - min) * (max_range - min_range) + min_range
    Result: scales to [feature_range[0], feature_range[1]]
    """
    min_val = np.min(series)
    max_val = np.max(series)
    range_val = max_val - min_val
    
    min_range, max_range = feature_range
    
    normalized = (series - min_val) / range_val * (max_range - min_range) + min_range
    
    return normalized, min_val, max_val

print("
=== 2. MIN-MAX NORMALIZATION (0-1) - Manual ===")
df_minmax = df.copy()
for col in df.columns:
    normalized, min_val, max_val = min_max_scale_manual(df[col].values)
    df_minmax[f'{col}_norm'] = normalized
    print(f"
{col}:")
    print(f"  Min: {min_val:.2f}, Max: {max_val:.2f}")
    print(f"  Normalized min: {np.min(normalized):.6f}")
    print(f"  Normalized max: {np.max(normalized):.6f}")

print("
Normalized data (0-1):")
print(df_minmax[[c for c in df_minmax.columns if '_norm' in c]])

# 3. ROBUST SCALING - Manual
def robust_scale_manual(series):
    """
    Manual Robust scaling: (x - median) / IQR
    Uses median and interquartile range (resistant to outliers)
    """
    # Calculate median
    sorted_vals = np.sort(series)
    n = len(sorted_vals)
    if n % 2 == 0:
        median_val = (sorted_vals[n//2 - 1] + sorted_vals[n//2]) / 2
    else:
        median_val = sorted_vals[n//2]
    
    # Calculate Q1 (25th percentile) and Q3 (75th percentile)
    q1_idx = n // 4
    q3_idx = 3 * n // 4
    q1 = sorted_vals[q1_idx]
    q3 = sorted_vals[q3_idx]
    iqr = q3 - q1
    
    # Scale
    if iqr == 0:
        robust_scaled = series - median_val
    else:
        robust_scaled = (series - median_val) / iqr
    
    return robust_scaled, median_val, q1, q3, iqr

print("
=== 3. ROBUST SCALING - Manual ===")
df_robust = df.copy()
for col in df.columns:
    robust_scaled, median, q1, q3, iqr = robust_scale_manual(df[col].values)
    df_robust[f'{col}_robust'] = robust_scaled
    print(f"
{col}:")
    print(f"  Median: {median:.2f}, Q1: {q1:.2f}, Q3: {q3:.2f}, IQR: {iqr:.2f}")
    print(f"  Robust scaled median: {np.median(robust_scaled):.6f}")

print("
Robust scaled data:")
print(df_robust[[c for c in df_robust.columns if '_robust' in c]])

# 4. MAXABS SCALING - Manual
def maxabs_scale_manual(series):
    """
    Manual MaxAbs scaling: x / max(|x|)
    Preserves sparsity (0s stay 0)
    """
    max_abs = np.max(np.abs(series))
    scaled = series / max_abs
    return scaled, max_abs

print("
=== 4. MAXABS SCALING - Manual ===")
# Create sparse data for demonstration
sparse_data = [0, 0, 5, 0, 10, 0, 3, 0, 8, 0, 0]
maxabs_scaled, max_abs = maxabs_scale_manual(np.array(sparse_data))
print(f"Original: {sparse_data}")
print(f"MaxAbs scaled: {[round(x, 4) for x in maxabs_scaled]}")
print(f"Max absolute value: {max_abs}")
print(f"Zeros preserved: {sum(1 for x in maxabs_scaled if x == 0)} (was {sum(1 for x in sparse_data if x == 0)})")

# 5. UNIT VECTOR SCALING (L2 Normalization) - Manual
def unit_vector_scale_manual(series):
    """
    Manual Unit Vector scaling: x / ||x||_2
    Normalizes to unit length
    """
    l2_norm = np.sqrt(np.sum(series ** 2))
    scaled = series / l2_norm
    return scaled, l2_norm

print("
=== 5. UNIT VECTOR (L2) SCALING - Manual ===")
vector = np.array([3, 4])
unit_scaled, l2_norm = unit_vector_scale_manual(vector)
print(f"Original vector: {vector}")
print(f"L2 norm: {l2_norm}")
print(f"Unit vector: {unit_scaled}")
print(f"Verify length=1: {np.sqrt(np.sum(unit_scaled**2)):.6f}")

# 6. COMPARISON OF METHODS ON OUTLIER DATA
print("
=== 6. OUTLIER IMPACT COMPARISON ===")
col = 'income'
print(f"
Analyzing '{col}' with outlier ($1M income):")
print(f"Original: mean={df[col].mean():.0f}, std={df[col].std():.0f}")
print(f"  min={df[col].min():.0f}, max={df[col].max():.0f}")

# Standardization
std_vals = df_std[f'{col}_std']
print(f"
Standardized: mean={std_vals.mean():.4f}, std={std_vals.std():.4f}")
print(f"  min={std_vals.min():.2f}, max={std_vals.max():.2f} (outlier at {std_vals.iloc[-1]:.2f})")

# Min-Max
norm_vals = df_minmax[f'{col}_norm']
print(f"
Min-Max: min={norm_vals.min():.4f}, max={norm_vals.max():.4f}")
print(f"  All values squashed by outlier: most values near 0-0.05")

# Robust
robust_vals = df_robust[f'{col}_robust']
print(f"
Robust: median={np.median(robust_vals):.4f}")
print(f"  min={robust_vals.min():.2f}, max={robust_vals.max():.2f} (outlier at {robust_vals.iloc[-1]:.2f})")
print(f"  Regular values well-scaled despite outlier")

# 7. INVERSE TRANSFORM
print("
=== 7. INVERSE TRANSFORM ===")
def inverse_standardize(standardized, mean_val, std_val):
    """Convert standardized values back to original scale"""
    return standardized * std_val + mean_val

def inverse_minmax(normalized, min_val, max_val, feature_range=(0, 1)):
    """Convert normalized values back to original scale"""
    min_range, max_range = feature_range
    return (normalized - min_range) / (max_range - min_range) * (max_val - min_val) + min_val

# Test inverse
test_std = df_std['income_std'].iloc[0]
test_mean = df['income'].mean()
test_std_dev = df['income'].std(ddof=1)
original_recovered = inverse_standardize(test_std, test_mean, test_std_dev)
print(f"Original income[0]: {df['income'].iloc[0]}")
print(f"Standardized: {test_std:.4f}")
print(f"Recovered: {original_recovered:.2f}")

# 8. VISUALIZATION DATA
print("
=== 8. SCALING COMPARISON TABLE ===")
comparison = pd.DataFrame({
    'Original': df['income'],
    'Standardized': df_std['income_std'].round(4),
    'Min-Max': df_minmax['income_norm'].round(4),
    'Robust': df_robust['income_robust'].round(4)
})
print(comparison.to_string())

print("
=== KEY INSIGHTS ===")
print("1. Standardization: Best for normally distributed data, preserves outliers")
print("2. Min-Max: Best for bounded data, sensitive to outliers")
print("3. Robust: Best for data with outliers, uses median/IQR")
print("4. MaxAbs: Best for sparse data, preserves zeros")
print("5. Always scale train/test separately - fit on train, transform both!")

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler, MaxAbsScaler, Normalizer
from sklearn.preprocessing import QuantileTransformer, PowerTransformer
import matplotlib.pyplot as plt

# Create sample data
np.random.seed(42)
data = {
    'income': np.concatenate([[50000, 60000, 55000, 70000, 45000, 58000, 62000, 48000, 75000, 52000], 
                              np.random.normal(55000, 10000, 90)]),
    'age': np.concatenate([[25, 30, 28, 35, 22, 29, 32, 26, 38, 27], 
                           np.random.normal(30, 5, 90)]),
    'score': np.concatenate([[7.5, 8.2, 7.8, 9.1, 6.9, 8.0, 8.5, 7.2, 9.5, 7.6], 
                             np.random.normal(8.0, 0.8, 90)])
}
df = pd.DataFrame(data)

# Add outliers
df.loc[0, 'income'] = 500000  # Outlier
df.loc[1, 'age'] = 80  # Outlier

print("Dataset shape:", df.shape)
print("
Summary Statistics:")
print(df.describe())

# 1. STANDARD SCALER (Standardization)
print("
" + "="*60)
print("1. STANDARD SCALER (Z-score Standardization)")
print("="*60)

scaler = StandardScaler()
df_standardized = pd.DataFrame(
    scaler.fit_transform(df),
    columns=df.columns
)

print(f"Mean of scaled features: {df_standardized.mean().round(6).tolist()}")
print(f"Std of scaled features: {df_standardized.std().round(6).tolist()}")
print(f"
First 5 rows:")
print(df_standardized.head())

# Show scaler parameters
print(f"
Scaler parameters:")
print(f"  Means: {scaler.mean_}")
print(f"  Scales (std): {scaler.scale_}")

# 2. MIN-MAX SCALER
print("
" + "="*60)
print("2. MIN-MAX SCALER (Normalization)")
print("="*60)

# Default range [0, 1]
minmax_scaler = MinMaxScaler()
df_minmax = pd.DataFrame(
    minmax_scaler.fit_transform(df),
    columns=df.columns
)

print(f"Min of scaled features: {df_minmax.min().round(6).tolist()}")
print(f"Max of scaled features: {df_max.max().round(6).tolist()}")
print(f"
First 5 rows:")
print(df_minmax.head())

# Custom range [-1, 1]
minmax_custom = MinMaxScaler(feature_range=(-1, 1))
df_minmax_custom = pd.DataFrame(
    minmax_custom.fit_transform(df),
    columns=df.columns
)
print(f"
Custom range [-1, 1]:")
print(f"Min: {df_minmax_custom.min().round(2).tolist()}")
print(f"Max: {df_minmax_custom.max().round(2).tolist()}")

# 3. ROBUST SCALER
print("
" + "="*60)
print("3. ROBUST SCALER (Outlier-resistant)")
print("="*60)

robust_scaler = RobustScaler()
df_robust = pd.DataFrame(
    robust_scaler.fit_transform(df),
    columns=df.columns
)

print(f"Medians of scaled features: {df_robust.median().round(6).tolist()}")
print(f"IQRs of scaled features:")
for col in df_robust.columns:
    q1 = df_robust[col].quantile(0.25)
    q3 = df_robust[col].quantile(0.75)
    iqr = q3 - q1
    print(f"  {col}: IQR={iqr:.4f}")

print(f"
First 5 rows:")
print(df_robust.head())

# Compare outlier handling
print(f"
Outlier comparison (income):")
print(f"  Original: {df['income'].iloc[0]}")
print(f"  Standardized: {df_standardized['income'].iloc[0]:.2f}")
print(f"  Min-Max: {df_minmax['income'].iloc[0]:.4f}")
print(f"  Robust: {df_robust['income'].iloc[0]:.2f}")

# 4. MAX ABS SCALER
print("
" + "="*60)
print("4. MAX ABS SCALER (Sparse data)")
print("="*60)

# Create sparse data
sparse_df = pd.DataFrame({
    'feature1': [0, 0, 5, 0, 10, 0, 3, 0, 8, 0] * 10,
    'feature2': [0, 2, 0, 4, 0, 6, 0, 8, 0, 10] * 10
})

maxabs_scaler = MaxAbsScaler()
df_maxabs = pd.DataFrame(
    maxabs_scaler.fit_transform(sparse_df),
    columns=sparse_df.columns
)

print(f"Original sparsity (zeros): {(sparse_df == 0).sum().sum()} / {sparse_df.size}")
print(f"After MaxAbs sparsity: {(df_maxabs == 0).sum().sum()} / {df_maxabs.size}")
print(f"MaxAbs preserved sparsity: {(df_maxabs == 0).sum().sum() == (sparse_df == 0).sum().sum()}")

# 5. NORMALIZER (Row-wise unit vector)
print("
" + "="*60)
print("5. NORMALIZER (Row-wise L2 normalization)")
print("="*60)

normalizer = Normalizer(norm='l2')
df_normalized = pd.DataFrame(
    normalizer.fit_transform(df),
    columns=df.columns
)

# Verify unit vectors
row_norms = np.sqrt((df_normalized ** 2).sum(axis=1))
print(f"Row L2 norms (should be 1.0): {row_norms[:5].round(6).tolist()}")
print(f"
First 5 rows (each row is unit length):")
print(df_normalized.head())

# L1 normalization
normalizer_l1 = Normalizer(norm='l1')
df_l1 = pd.DataFrame(
    normalizer_l1.fit_transform(df),
    columns=df.columns
)
row_sums = df_l1.sum(axis=1)
print(f"
L1 normalized row sums: {row_sums[:5].round(6).tolist()}")

# 6. QUANTILE TRANSFORMER (Non-linear)
print("
" + "="*60)
print("6. QUANTILE TRANSFORMER (Uniform/Gaussian)")
print("="*60)

# Transform to uniform distribution
quantile_uni = QuantileTransformer(output_distribution='uniform', random_state=42)
df_quantile_uni = pd.DataFrame(
    quantile_uni.fit_transform(df),
    columns=df.columns
)

print("Uniform distribution (approx):")
print(f"  Min: {df_quantile_uni.min().round(3).tolist()}")
print(f"  Max: {df_quantile_uni.max().round(3).tolist()}")

# Transform to Gaussian
quantile_norm = QuantileTransformer(output_distribution='normal', random_state=42)
df_quantile_norm = pd.DataFrame(
    quantile_norm.fit_transform(df),
    columns=df.columns
)

print(f"
Gaussian distribution (approx):")
print(f"  Mean: {df_quantile_norm.mean().round(3).tolist()}")
print(f"  Std: {df_quantile_norm.std().round(3).tolist()}")

# 7. POWER TRANSFORMER (Box-Cox / Yeo-Johnson)
print("
" + "="*60)
print("7. POWER TRANSFORMER (Box-Cox / Yeo-Johnson)")
print("="*60)

# Yeo-Johnson (works with negative values)
power_yj = PowerTransformer(method='yeo-johnson')
df_power_yj = pd.DataFrame(
    power_yj.fit_transform(df),
    columns=df.columns
)

print("Yeo-Johnson transformed:")
print(f"  Mean: {df_power_yj.mean().round(3).tolist()}")
print(f"  Std: {df_power_yj.std().round(3).tolist()}")

# Positive values only for Box-Cox
df_positive = df[['income', 'score']].abs() + 1  # Make positive
power_bc = PowerTransformer(method='box-cox')
df_power_bc = pd.DataFrame(
    power_bc.fit_transform(df_positive),
    columns=['income_bc', 'score_bc']
)
print(f"
Box-Cox (positive data only):")
print(f"  Lambda (income): {power_bc.lambdas_[0]:.3f}")

# 8. PIPELINE INTEGRATION
print("
" + "="*60)
print("8. SKLEARN PIPELINE INTEGRATION")
print("="*60)

from sklearn.pipeline import Pipeline
from sklearn.linear_model import Ridge
from sklearn.model_selection import train_test_split

# Split data
X = df[['income', 'age']]
y = df['score']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Create pipeline with scaling
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('regressor', Ridge(alpha=1.0))
])

# Fit and evaluate
pipeline.fit(X_train, y_train)
score = pipeline.score(X_test, y_test)
print(f"Pipeline R² score: {score:.4f}")

# Access fitted scaler
fitted_scaler = pipeline.named_steps['scaler']
print(f"
Fitted scaler means: {fitted_scaler.mean_}")
print(f"Fitted scaler scales: {fitted_scaler.scale_}")

# 9. INVERSE TRANSFORM
print("
" + "="*60)
print("9. INVERSE TRANSFORM")
print("="*60)

# Transform
X_train_scaled = scaler.fit_transform(X_train)
print(f"Original X_train[0]: {X_train.iloc[0].values}")
print(f"Scaled X_train[0]: {X_train_scaled[0]}")

# Inverse transform
X_recovered = scaler.inverse_transform(X_train_scaled)
print(f"Recovered X_train[0]: {X_recovered[0]}")
print(f"Recovery error: {np.abs(X_train.iloc[0].values - X_recovered[0]).max():.10f}")

# 10. COMPARISON VISUALIZATION
print("
" + "="*60)
print("10. SCALING METHOD COMPARISON")
print("="*60)

comparison_data = {
    'Method': ['StandardScaler', 'MinMaxScaler', 'RobustScaler', 'MaxAbsScaler', 'Normalizer'],
    'Center': ['Mean=0', 'Min=0', 'Median=0', 'No', 'No'],
    'Scale': ['Std=1', 'Max=1', 'IQR=1', 'MaxAbs=1', 'Unit vector'],
    'Outlier Robust': ['No', 'No', 'Yes', 'Yes', 'No'],
    'Preserves Sparse': ['No', 'No', 'No', 'Yes', 'No'],
    'Output Range': ['(-∞, ∞)', '[0, 1]', '(-∞, ∞)', '[-1, 1]', '[-1, 1]'],
    'Best For': [
        'Normal dist, linear models',
        'Neural networks, bounded inputs',
        'Data with outliers',
        'Sparse data',
        'Text/NLP, row-wise normalization'
    ]
}

comparison_df = pd.DataFrame(comparison_data)
print(comparison_df.to_string(index=False))

# 11. BEST PRACTICES
print("
" + "="*60)
print("11. BEST PRACTICES & COMMON MISTAKES")
print("="*60)

print("DO:")
print("  ✓ Fit scaler ONLY on training data")
print("  ✓ Apply same scaler to train, validation, and test")
print("  ✓ Use RobustScaler if data has outliers")
print("  ✓ Scale features before regularization (L1/L2)")
print("  ✓ Consider QuantileTransformer for highly skewed data")

print("
DON'T:")
print("  ✗ Fit scaler on full dataset (data leakage!)")
print("  ✗ Use StandardScaler on data with extreme outliers")
print("  ✗ Scale target variable for most algorithms")
print("  ✗ Forget to handle new categories in production")
print("  ✗ Apply different scalers to train and test")

# 12. HANDLING NEW DATA IN PRODUCTION
print("
" + "="*60)
print("12. PRODUCTION PATTERN")
print("="*60)

# Save fitted scaler
import pickle

# Fit on training data
production_scaler = StandardScaler()
production_scaler.fit(X_train)

# Save for production
# with open('scaler.pkl', 'wb') as f:
#     pickle.dump(production_scaler, f)

# Load and transform new data
# with open('scaler.pkl', 'rb') as f:
#     loaded_scaler = pickle.load(f)
# new_data_scaled = loaded_scaler.transform(new_data)

print("Scaler fitted on training data and ready for production use")
print("
Code pattern for production:")
print("  # Training")
print("  scaler = StandardScaler()")
print("  X_train_scaled = scaler.fit_transform(X_train)" )
print("  joblib.dump(scaler, 'scaler.pkl')")
print("  ")
print("  # Production")
print("  scaler = joblib.load('scaler.pkl')")
print("  X_new_scaled = scaler.transform(X_new)  # NOT fit_transform!")