DeepARで複数時系列予測する方法を解説【外部変数あり】

2024/02/27

上野桃香

DeepARで複数時系列予測する方法を解説【外部変数あり】

DeepAR

はじめに

DeepARで時系列予測をする記事が日本ではあまりないですね。この記事では少しやってみたではなく、外部変数の取り組みや出力結果をどう変換するかなど１通りのやり方を記載しています。

データの変換の仕方も独特なので公式ドキュメントを読み解くことが大変でした…。

DeepARはAmazonで提供されたアルゴリズム

DeepARは、Amazon SageMakerなどのAWSプラットフォームで提供されるアルゴリズムですが、同様にAWS以外の環境でもGluonTSライブラリを使用して利用できます。このアルゴリズムは、RNNのような深層学習モデルを基にしており、時系列データに特化した深層学習モデルを構築することが可能です。

DeepARの特長の一つは、外部変数の組み込みが容易であることです。外部変数は時系列データに関連する補助的な情報であり、これをモデルに組み込むことで精度向上が期待できます。また、DeepARはユーザーフレンドリーで手軽にモデルを構築できるため、初心者でも利用しやすいとされています。初めて利用する際は理解が難しいかもしれませんが、その柔軟性とパフォーマンスから高い評価を受けています。

詳しい解説はこちらにあります。

【15%の精度向上】Amazonが開発した時系列モデル、DeepARが凄い | Data Driven Knowledgebase (since2020.jp)

ライブラリのインポート

全コードは一番下にあるので先に見たい人はスキップして下さい。

GluonTSにはmxnetモデルとtorchモデルの２種類ある＆DeepARもどちらにもあるようですが、私はtorchの方で一通り出来ました。他の人では両方使っている人などいました。

pip install "gluonts[torch]"

DeepARで学習させる用にデータの前処理を行う

def deepar_process_data_1(df,threshold_date='2022-11-24',id_column,target): #日付型に変換 df['date'] = pd.to_datetime(df['date']) #テストデータの期間の目的変数を消去（消しても消さなくても予測結果は変わらない) df.loc[df['date'] > threshold_date,target] = None list_pref = list(df[id_column].unique()) return df,list_pref

ここでは最低限のことしか書きませんが、dfを読みこみ、学習とテストデータの分割点となる日付を入力し、複数時系列データの識別IDのリストを作成します。

def df_to_deepar(df,id_column,sta_cal_list,dyn_real_list,taget):

#dyn_real_list列でnullがある行は削除

df_cleaned = df.dropna(subset=dyn_real_list, how='any')

df_cleaned = df_cleaned.reset_index(drop=True)



# カテゴリ変数をカテゴリ型に変換する

unique_counts = []



# カテゴリ変数をカテゴリ型に変換する

for column in sta_cal_list:

df_cleaned = df_cleaned.copy()

df_cleaned[column] = df_cleaned[column].astype('category')

unique_count = df[column].nunique()

unique_counts.append((unique_count))

print(f'{column}: {unique_count}')



calframes = [df_cleaned[sta_cal_list]]



#DeepAR用のデータセットに変換

result = PandasDataset.from_long_dataframe(df_cleaned, target=taget, item_id=id_column,

timestamp='date', freq='D',static_features=pd.concat(calframes, axis=1),

feat_dynamic_real = dyn_real_list)



# unique_countsには各カラムごとの (カラム名, ユニークなクラス数) が格納される

print("Unique counts for each column:", unique_counts)



return result ,unique_counts

nullがある行を削除し、カテゴリ変数をカテゴリ型に変換します。

そして、各カテゴリ列ごとにグループがいくつあるのかカウントし、リストで返します。（学習時に使用するのでreturnで返す）

そして、pandas型からPandasDataset型に変換します。（名前は似ているけど別物）

timestampやfreqなど必要に応じて変えてください。

DeepARで学習し、予測値を得る

def do_deepar(df,unique_counts,list_pref,id_column):

# 開始時間を記録

start_time = time.time()



# DeepAREstimatorの作成

estimator = DeepAREstimator(

freq='D',

prediction_length=366,

num_layers=5,

trainer_kwargs={'max_epochs': 1},

nonnegative_pred_samples=True,

context_length=366,

cardinality=unique_counts,

num_feat_dynamic_real=3

)



predictor = estimator.train(df)

print('学習完了')

forecast_it, ts_it = make_evaluation_predictions(dataset=df, predictor=predictor, num_samples=100)

forecast = list(forecast_it) # 将来予測値の情報が入ったリスト

tss = list(ts_it) #データに関するリスト

# tssをデータフレーム化

ts_entry = pd.concat([tss[i] for i in range(len(list_pref))], axis=1)

print(ts_entry)

ts_entry.columns = list_pref



# forecastを辞書に格納

forecast_entry = dict(zip(list_pref, forecast))



# 終了時間を記録

end_time = time.time()



# 経過時間を表示

elapsed_time = end_time - start_time

print(f"処理にかかった時間: {elapsed_time}秒")



return forecast,tss

cardinalityの引数で先ほどのリストを格納します。

そして、prediction_lengthには予測期間、num_feat_dynamic_realには動的連続の数値の特徴量の数を入れて下さい。基本context_lengthと予測期間は同じになるようです。

そしてmake_evaluation_predictionsで予測値を作成します。tssはテストデータであり、予測した値はforecastに格納されています。

例として予測期間を366日にしています。なのでもし1366日分のデータdfを入れて予測すると、最も最近の366日の予測が返ってきます。逆に言うと、直近の366日の目的変数の値は何にも使用されません。

