Chapter 11

Timeseries forecasting

NotebookPython 342 cells

This is a companion notebook for the book Deep Learning with Python, Third Edition. For readability, it only contains runnable code blocks and section titles, and omits everything else in the book: text paragraphs, figures, and pseudocode.

If you want to be able to follow what's going on, I recommend reading the notebook side by side with your copy of the book.

The book's contents are available online at deeplearningwithpython.io.

In [0]python · cell 2

python

!pip install keras keras-hub --upgrade -q

In [0]python · cell 3

python

import os
os.environ["KERAS_BACKEND"] = "jax"

In [0]python · cell 4

python

# @title
import os
from IPython.core.magic import register_cell_magic

@register_cell_magic
def backend(line, cell):
    current, required = os.environ.get("KERAS_BACKEND", ""), line.split()[-1]
    if current == required:
        get_ipython().run_cell(cell)
    else:
        print(
            f"This cell requires the {required} backend. To run it, change KERAS_BACKEND to "
            f"\"{required}\" at the top of the notebook, restart the runtime, and rerun the notebook."
        )

Timeseries forecasting

Different kinds of timeseries tasks

A temperature forecasting example

In [0]python · cell 8

python

!wget https://s3.amazonaws.com/keras-datasets/jena_climate_2009_2016.csv.zip
!unzip jena_climate_2009_2016.csv.zip

In [0]python · cell 9

python

import os

fname = os.path.join("jena_climate_2009_2016.csv")

with open(fname) as f:
    data = f.read()

lines = data.split("\n")
header = lines[0].split(",")
lines = lines[1:]
print(header)
print(len(lines))

In [0]python · cell 10

python

import numpy as np

temperature = np.zeros((len(lines),))
raw_data = np.zeros((len(lines), len(header) - 1))

for i, line in enumerate(lines):
    values = [float(x) for x in line.split(",")[1:]]
    temperature[i] = values[1]
    raw_data[i, :] = values[:]

In [0]python · cell 11

python

from matplotlib import pyplot as plt

plt.plot(range(len(temperature)), temperature)

In [0]python · cell 12

python

plt.plot(range(1440), temperature[:1440])

In [0]python · cell 13

python

num_train_samples = int(0.5 * len(raw_data))
num_val_samples = int(0.25 * len(raw_data))
num_test_samples = len(raw_data) - num_train_samples - num_val_samples
print("num_train_samples:", num_train_samples)
print("num_val_samples:", num_val_samples)
print("num_test_samples:", num_test_samples)

Preparing the data

In [0]python · cell 15

python

mean = raw_data[:num_train_samples].mean(axis=0)
raw_data -= mean
std = raw_data[:num_train_samples].std(axis=0)
raw_data /= std

In [0]python · cell 16

python

import numpy as np
import keras

int_sequence = np.arange(10)
dummy_dataset = keras.utils.timeseries_dataset_from_array(
    data=int_sequence[:-3],
    targets=int_sequence[3:],
    sequence_length=3,
    batch_size=2,
)

for inputs, targets in dummy_dataset:
    for i in range(inputs.shape[0]):
        print([int(x) for x in inputs[i]], int(targets[i]))

In [0]python · cell 17

python

sampling_rate = 6
sequence_length = 120
delay = sampling_rate * (sequence_length + 24 - 1)
batch_size = 256

train_dataset = keras.utils.timeseries_dataset_from_array(
    raw_data[:-delay],
    targets=temperature[delay:],
    sampling_rate=sampling_rate,
    sequence_length=sequence_length,
    shuffle=True,
    batch_size=batch_size,
    start_index=0,
    end_index=num_train_samples,
)

val_dataset = keras.utils.timeseries_dataset_from_array(
    raw_data[:-delay],
    targets=temperature[delay:],
    sampling_rate=sampling_rate,
    sequence_length=sequence_length,
    shuffle=True,
    batch_size=batch_size,
    start_index=num_train_samples,
    end_index=num_train_samples + num_val_samples,
)

