1. はじめに
みなさん、こんにちは。前回、Granite Time Series ModelをSPSS Modelerで利用するための準備を実施しました。
今回は、このGraniteモデルをそのまま使ってゼロショットで時系列予測をしていきたいと思います。
2. 参考にしたもの
今回のプログラムを実装するにあたり、以下のgithubにある各種サンプルを参考にしました。
3. SPSS Modelerでモデルを実装
1. 使用データ、ストリーム
使用するデータは、以下のHugging Faceにあるスペインのエネルギーデータです。
energy_dataset.csvをダウンロードしてください。
GitHubに今回使用したストリームをアップロードしています。自由にダウンロードしてください。
2. ストリーム全体
今回の作成したストリームです。
拡張の変換ノードでゼロショット時系列予測を実装しています。
3. まずは拡張の変換ノードの手前まで実装
①. ストリームオプション
まずは事前に、SPSS Modelerのメニューバーから、ツール >> ストリームのプロパティー >> オプション を選択します。
そして、全般メニューにある、名義型フィールドの最大メンバーのチェックを外しておきましょう。
チェックを外したら適用ボタンを忘れずにクリックしてください。
③. データの入力と絞り込み
"energy_dataset.csv"を特に変更はせず可変長ファイルノードで入力してください。
そして、サンプリングノードでデータ件数を減らします。（この処理は任意であり、データを減らさなくてもOKです。）
筆者のPCは非力なため、今回は約35,000件中、29,000件を削除しています。
③. 時間フィールドのデータをタイムスタンプに変換
※.この処理は任意です。次回以降でModelerの時系列モデルも利用して比較したいので、ここでデータ加工しています。
"time"フィールドに時刻データが格納されていますが、"2015-01-01 01:00:00+01:00"というように、+1:00というタイムゾーンの情報が付与されています。（スペインなので、+1:00）
このタイムゾーン情報は不要なため削除して、タイムスタンプ型に変換します。
"allbutlast"関数を使いデータの後ろから6文字分を削除し、"to_datetime"関数でタイムスタンプ型に変換しています。
④. 不要フィールドの削除
今回は、ゼロショットモデルを作るため時刻フィールドと予測対象フィールド以外は必要ないため、その他を削除します。
時刻フィールドは"time"、予測対象フィールドは"total load actual"。この二つ以外は除外しましょう。
4. 拡張の変換ノードでGranite Time Series Modelを実装
さて、いよいよPythonでモデルを実装していきます。
今回は、ゼロショットモデルを使うためモデルの構築はなく関数を呼び出すような感じでGraniteを利用します。
そのため拡張の変換ノードで実装します。
①. シンタックス全体
以下がシンタックス全体です。
全体
#-------------------------------------------------
# IBM Granite Time Series ( Tiny Time Mixer )
# HugingFace Granite Time Series を利用
# "ibm-granite/granite-timeseries-ttm-r2
# "512-96-ft-r2.1",
# Zero-Shot Modelを使って予測
#-------------------------------------------------
#-----------------------------------------------
# ライブラリ導入パート
# 1. Modeler用のライブラリ
# 2. データ加工に必要なライブラリ
# 3. HugingFace Granite Time Seriesに必要なライブラリ
# tsfm_public & torch
#-----------------------------------------------
# Modeler用ライブラリ
import modelerpy
# Pandas & Numpy
import numpy as np
import pandas as pd
# PyTorch
import torch
# Granite Time Series用ライブラリ
from tsfm_public import (
TimeSeriesForecastingPipeline, # 時系列予測のパイプライン
TimeSeriesPreprocessor, # 入力データの整形・スケーリング
TinyTimeMixerForPrediction, # 時系列予測モデル
)
#-----------------------------------------------
# 定数定義
# 1. モデル用カラム名定義
# 2. モデル用パラメータ定義
#-----------------------------------------------
# タイムスタンプカラム定義
timestamp_column = "time"
# 予測対象カラム名
target_column = "total load actual"
# モデルの予測値格納カラム名
prediciton_column = target_column + "_prediction"
# Modeler用予測値格納カラム名
predictField = "$GTS-" + target_column
#-----------------------------------------------
# モデル用パラメータ定数
# 使用モデルで変えること
#-----------------------------------------------
# モデルが「過去何ステップ分のデータを見て」学習・予測するか（ここでは過去512時間分）
context_length = 512 # the max context length for the 512-96 model
# モデルが「未来何ステップ先まで予測」するか（ここでは96時間先＝4日分）
prediction_length = 96 # the max forecast length for the 512-96 model
#-----------------------------------------------
# データモデル定義
#-----------------------------------------------
if modelerpy.isComputeDataModelOnly():
#データモデル取得
modelerDataModel = modelerpy.getDataModel()
#モデル用変数フィールドを追加
#predictField - 実数を格納予定
modelerDataModel.addField(modelerpy.Field( predictField, "real", measure="continuous"))
#修正したデータモデルをModelerへ戻します。
modelerpy.setOutputDataModel(modelerDataModel)
#-----------------------------------------------
# 時系列モデルパ―ド
# 1. データの入力
# 2. モデル初期処理
# 3. モデル設定
# 4. モデルの実行
# 5. モデルの精度確認(任意のためコメントアウト)
# 6. Modelerへのデータ出力処理
#-----------------------------------------------
else:
#-----------------------------------------------
# 1. データ入力
#-----------------------------------------------
#ModelerからPandasでデータを入力
modelerData = modelerpy.readPandasDataframe()
input_df = modelerData
#-----------------------------------------------
# 2. モデル初期処理
# ①. モデルで使用するカラムの指定
# ②. モデル用 Preprocessor の初期化と学習
#-----------------------------------------------
#-----------------------------------------------
# モデルで使用するカラムの指定
#-----------------------------------------------
# 予測対象カラムをリストで定義
target_columns = [ target_column ]
# 時系列カラム"time"をタイムスタンプ型に変更
input_df[ timestamp_column ] = pd.to_datetime(input_df[ timestamp_column ])
# 列の指定
column_specifiers = {
"timestamp_column": timestamp_column, # 時間
"id_columns": [], # 複数系列を識別するためのID（今回は1系列なので空）
"target_columns": target_columns, # 予測対象（"total load actual"）
"control_columns": [], # 説明変数（天気とか、曜日などの外部変数）? 今は無し
}
#-----------------------------------------------
# モデル用 Preprocessor の初期化と学習
#-----------------------------------------------
# 「この特定の電力データにあわせて、どう前処理すればよいか」をモデルに教えてあげる作業。
# 学習データからスケーラーなどのパラメータを学習（平均・分散など）。
# スケーリング（標準化）も有効化。
# 時系列のスライス作成（context + prediction window をスライスする処理）
tsp = TimeSeriesPreprocessor(
**column_specifiers,
context_length=context_length, #学習に使用するレコード数
prediction_length=prediction_length, #予測するレコード数
scaling=True, #データのスケーリングを有効化
scaler_type="standard", #標準化（平均0・分散1）を適用
)
#入力データで事前学習
trained_tsp = tsp.train(input_df)
#-----------------------------------------------
# 3. モデル設定
# ①. モデルの読み込み
# ②. デバイスの選択
#-----------------------------------------------
# モデルの読み込み（from Hugging Face）
zeroshot_model = TinyTimeMixerForPrediction.from_pretrained(
"ibm-granite/granite-timeseries-ttm-r2", # Hugging Face 上の IBMのGraniteシリーズ 時系列モデル（TinyTimeMixer）
revision="512-96-ft-r2.1", # "512-96-ft-r2.1" は学習済み重み（revision）で、入力長512・出力長96 に対応したモデル
num_input_channels=len(target_columns), # num_input_channels はターゲットの列数。今回は "total load actual" の1列なので 1
)
# デバイスの確認
# GPU（CUDA）が使えれば使う、無ければCPU。
device = "cuda" if torch.cuda.is_available() else "cpu"
#-----------------------------------------------
# 4. モデル実行
# ①. Pipeline定義
# ②. Pipeline実行
#-----------------------------------------------
# Pipline定義 モデル + 前処理器 + 推論設定 をまとめた便利なインターフェース。
# feature_extractor=tsp で、前に train() した前処理器が使われる。
# batch_size=512：推論時のバッチサイズ（データ件数が多いとメモリ効率が良くなる）
pipeline = TimeSeriesForecastingPipeline(
zeroshot_model, # 上記で設定したGraniteを指定
device=device, # CPU or GPUが指定される
feature_extractor=tsp, # 上記で設定したTimeSeriesPreprocessorを指定
batch_size=512, # バッチサイズ
freq="h", # h = 1時間単位
)
#-----------------------------------------------
# Pipeline実行
#-----------------------------------------------
# 戻り値 zeroshot_forecast は DataFrame 形式で、予測結果が入ってる。
zeroshot_forecast = pipeline(input_df)
#-----------------------------------------------
# 5. モデル精度確認
#-----------------------------------------------
# # MSE・RMSE・MAE 計算用関数
# def custom_metric(actual, prediction, column_header="results"):
# """Simple function to compute MSE"""
# # pandas の Series からリスト → numpy配列に変換。ndim が1次元でも、強制的に2次元配列にする関数で保険
# a = np.atleast_2d(np.asarray(actual.tolist()))
# p = np.atleast_2d(np.asarray(prediction.tolist()))
# # NaN（欠損値）を含む行は無視するマスクを作成。
# mask = ~np.any(np.isnan(a), axis=1)
# # MSE と MAE を計算。
# mse = np.mean(np.square(a[mask, :] - p[mask, :]))
# mae = np.mean(np.abs(a[mask, :] - p[mask, :]))
# # 評価結果を pandas の DataFrame に整形して返す。
# return pd.DataFrame(
# {
# column_header: {
# "mean_squared_error": mse,
# "root_mean_squared_error": np.sqrt(mse),
# "mean_absolute_error": mae,
# }
# }
# )
# print(custom_metric(
# zeroshot_forecast["total load actual"], zeroshot_forecast["total load actual_prediction"], "zero-shot forecast"
# ))
#------------------------------------------------------------
# 6. Modelerに戻すためのデータ加工パート
# 1. 過去データに対する予測値の取得し入力データに結合
# 戻り値の特性を考慮して、リストの先頭の値を一つ下のレコードへ配置
# 2. 最後のレコードから96レコード先までの予測を取得
# 3. 最終レコードより未来の値を1.のデータに結合させ、Modelerに戻す
#------------------------------------------------------------
#=====================================
# Step 1: 過去の1ステップ先予測を格納
# 過去の実績値に対してのモデルの予測値も求めておく
#=====================================
# モデルの戻り値をコピー
past_pred_df = zeroshot_forecast.copy()
# ndarrayから1つ目の値を取得する関数
def get_first_val(x):
if isinstance(x, (list, np.ndarray)) and len(x) > 0:
return x[0]
return np.nan
# 予測値の1ステップ先だけ取り出す（リストの1つ目を取得)
past_pred_df[ predictField ] = (
past_pred_df[ prediciton_column ].apply(get_first_val)
)
# 1ステップ先なので、「1時間後」にずらす（一つ先のレコードに格納）
past_pred_df[ predictField ] = past_pred_df[ predictField ].shift(1)
# input_df にマージする（timeで結合、inner join でOK）
merged_df = pd.merge(input_df, past_pred_df[[timestamp_column, predictField]], on=timestamp_column, how="left")
#=====================================
# 2: 96時間先の未来を予測する処理
# 新しいデータフレームに未来の値を格納
#=====================================
# 入力の最後の512件だけ取得（512はcontext_length）
input_tail = input_df.tail(context_length)
# 未来96時間分の空の DataFrame を作成
last_time = input_tail[timestamp_column].iloc[-1]
future_times = pd.date_range(start=last_time + pd.Timedelta(hours=1), periods=prediction_length, freq="h")
# 空のDataFrameを作る（NaNが入る）
future_df = pd.DataFrame({timestamp_column: future_times})
# 入力+未来のDataFrameを作って予測（前処理器が内部で補完など行う）
# TinyTimeMixerは、過去context_lengthの系列を使ってprediction_length先を出すモデル
input_for_forecast = pd.concat([input_tail, future_df], ignore_index=True)
# 予測を実行
future_forecast = pipeline(input_for_forecast)
# 未来予測部分だけを切り出して新しいDataFrameに
# 各行の total load actual_prediction は 96個の予測値が入ったリストなので、
# 最初の行だけ使って 1レコードに展開
pred_list = future_forecast[ prediciton_column ].iloc[0]
future_df = pd.DataFrame({
timestamp_column: future_times,
predictField: pred_list,
})
#=====================================
# 3: 過去データの予測値と未来の値を結合してModelerに戻す
# 新しいデータフレームに未来の値を格納
#=====================================
# 他のカラム（total load actual など）は input_df と同じ構造にするため、NaN で埋める
for col in merged_df.columns:
if col not in future_df.columns:
future_df[col] = np.nan
# カラム順を merged_df に合わせて並べ替え
future_df = future_df[merged_df.columns]
# merged_df + future_df を縦に結合
input_df_with_pred = pd.concat([merged_df, future_df], ignore_index=True)
# 結果確認
print(input_df_with_pred.tail(100)) # 最後の100行を表示
# Modelerにデータを戻す
modelerpy.writePandasDataframe(input_df_with_pred)
②. シンタックス詳細
では、詳細をみていきましょう。
a. ライブラリ導入パート
必要なライブラリをインポートしています。
Granite Time Series用のライブラリは"tsfm_public"から各種インポートします。
ライブラリ導入パート
#-----------------------------------------------
# ライブラリ導入パート
# 1. Modeler用のライブラリ
# 2. データ加工に必要なライブラリ
# 3. HugingFace Granite Time Seriesに必要なライブラリ
# tsfm_public & torch
#-----------------------------------------------
# Modeler用ライブラリ
import modelerpy
# Pandas & Numpy
import numpy as np
import pandas as pd
# PyTorch
import torch
# Granite Time Series用ライブラリ
from tsfm_public import (
TimeSeriesForecastingPipeline, # 時系列予測のパイプライン
TimeSeriesPreprocessor, # 入力データの整形・スケーリング
TinyTimeMixerForPrediction, # 時系列予測モデル
)
b. 定数定義パート
このプログラムで使用する定数を定義しています。
タイムスタンプ、予測対象のカラム、モデルからの出力値を格納するカラムをそれぞれ定義しています。
モデルの戻り値には予測結果がリスト形式で格納されてきます。そのカラム名は、
予測対象カラム名 + "_prediction" となります。
またModelerに戻した時、予測結果のカラム名は "$XX-予測対象カラム名" としておくと、精度確認の際に便利です。
モデル用のパラメータ値には、今回利用するモデルの値を設定します。
granite-ttm-512-96-r2 使用するため、
学習レコード数 = 512
予測レコード数 = 96
を設定します。
定数定義パート
#-----------------------------------------------
# 定数定義
# 1. モデル用カラム名定義
# 2. モデル用パラメータ定義
#-----------------------------------------------
# タイムスタンプカラム定義
timestamp_column = "time"
# 予測対象カラム名
target_column = "total load actual"
# モデルの予測値格納カラム名
prediciton_column = target_column + "_prediction"
# Modeler用予測値格納カラム名
predictField = "$GTS-" + target_column
#-----------------------------------------------
# モデル用パラメータ定数
# 使用モデルで変えること
#-----------------------------------------------
# モデルが「過去何ステップ分のデータを見て」学習・予測するか（ここでは過去512時間分）
context_length = 512 # the max context length for the 512-96 model
# モデルが「未来何ステップ先まで予測」するか（ここでは96時間先＝4日分）
prediction_length = 96 # the max forecast length for the 512-96 model
c. データモデル定義
後続ノードへデータを渡すためにデータモデル定義をしています。
データモデルを取得して、予測値を格納するためのフィールドを追加しています。
if modelerpy.isComputeDataModelOnly(): 以下に処理を記述します。
データモデル定義パート
#-----------------------------------------------
# データモデル定義
#-----------------------------------------------
if modelerpy.isComputeDataModelOnly():
#データモデル取得
modelerDataModel = modelerpy.getDataModel()
#モデル用変数フィールドを追加
#predictField - 実数を格納予定
modelerDataModel.addField(modelerpy.Field( predictField, "real", measure="continuous"))
#修正したデータモデルをModelerへ戻します。
modelerpy.setOutputDataModel(modelerDataModel)
d. モデル準備パート
else以下の部分をみていきます。
ここからは、モデルを実行して予測結果を取得、Modelerへデータを返す処理になります。
まずは、モデルの準備パート
データを入力する部分はいつもの通りですね。
次にモデルにおけるカラムのロールを定義しますが、
時系列、対象カラムを定義したりするだけで、特段難しくはないですね。
次の Preprocessor のところが重要ですね。
事前学習済みのモデルですが、入力データに合わせて標準化するなどの定義をしています。
そのあと、train()関数で学習させていますね。
この処理がないと、モデルが正確な予測をできません。
モデル準備パート
else:
#-----------------------------------------------
# 1. データ入力
#-----------------------------------------------
#ModelerからPandasでデータを入力
modelerData = modelerpy.readPandasDataframe()
input_df = modelerData
#-----------------------------------------------
# 2. モデル初期処理
# ①. モデルで使用するカラムの指定
# ②. モデル用 Preprocessor の初期化と学習
#-----------------------------------------------
#-----------------------------------------------
# モデルで使用するカラムの指定
#-----------------------------------------------
# 予測対象カラムをリストで定義
target_columns = [ target_column ]
# 時系列カラム"time"をタイムスタンプ型に変更
input_df[ timestamp_column ] = pd.to_datetime(input_df[ timestamp_column ])
# 列の指定
column_specifiers = {
"timestamp_column": timestamp_column, # 時間
"id_columns": [], # 複数系列を識別するためのID（今回は1系列なので空）
"target_columns": target_columns, # 予測対象（"total load actual"）
"control_columns": [], # 説明変数（天気とか、曜日などの外部変数）? 今は無し
}
#-----------------------------------------------
# モデル用 Preprocessor の初期化と学習
#-----------------------------------------------
# 「この特定の電力データにあわせて、どう前処理すればよいか」をモデルに教えてあげる作業。
# 学習データからスケーラーなどのパラメータを学習（平均・分散など）。
# スケーリング（標準化）も有効化。
# 時系列のスライス作成（context + prediction window をスライスする処理）
tsp = TimeSeriesPreprocessor(
**column_specifiers,
context_length=context_length, #学習に使用するレコード数
prediction_length=prediction_length, #予測するレコード数
scaling=True, #データのスケーリングを有効化
scaler_type="standard", #標準化（平均0・分散1）を適用
)
#入力データで事前学習
trained_tsp = tsp.train(input_df)
e. モデル実行パート
モデルの準備が終わったら、入力データを使ってモデルを実行します。
使うモデルは、granite-ttm-512-96-r2 です。
デバイスは、NVIDIAのGPUが搭載された環境ならCUDAになりますね。
私の環境はそんな高価PCではないので、CPUになります。とほほ。
Pipelineは、これまで定義してきた列やデバイス、Preprocessorなどを指定します。
trained_tsp = tsp.train(input_df)と記述して、学習させたのに、
trained_tspではなく tsp を指定しています。ここはtspでよいみたいです。きちんと学習内容は保存されています。
時系列の区分は今回のデータは1時間ごとのデータのため、"h"を指定しています。
最後にPipelineを実行して予測結果を取得しています。
モデル実行パート
#-----------------------------------------------
# 3. モデル設定
# ①. モデルの読み込み
# ②. デバイスの選択
#-----------------------------------------------
# モデルの読み込み（from Hugging Face）
zeroshot_model = TinyTimeMixerForPrediction.from_pretrained(
"ibm-granite/granite-timeseries-ttm-r2", # Hugging Face 上の IBMのGraniteシリーズ 時系列モデル（TinyTimeMixer）
revision="512-96-ft-r2.1", # "512-96-ft-r2.1" は学習済み重み（revision）で、入力長512・出力長96 に対応したモデル
num_input_channels=len(target_columns), # num_input_channels はターゲットの列数。今回は "total load actual" の1列なので 1
)
# デバイスの確認
# GPU（CUDA）が使えれば使う、無ければCPU。
device = "cuda" if torch.cuda.is_available() else "cpu"
#-----------------------------------------------
# 4. モデル実行
# ①. Pipeline定義
# ②. Pipeline実行
#-----------------------------------------------
# Pipline定義 モデル + 前処理器 + 推論設定 をまとめた便利なインターフェース。
# feature_extractor=tsp で、前に train() した前処理器が使われる。
# batch_size=512：推論時のバッチサイズ（データ件数が多いとメモリ効率が良くなる）
pipeline = TimeSeriesForecastingPipeline(
zeroshot_model, # 上記で設定したGraniteを指定
device=device, # CPU or GPUが指定される
feature_extractor=tsp, # 上記で設定したTimeSeriesPreprocessorを指定
batch_size=512, # バッチサイズ
freq="h", # h = 1時間単位
)
#-----------------------------------------------
# Pipeline実行
#-----------------------------------------------
# 戻り値 zeroshot_forecast は DataFrame 形式で、予測結果が入ってる。
zeroshot_forecast = pipeline(input_df)
f. モデル精度確認パート
モデルの精度を確認します。
今回は、Modelerの機能を使って確認する予定ですので、コメントアウトしています。
任意でコメントを外して実行してみてください。
以下の紹介では、コメントアウトされていないので、こちらをコピーして活用してください。
精度家訓パート
#-----------------------------------------------
# 5. モデル精度確認
#-----------------------------------------------
# MSE・RMSE・MAE 計算用関数
def custom_metric(actual, prediction, column_header="results"):
"""Simple function to compute MSE"""
# pandas の Series からリスト → numpy配列に変換。ndim が1次元でも、強制的に2次元配列にする関数で保険
a = np.atleast_2d(np.asarray(actual.tolist()))
p = np.atleast_2d(np.asarray(prediction.tolist()))
# NaN（欠損値）を含む行は無視するマスクを作成。
mask = ~np.any(np.isnan(a), axis=1)
# MSE と MAE を計算。
mse = np.mean(np.square(a[mask, :] - p[mask, :]))
mae = np.mean(np.abs(a[mask, :] - p[mask, :]))
# 評価結果を pandas の DataFrame に整形して返す。
return pd.DataFrame(
{
column_header: {
"mean_squared_error": mse,
"root_mean_squared_error": np.sqrt(mse),
"mean_absolute_error": mae,
}
}
)
print(custom_metric(
zeroshot_forecast["total load actual"], zeroshot_forecast["total load actual_prediction"], "zero-shot forecast"
))
g. 予測結果の整形処理
さて、ここからがちょっと面倒くさいことをしています。
g-1. 私のやりたいこと
Modelerへ予測結果を戻したいのですが、私のイメージを説明します。
①. 入力データは、以下のような時系列データです。
・input_df
| time | total load actual |
|---|---|
| 2018-05-14 16:00:00 | 29598 |
| 2018-05-14 17:00:00 | 29422 |
| 2018-05-14 18:00:00 | 28956 |
| 2018-05-14 19:00:00 | 29499 |
②. モデルからの戻り値は以下のようになっています。
・zeroshot_forecast
| time | total load actual_prediction | total load actual |
|---|---|---|
| 2018-05-14 16:00:00 | 29782, 29904, 30401 | 29422, 28956, 28988 |
| 2018-05-14 17:00:00 | 29811, 30104, 30893 | 28956, 28988, 29499 |
| 2018-05-14 18:00:00 | 29871, 30471, 30949 | 28988, 29499, 30870 |
③. Modelerへ戻すイメージ
ここで、私はこの予測値であるtotal load actual_predictionの値を入力データ(input_df)に結合して
Modelerに戻したいのです。
total load actual_prediction は96個の予測値が格納されたリストです。
これは予測値なので、一つ先のレコードからはじまって、96個先のレコードまでの予測値となっています。
なので、"2018-05-14 16:00:00"の行の"29782, 29904, 30401"という値は以下のような関係性となります。
| time | total load actual_prediction 改 |
|---|---|
| 2018-05-14 16:00:00 | |
| 2018-05-14 17:00:00 | 29782 |
| 2018-05-14 18:00:00 | 29904 |
| 2018-05-14 19:00:00 | 30401 |
なので、下記のように予測値をリストではなく、各レコードに再配置しようとしているのです。
ポイントは各レコードで予測値が作成されているので、各レコードの先頭のリスト値を一つ下のレコードの予測値として格納する点です。
さらに未来の予測値も最後に結合して戻します。
g-2. 過去データについて予測値を各レコードに格納する
先ほどのイメージをコードで書いていきます。
戻り値をコピーして、各リストの一つ目の値を取り出します。
取り出した値を1レコード下のカラムに配置します。
そして、入力データを"time"の値をキーにして結合しています。
過去データの予測値取得
#=====================================
# Step 1: 過去の1ステップ先予測を格納
# 過去の実績値に対してのモデルの予測値も求めておく
#=====================================
# モデルの戻り値をコピー
past_pred_df = zeroshot_forecast.copy()
# ndarrayから1つ目の値を取得する関数
def get_first_val(x):
if isinstance(x, (list, np.ndarray)) and len(x) > 0:
return x[0]
return np.nan
# 予測値の1ステップ先だけ取り出す（リストの1つ目を取得)
past_pred_df[ predictField ] = (
past_pred_df[ prediciton_column ].apply(get_first_val)
)
# 1ステップ先なので、「1時間後」にずらす（一つ先のレコードに格納）
past_pred_df[ predictField ] = past_pred_df[ predictField ].shift(1)
# input_df にマージする（timeで結合、inner join でOK）
merged_df = pd.merge(input_df, past_pred_df[[timestamp_column, predictField]], on=timestamp_column, how="left")
この処理をすると以下のような感じでデータが作成される。
| time | total load actual | $GTS-total load actual |
|---|---|---|
| 2018-05-14 16:00:00 | 29598 | |
| 2018-05-14 17:00:00 | 29422 | 29782 |
| 2018-05-14 18:00:00 | 28956 | 29811 |
g-4. 未来の予測データを取得する。
未来の予測値を取得します。
最後の512レコードで96レコード先まで予測して"2019-01-01 00:00:00"～"2019-01-04 23:00:00"までのデータフレームを作りそこに格納します。
未来の予測値取得
#=================================
# 2: 96時間先の未来を予測する処理
# 新しいデータフレームに未来の値を格納
#=====================================
# 入力の最後の512件だけ取得（512はcontext_length）
input_tail = input_df.tail(context_length)
# 未来96時間分の空の DataFrame を作成
last_time = input_tail[timestamp_column].iloc[-1]
future_times = pd.date_range(start=last_time + pd.Timedelta(hours=1), periods=prediction_length, freq="h")
# 空のDataFrameを作る（NaNが入る）
future_df = pd.DataFrame({timestamp_column: future_times})
# 入力+未来のDataFrameを作って予測（前処理器が内部で補完など行う）
# TinyTimeMixerは、過去context_lengthの系列を使ってprediction_length先を出すモデル
input_for_forecast = pd.concat([input_tail, future_df], ignore_index=True)
# 予測を実行
future_forecast = pipeline(input_for_forecast)
# 未来予測部分だけを切り出して新しいDataFrameに
# 各行の total load actual_prediction は 96個の予測値が入ったリストなので、
# 最初の行だけ使って 1レコードに展開
pred_list = future_forecast[ prediciton_column ].iloc[0]
future_df = pd.DataFrame({
timestamp_column: future_times,
predictField: pred_list,
})
この処理で、以下のようなデータフレームが出来上がります。
| time | $GTS-total load actual |
|---|---|
| 2019-01-01 00:00:00 | 22962 |
| 2019-01-01 01:00:00 | 21492 |
| 2019-01-01 02:00:00 | 20559 |
g-5. 未来の予測データを結合する。
最後にデータを結合し、Modelerに戻します。
未来の予測値を格納したデータフレームのカラムを入力データに合わせた後、
それを、レコード追加する形で結合する。
Modelerへのデータ出力
#=====================================
# 3: 過去データの予測値と未来の値を結合してModelerに戻す
# 新しいデータフレームに未来の値を格納
#=====================================
# 他のカラム（total load actual など）は input_df と同じ構造にするため、NaN で埋める
for col in merged_df.columns:
if col not in future_df.columns:
future_df[col] = np.nan
# カラム順を merged_df に合わせて並べ替え
future_df = future_df[merged_df.columns]
# merged_df + future_df を縦に結合
input_df_with_pred = pd.concat([merged_df, future_df], ignore_index=True)
# 結果確認
print(input_df_with_pred.tail(100)) # 最後の100行を表示
# Modelerにデータを戻す
modelerpy.writePandasDataframe(input_df_with_pred)
最終的に以下のようなデータがModelerへ戻されます。
| time | total load actual | $GTS-total load actual |
|---|---|---|
| 2018-05-14 16:00:00 | 29598 | |
| 2018-05-14 17:00:00 | 29422 | 29782 |
| 2018-05-14 18:00:00 | 28956 | 29811 |
| ・・ | ・・ | ・・ |
| 2018-12-31 23:00:00 | 24455 | 25116 |
| 2019-01-01 00:00:00 | NaN | 22962 |
| 2019-01-01 01:00:00 | NaN | 21492 |
| 2019-01-01 02:00:00 | NaN | 20559 |
5. データを確認して、精度の確認
テーブルノードでデータを確認します。問題なさそうですね。
精度分析ノードで精度を確認します。
予測フィールドの検出設定を、フィールド名の形式 を選択して実行します。
$GTS-total load actualという名称にしたのはこのためです。
精度は以下のとおり
まぁまぁではないでしょうか。精度をpythonで確認した場合は以下のとおりでした。
mean_squared_error ：8.675437e+06
root_mean_squared_error ：2.945409e+03
mean_absolute_error ：2.232362e+03
6. 時系列グラフで確認
データ数が多いため条件抽出ノードで"2018-12-15"以降に絞り込みます。
その後、時系列グラフノードを設定します。
系列には、実測値(total load actual)と予測値($GTS-total load actual)を選択します。
X軸には、"time"を設定。
別パネルに時系列を表示、正規化のチェックを 外して ください。
以下が出力結果です。2019-01-01以降も予測されていますね。
4. 最後に
さて、いかがでしたでしょうか。
ZeroShotモデルなのに、素晴らしい精度だと思いました。
今回使用したHugging Face経由でのモデルは、"Fine Tuning(モデルをチューニング)"して予測値を取得することもできるため、そちらもチャレンジしてみようと思います。
また、Modelerの時系列モデルとの結果の比較もしてみたいですね。
今回は以上となります。
参考
SPSS Modeler ノードリファレンス目次
SPSS Modeler 逆引きストリーム集
SPSS funさん記事集
SPSS連載ブログバックナンバー
SPSSヒモトクブログなどは以下のTechXchangeのコミュニティに統合されました。
ご興味がある方は、ぜひiBM IDを登録して参加してみてください！！！お待ちしています。
IBM TechXchange Data Science Japan