pytorch_forecasting

최근 아시는 분에게 패키지를 추천 받아서 적용 예제 코드를 만들어 보았다. 

상세 내용을 포스팅 하고 싶었으나 시간이 너무 소요되어, 다음에 포스팅 하고자 한다.

해당 알고리즘은 분류 모형과 회귀모형 모두를 지원한다. 

우선 데이터를 만들어 보자. 실행 가능한 예제 데이터가 없을 것 같아. 임의로 만들었으니 양해 바란다.

import pandas as pd
import numpy as np
import pytorch_lightning as pl
import pytorch_forecasting as pf
classification_train_data = pd.DataFrame(
    dict(
        target=np.random.choice(["A", "B", "C"], size=300),
        value=np.random.rand(300),  
        group=np.repeat(np.arange(3), 100),
        time_idx=np.tile(np.arange(100), 3),
    )
)
classification_valid_data = pd.DataFrame(
    dict(
        target=np.random.choice(["A", "B", "C"], size=300),
        value=np.random.rand(300),  
        group=np.repeat(np.arange(3), 100),
        time_idx=np.tile(np.arange(100), 3),
    )
)

classification_test_data = pd.DataFrame(
    dict(
        target=np.random.choice(["A", "B", "C"], size=300), 
        value=np.random.rand(300),  
        group=np.repeat(np.arange(3), 100),
        time_idx=np.tile(np.arange(100), 3),
    )
)

 

만들어진 데이터는 아래의 차원을 가지고 있다. 각각 train, valid, test 자료이다.

자료는 dataloader를 통해 전달되는 형식이다. 

TimeSeriesDataSet을 설정하는 부분이 가장 중요해 보인다. 

NaN Label Encoder는 없는 라벨이 생길 수 있다는 설정인것 같다. 

자세한 것은 아래를 참조하기 바란다.

https://pytorch-forecasting.readthedocs.io/en/latest/api/pytorch_forecasting.data.timeseries.TimeSeriesDataSet.html

TimeSeriesDataSet — pytorch-forecasting documentation

 

training = pf.TimeSeriesDataSet(
    classification_train_data,
    group_ids=["group"],
    target="target", 
    time_idx="time_idx",
    min_encoder_length=12,
    max_encoder_length=12,
    min_prediction_length=7,
    max_prediction_length=7,
    time_varying_unknown_reals=["value"],
    time_varying_known_reals=["time_idx"],
    target_normalizer=pf.data.encoders.NaNLabelEncoder(), 
    # Use the NaNLabelEncoder to encode categorical target
    add_relative_time_idx=False,
#     add_encoder_length=True,
#     add_target_scales=True,
#     randomize_length=None
)

#predict 를 True 로 주면 마지막 자료만 기억함.
validation=pf.TimeSeriesDataSet.from_dataset(training,
            classification_valid_data,predict=False,stop_randomization=True)


batch_size = 3
train_dataloader = training.to_dataloader(train=True, 
                                          batch_size=batch_size, num_workers=0)
val_dataloader   = validation.to_dataloader(train=False, 
                                          batch_size=batch_size, num_workers=0)

 

아래와 같이 학습 및 하이퍼 파라미터 초기화를 수행하면 된다.

pl.seed_everything(42)

lr_logger = pl.callbacks.LearningRateMonitor()  
logger = pl.loggers.TensorBoardLogger("lightning_logs") 
early_stop_callback = pl.callbacks.EarlyStopping(monitor="val_loss",min_delta=1e-4, patience=50, verbose=False, mode="min")

trainer=pl.Trainer(max_epochs=50,
                   gpus=0,
                   weights_summary='top',
                   gradient_clip_val=.1,
                   callbacks=[lr_logger,early_stop_callback])

tft=pf.TemporalFusionTransformer.from_dataset(
    training,
    learning_rate=0.03,
    hidden_size=32,
    attention_head_size=1,
    dropout=.1,
    hidden_continuous_size=8,
    output_size=7,
    loss=pf.metrics.CrossEntropy(),
    reduce_on_plateau_patience=4)
    
res = trainer.tuner.lr_find(
    tft,
    train_dataloader=train_dataloader,
    val_dataloaders=val_dataloader,
    max_lr=10.0,
    min_lr=1e-6,
)


trainer.fit(
    tft,
    train_dataloader=train_dataloader,
    val_dataloaders=val_dataloader)

예측은 아래와 같다.

