Approach: implement an EarlyStopping helper that tracks best validation loss, counts epochs with no sufficient improvement (min_delta), supports patience, and optionally restores best model params. Integrate into an SGD training loop for a logistic regression (numpy) showing checkpointing of weights.python
import numpy as np
class EarlyStopping:
def __init__(self, patience=5, min_delta=0.0, restore_best=False):
self.patience = patience
self.min_delta = min_delta
self.restore_best = restore_best
self.best_loss = np.inf
self.best_params = None
self.wait = 0
self.stopped_epoch = None
def step(self, current_loss, params, epoch):
improved = (self.best_loss - current_loss) > self.min_delta
if improved:
self.best_loss = current_loss
self.best_params = {k: v.copy() for k, v in params.items()}
self.wait = 0
else:
self.wait += 1
if self.wait >= self.patience:
self.stopped_epoch = epoch
return True # stop
return False # continue
def on_train_end(self, params):
if self.restore_best and self.best_params is not None:
for k in params:
params[k][...] = self.best_params[k]
# Simple logistic regression helper functions
def sigmoid(z): return 1 / (1 + np.exp(-z))
def compute_loss(X, y, w, b):
m = X.shape[0]
logits = X.dot(w) + b
p = sigmoid(logits)
# binary cross-entropy
eps = 1e-12
return -np.mean(y * np.log(p+eps) + (1-y)*np.log(1-p+eps))
# SGD training loop with early stopping
def train_sgd(X_train, y_train, X_val, y_val, lr=0.1, epochs=100, batch_size=32,
patience=5, min_delta=1e-4, restore_best=True):
n_features = X_train.shape[1]
w = np.zeros(n_features)
b = 0.0
params = {'w': w, 'b': np.array(b)} # b wrapped to allow .copy()
es = EarlyStopping(patience=patience, min_delta=min_delta, restore_best=restore_best)
for epoch in range(1, epochs+1):
# Shuffle
perm = np.random.permutation(X_train.shape[0])
X_sh, y_sh = X_train[perm], y_train[perm]
for i in range(0, X_train.shape[0], batch_size):
xb = X_sh[i:i+batch_size]; yb = y_sh[i:i+batch_size]
preds = sigmoid(xb.dot(w) + b)
grad_w = xb.T.dot(preds - yb) / xb.shape[0]
grad_b = np.mean(preds - yb)
w -= lr * grad_w
b -= lr * grad_b
params['w'] = w
params['b'][...] = b
val_loss = compute_loss(X_val, y_val, w, b)
if es.step(val_loss, params, epoch):
print(f"Early stopping at epoch {epoch}. Best val_loss={es.best_loss:.6f}")
break
es.on_train_end(params)
return params['w'], float(params['b']), es.best_loss, es.stopped_epoch
# Usage: w, b, best_loss, stop_epoch = train_sgd(X_train, y_train, X_val, y_val)
Key points:- We track validation loss, require improvement > min_delta to reset wait.- restore_best replaces parameters with the best seen on validation.- Checkpointing uses copies to avoid mutation during training.Complexity:- Per epoch: O(N * d) where N is dataset size and d features (same as normal SGD).Edge cases:- Validation loss NaN/inf (handle by clipping or skipping step).- Very small min_delta or patience=0 (stop immediately if no improvement).Alternatives:- Monitor validation accuracy instead of loss, use moving-average smoothing, or implement exponential moving best to reduce sensitivity to noisy val metrics.