時系列基盤amazon chronos-t5を触ってみた（Google Colab）

AI・機械学習
2024.06.04

はじめに

今回は、HuggingFaceで公開されている時系列予測用の強力なモデル「amazon chronos-t5」を使って、電力変圧器の温度予測に挑戦します！

chronos-t5は、大規模な時系列データセットで事前学習された、Transformerベースの言語モデルです。時系列データをテキストデータのように扱い、未来の値を予測します。大量のデータで学習されているため、少ないデータでも高い予測精度が期待できます。

この記事では、Google Colabを使って、chronos-t5の推論とFine Tuning（追加学習）を順を追って解説します。難しい数式や専門用語はなるべく避け、初心者の方でも理解しやすいように、コード例を交えながら丁寧に説明していきます。

amazon chronos-t5について

chronos-t5は、以下の特徴を持つ強力な時系列予測モデルです。

  1. 大規模な事前学習: 多種多様な時系列データで事前学習されており、幅広いタスクに対応できます。
  2. Transformerベース: 自然言語処理で成功を収めたTransformerアーキテクチャを採用し、時系列データの長期的な依存関係を捉えることができます。
  3. 確率的予測: 単一の予測値だけでなく、予測分布を出力するため、予測の不確実性を評価できます。
  4. 複数のモデルサイズ: small, base, largeなど、複数のモデルサイズが用意されており、計算資源に合わせて選択できます。
GitHub - amazon-science/chronos-forecasting: Chronos: Pretrained (Language) Models for Probabilistic Time Series Forecasting
Chronos: Pretrained (Language) Models for Probabilistic Time Series Forecasting - amazon-science/chronos-forecasting
github.com

GluonTSとは

GluonTSは、PyTorchやMXNetといった深層学習フレームワークをベースに、時系列データのモデリングを強力にサポートするPythonのパッケージです。特に、深層学習を用いたモデルに焦点を当てているのが特徴です。

GluonTSで始める深層学習による時系列予測
深層学習を用いた時系列予測の世界へようこそ！この記事では、PythonのパッケージGluonTSを用いて、初心者でも時系列予測を理解し、実践できるようになるための完全ガイドを提供します。時系列予測とは何か？時系列予測とは、過去のデータに基づ...
hamaruki.com
2024.06.04

1. 推論（予測）と可視化

ライブラリのインストール

chronos-t5を使用するために、必要なライブラリをインストールします。Google Colabで以下のコマンドを実行してください。

!pip install typer-config
!pip install git+https://github.com/amazon-science/chronos-forecasting.git
!pip install "gluonts[arrow]"
!git clone https://github.com/amazon-science/chronos-forecasting.git

データの準備

今回は、電力変圧器の温度に関するデータセット「ETTh1.csv」を使用します。このデータセットは、以下のURLからダウンロードできます。

!wget https://github.com/zhouhaoyi/ETDataset/blob/main/ETT-small/ETTh1.csv

ETTh1.csvには、複数のセンサーで計測された温度データ（HUFL, HULL, MUFL, MULL, LUFL, LULL）と、予測したい目標の温度（OT: Oil Temperature）が含まれています。データは1時間ごとの時系列データで、以下のような形式になっています。

date,HUFL,HULL,MUFL,MULL,LUFL,LULL,OT
2016-07-01 00:00:00,0.5058,0.4688,0.5552,0.5248,0.2896,0.3206,35.1495
2016-07-01 01:00:00,0.5298,0.4852,0.5647,0.5305,0.2914,0.3218,35.1066
...
%env CUDA_LAUNCH_BLOCKING=1

データの前処理

chronos-t5に入力するため、データを適切な形式に変換します。

import pandas as pd
import torch

