Tutorial 5: Time-Series Prediction & Fashion MNIST Classification

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader, TensorDataset
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler

# Set random seeds for reproducibility
torch.manual_seed(42)
np.random.seed(42)

if torch.backends.mps.is_available():
    device = torch.device("mps")
    print("Using MPS device")
else:
    device = torch.device("cpu")
    print("MPS not available, using CPU")

# Generate synthetic temperature data
def generate_temperature_data(n_days=1000):
    """
    Generate synthetic daily temperature data with trend, seasonality, and noise
    """
    time = np.arange(n_days)
    
    # Trend: gradual warming over time
    trend = 0.01 * time
    
    # Seasonality: yearly cycle (365 days)
    seasonality = 10 * np.sin(2 * np.pi * time / 365)
    
    # Random noise
    noise = np.random.normal(0, 2, n_days)
    
    # Base temperature around 20°C
    temperature = 20 + trend + seasonality + noise
    
    return temperature

# Generate data
temperatures = generate_temperature_data(1000)

# Visualize
plt.figure(figsize=(15, 4))
plt.plot(temperatures[:365], label='First Year')
plt.xlabel('Day')
plt.ylabel('Temperature (°C)')
plt.title('Synthetic Temperature Data')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

print(f"Generated {len(temperatures)} days of temperature data")
print(f"Temperature range: {temperatures.min():.2f}°C to {temperatures.max():.2f}°C")

def create_sequences(data, window_size):
    """
    Create input-output pairs for time-series prediction
    
    Args:
        data: 1D array of time-series values
        window_size: Number of past observations to use for prediction
    
    Returns:
        X: Input sequences (n_samples, window_size)
        y: Target values (n_samples,)
    """
    X, y = [], []
    
    for i in range(len(data) - window_size):
        # Input: window_size past observations
        X.append(data[i:i + window_size])
        # Output: next value
        y.append(data[i + window_size])
    
    return np.array(X), np.array(y)

# Create sequences with window size of 7 days
window_size = 7
X, y = create_sequences(temperatures, window_size)

print(f"Created {len(X)} sequences")
print(f"Input shape: {X.shape}")
print(f"Output shape: {y.shape}")
print(f"\nExample:")
print(f"Input (7 days): {X[0]}")
print(f"Output (next day): {y[0]:.2f}")

# Split into train and test (80-20)
train_size = int(0.8 * len(X))
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

# Normalize using MinMaxScaler
scaler_X = MinMaxScaler()
scaler_y = MinMaxScaler()

# Fit on training data only!
X_train_scaled = scaler_X.fit_transform(X_train)
X_test_scaled = scaler_X.transform(X_test)

y_train_scaled = scaler_y.fit_transform(y_train.reshape(-1, 1)).flatten()
y_test_scaled = scaler_y.transform(y_test.reshape(-1, 1)).flatten()

# Convert to PyTorch tensors
X_train_tensor = torch.FloatTensor(X_train_scaled)
y_train_tensor = torch.FloatTensor(y_train_scaled)
X_test_tensor = torch.FloatTensor(X_test_scaled)
y_test_tensor = torch.FloatTensor(y_test_scaled)

print(f"Training samples: {len(X_train_tensor)}")
print(f"Test samples: {len(X_test_tensor)}")
print(f"\nScaled value range: [{X_train_scaled.min():.3f}, {X_train_scaled.max():.3f}]")

# Create DataLoaders
train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
test_dataset = TensorDataset(X_test_tensor, y_test_tensor)

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

print(f"Created DataLoaders with batch size 32")
print(f"Number of batches in train_loader: {len(train_loader)}")

class TimeSeriesNet(nn.Module):
    def __init__(self, input_size, hidden_size1=64, hidden_size2=32):
        super(TimeSeriesNet, self).__init__()
        
        self.network = nn.Sequential(
            nn.Linear(input_size, hidden_size1),
            nn.ReLU(),
            nn.Linear(hidden_size1, hidden_size2),
            nn.ReLU(),
            nn.Linear(hidden_size2, 1)
        )
    
    def forward(self, x):
        return self.network(x).squeeze()

# Instantiate the model
model = TimeSeriesNet(input_size=window_size)
print(model)

# Count parameters
total_params = sum(p.numel() for p in model.parameters())
print(f"\nTotal parameters: {total_params:,}")

# Define loss and optimizer
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# Training loop
num_epochs = 100
train_losses = []
val_losses = []

