Ml Model Training

ML Model Training

Train machine learning models with proper data handling and evaluation.

Training Workflow

1. Data Preparation → 2. Feature Engineering → 3. Model Selection → 4. Training → 5. Evaluation

Data Preparation

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder

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

# Encode categorical variables
le = LabelEncoder()
df['category'] = le.fit_transform(df['category'])

# Split data (70/15/15)
X = df.drop('target', axis=1)
y = df['target']
X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.3)
X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.5)

# Scale features
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_val = scaler.transform(X_val)
X_test = scaler.transform(X_test)

Scikit-learn Training

from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, accuracy_score

model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

y_pred = model.predict(X_val)
print(classification_report(y_val, y_pred))

PyTorch Training

import torch
import torch.nn as nn

class Model(nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 1),
            nn.Sigmoid()
        )

    def forward(self, x):
        return self.layers(x)

model = Model(X_train.shape[1])
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.BCELoss()

for epoch in range(100):
    model.train()
    optimizer.zero_grad()
    output = model(X_train_tensor)
    loss = criterion(output, y_train_tensor)
    loss.backward()
    optimizer.step()

Evaluation Metrics

Complete Framework Examples

• PyTorch: See references/pytorch-training.md for complete training with:
• Custom model classes with BatchNorm and Dropout
• Training/validation loops with early stopping
• Learning rate scheduling
• Model checkpointing
• Full evaluation with classification report

• TensorFlow/Keras: See references/tensorflow-keras.md for:
• Sequential model architecture
• Callbacks (EarlyStopping, ReduceLROnPlateau, ModelCheckpoint, TensorBoard)
• Training history visualization
• TFLite conversion for mobile deployment
• Custom training loops

Best Practices

Do:

• Use cross-validation for robust evaluation
• Track experiments with MLflow
• Save model checkpoints regularly
• Monitor for overfitting
• Document hyperparameters
• Use 70/15/15 train/val/test split

Don't:

• Train without a validation set
• Ignore class imbalance
• Skip feature scaling
• Use test set for hyperparameter tuning
• Forget to set random seeds

Known Issues Prevention

1. Data Leakage

Problem: Scaling or transforming data before splitting leads to test set information leaking into training.

Solution: Always split data first, then fit transformers only on training data:

# ✅ Correct: Fit on train, transform train/val/test
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_val = scaler.transform(X_val)  # Only transform
X_test = scaler.transform(X_test)  # Only transform

# ❌ Wrong: Fitting on all data
X_all = scaler.fit_transform(X)  # Leaks test info!

2. Class Imbalance Ignored

Problem: Training on imbalanced datasets (e.g., 95% class A, 5% class B) leads to models that predict only the majority class.

Solution: Use class weights or resampling:

from sklearn.utils.class_weight import compute_class_weight

# Compute class weights
class_weights = compute_class_weight('balanced', classes=np.unique(y_train), y=y_train)
model = RandomForestClassifier(class_weight='balanced')

# Or use SMOTE for oversampling minority class
from imblearn.over_sampling import SMOTE
smote = SMOTE()
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)

3. Overfitting Due to No Regularization

Problem: Complex models memorize training data, perform poorly on validation/test sets.

Solution: Add regularization techniques:

# Dropout in PyTorch
nn.Dropout(0.3)

# L2 regularization in scikit-learn
RandomForestClassifier(max_depth=10, min_samples_split=20)

# Early stopping in Keras
from tensorflow.keras.callbacks import EarlyStopping
early_stop = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)
model.fit(X_train, y_train, validation_data=(X_val, y_val), callbacks=[early_stop])

4. Not Setting Random Seeds

Problem: Results are not reproducible across runs, making debugging and comparison impossible.

Solution: Set all random seeds:

import random
import numpy as np
import torch

random.seed(42)
np.random.seed(42)
torch.manual_seed(42)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(42)

5. Using Test Set for Hyperparameter Tuning

Problem: Optimizing hyperparameters on test set leads to overfitting to test data.

Solution: Use validation set for tuning, test set only for final evaluation:

from sklearn.model_selection import GridSearchCV

# ✅ Correct: Tune on train+val, evaluate on test
param_grid = {'n_estimators': [50, 100, 200], 'max_depth': [5, 10, 15]}
grid_search = GridSearchCV(RandomForestClassifier(), param_grid, cv=5)
grid_search.fit(X_train, y_train)  # Cross-validation on training set
best_model = grid_search.best_estimator_

# Final evaluation on held-out test set
final_score = best_model.score(X_test, y_test)

When to Load References

Load reference files when you need:

• PyTorch implementation details: Load references/pytorch-training.md for complete training loops with early stopping, learning rate scheduling, and checkpointing
• TensorFlow/Keras patterns: Load references/tensorflow-keras.md for callback usage, custom training loops, and mobile deployment with TFLite

PyTorch Model Training

Complete neural network training with PyTorch.

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import accuracy_score, roc_auc_score, classification_report
import matplotlib.pyplot as plt

# Set device
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"Using device: {device}")

# Generate sample data
np.random.seed(42)
torch.manual_seed(42)