def prepare_data(file_path, context_length, forecast_horizon):
    """
    時系列データを読み込み、学習用とテスト用に分割し、PyTorchテンソルに変換する関数

    Args:
        file_path (str): データファイルのパス
        context_length (int): モデルに入力するデータの長さ
        forecast_horizon (int): 予測する長さ

    Returns:
        train_tensor (torch.Tensor): 学習用データテンソル
        test_tensor (torch.Tensor): テスト用データテンソル
    """

    df = pd.read_csv(file_path)

    # データセットの分割 (最後のforecast_horizon分をテストデータとする)
    df_train = df.iloc[-(context_length + forecast_horizon):-forecast_horizon]
    df_test = df.iloc[-forecast_horizon:]

    # データ形式の変換 (PyTorchのテンソルに変換し、転置)
    train_tensor = torch.tensor(df_train[["HUFL", "HULL", "MUFL", "MULL", "LUFL", "LULL", "OT"]].values, dtype=torch.float).t()
    test_tensor = torch.tensor(df_test[["HUFL", "HULL", "MUFL", "MULL", "LUFL", "LULL", "OT"]].values, dtype=torch.float).t()

    return train_tensor, test_tensor, df_train, df_test, df

モデルの読み込みと予測

次に、chronos-t5のモデルを読み込み、予測を行う関数と、結果を可視化する関数を定義します。

from chronos import ChronosPipeline
import matplotlib.pyplot as plt

def predict_with_chronos(train_tensor, forecast_horizon, model_name="amazon/chronos-t5-large", device_map="cuda"):
    """
    chronos-t5モデルで予測を行う関数

    Args:
        train_tensor (torch.Tensor): 学習用データテンソル
        forecast_horizon (int): 予測する長さ
        model_name (str): モデル名 (デフォルト: "amazon/chronos-t5-large")
        device_map (str): デバイス ("cuda" or "cpu")

    Returns:
        forecast_median_tensor (torch.Tensor): 予測結果の中央値テンソル
    """

    pipeline = ChronosPipeline.from_pretrained(
        model_name,
        device_map=device_map,
        torch_dtype=torch.bfloat16,  # 計算精度を指定
    )

    # 予測の実行 (limit_prediction_length=Falseで予測長を制限しない)
    forecast = pipeline.predict(train_tensor, forecast_horizon, limit_prediction_length=False)
    forecast_median_tensor, _ = torch.median(forecast, dim=1)  # 予測結果の中央値を計算

    return forecast_median_tensor, forecast

def visualize_prediction(train_tensor, test_tensor, forecast_median_tensor, channel_idx=6):
    """
    予測結果を可視化する関数

    Args:
        train_tensor (torch.Tensor): 学習用データテンソル
        test_tensor (torch.Tensor): テスト用データテンソル
        forecast_median_tensor (torch.Tensor): 予測結果の中央値テンソル
        channel_idx (int): 可視化するチャンネルのインデックス (デフォルト: 6, OT)
    """

    history = train_tensor[channel_idx, :].detach().numpy()  # 学習用データ
    true = test_tensor[channel_idx, :].detach().numpy()      # 実測値
    pred = forecast_median_tensor[channel_idx, :].detach().numpy()  # 予測値

    plt.figure(figsize=(12, 4))
    plt.plot(range(len(history)), history, label='History (512 timesteps)', c='darkblue')
    plt.plot(range(len(history), len(history) + len(true)), true, label='Ground Truth (96 timesteps)', color='darkblue', linestyle='--', alpha=0.5)
    plt.plot(range(len(history), len(history) + len(pred)), pred, label='Forecast (96 timesteps)', color='red', linestyle='--')
    plt.title(f"ETTh1 (Hourly) - Channel {channel_idx}", fontsize=18)
    plt.xlabel('Time', fontsize=14)
    plt.ylabel('Value', fontsize=14)
    plt.legend(fontsize=14)
    plt.show()

実行

context_length = 512
forecast_horizon = 96
# データの準備と前処理

train_tensor, test_tensor, df_train, df_test, df = prepare_data("ETT_small_h1.csv", context_length=context_length, forecast_horizon=forecast_horizon)

# 予測の実行
forecast_median_tensor, forecast = predict_with_chronos(train_tensor, forecast_horizon=forecast_horizon)

# 予測結果の可視化
visualize_prediction(train_tensor, test_tensor, forecast_median_tensor)

import matplotlib.pyplot as plt
import numpy as np

channel_idx = 6
time_index = 0