def return_all_preds(pred,id_column):

all_preds = list()



for item in pred:

ID = item.item_id

p = item.samples.mean(axis=0)

p10 = np.percentile(item.samples, 10, axis=0)

p90 = np.percentile(item.samples, 90, axis=0)

dates = pd.date_range(start=item.start_date.to_timestamp(), periods=len(p), freq='D')

ID_pred = pd.DataFrame({'date': dates, id_column: ID, 'prediction': p, 'p10': p10, 'p90': p90})

all_preds += [ID_pred]



all_preds = pd.concat(all_preds, ignore_index=True)

all_preds[id_column] = all_preds[id_column].astype(int)



return all_preds

最後に予測値の最低限の加工を行って終了です。

全コード

全コードはこのようになっています。

!pip install "gluonts[torch]"



import pandas as pd

import numpy as np

import matplotlib.pyplot as plt

import seaborn as sns

import time

import copy

from gluonts.torch.model import deepar

from gluonts.dataset.pandas import PandasDataset

from gluonts.torch.model.deepar import DeepAREstimator

from gluonts.evaluation.backtest import make_evaluation_predictions



def deepar_process_data_1(df,threshold_date='2022-11-24',id_column,taget):

#日付型に変換

df['date'] = pd.to_datetime(df['date'])

#テストデータの期間の目的変数を消去（消しても消さなくても予測結果は変わらない)

df.loc[df['date'] > threshold_date,taget] = None

list_pref = list(df[id_column].unique())

return df,list_pref



def df_to_deepar(df,id_column,sta_cal_list,dyn_real_list,taget):

#dyn_real_list列でnullがある行は削除

df_cleaned = df.dropna(subset=dyn_real_list, how='any')

df_cleaned = df_cleaned.reset_index(drop=True)



# カテゴリ変数をカテゴリ型に変換する

unique_counts = []



# カテゴリ変数をカテゴリ型に変換する

for column in sta_cal_list:

df_cleaned = df_cleaned.copy()

df_cleaned[column] = df_cleaned[column].astype('category')

unique_count = df[column].nunique()

unique_counts.append((unique_count))

print(f'{column}: {unique_count}')



calframes = [df_cleaned[sta_cal_list]]



#DeepAR用のデータセットに変換

result = PandasDataset.from_long_dataframe(df_cleaned, target=taget, item_id=id_column,

timestamp='date', freq='D',static_features=pd.concat(calframes, axis=1),

feat_dynamic_real = dyn_real_list)



# unique_countsには各カラムごとの (カラム名, ユニークなクラス数) が格納される

print("Unique counts for each column:", unique_counts)



return result ,unique_counts



def do_deepar(df,unique_counts,list_pref,id_column):

# 開始時間を記録

start_time = time.time()



# DeepAREstimatorの作成

estimator = DeepAREstimator(

freq='D',

prediction_length=366,

num_layers=5,

trainer_kwargs={'max_epochs': 1},

nonnegative_pred_samples=True,

context_length=366,

cardinality=unique_counts,

num_feat_dynamic_real=3

)



predictor = estimator.train(df)

print('学習完了')

forecast_it, ts_it = make_evaluation_predictions(dataset=df, predictor=predictor, num_samples=100)

forecast = list(forecast_it) # 将来予測値の情報が入ったリスト

tss = list(ts_it) # テストデータに関するリスト

# tssをデータフレーム化

ts_entry = pd.concat([tss[i] for i in range(len(list_pref))], axis=1)

print(ts_entry)

ts_entry.columns = list_pref



# forecastを辞書に格納

forecast_entry = dict(zip(list_pref, forecast))



# 終了時間を記録

end_time = time.time()



# 経過時間を表示

elapsed_time = end_time - start_time

print(f"処理にかかった時間: {elapsed_time}秒")



return forecast,tss



def return_all_preds(pred,id_column):

all_preds = list()



for item in pred:

ID = item.item_id

p = item.samples.mean(axis=0)

p10 = np.percentile(item.samples, 10, axis=0)

p90 = np.percentile(item.samples, 90, axis=0)

dates = pd.date_range(start=item.start_date.to_timestamp(), periods=len(p), freq='D')

ID_pred = pd.DataFrame({'date': dates, id_column: ID, 'prediction': p, 'p10': p10, 'p90': p90})

all_preds += [ID_pred]



all_preds = pd.concat(all_preds, ignore_index=True)

all_preds[id_column] = all_preds[id_column].astype(int)



return all_preds



#予測値の加工（※必要であれば）

def process_all_preds(all_preds,df_original):

return all_preds





def main():

#データを取得

df = pd.read_csv('ファイル名')

df_original = df.copy()

#時系列の識別IDの設定

id_column = 'ID'

taget = 'sales'

#静的カテゴリ型のカラム

sta_cal_list = ['area','customers']

#動的連続型のカラム

dyn_real_list = ['lag_8','lag_9','lag_10']

#データの加工

df,list_pref = deepar_process_data_1(df,id_column,taget)

#pandas型を学習用に変換

df,unique_counts= df_to_deepar(df,id_column,sta_cal_list,dyn_real_list,taget)

#学習し、予測値を返す

forecast,tss = do_deepar(df,unique_counts,list_pref,id_column)

#予測値をpanda型に変換

all_preds = return_all_preds(forecast,id_column)

#予測値の加工（※必要であれば追加）

result = process_all_preds(all_preds,df_original)



return result



result = main()