Skip to content

Traffic data fluctuates constantly or is affected by time. Predicting it can be challenging, but this task will help sharpen your time-series skills. With deep learning, you can use abstract patterns in data that can help boost predictability.

Your task is to build a system that can be applied to help you predict traffic volume or the number of vehicles passing at a specific point and time. Determining this can help reduce road congestion, support new designs for roads or intersections, improve safety, and more! Or, you can use to help plan your commute to avoid traffic!

The dataset provided contains the hourly traffic volume on an interstate highway in Minnesota, USA. It also includes weather features and holidays, which often impact traffic volume.

Time to predict some traffic!

The data:

The dataset is collected and maintained by UCI Machine Learning Repository. The target variable is traffic_volume. The dataset contains the following and has already been normalized and saved into training and test sets:

train_scaled.csv, test_scaled.csv

ColumnTypeDescription
tempNumericAverage temp in kelvin
rain_1hNumericAmount in mm of rain that occurred in the hour
snow_1hNumericAmount in mm of snow that occurred in the hour
clouds_allNumericPercentage of cloud cover
date_timeDateTimeHour of the data collected in local CST time
holiday_ (11 columns)CategoricalUS National holidays plus regional holiday, Minnesota State Fair
weather_main_ (11 columns)CategoricalShort textual description of the current weather
weather_description_ (35 columns)CategoricalLonger textual description of the current weather
traffic_volumeNumericHourly I-94 ATR 301 reported westbound traffic volume
hour_of_dayNumericThe hour of the day
day_of_weekNumericThe day of the week (0=Monday, Sunday=6)
day_of_monthNumericThe day of the month
monthNumericThe number of the month
traffic_volumeNumericHourly I-94 ATR 301 reported westbound traffic volume
# Import the relevant libraries
import numpy as np
import pandas as pd

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import TensorDataset, DataLoader
# Read the traffic data from the CSV training and test files
train_scaled_df = pd.read_csv('train_scaled.csv')
test_scaled_df = pd.read_csv('test_scaled.csv')

# Convert the DataFrame to NumPy arrays
train_scaled = train_scaled_df.to_numpy()
test_scaled = test_scaled_df.to_numpy()
# Split the data into features and target
X_train = train_scaled[:, :-1]  # All columns except the last one
y_train = train_scaled[:, -1]   # The last column is the target

X_test = test_scaled[:, :-1]    # All columns except the last one
y_test = test_scaled[:, -1]     # The last column is the target

# Convert to PyTorch tensors
X_train_tensor = torch.tensor(X_train, dtype=torch.float32)
y_train_tensor = torch.tensor(y_train, dtype=torch.float32).view(-1, 1)

X_test_tensor = torch.tensor(X_test, dtype=torch.float32)
y_test_tensor = torch.tensor(y_test, dtype=torch.float32).view(-1, 1)
# Reshape data for LSTM (batch_size, sequence_length, input_size)
# Since this is hourly data, we can treat each hour as a time step
sequence_length = 1  # Each row is a single time step
X_train_tensor = X_train_tensor.unsqueeze(1)  # Add sequence dimension
X_test_tensor = X_test_tensor.unsqueeze(1)    # Add sequence dimension

# Create TensorDatasets and 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=64, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)
# Define the LSTM model
class TrafficLSTM(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size):
        super(TrafficLSTM, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        # Initialize hidden state and cell state
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
        
        # Forward propagate LSTM
        out, _ = self.lstm(x, (h0, c0))
        
        # Decode the hidden state of the last time step
        out = self.fc(out[:, -1, :])
        return out
# Initialize the LSTM model
input_size = X_train.shape[1]  # Number of features
hidden_size = 64               # Number of hidden units
num_layers = 2                  # Number of LSTM layers
output_size = 1                 # Predicting a single value (traffic volume)

traffic_model = TrafficLSTM(input_size, hidden_size, num_layers, output_size)

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

# Training loop
num_epochs = 20
for epoch in range(num_epochs):
    traffic_model.train()
    for i, (inputs, labels) in enumerate(train_loader):
        # Forward pass
        outputs = traffic_model(inputs)
        loss = criterion(outputs, labels)
        
        # Backward pass and optimization
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    
    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')

# Save the final training loss
final_training_loss = torch.tensor(loss.item())
# Evaluate the model on the test set
traffic_model.eval()
with torch.no_grad():
    test_outputs = traffic_model(X_test_tensor)
    test_mse = criterion(test_outputs, y_test_tensor)

# Save the test MSE
test_mse = torch.tensor(test_mse.item())

# Print the final training loss and test MSE
print(f'Final Training Loss: {final_training_loss.item():.4f}')
print(f'Test MSE: {test_mse.item():.4f}')

# Save the model
torch.save(traffic_model.state_dict(), 'traffic_model.pth')