history = train_tensor[channel_idx, :].detach().numpy()
true = test_tensor[channel_idx, :].detach().numpy()
pred = forecast_median_tensor[channel_idx, :].detach().numpy()
low, median, high = np.quantile(forecast[channel_idx].numpy(), [0.1, 0.5, 0.9], axis=0)

plt.figure(figsize=(12, 4))

# Plotting the first time series from history
plt.plot(range(len(history)), history, label=f'History ({context_length} timesteps)', c='darkblue')

# Plotting ground truth and prediction
num_forecasts = len(true)

offset = len(history)
plt.plot(range(offset, offset + len(true)), true, label='Ground Truth (96 timesteps)', color='darkblue', linestyle='--', alpha=0.5)
plt.plot(range(offset, offset + len(pred)), pred, label='Forecast (96 timesteps)', color='red', linestyle='--')
plt.fill_between(range(offset, offset + len(pred)), low, high, color="tomato", alpha=0.3, label="80% prediction interval")

plt.title(f"ETTh1 (Hourly) -- (idx={time_index}, channel={channel_idx})", fontsize=18)
plt.xlabel('Time', fontsize=14)
plt.ylabel('Value', fontsize=14)
plt.legend(fontsize=14)
plt.show()

2. Fine Tuning（追加学習）

chronos-t5は、事前学習済みのモデルですが、自分のデータでFine Tuningを行うことで、さらに予測精度を向上させることができます。

データの準備

Fine Tuning用のデータセットを作成します。ここでは、元のデータセットを8:2の割合で学習用とテスト用に分割します。

# データを前8割をトレーニング用、後ろ2割をテスト用に分割
train_size = int(0.8 * len(df))
df_train = df.iloc[:train_size]
df_test = df.iloc[train_size:-(context_length+forecast_horizon)]  # 学習用データの残りの部分をテストデータとする

# GluonTS形式に変換
from pathlib import Path
from typing import List, Optional, Union

import numpy as np
from gluonts.dataset.arrow import ArrowWriter

def convert_to_arrow(
    path: Union[str, Path],
    time_series: Union[List[np.ndarray], np.ndarray],
    start_times: Optional[Union[List[np.datetime64], np.ndarray]] = None,
    compression: str = "lz4",
):
    """
    時系列データをGluonTSのArrow形式に変換する関数

    Args:
        path (str or Path): 変換後のデータファイルのパス
        time_series (List[np.ndarray] or np.ndarray): 時系列データ
        start_times (Optional[List[np.datetime64] or np.ndarray]): 各時系列の開始時刻 (デフォルト: None)
        compression (str): 圧縮形式 (デフォルト: "lz4")
    """
    if start_times is None:
        # 開始時刻が指定されていない場合は、任意の開始時刻を設定
        start_times = [np.datetime64("2000-01-01 00:00", "s")] * len(time_series)

    assert len(time_series) == len(start_times)

    dataset = [
        {"start": start, "target": ts} for ts, start in zip(time_series, start_times)
    ]
    ArrowWriter(compression=compression).write_to_file(
        dataset,
        path=path,
    )

# データをGluonTSのArrow形式に変換
cols = ["HUFL", "HULL", "MUFL", "MULL", "LUFL", "LULL", "OT"]  # 使用する特徴量
convert_to_arrow(
    path="./etth1-train.arrow",  # 変換後のファイル名
    time_series=[np.array(df_train[col]) for col in cols],
    start_times=[pd.to_datetime(df_train["date"]).values[0]] * len(cols),
)

モデルの準備とFine Tuningの実行

Fine Tuning用の設定ファイルを作成し、学習を実行します。

import yaml

batch_size = 4  # バッチサイズ
num_steps = train_size // batch_size  # 学習ステップ数

# Fine Tuningの設定
config_data = {
    'training_data_paths': ["./etth1-train.arrow"],  # 学習データファイルのパス
    'probability': [1.0],
    'output_dir': './output/',  # 学習結果の出力ディレクトリ
    'context_length': context_length,
    'prediction_length': forecast_horizon,
    'max_steps': num_steps,
    'per_device_train_batch_size': batch_size,
    'learning_rate': 0.001,
    #'model_id': 'amazon/chronos-t5-large',
    'model_id': 'amazon/chronos-t5-base',
    'random_init': False,  # 事前学習済みモデルを使用
    'tf32': True,        # NVIDIA GPUの場合Trueにする
    'gradient_accumulation_steps':2,
}