test=pf.TimeSeriesDataSet.from_dataset(training,
            classification_test_data,predict=False,stop_randomization=True)

test_dataloader   = test.to_dataloader(train=False, 
                                          batch_size=batch_size, num_workers=0)

 

회귀 예측은 다음과 같다.

import torch
import numpy as np
import pandas as pd
import pytorch_lightning as pl
import pytorch_forecasting as pf
from pytorch_forecasting.data.examples import generate_ar_data
import statsmodels.api as sm
import statsmodels as stat
import matplotlib.pyplot as plt
print(f'GPU : {torch.cuda.is_available()}')

#자료 생성
data=sm.datasets.get_rdataset('AirPassengers').data
data['time']=pd.date_range(pd.to_datetime('1949-01-01'),pd.to_datetime('1961-01-01'),freq='M')
data=data.set_index('time')
dcp=stat.tsa.seasonal.seasonal_decompose(data['value'],model='additive')
data=pd.DataFrame([data.index,dcp.seasonal,dcp.trend,dcp.resid,data.value]).T

data.columns=['time','seasonal','trend','resid','value']
data=data.fillna(0)
data=data.set_index('time')
data['target']=data.value.shift(-1)
data=data.drop('value',axis=1)
data=data[data['trend']!=0]

#group이 1개이면 1 할당
data['group_ids']=1
data['time_idx']=[int(i) for i in range(1,data.shape[0]+1)]

train = data[:80]
valid = data[80:110]

training=pf.TimeSeriesDataSet(
    train.dropna(),
    time_idx='time_idx',
    target='target',
    group_ids=['group_ids'],
    static_reals=['resid'],
    min_encoder_length=12,
    max_encoder_length=12,
    min_prediction_length=12,
    max_prediction_length=12,
    lags={'seasonal':[1,2,3,4,5,6]},
    time_varying_known_reals=["time_idx"],
    time_varying_unknown_reals=['seasonal','trend'],
    add_relative_time_idx=False,
    add_target_scales=True,
#     add_encoder_length=False,
    randomize_length=None)

validation=pf.TimeSeriesDataSet.from_dataset(training,
            valid.dropna(),predict=False,stop_randomization=True)

batch_size = 1
train_dataloader = training.to_dataloader(train=True, 
                                          batch_size=batch_size, num_workers=0)
val_dataloader   = validation.to_dataloader(train=False, 
                                          batch_size=batch_size, num_workers=0)
                                          
#베이스라인 성능                                          
actuals = torch.cat([y for x, (y, weight) in iter(val_dataloader)])
baseline_predictions = pf.Baseline().predict(val_dataloader)
(actuals - baseline_predictions).abs().mean().item()

#시드 고정
pl.seed_everything(42)
#로그 저장
lr_logger = pl.callbacks.LearningRateMonitor()  # log the learning rate
logger = pl.loggers.TensorBoardLogger("lightning_logs")  # logging results to a tensorboard
early_stop_callback = pl.callbacks.EarlyStopping(monitor="val_loss", min_delta=1e-4, patience=50, verbose=False, mode="min")

trainer=pl.Trainer(max_epochs=5000,
                   gpus='0',
                   weights_summary='top',
                   gradient_clip_val=.1,
                   callbacks=[lr_logger,early_stop_callback])

tft=pf.TemporalFusionTransformer.from_dataset(
    training,
    learning_rate=0.03,
    hidden_size=32,
    attention_head_size=1,
    dropout=.1,
    hidden_continuous_size=8,
    output_size=7,
    loss=pf.metrics.QuantileLoss(),
    reduce_on_plateau_patience=4)

##학습률 조절    
res = trainer.tuner.lr_find(
    tft,
    train_dataloader=train_dataloader,
    val_dataloaders=val_dataloader,
    max_lr=10.0,
    min_lr=1e-6,
)

#시각화
print(f"suggested learning rate: {res.suggestion()}")
fig = res.plot(show=True, suggest=True)
fig.show()

#학습 수행
trainer.fit(
    tft,
    train_dataloader=train_dataloader,
    val_dataloaders=val_dataloader)
    
#검증 자료 생성
test  = data[100:]
test      =pf.TimeSeriesDataSet.from_dataset(training,
             test.dropna(),predict=False,stop_randomization=True)
test_dataloader  =test.to_dataloader(train=False, 
                                          batch_size=batch_size, num_workers=0)
                                          
#예측자료 생성
pred=tft.predict(test_dataloader)