KerasでTransfomerモデルによる時系列データの分類をしてみる

KerasでTransfomerモデルを構築し、時系列データの分類を行います。
下記のサンプルを参考にしています。

Keras documentation: Timeseries classification with a Transformer model

ライブラリのインポート

import numpy as np
import keras
from keras import layers

 データの読み込み

def readucr(filename):
    data = np.loadtxt(filename, delimiter="\t")
    y = data[:, 0]
    x = data[:, 1:]
    return x, y.astype(int)


root_url = "https://raw.githubusercontent.com/hfawaz/cd-diagram/master/FordA/"

x_train, y_train = readucr(root_url + "FordA_TRAIN.tsv")
x_test, y_test = readucr(root_url + "FordA_TEST.tsv")

GitHubのリポジトリから「FordA」の訓練データとテストデータを読み込みます。「FordA」データセットは、車両の動作や状態を監視するために使用されるセンサーから収集されたデータを基にしています。機械学習やデータサイエンスのコミュニティで時系列分析の研究や教育目的で使用されます。

 データの整形

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

n_classes = len(np.unique(y_train))

idx = np.random.permutation(len(x_train))
x_train = x_train[idx]
y_train = y_train[idx]

y_train[y_train == -1] = 0
y_test[y_test == -1] = 0

モデルの構築

def transformer_encoder(inputs, head_size, num_heads, ff_dim, dropout=0):
    x = layers.MultiHeadAttention(
        key_dim=head_size, num_heads=num_heads, dropout=dropout
    )(inputs, inputs)
    x = layers.Dropout(dropout)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    res = x + inputs

    x = layers.Conv1D(filters=ff_dim, kernel_size=1, activation="relu")(res)
    x = layers.Dropout(dropout)(x)
    x = layers.Conv1D(filters=inputs.shape[-1], kernel_size=1)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    return x + res

まずはトランスフォーマーモデルのエンコーダ部分を定義しています。

def build_model(
    input_shape,
    head_size,
    num_heads,
    ff_dim,
    num_transformer_blocks,
    mlp_units,
    dropout=0,
    mlp_dropout=0,
):
    inputs = keras.Input(shape=input_shape)
    x = inputs
    for _ in range(num_transformer_blocks):
        x = transformer_encoder(x, head_size, num_heads, ff_dim, dropout)

    x = layers.GlobalAveragePooling1D(data_format="channels_last")(x)
    for dim in mlp_units:
        x = layers.Dense(dim, activation="relu")(x)
        x = layers.Dropout(mlp_dropout)(x)
    outputs = layers.Dense(n_classes, activation="softmax")(x)
    return keras.Model(inputs, outputs)


input_shape = x_train.shape[1:]

model = build_model(
    input_shape,
    head_size=256,
    num_heads=4,
    ff_dim=4,
    num_transformer_blocks=4,
    mlp_units=[128],
    mlp_dropout=0.4,
    dropout=0.25,
)

build_model関数は、トランスフォーマーベースのモデルを構築するためのものです。

指定された数のトランスフォーマーブロック（num_transformer_blocks）をモデルに追加します。各ブロックは先ほどのtransformer_encoder関数を呼び出して実装されます。

モデルのコンパイル

モデルをコンパイルする際に損失関数とsparse_categorical_crossentropyを使用しています。これは、クラスラベルが整数の形式で直接提供される多クラス分類問題に適しています。

model.compile(
    loss="sparse_categorical_crossentropy",
    optimizer=keras.optimizers.Adam(learning_rate=1e-4),
    metrics=["sparse_categorical_accuracy"],
)
model.summary()

学習

callbacks = [keras.callbacks.EarlyStopping(patience=10, restore_best_weights=True)]

model.fit(
    x_train,
    y_train,
    validation_split=0.2,
    epochs=150,
    batch_size=64,
    callbacks=callbacks,
)

訓練データの20%を検証用データとして最大150エポックで学習しています。また、EarlyStoppingコールバックが指定されており、10エポック改善がない場合に学習を停止します。

評価

model.evaluate(x_test, y_test, verbose=1)