Chapter 45

Deep learning for timeseries

NotebookPython 361 cells

This is a companion notebook for the book Deep Learning with Python, Second 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.

This notebook was generated for TensorFlow 2.6.

Deep learning for timeseries

Different kinds of timeseries tasks

A temperature-forecasting example

In [0]python · cell 5

python

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

Inspecting the data of the Jena weather dataset

In [0]python · cell 7

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))

Parsing the data

In [0]python · cell 9

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[:]

Plotting the temperature timeseries

In [0]python · cell 11

python

from matplotlib import pyplot as plt
plt.plot(range(len(temperature)), temperature)

Plotting the first 10 days of the temperature timeseries

In [0]python · cell 13

python

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

Computing the number of samples we'll use for each data split

In [0]python · cell 15

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

Normalizing the data

In [0]python · cell 18

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 19

python

import numpy as np
from tensorflow 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]))

Instantiating datasets for training, validation, and testing

In [0]python · cell 21

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)

Inspecting the output of one of our datasets

In [0]python · cell 23

python

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

A common-sense, non-machine-learning baseline

Computing the common-sense baseline MAE

In [0]python · cell 26

python

def evaluate_naive_method(dataset):
    total_abs_err = 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

Training and evaluating a densely connected model

In [0]python · cell 29

python

from tensorflow import keras
from tensorflow.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="rmsprop", 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}")

Plotting results

In [0]python · cell 31

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, "bo", 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 33

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="rmsprop", 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}")

A first recurrent baseline

A simple LSTM-based model

In [0]python · cell 36

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="rmsprop", 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(f"Test MAE: {model.evaluate(test_dataset)[1]:.2f}")

Understanding recurrent neural networks

NumPy implementation of a simple RNN

In [0]python · cell 39

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.stack(successive_outputs, axis=0)

A recurrent layer in Keras

An RNN layer that can process sequences of any length

In [0]python · cell 42

python

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

An RNN layer that returns only its last output step

In [0]python · cell 44

python

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

An RNN layer that returns its full output sequence

In [0]python · cell 46

python

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

Stacking RNN layers

In [0]python · cell 48

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)

Advanced use of recurrent neural networks

Using recurrent dropout to fight overfitting

Training and evaluating a dropout-regularized LSTM

In [0]python · cell 52

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="rmsprop", loss="mse", metrics=["mae"])
history = model.fit(train_dataset,
                    epochs=50,
                    validation_data=val_dataset,
                    callbacks=callbacks)

In [0]python · cell 53

python

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.LSTM(32, recurrent_dropout=0.2, unroll=True)(inputs)

Stacking recurrent layers

Training and evaluating a dropout-regularized, stacked GRU model

In [0]python · cell 56

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="rmsprop", 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

Training and evaluating a bidirectional LSTM

In [0]python · cell 59

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="rmsprop", loss="mse", metrics=["mae"])
history = model.fit(train_dataset,
                    epochs=10,
                    validation_data=val_dataset)

Going even further

Summary

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 from tensorflow 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. 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}")

from tensorflow import keras from tensorflow.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="rmsprop", 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, "bo", 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="rmsprop", 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="rmsprop", 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(f"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.stack(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="rmsprop", 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="rmsprop", 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="rmsprop", loss="mse", metrics=["mae"]) history = model.fit(train_dataset, epochs=10, validation_data=val_dataset)