The following implementation demonstrates a deep learning approach to forecasting meteorological data. It utilizes a hybrid architecture combining Long Short-Term Memory (LSTM) networks with a Multi-Head Self-Attention mechanism to capture temporal dependencies effectively.

Library Imports and Configuration

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from sklearn.preprocessing import StandardScaler
from torch.utils.data import DataLoader, TensorDataset

# Hyperparameter Configuration
CONFIG = {
    "seq_len": 1,          # Length of the input sequence window
    "batch_sz": 16,        # Number of samples per gradient update
    "n_features": 14,      # Number of input features per timestep
    "hidden_dim": 64,      # Number of neurons in the hidden layer
    "output_dim": 1,       # Dimension of the output (scalar for regression)
    "n_layers": 3,         # Number of stacked LSTM layers
    "epochs": 10,          # Total number of training epochs
    "learning_rate": 0.01, # Step size for optimizer
    "model_path": "weather_attn_lstm.pth"
}

Data Loading and Preprocessing

# Load dataset
df = pd.read_csv('data/weather_data.csv', index_col=0)

# Initialize scalers
feature_scaler = StandardScaler()
target_scaler = StandardScaler()

# Fit and transform the entire dataset
scaled_data = feature_scaler.fit_transform(df.values)

# Fit target scaler specifically on the temperature column (assumed 'T (degC)')
target_scaler.fit(df['T (degC)'].values.reshape(-1, 1))

def prepare_sequences(dataset, window_size, n_vars):
    """Creates sequences for supervised learning."""
    X, y = [], []
    total_len = len(dataset)
    
    for i in range(total_len - window_size):
        # Extract features for the current window
        X.append(dataset[i : i + window_size])
        # Extract the target (temperature at the next step)
        y.append(dataset[i + window_size][1]) 
    
    X = np.array(X)
    y = np.array(y)
    
    # Split into training and testing sets (80/20 split)
    split_idx = int(0.8 * X.shape[0])
    
    x_train = X[:split_idx].reshape(-1, window_size, n_vars)
    y_train = y[:split_idx]
    x_test = X[split_idx:].reshape(-1, window_size, n_vars)
    y_test = y[split_idx:]
    
    return x_train, y_train, x_test, y_test

# Generate datasets
x_train, y_train, x_test, y_test = prepare_sequences(
    scaled_data, 
    CONFIG["seq_len"], 
    CONFIG["n_features"]
)

# Convert to PyTorch Tensors
train_dataset = TensorDataset(
    torch.from_numpy(x_train).float(), 
    torch.from_numpy(y_train).float()
)
test_dataset = TensorDataset(
    torch.from_numpy(x_test).float(), 
    torch.from_numpy(y_test).float()
)

# Create DataLoaders
train_loader = DataLoader(train_dataset, batch_size=CONFIG["batch_sz"], shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=CONFIG["batch_sz"], shuffle=False)

Model Architecture Definition

class AttnLSTMNet(nn.Module):
    def __init__(self, input_sz, hidden_sz, num_layers, out_sz):
        super(AttnLSTMNet, self).__init__()
        self.hidden_sz = hidden_sz
        self.num_layers = num_layers
        
        # Multi-Head Attention layer
        self.attention_layer = nn.MultiheadAttention(
            embed_dim=input_sz, 
            num_heads=2
        )
        
        # LSTM layer for sequence processing
        self.lstm_layer = nn.LSTM(
            input_sz, 
            hidden_sz, 
            num_layers, 
            batch_first=True
        )
        
        # Fully connected layer for final prediction
        self.fc_layer = nn.Linear(hidden_sz, out_sz)

    def forward(self, x):
        # Input shape: (batch_size, seq_len, input_dim)
        
        # Self-Attention mechanism
        attn_output, attn_weights = self.attention_layer(x, x, x)
        
        # LSTM processing
        lstm_out, (h_n, c_n) = self.lstm_layer(attn_output)
        
        # Reshape for fully connected layer
        batch_size, seq_len, hidden_dim = lstm_out.shape
        lstm_flat = lstm_out.reshape(-1, hidden_dim)
        
        # Apply linear transformation
        predictions = self.fc_layer(lstm_flat)
        
        # Restore sequence dimensions and return last timestep
        predictions = predictions.reshape(seq_len, batch_size, -1)
        return predictions[-1]

# Initialize model, loss, and optimizer
model = AttnLSTMNet(
    CONFIG["n_features"], 
    CONFIG["hidden_dim"], 
    CONFIG["n_layers"], 
    CONFIG["output_dim"]
)
loss_criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=CONFIG["learning_rate"])

Model Training and Validation

min_val_loss = float('inf')

for epoch in range(CONFIG["epochs"]):
    model.train()
    running_loss = 0.0
    
    # Training phase
    for inputs, targets in train_loader:
        optimizer.zero_grad()
        
        # Forward pass
        outputs = model(inputs)
        
        # Calculate loss
        loss = loss_criterion(outputs, targets.unsqueeze(1))
        
        # Backward pass and optimization
        loss.backward()
        optimizer.step()
        
        running_loss += loss.item()
    
    # Validation phase
    model.eval()
    val_loss = 0.0
    with torch.no_grad():
        for inputs, targets in test_loader:
            outputs = model(inputs)
            loss = loss_criterion(outputs, targets.unsqueeze(1))
            val_loss += loss.item()
    
    avg_val_loss = val_loss / len(test_loader)
    print(f"Epoch [{epoch+1}/{CONFIG["epochs"]}] Train Loss: {running_loss/len(train_loader):.4f} | Val Loss: {avg_val_loss:.4f}")
    
    # Save best model
    if avg_val_loss < min_val_loss:
        min_val_loss = avg_val_loss
        torch.save(model.state_dict(), CONFIG["model_path"])

Result Visualization

model.eval()
with torch.no_grad():
    # Predict on first 10,000 training samples
    sample_input = torch.tensor(x_train[:10000], dtype=torch.float32)
    predictions = model(sample_input).numpy()

plt.figure(figsize=(12, 8))

# Inverse transform to original scale
plt.plot(
    target_scaler.inverse_transform(predictions.reshape(-1, 1)), 
    "b", 
    label="Predicted Values"
)
plt.plot(
    target_scaler.inverse_transform(y_train[:10000].reshape(-1, 1)), 
    "r", 
    label="Actual Values"
)

plt.legend()
plt.show()