X = np.random.randn(1000, 20).astype(np.float32)
y = (X[:, 0] + X[:, 1] * 2 + np.random.randn(1000) * 0.1 > 0).astype(np.float32)

# Split and scale
X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.3, random_state=42)
X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.5, random_state=42)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_val = scaler.transform(X_val)
X_test = scaler.transform(X_test)

# Create DataLoaders
train_dataset = TensorDataset(
    torch.tensor(X_train, dtype=torch.float32),
    torch.tensor(y_train, dtype=torch.float32).unsqueeze(1)
)
val_dataset = TensorDataset(
    torch.tensor(X_val, dtype=torch.float32),
    torch.tensor(y_val, dtype=torch.float32).unsqueeze(1)
)
test_dataset = TensorDataset(
    torch.tensor(X_test, dtype=torch.float32),
    torch.tensor(y_test, dtype=torch.float32).unsqueeze(1)
)

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=32)
test_loader = DataLoader(test_dataset, batch_size=32)


class NeuralNetwork(nn.Module):
    def __init__(self, input_dim, hidden_dims=[64, 32, 16], dropout=0.3):
        super().__init__()

        layers = []
        prev_dim = input_dim

        for hidden_dim in hidden_dims:
            layers.extend([
                nn.Linear(prev_dim, hidden_dim),
                nn.BatchNorm1d(hidden_dim),
                nn.ReLU(),
                nn.Dropout(dropout)
            ])
            prev_dim = hidden_dim

        layers.append(nn.Linear(prev_dim, 1))
        layers.append(nn.Sigmoid())

        self.model = nn.Sequential(*layers)

    def forward(self, x):
        return self.model(x)


class Trainer:
    def __init__(self, model, device, learning_rate=0.001):
        self.model = model.to(device)
        self.device = device
        self.criterion = nn.BCELoss()
        self.optimizer = optim.Adam(model.parameters(), lr=learning_rate)
        self.scheduler = optim.lr_scheduler.ReduceLROnPlateau(
            self.optimizer, mode='min', factor=0.2, patience=5
        )
        self.history = {'train_loss': [], 'val_loss': [], 'train_acc': [], 'val_acc': []}

    def train_epoch(self, train_loader):
        self.model.train()
        total_loss = 0
        predictions = []
        targets = []

        for X_batch, y_batch in train_loader:
            X_batch, y_batch = X_batch.to(self.device), y_batch.to(self.device)

            self.optimizer.zero_grad()
            outputs = self.model(X_batch)
            loss = self.criterion(outputs, y_batch)
            loss.backward()
            self.optimizer.step()

            total_loss += loss.item()
            predictions.extend((outputs > 0.5).cpu().numpy())
            targets.extend(y_batch.cpu().numpy())

        avg_loss = total_loss / len(train_loader)
        accuracy = accuracy_score(targets, predictions)
        return avg_loss, accuracy

    def validate(self, val_loader):
        self.model.eval()
        total_loss = 0
        predictions = []
        targets = []

        with torch.no_grad():
            for X_batch, y_batch in val_loader:
                X_batch, y_batch = X_batch.to(self.device), y_batch.to(self.device)
                outputs = self.model(X_batch)
                loss = self.criterion(outputs, y_batch)

                total_loss += loss.item()
                predictions.extend((outputs > 0.5).cpu().numpy())
                targets.extend(y_batch.cpu().numpy())

        avg_loss = total_loss / len(val_loader)
        accuracy = accuracy_score(targets, predictions)
        return avg_loss, accuracy

    def fit(self, train_loader, val_loader, epochs=100, patience=10):
        best_val_loss = float('inf')
        patience_counter = 0

        for epoch in range(epochs):
            train_loss, train_acc = self.train_epoch(train_loader)
            val_loss, val_acc = self.validate(val_loader)

            self.history['train_loss'].append(train_loss)
            self.history['val_loss'].append(val_loss)
            self.history['train_acc'].append(train_acc)
            self.history['val_acc'].append(val_acc)

            self.scheduler.step(val_loss)

            print(f"Epoch {epoch+1}/{epochs} - "
                  f"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}, "
                  f"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}")

            # Early stopping
            if val_loss < best_val_loss:
                best_val_loss = val_loss
                patience_counter = 0
                torch.save(self.model.state_dict(), 'best_model.pt')
            else:
                patience_counter += 1
                if patience_counter >= patience:
                    print(f"Early stopping at epoch {epoch + 1}")
                    break

        # Load best model
        self.model.load_state_dict(torch.load('best_model.pt'))

    def evaluate(self, test_loader):
        self.model.eval()
        predictions = []
        probabilities = []
        targets = []

        with torch.no_grad():
            for X_batch, y_batch in test_loader:
                X_batch = X_batch.to(self.device)
                outputs = self.model(X_batch)
                probabilities.extend(outputs.cpu().numpy())
                predictions.extend((outputs > 0.5).cpu().numpy())
                targets.extend(y_batch.numpy())

        print("\nTest Results:")
        print(classification_report(targets, predictions))
        print(f"AUC-ROC: {roc_auc_score(targets, probabilities):.4f}")