for epoch in range(num_epochs):
    # Training phase
    model.train()
    train_loss = 0.0
    
    for inputs, targets in train_loader:
        # Forward pass
        outputs = model(inputs)
        loss = criterion(outputs, targets)
        
        # Backward pass and optimize
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        train_loss += loss.item() * inputs.size(0)
    
    train_loss = train_loss / len(train_loader.dataset)
    train_losses.append(train_loss)
    
    # Validation phase
    model.eval()
    val_loss = 0.0
    
    with torch.no_grad():
        for inputs, targets in test_loader:
            outputs = model(inputs)
            loss = criterion(outputs, targets)
            val_loss += loss.item() * inputs.size(0)
    
    val_loss = val_loss / len(test_loader.dataset)
    val_losses.append(val_loss)
    
    if (epoch + 1) % 5 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Train Loss: {train_loss:.6f}, Val Loss: {val_loss:.6f}')

print("\nTraining complete!")

import plotly.express as px
import pandas as pd

# Create a DataFrame with the training history
history_df = pd.DataFrame({
    'Epoch': list(range(1, len(train_losses) + 1)),
    'Training Loss': train_losses,
    'Validation Loss': val_losses
})

# Melt the DataFrame for easier plotting
history_melted = history_df.melt(
    id_vars=['Epoch'],
    value_vars=['Training Loss', 'Validation Loss'],
    var_name='Loss Type',
    value_name='Loss (MSE)'
)

# Create interactive plot
fig = px.line(
    history_melted,
    x='Epoch',
    y='Loss (MSE)',
    color='Loss Type',
    title='Training History - Time-Series Model',
    labels={'Loss (MSE)': 'Loss (MSE)', 'Epoch': 'Epoch'},
    template='plotly_white'
)

fig.update_traces(line=dict(width=2))
fig.update_layout(
    hovermode='x unified',
    legend=dict(yanchor="top", y=0.99, xanchor="right", x=0.99)
)

fig.show()

# Make predictions on test set
model.eval()
with torch.no_grad():
    predictions_scaled = model(X_test_tensor).numpy()

# Inverse transform to get actual temperature values
predictions = scaler_y.inverse_transform(predictions_scaled.reshape(-1, 1)).flatten()
actuals = scaler_y.inverse_transform(y_test_scaled.reshape(-1, 1)).flatten()

# Calculate metrics
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

mse = mean_squared_error(actuals, predictions)
rmse = np.sqrt(mse)
mae = mean_absolute_error(actuals, predictions)
r2 = r2_score(actuals, predictions)

print("=" * 50)
print("Model Performance on Test Set")
print("=" * 50)
print(f"Mean Squared Error (MSE): {mse:.4f}")
print(f"Root Mean Squared Error (RMSE): {rmse:.4f}°C")
print(f"Mean Absolute Error (MAE): {mae:.4f}°C")
print(f"R² Score: {r2:.4f}")
print("=" * 50)

# Visualize predictions vs actual values
plt.figure(figsize=(15, 5))

# Plot first 100 predictions
n_plot = 100
plt.plot(actuals[:n_plot], label='Actual', marker='o', markersize=3, alpha=0.7)
plt.plot(predictions[:n_plot], label='Predicted', marker='x', markersize=3, alpha=0.7)
plt.xlabel('Time Step')
plt.ylabel('Temperature (°C)')
plt.title('Temperature Prediction: Actual vs Predicted')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

# Scatter plot
plt.figure(figsize=(8, 8))
plt.scatter(actuals, predictions, alpha=0.5)
plt.plot([actuals.min(), actuals.max()], [actuals.min(), actuals.max()], 'r--', lw=2)
plt.xlabel('Actual Temperature (°C)')
plt.ylabel('Predicted Temperature (°C)')
plt.title('Prediction Accuracy')
plt.grid(True, alpha=0.3)
plt.axis('equal')
plt.show()

from torchvision import datasets, transforms

# Define transformations
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])

# Load Fashion MNIST
fashion_train = datasets.FashionMNIST(
    root='./data',
    train=True,
    download=True,
    transform=transform
)

fashion_test = datasets.FashionMNIST(
    root='./data',
    train=False,
    download=True,
    transform=transform
)

# Create data loaders
fashion_train_loader = DataLoader(fashion_train, batch_size=256, shuffle=True, num_workers=4, persistent_workers=True)
fashion_test_loader = DataLoader(fashion_test, batch_size=256, shuffle=False, num_workers=4, persistent_workers=True)

print(f"Training samples: {len(fashion_train)}")
print(f"Test samples: {len(fashion_test)}")

# Class names
class_names = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat',
               'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']

# Visualize some samples
fig, axes = plt.subplots(2, 5, figsize=(12, 5))
axes = axes.ravel()

for i in range(10):
    img, label = fashion_train[i]
    axes[i].imshow(img.squeeze(), cmap='gray')
    axes[i].set_title(f'{class_names[label]}')
    axes[i].axis('off')

