Introduction.
Multi-step time series forecasting involves predicting multiple future time steps in a time series sequence. Mathematically, let represent the value of the time series at time . Multi-step forecasting aims to predict the future values of the time series over a horizon of time steps. Therefore, the forecasted values can be represented as , where denotes the predicted value at time for .
Multi-variate time series forecasting involves predicting the future values of a time series using multiple input variables, where each variable can influence the target time series. Mathematically, let represent input variables at time , and represent the target time series. The goal of multi-variate time series forecasting is to predict the future values of the target time series, , based on the input variables. This can be represented as a function such that for .
This article contains the following topics with supported video tutorial and code.
- Multivariate Multi-Step Multi-Output Time series Forecasting
- Strategy to prepare dataset.
- Multivariate Single-Step Multi-Output Time series Forecasting
- Strategy to prepare dataset.
- Strategy for the Future Enhancements.
Part-1.
Part-2.
Code-1. [Single-Step Multi-Output]
from keras import Model
from keras.layers import Input, Dense, Bidirectional, LSTM, RepeatVector, TimeDistributed
from sklearn.preprocessing import MinMaxScaler
from numpy import array , hstack
import numpy as np
data_X1 = [21, 26, 31, 36, 41, 46, 51, 56, 61, 65, 70, 75, 80, 85, 90, 95, 100, 105, 110, 115]
data_X2 = [2, 5, 8, 11, 14, 17, 20, 23, 26, 29, 32, 35, 38, 41, 44, 47, 50, 53, 56, 59]
data_X3 = [13, 17, 21, 25, 29, 33, 37, 41, 44, 48, 52, 56, 60, 64, 68, 72, 76, 80, 84, 88]
data_X1 = np.array(data_X1)
data_X2 = np.array(data_X2)
data_X3 = np.array(data_X3)
# Step 1 : convert to [rows, columns] structure
data_X1 = data_X1.reshape((len(data_X1), 1))
data_X2 = data_X2.reshape((len(data_X2), 1))
data_X3 = data_X3.reshape((len(data_X3), 1))
print ("data_X1.shape" , data_X1.shape)
print ("data_X2.shape" , data_X2.shape)
print ("data_X3.shape" , data_X3.shape)
data_scalar = MinMaxScaler(feature_range=(0,1))
data_X1_scaled = data_scalar.fit_transform(data_X1)
data_X2_scaled = data_scalar.fit_transform(data_X2)
data_X3_scaled = data_scalar.fit_transform(data_X3)
# Step 3 : horizontally stack columns
dataset_stacked = hstack((data_X1_scaled, data_X2_scaled, data_X3_scaled))
dataset_stacked = np.array(dataset_stacked)
print ("dataset_stacked.shape" , dataset_stacked.shape)
print("final dataset => ",dataset_stacked)
# prepare the dataset
input_timesteps=3
input_features=3
output_timesteps=1
output_features=3
data_Y = []
data_X = []
for i in range(0, ((len(data_X1))-(input_timesteps+output_timesteps))):
print("---------")
tmpY2d = []
for i_row in range(i,(i + output_timesteps)):
tmpYr = []
print(i_row)
for i_col in range(0,output_features):
tmpYr.append(dataset_stacked[i_row][i_col])
tmpY2d.append(tmpYr)
data_Y.append(tmpY2d)
print("---------")
tmpX2d = []
for j_row in range((i+output_timesteps), (i + output_timesteps + input_timesteps)):
tmpXr = []
print(j_row)
for j_col in range(0, input_features):
tmpXr.append(dataset_stacked[j_row][j_col])
tmpX2d.append(tmpXr)
data_X.append(tmpX2d)
data_X = np.array(data_X)
data_Y = np.array(data_Y)
print("shape of input data ",data_X.shape)
print("input data => ",data_X)
print("shape of output data => ", data_Y.shape)
print("Output data => ",data_Y)
def define_model():
# Define the Input data shape
encoder_inputs = Input(shape=(input_timesteps, input_features))
# Use single BiLSTM as Encoder
# Here we can use bigger network also like one BiLSTM with return_sequences=True and
# other BiLSTM with return_sequences=False
# OR CNN, CNN+LSTM and so many
encoder = Bidirectional(LSTM(units=16, return_sequences=True))(encoder_inputs)
# Apply RepeatVector to get the result for multiple time steps (here our output_timesteps =2)
# For this step Decoder operation starts
# repeat_output = RepeatVector(output_timesteps)(encoder)
decoder = Bidirectional(LSTM(units=16, return_sequences=False))(encoder)
# Use TimeDistributed layer to get multiple Output features
out = Dense(output_features)(decoder)
# out = TimeDistributed(Dense(output_features))(decoder)
model = Model(encoder_inputs, out)
# Compile the model
model.compile(loss='mae', optimizer='adam', metrics=['mae'])
model.summary()
return model
# Call the model
model = define_model()
# Fit the model
model.fit(data_X,data_Y,epochs=4,batch_size=2,verbose=1)
# Take a test data to test the working of the model
test_dataX = []
test_dataX.append(dataset_stacked[0])
test_dataX.append(dataset_stacked[1])
test_dataX.append(dataset_stacked[2])
test_dataX = np.array(test_dataX)
# Reshape the data into 3-D numpy array
test_dataX = np.reshape(test_dataX,(1,input_timesteps,input_features))
print("test dataset => ",test_dataX)
# Run for the prediction output
pred_output = model.predict(test_dataX)
print("prediction output => ",pred_output)
# invert the scaling to get forecast values
inverted_output = data_scalar.inverse_transform(pred_output)
print("obtained prediction => ",inverted_output)
Code-2. [Multi-Step and Multi-Output]
from keras import Model
from keras.layers import Input, Dense, Bidirectional, LSTM, RepeatVector, TimeDistributed
from sklearn.preprocessing import MinMaxScaler
from numpy import array , hstack
import numpy as np
data_X1 = [21, 26, 31, 36, 41, 46, 51, 56, 61, 65, 70, 75, 80, 85, 90, 95, 100, 105, 110, 115]
data_X2 = [2, 5, 8, 11, 14, 17, 20, 23, 26, 29, 32, 35, 38, 41, 44, 47, 50, 53, 56, 59]
data_X3 = [13, 17, 21, 25, 29, 33, 37, 41, 44, 48, 52, 56, 60, 64, 68, 72, 76, 80, 84, 88]
data_X1 = np.array(data_X1)
data_X2 = np.array(data_X2)
data_X3 = np.array(data_X3)
# Step 1 : convert to [rows, columns] structure
data_X1 = data_X1.reshape((len(data_X1), 1))
data_X2 = data_X2.reshape((len(data_X2), 1))
data_X3 = data_X3.reshape((len(data_X3), 1))
print ("data_X1.shape" , data_X1.shape)
print ("data_X2.shape" , data_X2.shape)
print ("data_X3.shape" , data_X3.shape)
# Step 2: Do scaling
data_scalar = MinMaxScaler(feature_range=(0,1))
data_X1_scaled = data_scalar.fit_transform(data_X1)
data_X2_scaled = data_scalar.fit_transform(data_X2)
data_X3_scaled = data_scalar.fit_transform(data_X3)
# Step 3 : horizontally stack columns
dataset_stacked = hstack((data_X1_scaled, data_X2_scaled, data_X3_scaled))
dataset_stacked = np.array(dataset_stacked)
print ("dataset_stacked.shape" , dataset_stacked.shape)
print("final dataset => ",dataset_stacked)
# prepare the dataset
input_timesteps=3
input_features=3
output_timesteps=2
output_features=3
data_Y = []
data_X = []
for i in range(0, ((len(data_X1))-(input_timesteps+output_timesteps))):
tmpY2d = []
for i_row in range(i,(i + output_timesteps)):
tmpYr = []
for i_col in range(0,output_features):
tmpYr.append(dataset_stacked[i_row][i_col])
tmpY2d.append(tmpYr)
data_Y.append(tmpY2d)
tmpX2d = []
for j_row in range((i+output_timesteps), (i + output_timesteps + input_timesteps)):
tmpXr = []
for j_col in range(0, input_features):
tmpXr.append(dataset_stacked[j_row][j_col])
tmpX2d.append(tmpXr)
data_X.append(tmpX2d)
data_X = np.array(data_X)
data_Y = np.array(data_Y)
print("shape of input data ",data_X.shape)
print("input data => ",data_X)
print("shape of output data => ", data_Y.shape)
print("Output data => ",data_Y)
def define_model():
# Define the Input data shape
encoder_inputs = Input(shape=(input_timesteps, input_features))
# Use single BiLSTM as Encoder
# Here we can use bigger network also like one BiLSTM with return_sequences=True and
# other BiLSTM with return_sequences=False
# OR CNN, CNN+LSTM and so many
encoder = Bidirectional(LSTM(units=16, return_sequences=False))(encoder_inputs)
# Apply RepeatVector to get the result for multiple time steps (here our output_timesteps =2)
# For this step Decoder operation starts
repeat_output = RepeatVector(output_timesteps)(encoder)
decoder = Bidirectional(LSTM(units=16, return_sequences=True))(repeat_output)
# Use TimeDistributed layer to get multiple Output features
out = TimeDistributed(Dense(output_features))(decoder)
model = Model(encoder_inputs, out)
# Compile the model
model.compile(loss='mae', optimizer='adam', metrics=['mae'])
model.summary()
return model
# Call the model
model = define_model()
# Fit the model
model.fit(data_X,data_Y,epochs=4,batch_size=2,verbose=1)
# Take a test data to test the working of the model
test_dataX = []
test_dataX.append(dataset_stacked[0])
test_dataX.append(dataset_stacked[1])
test_dataX.append(dataset_stacked[2])
test_dataX = np.array(test_dataX)
# Reshape the data into 3-D numpy array
test_dataX = np.reshape(test_dataX,(1,input_timesteps,input_features))
print("test dataset => ",test_dataX)
# Run for the prediction output
pred_output = model.predict(test_dataX)
print("prediction output => ",pred_output)
# invert the scaling to get forecast values
inverted_output = data_scalar.inverse_transform(pred_output[0])
print("obtained prediction => ",inverted_output)
Note. This code is not intended for any commercial use. It is created solely for simple educational purposes.
Niraj Kumar