# Train
model = NeuralNetwork(input_dim=20)
trainer = Trainer(model, device)
trainer.fit(train_loader, val_loader, epochs=100, patience=10)
trainer.evaluate(test_loader)

Dependencies

torch>=2.0.0
scikit-learn>=1.3.0
matplotlib>=3.7.0
numpy>=1.24.0

TensorFlow/Keras Model Training

Complete neural network training with Keras API.

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers, callbacks
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt

# Generate sample data
np.random.seed(42)
X = np.random.randn(1000, 20)
y = (X[:, 0] + X[:, 1] * 2 + np.random.randn(1000) * 0.1 > 0).astype(int)

# Split and scale
X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.3, random_state=42)
X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.5, random_state=42)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_val = scaler.transform(X_val)
X_test = scaler.transform(X_test)


def create_model(input_dim, learning_rate=0.001):
    """Create a neural network model."""
    model = keras.Sequential([
        layers.Input(shape=(input_dim,)),
        layers.Dense(64, activation='relu'),
        layers.BatchNormalization(),
        layers.Dropout(0.3),
        layers.Dense(32, activation='relu'),
        layers.BatchNormalization(),
        layers.Dropout(0.2),
        layers.Dense(16, activation='relu'),
        layers.Dense(1, activation='sigmoid')
    ])

    model.compile(
        optimizer=keras.optimizers.Adam(learning_rate=learning_rate),
        loss='binary_crossentropy',
        metrics=['accuracy', keras.metrics.AUC(name='auc')]
    )

    return model


# Define callbacks
early_stopping = callbacks.EarlyStopping(
    monitor='val_loss',
    patience=10,
    restore_best_weights=True
)

reduce_lr = callbacks.ReduceLROnPlateau(
    monitor='val_loss',
    factor=0.2,
    patience=5,
    min_lr=0.0001
)

model_checkpoint = callbacks.ModelCheckpoint(
    'best_model.keras',
    monitor='val_auc',
    mode='max',
    save_best_only=True
)

tensorboard = callbacks.TensorBoard(
    log_dir='./logs',
    histogram_freq=1
)


# Train model
model = create_model(X_train.shape[1])

history = model.fit(
    X_train, y_train,
    validation_data=(X_val, y_val),
    epochs=100,
    batch_size=32,
    callbacks=[early_stopping, reduce_lr, model_checkpoint, tensorboard],
    verbose=1
)


# Evaluate
test_results = model.evaluate(X_test, y_test, verbose=0)
print(f"Test Loss: {test_results[0]:.4f}")
print(f"Test Accuracy: {test_results[1]:.4f}")
print(f"Test AUC: {test_results[2]:.4f}")


# Plot training history
def plot_history(history):
    fig, axes = plt.subplots(1, 3, figsize=(15, 4))

    # Loss
    axes[0].plot(history.history['loss'], label='Train')
    axes[0].plot(history.history['val_loss'], label='Validation')
    axes[0].set_title('Loss')
    axes[0].legend()

    # Accuracy
    axes[1].plot(history.history['accuracy'], label='Train')
    axes[1].plot(history.history['val_accuracy'], label='Validation')
    axes[1].set_title('Accuracy')
    axes[1].legend()

    # AUC
    axes[2].plot(history.history['auc'], label='Train')
    axes[2].plot(history.history['val_auc'], label='Validation')
    axes[2].set_title('AUC')
    axes[2].legend()

    plt.tight_layout()
    plt.savefig('training_history.png')
    plt.show()


plot_history(history)


# Save model for production
model.save('final_model.keras')

# Convert to TFLite for mobile deployment
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()
with open('model.tflite', 'wb') as f:
    f.write(tflite_model)

Custom Training Loop

@tf.function
def train_step(model, x, y, optimizer, loss_fn, train_acc_metric):
    with tf.GradientTape() as tape:
        predictions = model(x, training=True)
        loss = loss_fn(y, predictions)

    gradients = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(gradients, model.trainable_variables))

    train_acc_metric.update_state(y, predictions)
    return loss


def custom_training(model, train_dataset, val_dataset, epochs=50):
    optimizer = keras.optimizers.Adam(learning_rate=0.001)
    loss_fn = keras.losses.BinaryCrossentropy()
    train_acc_metric = keras.metrics.BinaryAccuracy()
    val_acc_metric = keras.metrics.BinaryAccuracy()

    for epoch in range(epochs):
        # Training
        for x_batch, y_batch in train_dataset:
            loss = train_step(model, x_batch, y_batch, optimizer, loss_fn, train_acc_metric)

        train_acc = train_acc_metric.result()
        train_acc_metric.reset_states()

        # Validation
        for x_batch, y_batch in val_dataset:
            predictions = model(x_batch, training=False)
            val_acc_metric.update_state(y_batch, predictions)

        val_acc = val_acc_metric.result()
        val_acc_metric.reset_states()

        print(f"Epoch {epoch + 1}: Train Acc = {train_acc:.4f}, Val Acc = {val_acc:.4f}")

Dependencies

tensorflow>=2.15.0
scikit-learn>=1.3.0
matplotlib>=3.7.0
numpy>=1.24.0