test_dataset = keras.utils.timeseries_dataset_from_array(
    raw_data[:-delay],
    targets=temperature[delay:],
    sampling_rate=sampling_rate,
    sequence_length=sequence_length,
    shuffle=True,
    batch_size=batch_size,
    start_index=num_train_samples + num_val_samples,
)

In [0]python · cell 18

python

for samples, targets in train_dataset:
    print("samples shape:", samples.shape)
    print("targets shape:", targets.shape)
    break

A commonsense, non-machine-learning baseline

In [0]python · cell 20

python

def evaluate_naive_method(dataset):
    total_abs_err = 0.0
    samples_seen = 0
    for samples, targets in dataset:
        preds = samples[:, -1, 1] * std[1] + mean[1]
        total_abs_err += np.sum(np.abs(preds - targets))
        samples_seen += samples.shape[0]
    return total_abs_err / samples_seen

print(f"Validation MAE: {evaluate_naive_method(val_dataset):.2f}")
print(f"Test MAE: {evaluate_naive_method(test_dataset):.2f}")

Let's try a basic machine learning model

In [0]python · cell 22

python

import keras
from keras import layers

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1]))
x = layers.Flatten()(inputs)
x = layers.Dense(16, activation="relu")(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

callbacks = [
    keras.callbacks.ModelCheckpoint("jena_dense.keras", save_best_only=True)
]
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=10,
    validation_data=val_dataset,
    callbacks=callbacks,
)

model = keras.models.load_model("jena_dense.keras")
print(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

In [0]python · cell 23

python

import matplotlib.pyplot as plt

loss = history.history["mae"]
val_loss = history.history["val_mae"]
epochs = range(1, len(loss) + 1)
plt.figure()
plt.plot(epochs, loss, "r--", label="Training MAE")
plt.plot(epochs, val_loss, "b", label="Validation MAE")
plt.title("Training and validation MAE")
plt.legend()
plt.show()

Let's try a 1D convolutional model

In [0]python · cell 25

python

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1]))
x = layers.Conv1D(8, 24, activation="relu")(inputs)
x = layers.MaxPooling1D(2)(x)
x = layers.Conv1D(8, 12, activation="relu")(x)
x = layers.MaxPooling1D(2)(x)
x = layers.Conv1D(8, 6, activation="relu")(x)
x = layers.GlobalAveragePooling1D()(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

callbacks = [
    keras.callbacks.ModelCheckpoint("jena_conv.keras", save_best_only=True)
]
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=10,
    validation_data=val_dataset,
    callbacks=callbacks,
)

model = keras.models.load_model("jena_conv.keras")
print(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

Recurrent neural networks

In [0]python · cell 27

python

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1]))
x = layers.LSTM(16)(inputs)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

callbacks = [
    keras.callbacks.ModelCheckpoint("jena_lstm.keras", save_best_only=True)
]
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=10,
    validation_data=val_dataset,
    callbacks=callbacks,
)

model = keras.models.load_model("jena_lstm.keras")
print("Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

Understanding recurrent neural networks

In [0]python · cell 29

python

import numpy as np

timesteps = 100
input_features = 32
output_features = 64
inputs = np.random.random((timesteps, input_features))
state_t = np.zeros((output_features,))
W = np.random.random((output_features, input_features))
U = np.random.random((output_features, output_features))
b = np.random.random((output_features,))
successive_outputs = []
for input_t in inputs:
    output_t = np.tanh(np.dot(W, input_t) + np.dot(U, state_t) + b)
    successive_outputs.append(output_t)
    state_t = output_t
final_output_sequence = np.concatenate(successive_outputs, axis=0)

A recurrent layer in Keras

In [0]python · cell 31

python

num_features = 14
inputs = keras.Input(shape=(None, num_features))
outputs = layers.SimpleRNN(16)(inputs)

In [0]python · cell 32

python

num_features = 14
steps = 120
inputs = keras.Input(shape=(steps, num_features))
outputs = layers.SimpleRNN(16, return_sequences=False)(inputs)
print(outputs.shape)

In [0]python · cell 33

python

num_features = 14
steps = 120
inputs = keras.Input(shape=(steps, num_features))
outputs = layers.SimpleRNN(16, return_sequences=True)(inputs)
print(outputs.shape)

In [0]python · cell 34

python

inputs = keras.Input(shape=(steps, num_features))
x = layers.SimpleRNN(16, return_sequences=True)(inputs)
x = layers.SimpleRNN(16, return_sequences=True)(x)
outputs = layers.SimpleRNN(16)(x)

Getting the most out of recurrent neural networks

Using recurrent dropout to fight overfitting

In [0]python · cell 37

python

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1]))
x = layers.LSTM(32, recurrent_dropout=0.25)(inputs)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