# 設定ファイルをYAML形式で保存
config_file_path = './ft_config.yaml'
with open(config_file_path, 'w') as file:
    yaml.dump(config_data, file)

def fine_tune_chronos(train_file_path, config_file_path):
    """
    chronos-t5モデルをFine Tuningする関数

    Args:
        train_file_path (str): 学習用データファイルのパス
        config_file_path (str): Fine Tuning設定ファイルのパス
    """

    # Fine Tuningの実行
    !CUDA_VISIBLE_DEVICES=0 python chronos-forecasting/scripts/training/train.py --config {config_file_path}

# Fine Tuningの実行
fine_tune_chronos("./etth1-train.arrow", config_file_path)

Fine Tuning後のモデルの保存と予測

Fine Tuning後のモデルをHuggingFaceに保存し、予測を行います。

# Fine Tuning後のモデルで予測
forecast_median_tensor_ft, forecast_ft = predict_with_chronos(train_tensor, forecast_horizon=forecast_horizon, model_name="./output/run-0/checkpoint-final/")

# 予測結果の可視化 (Fine Tuning後)
visualize_prediction(train_tensor, test_tensor, forecast_median_tensor_ft)

import matplotlib.pyplot as plt
import numpy as np

channel_idx = 6
time_index = 0

history = train_tensor[channel_idx, :].detach().numpy()
true = test_tensor[channel_idx, :].detach().numpy()
pred = forecast_median_tensor_ft[channel_idx, :].detach().numpy()
low, median, high = np.quantile(forecast_ft[channel_idx].numpy(), [0.1, 0.5, 0.9], axis=0)

plt.figure(figsize=(12, 4))

# Plotting the first time series from history
plt.plot(range(len(history)), history, label=f'History ({context_length} timesteps)', c='darkblue')

# Plotting ground truth and prediction
num_forecasts = len(true)

offset = len(history)
plt.plot(range(offset, offset + len(true)), true, label='Ground Truth (96 timesteps)', color='darkblue', linestyle='--', alpha=0.5)
plt.plot(range(offset, offset + len(pred)), pred, label='Forecast (96 timesteps)', color='red', linestyle='--')
plt.fill_between(range(offset, offset + len(pred)), low, high, color="tomato", alpha=0.3, label="80% prediction interval")

plt.title(f"ETTh1 (Hourly) -- (idx={time_index}, channel={channel_idx})", fontsize=18)
plt.xlabel('Time', fontsize=14)
plt.ylabel('Value', fontsize=14)
plt.legend(fontsize=14)
plt.show()

まとめ

この記事では、時系列基盤モデル「amazon chronos-t5」を使って、電力変圧器の温度予測を行いました。事前学習済みのモデルだけでも高い予測精度が得られましたが、Fine Tuningによってさらに精度を向上させることができました。

chronos-t5は、他の時系列モデルと比べて、大規模なデータで学習されており、表現力が高いという特徴があります。また、確率予測を行うため、予測の不確実性を考慮した分析が可能になります。

この記事が、時系列予測に興味のある方の参考になれば幸いです。

GoogleColabノートブック

Google Colab
colab.research.google.com

参考資料

Google Colabで時系列基盤モデルを試す④：amazon chronos-t5｜はち
はじめに Google Timesfm、Moment、IBM Graniteに引き続き、HuggingFaceにある商用可能なライセンスの時系列基盤モデルを4つ試し、比較していきたいと思います。 利用するデータはETTh1という電力変圧器温度に関する多変量時系列データセットです。事前学習にこのデータが含まれる可能性が...
note.com
amazon/chronos-t5-large · Hugging Face
We’re on a journey to advance and democratize artificial intelligence through open source and open science.
huggingface.co
GitHub - amazon-science/chronos-forecasting: Chronos: Pretrained (Language) Models for Probabilistic Time Series Forecasting
Chronos: Pretrained (Language) Models for Probabilistic Time Series Forecasting - amazon-science/chronos-forecasting
github.com

コメント

タイトルとURLをコピーしました