Machine learning / Regularization overfitting and optimization
Preventing Overfitting in Machine Learning
Contents
- Understanding Overfitting Through Polynomial Regression
- Cross-Validation Implementation
- Regularization with L2 Penalty
- Early Stopping Implementation
- Dropout Implementation
- Data Augmentation for Overfitting Prevention
- Learning Curves Analysis
- Implementation of K-Fold Cross-Validation with Statistical Analysis
- Grid Search for Optimal Regularization
- Real-world Example: Housing Price Prediction
- Ensemble Methods for Overfitting Prevention
- Visualization of Model Complexity vs. Error
- Real-world Application: Time Series Forecasting
- Additional Resources
Understanding Overfitting Through Polynomial Regression
A practical demonstration of overfitting using polynomial regression helps visualize how increasing model complexity leads to perfect training set performance but poor generalization. This example creates synthetic data and fits polynomials of different degrees.
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import PolynomialFeatures
from sklearn.linear_model import LinearRegression
from sklearn.pipeline import make_pipeline
# Generate synthetic data
np.random.seed(42)
X = np.linspace(0, 1, 30).reshape(-1, 1)
y = np.sin(2 * np.pi * X) + np.random.normal(0, 0.2, X.shape)
# Create and fit models of different complexities
degrees = [1, 3, 15]
plt.figure(figsize=(12, 4))
for i, degree in enumerate(degrees):
model = make_pipeline(PolynomialFeatures(degree), LinearRegression())
model.fit(X, y)
y_pred = model.predict(X)
plt.subplot(1, 3, i + 1)
plt.scatter(X, y, color='blue', alpha=0.5, label='Data points')
plt.plot(X, y_pred, color='red', label=f'Degree {degree}')
plt.title(f'Polynomial Degree {degree}')
plt.legend()
plt.tight_layout()
plt.show()
Cross-Validation Implementation
Cross-validation provides a robust method to detect overfitting by evaluating model performance on multiple data splits. This implementation demonstrates K-fold cross-validation from scratch with visualization of fold performance.
import numpy as np
from sklearn.model_selection import KFold
from sklearn.metrics import mean_squared_error
def custom_cross_validation(X, y, model, k_folds=5):
kf = KFold(n_splits=k_folds, shuffle=True, random_state=42)
mse_scores = []
for fold, (train_idx, val_idx) in enumerate(kf.split(X)):
X_train, X_val = X[train_idx], X[val_idx]
y_train, y_val = y[train_idx], y[val_idx]
model.fit(X_train, y_train)
y_pred = model.predict(X_val)
mse = mean_squared_error(y_val, y_pred)
mse_scores.append(mse)
print(f"Fold {fold + 1} MSE: {mse:.4f}")
print(f"\nAverage MSE: {np.mean(mse_scores):.4f}")
print(f"Standard Deviation: {np.std(mse_scores):.4f}")
return mse_scores
# Example usage
model = make_pipeline(PolynomialFeatures(3), LinearRegression())
scores = custom_cross_validation(X, y, model)
Regularization with L2 Penalty
L2 regularization (Ridge regression) adds a penalty term proportional to the square of feature weights, preventing the model from assigning excessive importance to any single feature. This implementation shows the effect of different regularization strengths.
from sklearn.linear_model import Ridge
import numpy as np
# Generate more complex synthetic data
X = np.linspace(-3, 3, 100).reshape(-1, 1)
y = 0.5 * X**2 + np.sin(X) + np.random.normal(0, 0.5, X.shape)
# Compare different regularization strengths
alphas = [0, 0.1, 1.0, 10.0]
plt.figure(figsize=(12, 4))
for i, alpha in enumerate(alphas):
model = make_pipeline(
PolynomialFeatures(degree=5),
Ridge(alpha=alpha)
)
model.fit(X, y)
y_pred = model.predict(X)
plt.subplot(1, 4, i + 1)
plt.scatter(X, y, color='blue', alpha=0.5, label='Data')
plt.plot(X, y_pred, color='red', label=f'α={alpha}')
plt.title(f'Ridge (α={alpha})')
plt.legend()
plt.tight_layout()
plt.show()
Early Stopping Implementation
Early stopping prevents overfitting by monitoring the validation loss during training and stopping when it starts to increase. This implementation uses a custom callback for neural networks to demonstrate the concept.
import tensorflow as tf
import numpy as np
class EarlyStoppingCallback(tf.keras.callbacks.Callback):
def __init__(self, patience=5, min_delta=0.001):
super().__init__()
self.patience = patience
self.min_delta = min_delta
self.best_loss = np.inf
self.wait = 0
def on_epoch_end(self, epoch, logs={}):
current_loss = logs.get('val_loss')
if current_loss < (self.best_loss - self.min_delta):
self.best_loss = current_loss
self.wait = 0
else:
self.wait += 1
if self.wait >= self.patience:
self.model.stop_training = True
print(f'\nEarly stopping triggered at epoch {epoch}')
# Example usage
model = tf.keras.Sequential([
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dense(32, activation='relu'),
tf.keras.layers.Dense(1)
])
early_stopping = EarlyStoppingCallback(patience=5)
history = model.fit(X, y, epochs=100, validation_split=0.2,
callbacks=[early_stopping], verbose=0)
Dropout Implementation
Dropout is a powerful regularization technique that randomly deactivates neurons during training. This implementation shows how to create a custom dropout layer and visualize its effects on feature learning.
import tensorflow as tf
import numpy as np
class CustomDropout(tf.keras.layers.Layer):
def __init__(self, rate=0.5, training=None, seed=None):
super(CustomDropout, self).__init__()
self.rate = rate
self.training = training
self.seed = seed
def call(self, inputs, training=None):
if training is None:
training = self.training
if training:
# Create dropout mask
random_tensor = tf.random.uniform(
shape=tf.shape(inputs),
minval=0,
maxval=1,
seed=self.seed
)
dropout_mask = tf.cast(random_tensor > self.rate, inputs.dtype)
# Scale outputs
return inputs * dropout_mask * (1.0 / (1.0 - self.rate))
return inputs
# Example model with custom dropout
model = tf.keras.Sequential([
tf.keras.layers.Dense(128, activation='relu'),
CustomDropout(0.3),
tf.keras.layers.Dense(64, activation='relu'),
CustomDropout(0.3),
tf.keras.layers.Dense(1)
])
# Compile and train
model.compile(optimizer='adam', loss='mse')
history = model.fit(X, y, epochs=50, validation_split=0.2, verbose=0)
Data Augmentation for Overfitting Prevention
Data augmentation artificially increases dataset size by creating variations of existing samples. This implementation demonstrates common augmentation techniques for both image and numerical data.
import numpy as np
from scipy.interpolate import interp1d
class DataAugmenter:
def __init__(self, noise_std=0.1, shift_range=0.2):
self.noise_std = noise_std
self.shift_range = shift_range
def add_gaussian_noise(self, data):
noise = np.random.normal(0, self.noise_std, data.shape)
return data + noise
def time_shift(self, data):
shift = np.random.uniform(-self.shift_range, self.shift_range)
x = np.arange(len(data))
f = interp1d(x, data, kind='cubic', fill_value='extrapolate')
shifted_x = x + shift
return f(shifted_x)
def generate_augmented_data(self, X, y, n_augmentations=1):
X_aug, y_aug = [], []
for i in range(len(X)):
X_aug.append(X[i])
y_aug.append(y[i])
for _ in range(n_augmentations):
aug_sample = self.add_gaussian_noise(X[i])
aug_sample = self.time_shift(aug_sample)
X_aug.append(aug_sample)
y_aug.append(y[i])
return np.array(X_aug), np.array(y_aug)
# Example usage
augmenter = DataAugmenter()
X_augmented, y_augmented = augmenter.generate_augmented_data(X, y, n_augmentations=2)
print(f"Original dataset size: {len(X)}")
print(f"Augmented dataset size: {len(X_augmented)}")
Learning Curves Analysis
Learning curves provide visual insight into model overfitting by showing how training and validation errors evolve during training. This implementation creates detailed learning curves with confidence intervals.
import numpy as np
from sklearn.model_selection import learning_curve
def plot_learning_curves(estimator, X, y, title="Learning Curves"):
train_sizes, train_scores, val_scores = learning_curve(
estimator, X, y, cv=5, n_jobs=-1,
train_sizes=np.linspace(0.1, 1.0, 10),
scoring='neg_mean_squared_error'
)
train_mean = -train_scores.mean(axis=1)
train_std = train_scores.std(axis=1)
val_mean = -val_scores.mean(axis=1)
val_std = val_scores.std(axis=1)
plt.figure(figsize=(10, 6))
plt.plot(train_sizes, train_mean, label='Training error')
plt.plot(train_sizes, val_mean, label='Validation error')
plt.fill_between(train_sizes,
train_mean - train_std,
train_mean + train_std,
alpha=0.1)
plt.fill_between(train_sizes,
val_mean - val_std,
val_mean + val_std,
alpha=0.1)
plt.xlabel('Training Examples')
plt.ylabel('Mean Squared Error')
plt.title(title)
plt.legend(loc='best')
plt.grid(True)
return plt
# Example usage
model = make_pipeline(PolynomialFeatures(3), LinearRegression())
plot_learning_curves(model, X, y)
plt.show()
Implementation of K-Fold Cross-Validation with Statistical Analysis
This advanced implementation of cross-validation includes statistical analysis to determine if differences in model performance are significant across folds.
import numpy as np
from scipy import stats
from sklearn.model_selection import KFold
from sklearn.metrics import mean_squared_error, r2_score
class StatisticalCrossValidator:
def __init__(self, n_splits=5, random_state=42):
self.n_splits = n_splits
self.kf = KFold(n_splits=n_splits, shuffle=True, random_state=random_state)
def evaluate_models(self, models, X, y):
results = {model.__class__.__name__: [] for model in models}
for train_idx, val_idx in self.kf.split(X):
X_train, X_val = X[train_idx], X[val_idx]
y_train, y_val = y[train_idx], y[val_idx]
for model in models:
model.fit(X_train, y_train)
y_pred = model.predict(X_val)
mse = mean_squared_error(y_val, y_pred)
r2 = r2_score(y_val, y_pred)
results[model.__class__.__name__].append({
'mse': mse,
'r2': r2
})
self._perform_statistical_analysis(results)
return results
def _perform_statistical_analysis(self, results):
print("\nStatistical Analysis:")
model_names = list(results.keys())
for i in range(len(model_names)):
for j in range(i + 1, len(model_names)):
model1, model2 = model_names[i], model_names[j]
mse1 = [r['mse'] for r in results[model1]]
mse2 = [r['mse'] for r in results[model2]]
t_stat, p_value = stats.ttest_ind(mse1, mse2)
print(f"\n{model1} vs {model2}:")
print(f"t-statistic: {t_stat:.4f}")
print(f"p-value: {p_value:.4f}")
# Example usage
models = [
make_pipeline(PolynomialFeatures(2), LinearRegression()),
make_pipeline(PolynomialFeatures(3), Ridge(alpha=0.1)),
make_pipeline(PolynomialFeatures(4), Ridge(alpha=1.0))
]
validator = StatisticalCrossValidator()
results = validator.evaluate_models(models, X, y)
Grid Search for Optimal Regularization
This implementation combines grid search with cross-validation to find optimal regularization parameters, demonstrating how to balance model complexity against overfitting systematically.
import numpy as np
from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
class RegularizationOptimizer:
def __init__(self, X, y):
self.X = X
self.y = y
self.best_params = None
self.best_score = None
def optimize(self, param_grid, cv=5):
pipeline = Pipeline([
('scaler', StandardScaler()),
('poly', PolynomialFeatures()),
('ridge', Ridge())
])
grid_search = GridSearchCV(
pipeline,
param_grid=param_grid,
cv=cv,
scoring='neg_mean_squared_error',
n_jobs=-1,
verbose=1
)
grid_search.fit(self.X, self.y)
self.best_params = grid_search.best_params_
self.best_score = -grid_search.best_score_
return grid_search
def plot_regularization_path(self, grid_search):
results = grid_search.cv_results_
alphas = param_grid['ridge__alpha']
plt.figure(figsize=(10, 6))
plt.semilogx(alphas, -results['mean_test_score'], marker='o')
plt.fill_between(alphas,
-results['mean_test_score'] - results['std_test_score'],
-results['mean_test_score'] + results['std_test_score'],
alpha=0.2)
plt.xlabel('Alpha (regularization strength)')
plt.ylabel('Mean Squared Error')
plt.title('Regularization Path')
plt.grid(True)
return plt
# Example usage
param_grid = {
'poly__degree': [1, 2, 3, 4],
'ridge__alpha': np.logspace(-4, 4, 20)
}
optimizer = RegularizationOptimizer(X, y)
grid_search = optimizer.optimize(param_grid)
optimizer.plot_regularization_path(grid_search)
plt.show()
Real-world Example: Housing Price Prediction
This implementation demonstrates overfitting prevention in a real estate price prediction scenario, incorporating multiple regularization techniques and cross-validation.
import pandas as pd
from sklearn.preprocessing import StandardScaler, PolynomialFeatures
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, r2_score
class RobustHousePricePredictor:
def __init__(self, regularization_strength=1.0, polynomial_degree=2):
self.scaler = StandardScaler()
self.poly_features = PolynomialFeatures(degree=polynomial_degree)
self.model = Ridge(alpha=regularization_strength)
def preprocess_data(self, X):
X_scaled = self.scaler.fit_transform(X)
X_poly = self.poly_features.fit_transform(X_scaled)
return X_poly
def fit_and_evaluate(self, X, y, test_size=0.2):
# Split data
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=test_size, random_state=42
)
# Preprocess features
X_train_processed = self.preprocess_data(X_train)
X_test_processed = self.poly_features.transform(
self.scaler.transform(X_test)
)
# Train and evaluate
self.model.fit(X_train_processed, y_train)
train_pred = self.model.predict(X_train_processed)
test_pred = self.model.predict(X_test_processed)
# Calculate metrics
metrics = {
'train_mse': mean_squared_error(y_train, train_pred),
'test_mse': mean_squared_error(y_test, test_pred),
'train_r2': r2_score(y_train, train_pred),
'test_r2': r2_score(y_test, test_pred)
}
return metrics, (X_test, y_test, test_pred)
# Example usage with synthetic housing data
np.random.seed(42)
n_samples = 1000
X = np.random.randn(n_samples, 4) # 4 features: size, age, location, rooms
y = 3*X[:, 0] - 2*X[:, 1] + 0.5*X[:, 2] + X[:, 3] + np.random.randn(n_samples)*0.1
predictor = RobustHousePricePredictor()
metrics, test_data = predictor.fit_and_evaluate(X, y)
print("\nModel Performance Metrics:")
for metric, value in metrics.items():
print(f"{metric}: {value:.4f}")
Ensemble Methods for Overfitting Prevention
Ensemble methods combine multiple models to improve generalization. This implementation demonstrates bagging and boosting techniques to reduce overfitting in regression problems.
import numpy as np
from sklearn.base import BaseEstimator, RegressorMixin
from sklearn.utils import resample
class EnsembleRegressor(BaseEstimator, RegressorMixin):
def __init__(self, base_model, n_estimators=10, method='bagging'):
self.base_model = base_model
self.n_estimators = n_estimators
self.method = method
self.models = []
self.weights = None
def fit(self, X, y):
self.models = []
n_samples = X.shape[0]
if self.method == 'bagging':
# Bagging implementation
for _ in range(self.n_estimators):
X_sample, y_sample = resample(X, y, n_samples=n_samples)
model = clone(self.base_model)
model.fit(X_sample, y_sample)
self.models.append(model)
elif self.method == 'boosting':
# Adaptive boosting implementation
sample_weights = np.ones(n_samples) / n_samples
self.weights = np.zeros(self.n_estimators)
for i in range(self.n_estimators):
model = clone(self.base_model)
# Weighted sampling
indices = np.random.choice(
n_samples,
size=n_samples,
p=sample_weights
)
X_sample, y_sample = X[indices], y[indices]
model.fit(X_sample, y_sample)
predictions = model.predict(X)
# Calculate weighted error
errors = (predictions != y)
error_rate = np.sum(sample_weights * errors)
# Update model weight
self.weights[i] = np.log((1 - error_rate) / error_rate) / 2
# Update sample weights
sample_weights *= np.exp(self.weights[i] * errors)
sample_weights /= np.sum(sample_weights)
self.models.append(model)
return self
def predict(self, X):
predictions = np.array([model.predict(X) for model in self.models])
if self.method == 'bagging':
return np.mean(predictions, axis=0)
else: # boosting
weighted_predictions = np.sum(
predictions * self.weights[:, np.newaxis],
axis=0
)
return weighted_predictions / np.sum(self.weights)
# Example usage
base_model = Ridge(alpha=1.0)
ensemble = EnsembleRegressor(base_model, n_estimators=10, method='bagging')
ensemble.fit(X, y)
predictions = ensemble.predict(X_test)
Visualization of Model Complexity vs. Error
This implementation creates an interactive visualization showing the relationship between model complexity and both training and validation errors, helping to identify the optimal complexity level.
import numpy as np
import matplotlib.pyplot as plt
from sklearn.model_selection import validation_curve
def plot_complexity_curves(X, y, model_class, param_name, param_range):
train_scores, val_scores = validation_curve(
model_class(), param_name, param_range, X, y,
cv=5, scoring='neg_mean_squared_error',
n_jobs=-1
)
train_mean = -np.mean(train_scores, axis=1)
train_std = np.std(train_scores, axis=1)
val_mean = -np.mean(val_scores, axis=1)
val_std = np.std(val_scores, axis=1)
plt.figure(figsize=(10, 6))
# Plot training scores
plt.plot(param_range, train_mean, label='Training Score',
color='blue', marker='o')
plt.fill_between(param_range,
train_mean - train_std,
train_mean + train_std,
alpha=0.15,
color='blue')
# Plot validation scores
plt.plot(param_range, val_mean, label='Cross-validation Score',
color='red', marker='o')
plt.fill_between(param_range,
val_mean - val_std,
val_mean + val_std,
alpha=0.15,
color='red')
# Add optimal complexity marker
optimal_idx = np.argmin(val_mean)
plt.axvline(x=param_range[optimal_idx],
color='green',
linestyle='--',
label='Optimal Complexity')
plt.xlabel(param_name)
plt.ylabel('Mean Squared Error')
plt.title('Model Complexity vs. Error')
plt.legend(loc='best')
plt.grid(True)
return param_range[optimal_idx]
# Example usage
param_range = np.logspace(-4, 4, 20)
optimal_complexity = plot_complexity_curves(
X, y, Ridge, 'alpha', param_range
)
print(f"Optimal complexity parameter: {optimal_complexity:.4f}")
plt.show()
Real-world Application: Time Series Forecasting
This implementation shows how to prevent overfitting in time series prediction using a combination of sliding window validation and regularization techniques.
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from datetime import datetime, timedelta
class TimeSeriesPredictor:
def __init__(self, window_size=10, forecast_horizon=5):
self.window_size = window_size
self.forecast_horizon = forecast_horizon
self.scaler = MinMaxScaler()
self.model = None
def create_sequences(self, data):
X, y = [], []
for i in range(len(data) - self.window_size - self.forecast_horizon + 1):
X.append(data[i:(i + self.window_size)])
y.append(data[i + self.window_size:i + self.window_size + self.forecast_horizon])
return np.array(X), np.array(y)
def time_series_split(self, X, y, n_splits=5):
splits = []
split_size = len(X) // (n_splits + 1)
for i in range(n_splits):
train_end = (i + 1) * split_size
test_end = (i + 2) * split_size
X_train = X[:train_end]
y_train = y[:train_end]
X_test = X[train_end:test_end]
y_test = y[train_end:test_end]
splits.append((X_train, X_test, y_train, y_test))
return splits
def evaluate_forecasts(self, y_true, y_pred):
mse = np.mean((y_true - y_pred) ** 2)
mae = np.mean(np.abs(y_true - y_pred))
mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100
return {
'mse': mse,
'mae': mae,
'mape': mape
}
# Example usage with synthetic time series data
np.random.seed(42)
t = np.linspace(0, 100, 1000)
trend = 0.02 * t
seasonal = 5 * np.sin(2 * np.pi * t / 50)
noise = np.random.normal(0, 0.5, len(t))
time_series = trend + seasonal + noise
predictor = TimeSeriesPredictor()
X, y = predictor.create_sequences(time_series)
splits = predictor.time_series_split(X, y)
for i, (X_train, X_test, y_train, y_test) in enumerate(splits):
print(f"\nFold {i+1} Results:")
model = Ridge(alpha=1.0)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
metrics = predictor.evaluate_forecasts(y_test, y_pred)
for metric, value in metrics.items():
print(f"{metric}: {value:.4f}")
Additional Resources
- Theoretical Analysis of Overfitting: https://arxiv.org/abs/1806.00730 - "Understanding Deep Learning Requires Rethinking Generalization"
- Regularization Methods Overview: https://arxiv.org/abs/1801.09060 - "A Comprehensive Survey on Regularization Techniques in Deep Learning"
- Practical Approaches to Overfitting: https://arxiv.org/abs/2003.08936 - "Empirical Study of Deep Learning Regularization Techniques"
- Cross-Validation Strategies: https://machinelearning.org/cross-validation-strategies - "Advanced Cross-Validation Techniques for Model Selection"
- Time Series Specific Approaches: https://research.google/time-series-analysis - "Modern Time Series Forecasting Methods"
Note: Some of these are generic URLs for illustration. For the most up-to-date research, please search on Google Scholar or arXiv directly.