plt.tight_layout()
plt.show()

class SimpleFashionNet(nn.Module):
    """Simple 2-layer MLP"""
    def __init__(self):
        super(SimpleFashionNet, self).__init__()
        self.network = nn.Sequential(
            nn.Flatten(),
            nn.Linear(28*28, 128),
            nn.ReLU(),
            nn.Linear(128, 10)
        )
    
    def forward(self, x):
        return self.network(x)

class DeepFashionNet(nn.Module):
    """Deeper MLP with dropout"""
    def __init__(self):
        super(DeepFashionNet, self).__init__()
        self.network = nn.Sequential(
            nn.Flatten(),
            nn.Linear(28*28, 256),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 10)
        )
    
    def forward(self, x):
        return self.network(x)

# Create models
simple_model = SimpleFashionNet().to(device)
deep_model = DeepFashionNet().to(device)

print("Simple Model:")
print(simple_model)
print(f"Parameters: {sum(p.numel() for p in simple_model.parameters()):,}\n")

print("Deep Model:")
print(deep_model)
print(f"Parameters: {sum(p.numel() for p in deep_model.parameters()):,}")

def train_model(model, train_loader, test_loader, num_epochs=10, lr=0.001):
    """
    Train a model and return training history
    """
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=lr)
    
    train_losses = []
    val_losses = []
    val_accuracies = []
    
    for epoch in range(num_epochs):
        # Training phase
        model.train()
        train_loss = 0.0
        
        for images, labels in train_loader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            loss = criterion(outputs, labels)
            
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            train_loss += loss.item() * images.size(0)
        
        train_loss = train_loss / len(train_loader.dataset)
        train_losses.append(train_loss)
        
        # Validation phase
        model.eval()
        val_loss = 0.0
        correct = 0
        total = 0
        
        with torch.no_grad():
            for images, labels in test_loader:
                images, labels = images.to(device), labels.to(device)
                outputs = model(images)
                loss = criterion(outputs, labels)
                val_loss += loss.item() * images.size(0)
                
                _, predicted = torch.max(outputs, 1)
                total += labels.size(0)
                correct += (predicted == labels).sum().item()
        
        val_loss = val_loss / len(test_loader.dataset)
        val_accuracy = 100 * correct / total
        
        val_losses.append(val_loss)
        val_accuracies.append(val_accuracy)
        
        print(f'Epoch [{epoch+1}/{num_epochs}], '
              f'Train Loss: {train_loss:.4f}, '
              f'Val Loss: {val_loss:.4f}, '
              f'Val Acc: {val_accuracy:.2f}%')
    
    return train_losses, val_losses, val_accuracies

print("Training Simple Model...")
print("=" * 70)
simple_train_losses, simple_val_losses, simple_val_accs = train_model(
    simple_model, fashion_train_loader, fashion_test_loader, num_epochs=15
)
print("\n" + "=" * 70)

print("\nTraining Deep Model...")
print("=" * 70)
deep_train_losses, deep_val_losses, deep_val_accs = train_model(
    deep_model, fashion_train_loader, fashion_test_loader, num_epochs=15
)
print("\n" + "=" * 70)

import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pandas as pd

# Create a DataFrame with all the data
comparison_df = pd.DataFrame({
    'Epoch': list(range(1, len(simple_train_losses) + 1)),
    'Simple_Train': simple_train_losses,
    'Simple_Val': simple_val_losses,
    'Simple_Acc': simple_val_accs,
    'Deep_Train': deep_train_losses,
    'Deep_Val': deep_val_losses,
    'Deep_Acc': deep_val_accs
})

# Create subplots
fig = make_subplots(
    rows=1, cols=3,
    subplot_titles=('Training Loss Comparison', 
                    'Validation Loss Comparison', 
                    'Validation Accuracy Comparison'),
    horizontal_spacing=0.1
)

# Plot 1: Training Loss
fig.add_trace(
    go.Scatter(x=comparison_df['Epoch'], y=comparison_df['Simple_Train'],
               mode='lines+markers', name='Simple Model',
               marker=dict(size=4), line=dict(width=2),
               legendgroup='simple'),
    row=1, col=1
)
fig.add_trace(
    go.Scatter(x=comparison_df['Epoch'], y=comparison_df['Deep_Train'],
               mode='lines+markers', name='Deep Model',
               marker=dict(size=4, symbol='square'), line=dict(width=2),
               legendgroup='deep'),
    row=1, col=1
)

