|
時系列予測は永続的なトピックです。自然言語処理の分野での成功に触発されて、トランスフォーマー モデルも時系列予測で大きな進歩を遂げました。この記事は、Transformer モデルを使用した時系列予測を学習するための出発点として役立ちます。 ここでは、Kaggle の Store Sales — Time Series Forecasting をデータとして直接使用します。このコンテストでは、今後 16 日間にわたる 54 店舗のさまざまな製品ラインの売上を予測し、合計 1782 の時系列を作成する必要がありました。データは 2013 年 1 月 1 日から 2017 年 8 月 15 日までのもので、今後 16 日間の売上を予測することが目標です。簡潔にするために簡略化していますが、モデルへの入力には 20 列に 3,029,400 個のデータ ポイントが含まれています。各行のキー列は、「store_nbr」、「family」、および「date」です。データは、次の 3 つの変数カテゴリに分類されます。 1. 最後のトレーニング データの日付 (2017 年 8 月 15 日) 時点でわかっている時間関連の変数。これらの変数には、特定の店舗での特定の製品ラインの売上を表す「sales」、店舗での合計取引数を表す「transactions」、その店舗の合計売上を表す「store_sales」、その製品ラインの合計売上を表す「family_sales」などの数値変数が含まれます。 2. トレーニングの期限（2017 年 8 月 31 日）は、「onpromotion」（製品ライン内のプロモーション対象製品の数）や「dcoilwtico」などの変数を含め、以前からわかっていました。これらの数値列は、「休日」列によって補完されます。この列は休日またはイベントの存在を示し、整数としてカテゴリ別にコード化されます。さらに、「time_idx」、「week_day」、「month_day」、「month」、および「year」の列は、整数としてエンコードされた時間コンテキストを提供します。私たちのモデルはエンコーダのみですが、デコーダがない場合の将来の情報を含めるために、16日間の移動値「onpromotion」と「dcoilwtico」が追加されました。 3. 静的共変量は時間の経過とともに一定であり、「store_nbr」、「family」などの識別子や、「city」、「state」、「type」、「cluster」（店舗の特性の詳細）などのカテゴリ変数が含まれ、これらはすべて整数でエンコードされています。 最終的に生成される df は「data_all」という名前で、次の構造を持ちます。 categorical_covariates = ['time_idx','week_day','month_day','month','year','holiday'] categorical_covariates_num_embeddings = [] for col in categorical_covariates: data_all[col] = data_all[col].astype('category').cat.codes categorical_covariates_num_embeddings.append(data_all[col].nunique()) categorical_static = ['store_nbr','city','state','type','cluster','family_int'] categorical_static_num_embeddings = [] for col in categorical_static: data_all[col] = data_all[col].astype('category').cat.codes categorical_static_num_embeddings.append(data_all[col].nunique()) numeric_covariates = ['sales','dcoilwtico','dcoilwtico_future','onpromotion','onpromotion_future','store_sales','transactions','family_sales'] target_idx = np.where(np.array(numeric_covariates)=='sales')[0][0] データを PyTorch モデルに適したテンソルに変換する前に、データをトレーニング セットと検証セットに分割する必要があります。ウィンドウ サイズは、各トレーニング サンプルのシーケンスの長さを表す重要なハイパーパラメータです。さらに、「num_val」は使用される検証フォールドの数を示し、このコンテキストでは 2 に設定されています。 2013年1月1日から2017年6月28日までの観測データをトレーニングデータセットとして指定し、2017年6月29日から2017年7月14日までと2017年7月15日から2017年7月30日までを検証期間として指定しました。 同時に、データがスケーリングされます。完全なコードは次のとおりです。 def dataframe_to_tensor(series,numeric_covariates,categorical_covariates,categorical_static,target_idx): numeric_cov_arr = np.array(series[numeric_covariates].values.tolist()) category_cov_arr = np.array(series[categorical_covariates].values.tolist()) static_cov_arr = np.array(series[categorical_static].values.tolist()) x_numeric = torch.tensor(numeric_cov_arr,dtype=torch.float32).transpose(2,1) x_numeric = torch.log(x_numeric+1e-5) x_category = torch.tensor(category_cov_arr,dtype=torch.long).transpose(2,1) x_static = torch.tensor(static_cov_arr,dtype=torch.long) y = torch.tensor(numeric_cov_arr[:,target_idx,:],dtype=torch.float32) return x_numeric, x_category, x_static, y window_size = 16 forecast_length = 16 num_val = 2 val_max_date = '2017-08-15' train_max_date = str((pd.to_datetime(val_max_date) - pd.Timedelta(days=window_size*num_val+forecast_length)).date()) train_final = data_all[data_all['date']<=train_max_date] val_final = data_all[(data_all['date']>train_max_date)&(data_all['date']<=val_max_date)] train_series = train_final.groupby(categorical_static+['family']).agg(list).reset_index() val_series = val_final.groupby(categorical_static+['family']).agg(list).reset_index() x_numeric_train_tensor, x_category_train_tensor, x_static_train_tensor, y_train_tensor = dataframe_to_tensor(train_series,numeric_covariates,categorical_covariates,categorical_static,target_idx) x_numeric_val_tensor, x_category_val_tensor, x_static_val_tensor, y_val_tensor = dataframe_to_tensor(val_series,numeric_covariates,categorical_covariates,categorical_static,target_idx)
データをロードする際、データ ローダーは、モデルがさまざまなシーケンス セグメントに公開されるように、各時系列をウィンドウ範囲内のランダム インデックスから始まる時間チャンクに分割する必要があります。 バイアスを減らすために、データをランダムにシャッフルするのではなく、チャンクの開始時間に基づいてデータセットをソートする追加のハイパーパラメータ設定が導入されました。その後、データは 5 年間のデータセットを反映して 5 つの部分に分割され、各部分は内部でシャッフルされ、最終バッチには昨年の観測値が含まれますが、ランダム性は維持されます。モデルの最終的な勾配更新は最新の年の影響を受け、理論的には最新の期間の予測が改善されます。 def divide_shuffle(df,div_num): space = df.shape[0]//div_num division = np.arange(0,df.shape[0],space) return pd.concat([df.iloc[division[i]:division[i]+space,:].sample(frac=1) for i in range(len(division))]) def create_time_blocks(time_length,window_size): start_idx = np.random.randint(0,window_size-1) end_idx = time_length-window_size-16-1 time_indices = np.arange(start_idx,end_idx+1,window_size)[:-1] time_indices = np.append(time_indices,end_idx) return time_indices def data_loader(x_numeric_tensor, x_category_tensor, x_static_tensor, y_tensor, batch_size, time_shuffle): num_series = x_numeric_tensor.shape[0] time_length = x_numeric_tensor.shape[1] index_pd = pd.DataFrame({'serie_idx':range(num_series)}) index_pd['time_idx'] = [create_time_blocks(time_length,window_size) for n in range(index_pd.shape[0])] if time_shuffle: index_pd = index_pd.explode('time_idx') index_pd = index_pd.sample(frac=1) else: index_pd = index_pd.explode('time_idx').sort_values('time_idx') index_pd = divide_shuffle(index_pd,5) indices = np.array(index_pd).astype(int) for batch_idx in np.arange(0,indices.shape[0],batch_size): cur_indices = indices[batch_idx:batch_idx+batch_size,:] x_numeric = torch.stack([x_numeric_tensor[n[0],n[1]:n[1]+window_size,:] for n in cur_indices]) x_category = torch.stack([x_category_tensor[n[0],n[1]:n[1]+window_size,:] for n in cur_indices]) x_static = torch.stack([x_static_tensor[n[0],:] for n in cur_indices]) y = torch.stack([y_tensor[n[0],n[1]+window_size:n[1]+window_size+forecast_length] for n in cur_indices]) yield x_numeric.to(device), x_category.to(device), x_static.to(device), y.to(device) def val_loader(x_numeric_tensor, x_category_tensor, x_static_tensor, y_tensor, batch_size, num_val): num_time_series = x_numeric_tensor.shape[0] for i in range(num_val): for batch_idx in np.arange(0,num_time_series,batch_size): x_numeric = x_numeric_tensor[batch_idx:batch_idx+batch_size,window_size*i:window_size*(i+1),:] x_category = x_category_tensor[batch_idx:batch_idx+batch_size,window_size*i:window_size*(i+1),:] x_static = x_static_tensor[batch_idx:batch_idx+batch_size] y_val = y_tensor[batch_idx:batch_idx+batch_size,window_size*(i+1):window_size*(i+1)+forecast_length] yield x_numeric.to(device), x_category.to(device), x_static.to(device), y_val.to(device)
モデルここでは、Pytorch を使用して、「Attention is All You Need」(2017)² で説明されている Transformer アーキテクチャを簡単に実装します。時系列予測であるため、注意メカニズムには因果関係は必要ありません。つまり、注意ブロックにマスキングは適用されません。 入力から始めます。カテゴリ特徴は埋め込みレイヤーを通過して密な形式で表現され、その後 Transformer ブロックに送られます。多層パーセプトロン (MLP) は、最終的なエンコードされた入力を受け取り、予測を生成します。埋め込み次元、各 Transformer ブロック内の注意ヘッドの数、およびドロップアウト確率は、モデルの主なハイパーパラメータです。複数の Transformer ブロックのスタックは、「num_blocks」ハイパーパラメータによって制御されます。 以下は、単一の Transformer ブロックと全体的な予測モデルの実装です。 class transformer_block(nn.Module): def __init__(self,embed_size,num_heads): super(transformer_block, self).__init__() self.attention = nn.MultiheadAttention(embed_size, num_heads, batch_first=True) self.fc = nn.Sequential(nn.Linear(embed_size, 4 * embed_size), nn.LeakyReLU(), nn.Linear(4 * embed_size, embed_size)) self.dropout = nn.Dropout(drop_prob) self.ln1 = nn.LayerNorm(embed_size, eps=1e-6) self.ln2 = nn.LayerNorm(embed_size, eps=1e-6) def forward(self, x): attn_out, _ = self.attention(x, x, x, need_weights=False) x = x + self.dropout(attn_out) x = self.ln1(x) fc_out = self.fc(x) x = x + self.dropout(fc_out) x = self.ln2(x) return x class transformer_forecaster(nn.Module): def __init__(self,embed_size,num_heads,num_blocks): super(transformer_forecaster, self).__init__() num_len = len(numeric_covariates) self.embedding_cov = nn.ModuleList([nn.Embedding(n,embed_size-num_len) for n in categorical_covariates_num_embeddings]) self.embedding_static = nn.ModuleList([nn.Embedding(n,embed_size-num_len) for n in categorical_static_num_embeddings]) self.blocks = nn.ModuleList([transformer_block(embed_size,num_heads) for n in range(num_blocks)]) self.forecast_head = nn.Sequential(nn.Linear(embed_size, embed_size*2), nn.LeakyReLU(), nn.Dropout(drop_prob), nn.Linear(embed_size*2, embed_size*4), nn.LeakyReLU(), nn.Linear(embed_size*4, forecast_length), nn.ReLU()) def forward(self, x_numeric, x_category, x_static): tmp_list = [] for i,embed_layer in enumerate(self.embedding_static): tmp_list.append(embed_layer(x_static[:,i])) categroical_static_embeddings = torch.stack(tmp_list).mean(dim=0).unsqueeze(1) tmp_list = [] for i,embed_layer in enumerate(self.embedding_cov): tmp_list.append(embed_layer(x_category[:,:,i])) categroical_covariates_embeddings = torch.stack(tmp_list).mean(dim=0) T = categroical_covariates_embeddings.shape[1] embed_out = (categroical_covariates_embeddings + categroical_static_embeddings.repeat(1,T,1))/2 x = torch.concat((x_numeric,embed_out),dim=-1) for block in self.blocks: x = block(x) x = x.mean(dim=1) x = self.forecast_head(x) return x
修正したトランスフォーマーのアーキテクチャを以下に示します。 このモデルは、数値特徴、カテゴリ特徴、静的特徴の 3 つの個別の入力テンソルを受け入れます。カテゴリカルおよび静的特徴埋め込みは平均化され、数値特徴と結合されて、Transformer ブロックの準備が整った形状 (batch_size、window_size、embedding_size) のテンソルが形成されます。この複合テンソルには、必要な位置情報を提供する埋め込み時間変数も含まれています。 Transformer ブロックは、順次情報を抽出し、結果のテンソルを時間次元に沿って集約し、それを MLP に渡して最終的な予測 (batch_size、forecast_length) を生成します。このコンテストでは、評価指標として二乗平均対数誤差 (RMSLE) を使用します。式は次のとおりです。 予測は対数変換されるため、-1 未満の負の売上予測 (未定義のエラーが発生します) を処理する必要があります。そのため、負の売上予測とそれに伴う NaN 損失値を回避するために、MLP レイヤーの後に ReLU アクティベーション レイヤーを追加して、負でない予測を保証します。 class RMSLELoss(nn.Module): def __init__(self): super().__init__() self.mse = nn.MSELoss() def forward(self, pred, actual): return torch.sqrt(self.mse(torch.log(pred + 1), torch.log(actual + 1)))
トレーニングと検証モデルをトレーニングするときに設定する必要があるハイパーパラメータがいくつかあります。ウィンドウ サイズ、シャッフル時間の有無、埋め込みサイズ、ヘッド数、ブロック数、ドロップアウト、バッチ サイズ、学習率です。次の構成は有効ですが、最適であることは保証されません。 num_epoch = 1000 min_val_loss = 999 num_blocks = 1 embed_size = 500 num_heads = 50 batch_size = 128 learning_rate = 3e-4 time_shuffle = False drop_prob = 0.1 model = transformer_forecaster(embed_size,num_heads,num_blocks).to(device) criterion = RMSLELoss() optimizer = torch.optim.Adam(model.parameters(),lr=learning_rate) scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=50, gamma=0.5) ここでは、adam オプティマイザーと学習率スケジュールを使用して、トレーニング中に学習率が徐々に調整されるようにします。 for epoch in range(num_epoch): batch_loader = data_loader(x_numeric_train_tensor, x_category_train_tensor, x_static_train_tensor, y_train_tensor, batch_size, time_shuffle) train_loss = 0 counter = 0 model.train() for x_numeric, x_category, x_static, y in batch_loader: optimizer.zero_grad() preds = model(x_numeric, x_category, x_static) loss = criterion(preds, y) train_loss += loss.item() counter += 1 loss.backward() optimizer.step() train_loss = train_loss/counter print(f'Epoch {epoch} training loss: {train_loss}') model.eval() val_batches = val_loader(x_numeric_val_tensor, x_category_val_tensor, x_static_val_tensor, y_val_tensor, batch_size, num_val) val_loss = 0 counter = 0 for x_numeric_val, x_category_val, x_static_val, y_val in val_batches: with torch.no_grad(): preds = model(x_numeric_val,x_category_val,x_static_val) loss = criterion(preds,y_val).item() val_loss += loss counter += 1 val_loss = val_loss/counter print(f'Epoch {epoch} validation loss: {val_loss}') if val_loss<min_val_loss: print('saved...') torch.save(model,data_folder+'best.model') min_val_loss = val_loss scheduler.step() 結果トレーニング後、最高のパフォーマンスを示したモデルのトレーニング損失は 0.387、検証損失は 0.457 でした。テスト セットに適用すると、モデルは RMSLE 0.416 を達成し、競争で 89 位 (上位 10%) にランクされました。 埋め込みが大きく、アテンション ヘッドが多いほどパフォーマンスが向上するように見えますが、最良の結果は単一の Transformer で達成され、データが限られている場合はシンプルさが利点となることを示唆しています。グローバル シャッフルを放棄してローカル シャッフルを選択すると、効果が向上します。わずかな時間バイアスを導入すると、予測の精度が向上します。 |
