Imputation Strategies: From Simple to Advanced Techniques

import numpy as np
import pandas as pd
from collections import Counter

# Create sample data with missing values
np.random.seed(42)
data = {
    'age': [25, 30, np.nan, 35, 40, np.nan, 45, 50, 28, 32],
    'income': [50000, np.nan, 60000, 70000, np.nan, 55000, 80000, np.nan, 52000, 65000],
    'score': [85, 90, 78, 92, np.nan, 88, 95, 87, 82, 91],
    'category': ['A', 'B', 'A', np.nan, 'B', 'A', 'B', 'A', np.nan, 'B']
}
df = pd.DataFrame(data)

print("Original Dataset:")
print(df)

# 1. MEAN IMPUTATION (Manual)
def mean_imputation_manual(series):
    """Manual mean imputation for numeric series"""
    # Calculate mean of non-missing values
    observed = series[~np.isnan(series)]
    mean_val = np.sum(observed) / len(observed)
    
    # Replace missing with mean
    imputed = series.copy()
    imputed[np.isnan(imputed)] = mean_val
    
    return imputed, mean_val

print("
=== 1. MEAN IMPUTATION (Manual) ===")
df_mean = df.copy()
age_imputed, age_mean = mean_imputation_manual(df['age'].values)
income_imputed, income_mean = mean_imputation_manual(df['income'].values)
df_mean['age'] = age_imputed
df_mean['income'] = income_imputed
print(f"Age mean: {age_mean:.2f}")
print(f"Income mean: {income_mean:.2f}")
print("
Mean-imputed dataset:")
print(df_mean)

# 2. MEDIAN IMPUTATION (Manual)
def median_imputation_manual(series):
    """Manual median imputation using sorting"""
    observed = series[~np.isnan(series)]
    sorted_vals = np.sort(observed)
    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]
    
    imputed = series.copy()
    imputed[np.isnan(imputed)] = median_val
    
    return imputed, median_val

print("
=== 2. MEDIAN IMPUTATION (Manual) ===")
df_median = df.copy()
age_med_imp, age_median = median_imputation_manual(df['age'].values)
income_med_imp, income_median = median_imputation_manual(df['income'].values)
df_median['age'] = age_med_imp
df_median['income'] = income_med_imp
print(f"Age median: {age_median:.2f}")
print(f"Income median: {income_median:.2f}")

# 3. MODE IMPUTATION (Manual)
def mode_imputation_manual(series):
    """Manual mode imputation for categorical data"""
    # Count frequencies
    observed = series[~pd.isna(series)]
    counter = Counter(observed)
    mode_val = counter.most_common(1)[0][0]
    
    imputed = series.copy()
    imputed[pd.isna(imputed)] = mode_val
    
    return imputed, mode_val

print("
=== 3. MODE IMPUTATION (Manual) ===")
df_mode = df.copy()
cat_imputed, mode_val = mode_imputation_manual(df['category'])
df_mode['category'] = cat_imputed
print(f"Category mode: {mode_val}")
print("
Mode-imputed dataset:")
print(df_mode)

# 4. KNN IMPUTATION (Manual - Simplified)
def euclidean_distance(row1, row2, cols):
    """Calculate Euclidean distance between two rows"""
    dist_sq = 0
    count = 0
    for col in cols:
        if not pd.isna(row1[col]) and not pd.isna(row2[col]):
            dist_sq += (row1[col] - row2[col]) ** 2
            count += 1
    return np.sqrt(dist_sq) if count > 0 else float('inf')

def knn_imputation_manual(df, target_col, k=2, numeric_cols=None):
    """
    Simplified KNN imputation for a single column.
    Uses available numeric columns for distance calculation.
    """
    if numeric_cols is None:
        numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    
    df_result = df.copy()
    
    for idx in df.index:
        if pd.isna(df.loc[idx, target_col]):
            # Calculate distances to all complete observations
            distances = []
            for other_idx in df.index:
                if other_idx != idx and not pd.isna(df.loc[other_idx, target_col]):
                    dist = euclidean_distance(df.loc[idx], df.loc[other_idx], numeric_cols)
                    if dist != float('inf'):
                        distances.append((other_idx, dist))
            
            # Get k nearest neighbors
            distances.sort(key=lambda x: x[1])
            neighbors = distances[:k]
            
            if neighbors:
                # Weighted average by inverse distance
                weights = [1/(d[1] + 0.001) for d in neighbors]  # Add small epsilon
                values = [df.loc[d[0], target_col] for d in neighbors]
                imputed_val = np.average(values, weights=weights)
                df_result.loc[idx, target_col] = imputed_val
    
    return df_result

print("
=== 4. KNN IMPUTATION (Manual - k=2) ===")
df_knn = df.copy()
# Fill category first for distance calculation
df_knn['category'] = df_knn['category'].fillna('A')  # Simple fill for demo
df_knn = knn_imputation_manual(df_knn, 'income', k=2)
print("KNN-imputed dataset:")
print(df_knn[['age', 'income', 'score']])

# 5. INTERPOLATION (Time-series)
def linear_interpolation_manual(series):
    """Manual linear interpolation for time-series gaps"""
    result = series.copy()
    
    for i in range(len(series)):
        if pd.isna(series.iloc[i]):
            # Find previous valid value
            prev_idx = None
            for j in range(i-1, -1, -1):
                if not pd.isna(series.iloc[j]):
                    prev_idx = j
                    break
            
            # Find next valid value
            next_idx = None
            for j in range(i+1, len(series)):
                if not pd.isna(series.iloc[j]):
                    next_idx = j
                    break
            
            if prev_idx is not None and next_idx is not None:
                # Linear interpolation
                prev_val = series.iloc[prev_idx]
                next_val = series.iloc[next_idx]
                weight = (i - prev_idx) / (next_idx - prev_idx)
                result.iloc[i] = prev_val + weight * (next_val - prev_val)
            elif prev_idx is not None:
                result.iloc[i] = series.iloc[prev_idx]  # Forward fill
            elif next_idx is not None:
                result.iloc[i] = series.iloc[next_idx]  # Backward fill
    
    return result

print("
=== 5. LINEAR INTERPOLATION (Manual) ===")
ts_data = pd.Series([10, np.nan, np.nan, 25, 30, np.nan, 45])
print("Original:", ts_data.tolist())
ts_interpolated = linear_interpolation_manual(ts_data)
print("Interpolated:", ts_interpolated.tolist())

# Compare imputation effects
print("
=== COMPARISON OF IMPUTATION METHODS ===")
print("Original age std:", df['age'].std())
print("Mean-imputed age std:", df_mean['age'].std())
print("Median-imputed age std:", df_median['age'].std())
print("
Note: Simple imputation reduces variance, underestimating uncertainty")

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.experimental import enable_iterative_imputer
from sklearn.impute import IterativeImputer
from sklearn.ensemble import RandomForestRegressor
import matplotlib.pyplot as plt

# Create sample data
np.random.seed(42)
data = {
    'age': [25, 30, np.nan, 35, 40, np.nan, 45, 50, 28, 32, 38, 42],
    'income': [50000, 45000, 60000, 70000, np.nan, 55000, 80000, 75000, 52000, 65000, np.nan, 72000],
    'score': [85, 90, 78, 92, 88, 88, 95, 87, 82, 91, 89, 93],
    'tenure': [2, 5, 3, 4, 6, np.nan, 8, 7, 1, 5, 4, np.nan],
    'satisfaction': [4, 5, 3, np.nan, 4, 4, 5, 4, 3, 5, 4, np.nan]
}
df = pd.DataFrame(data)

print("Original Dataset:")
print(df)
print(f"
Missing values per column:")
print(df.isnull().sum())

# 1. SIMPLEIMPUTER - Univariate methods
print("
" + "="*60)
print("1. SIMPLEIMPUTER - Univariate Methods")
print("="*60)

# Mean imputation
mean_imp = SimpleImputer(strategy='mean')
df_mean = pd.DataFrame(
    mean_imp.fit_transform(df),
    columns=df.columns
)
print("
Mean imputation:")
print(df_mean.head())

# Median imputation
median_imp = SimpleImputer(strategy='median')
df_median = pd.DataFrame(
    median_imp.fit_transform(df),
    columns=df.columns
)
print("
Median imputation (robust to outliers):")
print(df_median.head())

# Most frequent imputation
mode_imp = SimpleImputer(strategy='most_frequent')
df_mode = pd.DataFrame(
    mode_imp.fit_transform(df),
    columns=df.columns
)
print("
Most frequent imputation:")
print(df_mode.head())

# Constant imputation
constant_imp = SimpleImputer(strategy='constant', fill_value=-999)
df_constant = pd.DataFrame(
    constant_imp.fit_transform(df),
    columns=df.columns
)
print("
Constant imputation (fill_value=-999):")
print(df_constant.head())

# 2. KNN IMPUTER
print("
" + "="*60)
print("2. KNN IMPUTER - Multivariate Method")
print("="*60)

knn_imp = KNNImputer(n_neighbors=3, weights='distance')
df_knn = pd.DataFrame(
    knn_imp.fit_transform(df),
    columns=df.columns
)
print(f"
KNN imputation (k=3, distance-weighted):")
print(df_knn)

# Compare KNN vs Mean for income
print("
Comparison for missing income values:")
missing_idx = df['income'].isna()
print(f"KNN estimates: {df_knn.loc[missing_idx, 'income'].values}")
print(f"Mean estimate: {df['income'].mean():.2f}")

# 3. ITERATIVE IMPUTER (MICE-like)
print("
" + "="*60)
print("3. ITERATIVE IMPUTER (MICE-like) - Multiple Imputation")
print("="*60)

# Use RandomForest as estimator for better non-linear relationships
estimator = RandomForestRegressor(n_estimators=10, random_state=42, max_depth=5)
iterative_imp = IterativeImputer(
    estimator=estimator,
    max_iter=10,
    random_state=42,
    sample_posterior=True  # Adds randomness to reflect uncertainty
)

df_iterative = pd.DataFrame(
    iterative_imp.fit_transform(df),
    columns=df.columns
)
print("
Iterative imputation (RandomForest estimator):")
print(df_iterative)

# 4. MULTIPLE IMPUTATION (Simulating MICE)
print("
" + "="*60)
print("4. MULTIPLE IMPUTATION - Multiple Plausible Values")
print("="*60)

# Create 5 imputed datasets
n_imputations = 5
imputed_datasets = []

for i in range(n_imputations):
    # Use different random state for each
    imp = IterativeImputer(
        estimator=RandomForestRegressor(n_estimators=10, random_state=42+i),
        max_iter=10,
        random_state=42+i,
        sample_posterior=True
    )
    df_imp = pd.DataFrame(
        imp.fit_transform(df),
        columns=df.columns
    )
    imputed_datasets.append(df_imp)

# Analyze uncertainty in imputed values
missing_income_idx = df['income'].isna()
print("
Uncertainty analysis for imputed income values:")
income_estimates = [d.loc[missing_income_idx, 'income'].values for d in imputed_datasets]
income_estimates = np.array(income_estimates)
print(f"Imputed income estimates across {n_imputations} datasets:")
print(income_estimates.T)
print(f"
Mean of estimates: {income_estimates.mean(axis=0)}")
print(f"Std of estimates: {income_estimates.std(axis=0)}")

# 5. TIME-SERIES SPECIFIC IMPUTATION
print("
" + "="*60)
print("5. TIME-SERIES IMPUTATION")
print("="*60)

ts_data = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=10, freq='D'),
    'value': [100, np.nan, 102, np.nan, np.nan, 108, 110, np.nan, 114, 116]
})
ts_data.set_index('date', inplace=True)

print("
Original time series:")
print(ts_data)

# Forward fill
ts_ffill = ts_data.ffill()
print("
Forward fill:")
print(ts_ffill)

# Backward fill
ts_bfill = ts_data.bfill()
print("
Backward fill:")
print(ts_bfill)

# Linear interpolation
ts_interp = ts_data.interpolate(method='linear')
print("
Linear interpolation:")
print(ts_interp)

# Time-based interpolation
ts_time_interp = ts_data.interpolate(method='time')
print("
Time-based interpolation:")
print(ts_time_interp)

# 6. ADVANCED: IMPUTATION WITH MISSING INDICATORS
print("
" + "="*60)
print("6. IMPUTATION WITH MISSING INDICATORS")
print("="*60)

from sklearn.impute import SimpleImputer

# Create imputer that adds missing indicators
imp_with_indicator = SimpleImputer(strategy='mean', add_indicator=True)
df_with_indicators = imp_with_indicator.fit_transform(df)

# Get feature names including indicators
n_features = len(df.columns)
indicator_features = [f'{col}_missing' for col in df.columns]
all_columns = list(df.columns) + indicator_features

df_indicators = pd.DataFrame(df_with_indicators, columns=all_columns)
print("
Imputed data with missing indicators:")
print(df_indicators)

# 7. VALIDATION: Compare imputation methods
print("
" + "="*60)
print("7. VALIDATION: Comparing Imputation Quality")
print("="*60)

# Artificially introduce missingness and compare
np.random.seed(42)
complete_data = np.random.randn(100, 3) + 5
df_complete = pd.DataFrame(complete_data, columns=['A', 'B', 'C'])

# Introduce 20% missing values
missing_mask = np.random.random((100, 3)) < 0.2
df_incomplete = df_complete.copy()
df_incomplete[missing_mask] = np.nan

# Impute and calculate RMSE
strategies = ['mean', 'median', 'most_frequent']
results = {}

for strategy in strategies:
    imp = SimpleImputer(strategy=strategy)
    df_imp = pd.DataFrame(imp.fit_transform(df_incomplete), columns=['A', 'B', 'C'])
    
    # Calculate RMSE on originally missing positions
    rmse = np.sqrt(np.mean((df_complete[missing_mask].values - df_imp[missing_mask].values) ** 2))
    results[strategy] = rmse

# KNN
knn_imp = KNNImputer(n_neighbors=5)
df_knn = pd.DataFrame(knn_imp.fit_transform(df_incomplete), columns=['A', 'B', 'C'])
knn_rmse = np.sqrt(np.mean((df_complete[missing_mask].values - df_knn[missing_mask].values) ** 2))
results['KNN(k=5)'] = knn_rmse

print("
RMSE by imputation strategy (lower is better):")
for strategy, rmse in sorted(results.items(), key=lambda x: x[1]):
    print(f"  {strategy}: {rmse:.4f}")

# Visualization
try:
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))
    
    # Distribution comparison for age
    axes[0, 0].hist(df['age'].dropna(), alpha=0.5, label='Original', bins=10)
    axes[0, 0].hist(df_mean['age'], alpha=0.5, label='Mean Imputed', bins=10)
    axes[0, 0].set_title('Mean Imputation: Variance Reduction')
    axes[0, 0].legend()
    
    # KNN scatter
    axes[0, 1].scatter(df['age'], df['income'], alpha=0.6, label='Original')
    axes[0, 1].scatter(df_knn['age'], df_knn['income'], alpha=0.6, label='KNN Imputed', marker='x')
    axes[0, 1].set_title('KNN Imputation Preserves Relationships')
    axes[0, 1].legend()
    
    # Method comparison
    strategies_list = list(results.keys())
    rmse_values = list(results.values())
    axes[1, 0].bar(strategies_list, rmse_values)
    axes[1, 0].set_title('Imputation Method Comparison (RMSE)')
    axes[1, 0].set_ylabel('RMSE')
    
    # Multiple imputation uncertainty
    axes[1, 1].boxplot([income_estimates[:, i] for i in range(income_estimates.shape[1])])
    axes[1, 1].set_title('Multiple Imputation Uncertainty')
    axes[1, 1].set_ylabel('Imputed Income')
    
    plt.tight_layout()
    plt.savefig('imputation_comparison.png', dpi=150, bbox_inches='tight')
    print("
Visualization saved!")
except Exception as e:
    print(f"
Visualization skipped: {e}")

print("
" + "="*60)
print("SUMMARY: Imputation Strategy Selection Guide")
print("="*60)
print("• MCAR + <5% missing: Mean/Median imputation")
print("• MAR + known predictors: KNN imputation")
print("• Complex relationships: Iterative/MICE imputation")
print("• Time series: Interpolation or forward/backward fill")
print("• Categorical: Mode or create 'Missing' category")
print("• Uncertainty quantification: Multiple imputation")