callbacks = [
    keras.callbacks.ModelCheckpoint(
        "jena_lstm_dropout.keras", save_best_only=True
    )
]
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=50,
    validation_data=val_dataset,
    callbacks=callbacks,
)

Stacking recurrent layers

In [0]python · cell 39

python

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1]))
x = layers.GRU(32, recurrent_dropout=0.5, return_sequences=True)(inputs)
x = layers.GRU(32, recurrent_dropout=0.5)(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

callbacks = [
    keras.callbacks.ModelCheckpoint(
        "jena_stacked_gru_dropout.keras", save_best_only=True
    )
]
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=50,
    validation_data=val_dataset,
    callbacks=callbacks,
)
model = keras.models.load_model("jena_stacked_gru_dropout.keras")
print(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

Using bidirectional RNNs

In [0]python · cell 41

python

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1]))
x = layers.Bidirectional(layers.LSTM(16))(inputs)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

model.compile(optimizer="adam", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=10,
    validation_data=val_dataset,
)

Going even further

# @title import os from IPython.core.magic import register_cell_magic @register_cell_magic def backend(line, cell): current, required = os.environ.get("KERAS_BACKEND", ""), line.split()[-1] if current == required: get_ipython().run_cell(cell) else: print( f"This cell requires the {required} backend. To run it, change KERAS_BACKEND to " f"\"{required}\" at the top of the notebook, restart the runtime, and rerun the notebook." )

import os fname = os.path.join("jena_climate_2009_2016.csv") with open(fname) as f: data = f.read() lines = data.split("\n") header = lines[0].split(",") lines = lines[1:] print(header) print(len(lines))

import numpy as np temperature = np.zeros((len(lines),)) raw_data = np.zeros((len(lines), len(header) - 1)) for i, line in enumerate(lines): values = [float(x) for x in line.split(",")[1:]] temperature[i] = values[1] raw_data[i, :] = values[:]

num_train_samples = int(0.5 * len(raw_data)) num_val_samples = int(0.25 * len(raw_data)) num_test_samples = len(raw_data) - num_train_samples - num_val_samples print("num_train_samples:", num_train_samples) print("num_val_samples:", num_val_samples) print("num_test_samples:", num_test_samples)

import numpy as np import keras int_sequence = np.arange(10) dummy_dataset = keras.utils.timeseries_dataset_from_array( data=int_sequence[:-3], targets=int_sequence[3:], sequence_length=3, batch_size=2, ) for inputs, targets in dummy_dataset: for i in range(inputs.shape[0]): print([int(x) for x in inputs[i]], int(targets[i]))

sampling_rate = 6 sequence_length = 120 delay = sampling_rate * (sequence_length + 24 - 1) batch_size = 256 train_dataset = keras.utils.timeseries_dataset_from_array( raw_data[:-delay], targets=temperature[delay:], sampling_rate=sampling_rate, sequence_length=sequence_length, shuffle=True, batch_size=batch_size, start_index=0, end_index=num_train_samples, ) val_dataset = keras.utils.timeseries_dataset_from_array( raw_data[:-delay], targets=temperature[delay:], sampling_rate=sampling_rate, sequence_length=sequence_length, shuffle=True, batch_size=batch_size, start_index=num_train_samples, end_index=num_train_samples + num_val_samples, ) test_dataset = keras.utils.timeseries_dataset_from_array( raw_data[:-delay], targets=temperature[delay:], sampling_rate=sampling_rate, sequence_length=sequence_length, shuffle=True, batch_size=batch_size, start_index=num_train_samples + num_val_samples, )