# Plot 2: Validation Loss
fig.add_trace(
    go.Scatter(x=comparison_df['Epoch'], y=comparison_df['Simple_Val'],
               mode='lines+markers', name='Simple Model',
               marker=dict(size=4), line=dict(width=2),
               legendgroup='simple', showlegend=False),
    row=1, col=2
)
fig.add_trace(
    go.Scatter(x=comparison_df['Epoch'], y=comparison_df['Deep_Val'],
               mode='lines+markers', name='Deep Model',
               marker=dict(size=4, symbol='square'), line=dict(width=2),
               legendgroup='deep', showlegend=False),
    row=1, col=2
)

# Plot 3: Validation Accuracy
fig.add_trace(
    go.Scatter(x=comparison_df['Epoch'], y=comparison_df['Simple_Acc'],
               mode='lines+markers', name='Simple Model',
               marker=dict(size=4), line=dict(width=2),
               legendgroup='simple', showlegend=False),
    row=1, col=3
)
fig.add_trace(
    go.Scatter(x=comparison_df['Epoch'], y=comparison_df['Deep_Acc'],
               mode='lines+markers', name='Deep Model',
               marker=dict(size=4, symbol='square'), line=dict(width=2),
               legendgroup='deep', showlegend=False),
    row=1, col=3
)

# Update axes labels
fig.update_xaxes(title_text="Epoch", row=1, col=1)
fig.update_xaxes(title_text="Epoch", row=1, col=2)
fig.update_xaxes(title_text="Epoch", row=1, col=3)

fig.update_yaxes(title_text="Loss", row=1, col=1)
fig.update_yaxes(title_text="Loss", row=1, col=2)
fig.update_yaxes(title_text="Accuracy (%)", row=1, col=3)

# Update layout
fig.update_layout(
    height=400,
    width=1400,
    template='plotly_white',
    hovermode='x unified',
    legend=dict(
        yanchor="top", 
        y=0.99, 
        xanchor="left", 
        x=0.01,
        orientation="v"
    )
)

fig.show()

# Print final results
print("\n" + "=" * 70)
print("FINAL RESULTS")
print("=" * 70)
print(f"Simple Model - Final Val Accuracy: {simple_val_accs[-1]:.2f}%")
print(f"Deep Model   - Final Val Accuracy: {deep_val_accs[-1]:.2f}%")
print(f"\nImprovement: {deep_val_accs[-1] - simple_val_accs[-1]:.2f}%")
print("=" * 70)

from sklearn.metrics import confusion_matrix
import seaborn as sns

def plot_confusion_matrix(model, test_loader, class_names):
    """
    Create and plot confusion matrix
    """
    model.eval()
    all_preds = []
    all_labels = []
    
    with torch.no_grad():
        for images, labels in test_loader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            _, predicted = torch.max(outputs, 1)
            all_preds.extend(predicted.cpu().numpy())
            all_labels.extend(labels.cpu().numpy())
    
    # Create confusion matrix
    cm = confusion_matrix(all_labels, all_preds)
    
    # Plot
    plt.figure(figsize=(12, 10))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
                xticklabels=class_names,
                yticklabels=class_names)
    plt.xlabel('Predicted')
    plt.ylabel('Actual')
    plt.title('Confusion Matrix')
    plt.tight_layout()
    plt.show()
    
    return cm

print("Confusion Matrix for Deep Model:")
cm = plot_confusion_matrix(deep_model, fashion_test_loader, class_names)

def show_misclassifications(model, test_loader, class_names, num_examples=10):
    """
    Display examples of misclassified images
    """
    model.eval()
    misclassified = []
    
    with torch.no_grad():
        for images, labels in test_loader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            _, predicted = torch.max(outputs, 1)
            
            # Find misclassified examples
            mask = predicted != labels
            for i, is_wrong in enumerate(mask):
                if is_wrong:
                    misclassified.append({
                        'image': images[i].cpu(),  # Move to CPU here
                        'true': labels[i].item(),
                        'pred': predicted[i].item()
                    })
            
            if len(misclassified) >= num_examples:
                break
    
    # Plot
    fig, axes = plt.subplots(2, 5, figsize=(15, 6))
    axes = axes.ravel()
    
    for i in range(min(num_examples, len(misclassified))):
        img = misclassified[i]['image'].squeeze()
        true_label = class_names[misclassified[i]['true']]
        pred_label = class_names[misclassified[i]['pred']]
        
        axes[i].imshow(img, cmap='gray')
        axes[i].set_title(f'True: {true_label}\nPred: {pred_label}', fontsize=10)
        axes[i].axis('off')
    
    plt.tight_layout()
    plt.show()

print("Examples of Misclassified Images:")
show_misclassifications(deep_model, fashion_test_loader, class_names)