Predicting Time Series Data with Machine Learning, Generative AI, and Deep Learning
11 min readJun 19, 2024
Time series data prediction is a critical aspect of various industries, ranging from finance and healthcare to marketing and logistics. The ability to forecast future values based on historical data can drive significant improvements in decision-making processes and operational efficiency. With advancements in machine learning, generative AI, and deep learning, there are now more sophisticated methods available for tackling time series prediction problems. This blog will explore different approaches and models that can be used for time series data prediction.
Understanding Time Series Data
Time series data is a sequence of data points collected or recorded at specific time intervals. Examples include stock prices, weather data, sales figures, and sensor readings. The goal of time series prediction is to use past observations to predict future values, which can be challenging due to the inherent complexities and patterns within the data.
1. Machine Learning Approaches
1.1 ARIMA (AutoRegressive Integrated Moving Average)
- ARIMA is a classical statistical method for time series forecasting. It combines autoregressive (AR) models, differencing (to make the data stationary), and moving average (MA) models.
Example Usage :
import pandas as pd
from statsmodels.tsa.arima.model import ARIMA
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Fit ARIMA model
model = ARIMA(time_series_data['Value'], order=(5, 1, 0)) # (p,d,q)
model_fit = model.fit()
# Make predictions
predictions = model_fit.forecast(steps=10)
print(predictions)1.2 SARIMA (Seasonal ARIMA)
- SARIMA extends ARIMA by considering seasonal effects. It’s useful for data with seasonal patterns, such as monthly sales data.
Example Usage :
import pandas as pd
import numpy as np
from statsmodels.tsa.statespace.sarimax import SARIMAX
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Fit SARIMA model
model = SARIMAX(time_series_data['Value'], order=(1, 1, 1), seasonal_order=(1, 1, 1, 12)) # (p,d,q) (P,D,Q,s)
model_fit = model.fit(disp=False)
# Make predictions
predictions = model_fit.forecast(steps=10)
print(predictions)1.3 Prophet
- Developed by Facebook, Prophet is a powerful tool designed for forecasting time series data that can handle missing data and outliers and provide reliable uncertainty intervals.
Example Usage :
from fbprophet import Prophet
import pandas as pd
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.rename(columns={'Date': 'ds', 'Value': 'y'}, inplace=True)
# Fit Prophet model
model = Prophet()
model.fit(time_series_data)
# Make future dataframe and predictions
future = model.make_future_dataframe(periods=10)
forecast = model.predict(future)
print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']])1.4 XGBoost
- XGBoost is a gradient boosting framework that can be used for time series prediction by transforming the problem into a supervised learning task, treating previous time steps as features.
Example Usage :
import pandas as pd
import numpy as np
from xgboost import XGBRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for supervised learning
def create_lag_features(data, lag=1):
df = data.copy()
for i in range(1, lag + 1):
df[f'lag_{i}'] = df['Value'].shift(i)
return df.dropna()
lag = 5
data_with_lags = create_lag_features(time_series_data, lag=lag)
X = data_with_lags.drop('Value', axis=1)
y = data_with_lags['Value']
# Split the data into training and testing sets
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
# Fit XGBoost model
model = XGBRegressor(objective='reg:squarederror', n_estimators=1000)
model.fit(X_train, y_train)
# Make predictions
y_pred = model.predict(X_test)
mse = mean_squared_error(y_test, y_pred)
print(f'Mean Squared Error: {mse}')2. Generative AI Approaches
2.1 GANs (Generative Adversarial Networks)
- GANs consist of a generator and a discriminator. For time series prediction, GANs can generate plausible future sequences by learning the underlying data distribution.
Example Usage :
import numpy as np
import pandas as pd
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM, Conv1D, MaxPooling1D, Flatten, LeakyReLU, Reshape
from tensorflow.keras.optimizers import Adam
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for GAN
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
X_train, y_train = create_dataset(scaled_data, time_step)
X_train = X_train.reshape(X_train.shape[0], X_train.shape[1], 1)
# GAN components
def build_generator():
model = Sequential()
model.add(Dense(100, input_dim=time_step))
model.add(LeakyReLU(alpha=0.2))
model.add(Dense(time_step, activation='tanh'))
model.add(Reshape((time_step, 1)))
return model
def build_discriminator():
model = Sequential()
model.add(LSTM(50, input_shape=(time_step, 1)))
model.add(Dense(1, activation='sigmoid'))
return model
# Build and compile the discriminator
discriminator = build_discriminator()
discriminator.compile(loss='binary_crossentropy', optimizer=Adam(0.0002, 0.5), metrics=['accuracy'])
# Build the generator
generator = build_generator()
# The generator takes noise as input and generates data
z = Input(shape=(time_step,))
generated_data = generator(z)
# For the combined model, we will only train the generator
discriminator.trainable = False
# The discriminator takes generated data as input and determines validity
validity = discriminator(generated_data)
# The combined model (stacked generator and discriminator)
combined = Model(z, validity)
combined.compile(loss='binary_crossentropy', optimizer=Adam(0.0002, 0.5))
# Training the GAN
epochs = 10000
batch_size = 32
valid = np.ones((batch_size, 1))
fake = np.zeros((batch_size, 1))
for epoch in range(epochs):
# ---------------------
# Train Discriminator
# ---------------------
# Select a random batch of real data
idx = np.random.randint(0, X_train.shape[0], batch_size)
real_data = X_train[idx]
# Generate a batch of fake data
noise = np.random.normal(0, 1, (batch_size, time_step))
gen_data = generator.predict(noise)
# Train the discriminator
d_loss_real = discriminator.train_on_batch(real_data, valid)
d_loss_fake = discriminator.train_on_batch(gen_data, fake)
d_loss = 0.5 * np.add(d_loss_real, d_loss_fake)
# ---------------------
# Train Generator
# ---------------------
noise = np.random.normal(0, 1, (batch_size, time_step))
# Train the generator (to have the discriminator label samples as valid)
g_loss = combined.train_on_batch(noise, valid)
# Print the progress
if epoch % 1000 == 0:
print(f"{epoch} [D loss: {d_loss[0]} | D accuracy: {100*d_loss[1]}] [G loss: {g_loss}]")
# Make predictions
noise = np.random.normal(0, 1, (1, time_step))
generated_prediction = generator.predict(noise)
generated_prediction = scaler.inverse_transform(generated_prediction)
print(generated_prediction)2.2 WaveNet
- Developed by DeepMind, WaveNet is a deep generative model originally designed for audio generation but has been adapted for time series forecasting, especially in the domain of audio and speech.
Example Usage :
import numpy as np
import pandas as pd
import tensorflow as tf
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Conv1D, Add, Activation, Multiply, Lambda, Dense, Flatten
from tensorflow.keras.optimizers import Adam
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for WaveNet
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
X, y = create_dataset(scaled_data, time_step)
X = X.reshape(X.shape[0], X.shape[1], 1)
# Define WaveNet model
def residual_block(x, dilation_rate):
tanh_out = Conv1D(32, kernel_size=2, dilation_rate=dilation_rate, padding='causal', activation='tanh')(x)
sigm_out = Conv1D(32, kernel_size=2, dilation_rate=dilation_rate, padding='causal', activation='sigmoid')(x)
out = Multiply()([tanh_out, sigm_out])
out = Conv1D(32, kernel_size=1, padding='same')(out)
out = Add()([out, x])
return out
input_layer = Input(shape=(time_step, 1))
out = Conv1D(32, kernel_size=2, padding='causal', activation='tanh')(input_layer)
skip_connections = []
for i in range(10):
out = residual_block(out, 2**i)
skip_connections.append(out)
out = Add()(skip_connections)
out = Activation('relu')(out)
out = Conv1D(1, kernel_size=1, activation='relu')(out)
out = Flatten()(out)
out = Dense(1)(out)
model = Model(input_layer, out)
model.compile(optimizer=Adam(learning_rate=0.001), loss='mean_squared_error')
# Train the model
model.fit(X, y, epochs=10, batch_size=16)
# Make predictions
predictions = model.predict(X)
predictions = scaler.inverse_transform(predictions)
print(predictions)3. Deep Learning Approaches
3.1 LSTM (Long Short-Term Memory)
LSTM networks are a type of recurrent neural network (RNN) capable of learning long-term dependencies. They are widely used for time series prediction due to their ability to capture temporal patterns.
Example Usage :
import numpy as np
import pandas as pd
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from sklearn.preprocessing import MinMaxScaler
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for LSTM
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
train_size = int(len(scaled_data) * 0.8)
train_data = scaled_data[:train_size]
test_data = scaled_data[train_size:]
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
X_train, y_train = create_dataset(train_data, time_step)
X_test, y_test = create_dataset(test_data, time_step)
X_train = X_train.reshape(X_train.shape[0], X_train.shape[1], 1)
X_test = X_test.reshape(X_test.shape[0], X_test.shape[1], 1)
# Build LSTM model
model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(time_step, 1)))
model.add(LSTM(50, return_sequences=False))
model.add(Dense(25))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
model.fit(X_train, y_train, batch_size=1, epochs=1)
# Make predictions
train_predict = model.predict(X_train)
test_predict = model.predict(X_test)
train_predict = scaler.inverse_transform(train_predict)
test_predict = scaler.inverse_transform(test_predict)
print(test_predict)3.2 GRU (Gated Recurrent Unit)
GRU is a variant of LSTM that is simpler and often performs equally well for time series tasks. GRUs are used to model sequences and capture temporal dependencies.
Example Usage :
import numpy as np
import pandas as pd
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import GRU, Dense
from sklearn.preprocessing import MinMaxScaler
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for GRU
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
train_size = int(len(scaled_data) * 0.8)
train_data = scaled_data[:train_size]
test_data = scaled_data[train_size:]
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
X_train, y_train = create_dataset(train_data, time_step)
X_test, y_test = create_dataset(test_data, time_step)
X_train = X_train.reshape(X_train.shape[0], X_train.shape[1], 1)
X_test = X_test.reshape(X_test.shape[0], X_test.shape[1], 1)
# Build GRU model
model = Sequential()
model.add(GRU(50, return_sequences=True, input_shape=(time_step, 1)))
model.add(GRU(50, return_sequences=False))
model.add(Dense(25))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
model.fit(X_train, y_train, batch_size=1, epochs=1)
# Make predictions
train_predict = model.predict(X_train)
test_predict = model.predict(X_test)
train_predict = scaler.inverse_transform(train_predict)
test_predict = scaler.inverse_transform(test_predict)
print(test_predict)3.3 Transformer Models
Transformers, known for their success in NLP tasks, have been adapted for time series prediction. Models like the Temporal Fusion Transformer (TFT) leverage the attention mechanism to handle temporal data effectively.
Example Usage :
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM, Conv1D, MaxPooling1D, Flatten, MultiHeadAttention, LayerNormalization, Dropout
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
train_size = int(len(scaled_data) * 0.8)
train_data = scaled_data[:train_size]
test_data = scaled_data[train_size:]
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
X_train, y_train = create_dataset(train_data, time_step)
X_test, y_test = create_dataset(test_data, time_step)
X_train = X_train.reshape(X_train.shape[0], X_train.shape[1], 1)
X_test = X_test.reshape(X_test.shape[0], X_test.shape[1], 1)
# Build Transformer model
model = Sequential()
model.add(MultiHeadAttention(num_heads=4, key_dim=2, input_shape=(time_step, 1)))
model.add(LayerNormalization())
model.add(Dense(50, activation='relu'))
model.add(Dropout(0.1))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
model.fit(X_train, y_train, batch_size=1, epochs=1)
# Make predictions
train_predict = model.predict(X_train)
test_predict = model.predict(X_test)
train_predict = scaler.inverse_transform(train_predict)
test_predict = scaler.inverse_transform(test_predict)
print(test_predict)3.4 Seq2Seq (Sequence to Sequence)
Seq2Seq models are used for predicting sequences of data. Originally developed for language translation, they are effective for time series forecasting by learning the mapping from input sequences to output sequences.
Example Usage :
import numpy as np
import pandas as pd
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, LSTM, Dense
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for Seq2Seq
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
X, y = create_dataset(scaled_data, time_step)
X = X.reshape(X.shape[0], X.shape[1], 1)
# Define Seq2Seq model
encoder_inputs = Input(shape=(time_step, 1))
encoder = LSTM(50, return_state=True)
encoder_outputs, state_h, state_c = encoder(encoder_inputs)
decoder_inputs = Input(shape=(time_step, 1))
decoder_lstm = LSTM(50, return_sequences=True, return_state=True)
decoder_outputs, _, _ = decoder_lstm(decoder_inputs, initial_state=[state_h, state_c])
decoder_dense = Dense(1)
decoder_outputs = decoder_dense(decoder_outputs)
model = Model([encoder_inputs, decoder_inputs], decoder_outputs)
model.compile(optimizer='adam', loss='mean_squared_error')
# Train the model
model.fit([X, X], y, epochs=10, batch_size=16)
# Make predictions
predictions = model.predict([X, X])
predictions = scaler.inverse_transform(predictions)
print(predictions)3.5 TCN (Temporal Convolutional Networks)
TCNs use dilated convolutions to capture long-term dependencies in time series data. They provide a robust alternative to RNNs for sequential data modeling.
Example Usage :
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv1D, Dense, Flatten
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for TCN
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
X, y = create_dataset(scaled_data, time_step)
X = X.reshape(X.shape[0], X.shape[1], 1)
# Define TCN model
model = Sequential()
model.add(Conv1D(filters=64, kernel_size=2, dilation_rate=1, activation='relu', input_shape=(time_step, 1)))
model.add(Conv1D(filters=64, kernel_size=2, dilation_rate=2, activation='relu'))
model.add(Conv1D(filters=64, kernel_size=2, dilation_rate=4, activation='relu'))
model.add(Flatten())
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
# Train the model
model.fit(X, y, epochs=10, batch_size=16)
# Make predictions
predictions = model.predict(X)
predictions = scaler.inverse_transform(predictions)
print(predictions)3.6 DeepAR
Developed by Amazon, DeepAR is an autoregressive recurrent network designed for time series forecasting. It handles multiple time series and can capture complex patterns.
Example Usage :
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Flatten
# Load your time series data
time_series_data = pd.read_csv('time_series_data.csv')
time_series_data['Date'] = pd.to_datetime(time_series_data['Date'])
time_series_data.set_index('Date', inplace=True)
# Prepare data for DeepAR-like model
def create_dataset(dataset, time_step=1):
X, Y = [], []
for i in range(len(dataset)-time_step-1):
a = dataset[i:(i+time_step), 0]
X.append(a)
Y.append(dataset[i + time_step, 0])
return np.array(X), np.array(Y)
time_step = 10
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(time_series_data['Value'].values.reshape(-1, 1))
X, y = create_dataset(scaled_data, time_step)
X = X.reshape(X.shape[0], X.shape[1], 1)
# Define DeepAR-like model
model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(time_step, 1)))
model.add(LSTM(50))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
# Train the model
model.fit(X, y, epochs=10, batch_size=16)
# Make predictions
predictions = model.predict(X)
predictions = scaler.inverse_transform(predictions)
print(predictions)Time series data prediction is a complex yet fascinating field that benefits immensely from machine learning, generative AI, and deep learning advancements. By leveraging models like ARIMA, Prophet, LSTM, and Transformers, practitioners can uncover hidden patterns in data and make accurate forecasts. As technology continues to evolve, the tools and methods available for time series prediction will only become more sophisticated, offering new opportunities for innovation and improvement across various domains.