def evaluate_naive_method(dataset): total_abs_err = 0.0 samples_seen = 0 for samples, targets in dataset: preds = samples[:, -1, 1] * std[1] + mean[1] total_abs_err += np.sum(np.abs(preds - targets)) samples_seen += samples.shape[0] return total_abs_err / samples_seen print(f"Validation MAE: {evaluate_naive_method(val_dataset):.2f}") print(f"Test MAE: {evaluate_naive_method(test_dataset):.2f}")

import keras from keras import layers inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1])) x = layers.Flatten()(inputs) x = layers.Dense(16, activation="relu")(x) outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) callbacks = [ keras.callbacks.ModelCheckpoint("jena_dense.keras", save_best_only=True) ] model.compile(optimizer="adam", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=10, validation_data=val_dataset, callbacks=callbacks, ) model = keras.models.load_model("jena_dense.keras") print(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

import matplotlib.pyplot as plt loss = history.history["mae"] val_loss = history.history["val_mae"] epochs = range(1, len(loss) + 1) plt.figure() plt.plot(epochs, loss, "r--", label="Training MAE") plt.plot(epochs, val_loss, "b", label="Validation MAE") plt.title("Training and validation MAE") plt.legend() plt.show()

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1])) x = layers.Conv1D(8, 24, activation="relu")(inputs) x = layers.MaxPooling1D(2)(x) x = layers.Conv1D(8, 12, activation="relu")(x) x = layers.MaxPooling1D(2)(x) x = layers.Conv1D(8, 6, activation="relu")(x) x = layers.GlobalAveragePooling1D()(x) outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) callbacks = [ keras.callbacks.ModelCheckpoint("jena_conv.keras", save_best_only=True) ] model.compile(optimizer="adam", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=10, validation_data=val_dataset, callbacks=callbacks, ) model = keras.models.load_model("jena_conv.keras") print(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1])) x = layers.LSTM(16)(inputs) outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) callbacks = [ keras.callbacks.ModelCheckpoint("jena_lstm.keras", save_best_only=True) ] model.compile(optimizer="adam", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=10, validation_data=val_dataset, callbacks=callbacks, ) model = keras.models.load_model("jena_lstm.keras") print("Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

import numpy as np timesteps = 100 input_features = 32 output_features = 64 inputs = np.random.random((timesteps, input_features)) state_t = np.zeros((output_features,)) W = np.random.random((output_features, input_features)) U = np.random.random((output_features, output_features)) b = np.random.random((output_features,)) successive_outputs = [] for input_t in inputs: output_t = np.tanh(np.dot(W, input_t) + np.dot(U, state_t) + b) successive_outputs.append(output_t) state_t = output_t final_output_sequence = np.concatenate(successive_outputs, axis=0)

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1])) x = layers.LSTM(32, recurrent_dropout=0.25)(inputs) x = layers.Dropout(0.5)(x) outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) callbacks = [ keras.callbacks.ModelCheckpoint( "jena_lstm_dropout.keras", save_best_only=True ) ] model.compile(optimizer="adam", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=50, validation_data=val_dataset, callbacks=callbacks, )

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1])) x = layers.GRU(32, recurrent_dropout=0.5, return_sequences=True)(inputs) x = layers.GRU(32, recurrent_dropout=0.5)(x) x = layers.Dropout(0.5)(x) outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) callbacks = [ keras.callbacks.ModelCheckpoint( "jena_stacked_gru_dropout.keras", save_best_only=True ) ] model.compile(optimizer="adam", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=50, validation_data=val_dataset, callbacks=callbacks, ) model = keras.models.load_model("jena_stacked_gru_dropout.keras") print(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

inputs = keras.Input(shape=(sequence_length, raw_data.shape[-1])) x = layers.Bidirectional(layers.LSTM(16))(inputs) outputs = layers.Dense(1)(x) model = keras.Model(inputs, outputs) model.compile(optimizer="adam", loss="mse", metrics=["mae"]) history = model.fit( train_dataset, epochs=10, validation_data=val_dataset, )