3. 딥러닝을 이용한 시계열 예측 (TensorFlow, LSTM)

 

 

---

단변량 (univariate) & 단일 시점 (single step)

 

1. (단변량) 시계열 데이터셋 만들기 

• 우선 예제의 단변량 시계열 데이터의 형태는 아래와 같음. (시간에 따른 기온 데이터) 

• 시계열 데이터를 슬라이딩 윈도우 형태로 변환해 준다. 
• 윈도 형태 데이터셋 변환 함수에는 train, valid, test 데이터셋 구분을 위한 start ~ end index , window 사이즈, 예측하고자 하는 타깃시점 (= 몇 step 후를 예측할지)을 입력한다.
• 아래 코드는 single step 예측 기준. target_size 를 크게 잡을수록 먼 시점을 예측

# Define a specific window for training Neural Network
def univariate_data(dataset, start_index, end_index, history_size, target_size):
    data = []
    labels = []

    start_index = start_index + history_size   # start_index 에 윈도우사이즈를 더해줌.
    if end_index is None:
        end_index = len(dataset) - target_size   # end_index 안주면 끝까지

    for i in range(start_index, end_index):
        indices = range(i - history_size, i)   # start 부터 end index 에 해당하는 타임 윈도우에 대한 인덱스 만들어줌
        data.append(np.reshape(dataset[indices], (history_size, 1)))    # Reshape data from (history_size,) to (history_size, 1)
        labels.append(dataset[i+target_size])  # [i - history ~ i - 1] 까지를 학습데이터로 해서 [i+target_size] 를 예측
    return np.array(data), np.array(labels)
    
# Standardization
uni_data = uni_data.values  # series -> ndarray 변환
uni_train_mean = uni_data[:TRAIN_SPLIT].mean()
uni_train_std = uni_data[:TRAIN_SPLIT].std()
uni_data = (uni_data - uni_train_mean) / uni_train_std  # Standardization

# univariate_data 함수를 이용한 학습/검증 데이터셋 생성 
univariate_past_history = 20
univariate_future_target = 0

x_train_uni, y_train_uni = univariate_data(uni_data, 0, TRAIN_SPLIT,
                                         univariate_past_history,
                                         univariate_future_target)
x_val_uni, y_val_uni = univariate_data(uni_data, TRAIN_SPLIT, None,
                                     univariate_past_history,
                                     univariate_future_target)

print('Single window of past history')
print(x_train_uni[1])
print('\n Target temperature to predict')
print(y_train_uni[1])

 

• Tensor Flow의 iterable 한 Dataset 만들기

BATCH_SIZE = 256
BUFFER_SIZE = 10000

train_univariate = tf.data.Dataset.from_tensor_slices((x_train_uni, y_train_uni))
train_univariate = train_univariate.cache().shuffle(BUFFER_SIZE).batch(BATCH_SIZE).repeat()

# cache() 는 데이터를 메모리에 저장시켜, epoch 마다 데이터를 새로 읽을필요없이 캐시로 부터 저장된 값을 더 빠르게 불러올수 있게 해줌.
# repeat() 을 통해, 매 epoch 마다 batch 를 새로 shuffling 하게 해줌


val_univariate = tf.data.Dataset.from_tensor_slices((x_val_uni, y_val_uni))
val_univariate = val_univariate.batch(BATCH_SIZE)

# 한 epoch 내에서 총 배치 개수 회수만큼만 학습할거면 repeat()을 굳이 사용하지 않아도 된다.

 

2. LSTM 모델 생성, 학습, 검증

• LSTM을 생성할 때는 유닛의 개수 (= 출력 차원 크기)와 입력 데이터의 형태를 입력해 준다. 
• return sequence 가 False 이면 : (배치사이즈, 윈도우 사이즈, 피쳐 개수) →  (유닛 개수) 
• True 이면 : (배치사이즈, 윈도우윈도 사이즈, 피쳐 개수) →  (배치사이즈, 윈도 사이즈, 유닛 개수) 

# Keras 를 이용한 LSTM 모델 생성 

simple_lstm_model = tf.keras.models.Sequential([
    tf.keras.layers.LSTM(8, input_shape=np.array(x_train_uni).shape[-2:]),
    # LSTM 의 유닛 수 (8) 은 출력차원 크기와 같음
    # LSTM 의 입력데이터 shape은 기본적으로 (batch_size, time_steps, features) 인데, input_shape 은 뒤에 2개 (time_steps, features) 의미함
    # return_sequences 을 지정하지 않으면 디폴트로 False, return_sequences=True 로 하면 입력 텐서의 time_steps 마다의 output을 출력함
    # 즉 False 일때의 output = (batch_size, units) 의 2D 텐서이고, True 일때는 (batch_size, time_steps, units) 의 3D 텐서
    # 예를들어 다음날 주가를 예측하는 시계열 모델은 False 를 쓰면되고, 문장의 각 단어에 대한 번역을 필요로하는 모델은 True를 쓰면 됨
    tf.keras.layers.Dense(1)
])

simple_lstm_model.compile(optimizer='adam', loss='mae')

 

• 모델 학습 

EVALUATION_INTERVAL = 200
EPOCHS = 10

simple_lstm_model.fit(train_univariate, epochs=EPOCHS,
                      steps_per_epoch=EVALUATION_INTERVAL,
                      validation_data=val_univariate, validation_steps=50)

# step 수 지정은, 총 배치중에 몇개를 학습할지 뜻함. batch 가 1117개 인데 200으로 설정하면 200개만 학습함
# 위에서 train_univariate 를 repeat() 으로 무한대로 반복하게 만들어줬으면 steps_per_epoch 파라미터 필수로 지정해줘야함.
# repeat() 안했으면 steps_per_epoch 필수 지정 안해도 되는데, 이때는 batch 의 총 개수만큼 학습함

 

• 예측 및 성능 검증 

x_test_uni = x_val_uni[:10000]
y_test_uni = y_val_uni[:10000]
# 원래는 시작 단계부터 val, test 데이터셋 분리해야함 

print(x_test_uni.shape, y_test_uni.shape)

# 예측 성능 평가 (MAE)
mae_loss = tf.keras.losses.MeanAbsoluteError()
test_loss = mae_loss(y_test_uni, predictions.flatten()).numpy() # flatten()에 유의할것.. 타겟과 예측값 shape 이 동일해야함

print(f"Test MAE Loss: {test_loss}")

plt.figure(figsize=(10, 6))
plt.plot(y_test_uni[:100], label='True Values')
plt.plot(predictions[:100], label='Predictions')
plt.legend()
plt.show()

 

 

다변량 & 다중시점 

 

1. (다변량) 시계열 데이터셋 만들기 

• 앞선 같은 예제를 기준. 다변량 시계열 데이터셋의 모습은 아래와 같다. (기압,온도,밀도 3 변수에 타깃은 온도)

• 다변량 데이터 변수별 시각화 및 표준화 

# visualization
features.plot(subplots=True)

# standardization
dataset = features.values
data_mean = dataset[:TRAIN_SPLIT].mean(axis=0)
data_std = dataset[:TRAIN_SPLIT].std(axis=0)
dataset = (dataset-data_mean)/data_std
print(dataset, dataset.shape, type(dataset))

 

• 다변량 시계열 데이터셋을 슬라이딩 윈도우 형태로 생성 
• 앞선 예제와 동일하나, multi step (구간 예측 = 다중시점 예측)이 가능하도록 설정. 아래 코드에서는 과거 720개 시점을 6개 step으로 추출해서 타임 윈도를 구성하여 인풋으로 하고, 미래 72개 시점을 아웃풋으로 예측.

def multivariate_data(dataset, target, start_index, end_index, history_size, target_size, step, single_step=False):   # 다변량은 타겟을 따로 명시해줘야한다.
    data = []
    labels = []

    start_index = start_index + history_size
    if end_index is None:
        end_index = len(dataset) - target_size

    for i in range(start_index, end_index):
        indices = range(i - history_size, i, step)  # window size = history_size // step
        data.append(dataset[indices])

        if single_step:
            labels.append(target[i + target_size])   # single step 예측이라면
        else:
            labels.append(target[i:i + target_size])   # multi step 이라면 (=구간예측)
    return np.array(data), np.array(labels)
    
    past_history = 720
    
past_history = 720
future_target = 72
STEP = 6

x_train_multi, y_train_multi = multivariate_data(dataset, dataset[:, 1], 0,
                                                 TRAIN_SPLIT, past_history, future_target, STEP)
x_val_multi, y_val_multi = multivariate_data(dataset, dataset[:, 1],
                                             TRAIN_SPLIT, None, past_history, future_target, STEP)

print('Single window of past history : {}'.format(x_train_multi[0].shape))
print('\n Target temperature to predict : {}'.format(y_train_multi[0].shape)) # multi step 이기 때문에 target이 72개 값    

print(x_train_single.shape, y_train_single.shape, x_val_single.shape, y_val_single.shape)

BATCH_SIZE = 256
BUFFER_SIZE = 10000

train_multivariate = tf.data.Dataset.from_tensor_slices((x_train_multi, y_train_multi))
train_multivariate = train_multivariate.cache().shuffle(BUFFER_SIZE).batch(BATCH_SIZE).repeat()

val_multivariate = tf.data.Dataset.from_tensor_slices((x_val_multi, y_val_multi))
val_multivariate = val_multivariate.batch(BATCH_SIZE)

 

 

2. LSTM 모델 생성, 학습, 검증

• 다중시점 구간 예측이기 때문에 LSTM 의 dense lyaer의 최종 output의 차원은 예측하고자 하는 구간의 길이 여야함

multi_step_model = tf.keras.models.Sequential()
multi_step_model.add(tf.keras.layers.LSTM(32, input_shape=x_train_multi.shape[-2:]))
multi_step_model.add(tf.keras.layers.Dense(future_target))
multi_step_model.compile(optimizer=tf.keras.optimizers.RMSprop(), loss='mae')

for x, y in val_multivariate.take(1):     # Dateset을 배치단위로 만들었기 때문에, 검증 데이터셋의 첫번째 batch 에 대한 예측 결과를 출력함 => (256,72,1)
    print(multi_step_model.predict(x).shape)

print(f'예측과 타겟 shape 비교 : 타겟값 : {y_val_multi[0].shape} ')

 

• LSTM 모델 학습 및 예측 결과 확인
• epoch 별 학습,검증 데이터셋의 예측 오차는 History 객체로 반환할 수 있다. 

multi_step_history = multi_step_model.fit(train_multivariate, epochs=EPOCHS,
                                          steps_per_epoch=EVALUATION_INTERVAL,
                                          validation_data=val_multivariate,
                                          validation_steps=50)
                                          
# multi_step_history.history 로 검증오차 반환가능
                                          

# 예측값과 타겟값 형태 확인 
for x, y in val_multivariate.take(1):
  true_values = y
  multi_step_predictions = multi_step_model.predict(x)
  
print(true_values.shape, multi_step_predictions.shape, type(true_values), type(multi_step_predictions))

true_values = np.concatenate(true_values, axis=0)
predicted_values = np.concatenate(multi_step_predictions, axis=0)

print(true_values.shape, predicted_values .shape, type(true_values), type(predicted_values ))

mae = mae_loss(true_values, predicted_values)
print(f'Mean Absolute Error: {mae}')

 

• 예측 결과 시각화 

# Plotting the results
def plot_multi_step(history, true_future, prediction):
    plt.figure(figsize=(8, 4))
    num_in = list(range(-len(history), 0))
    num_out = len(true_future)

    plt.plot(num_in, np.array(history[:, 1]), label='History')
    plt.plot(np.arange(num_out), np.array(true_future), 'bo', label='True Future')
    if prediction.any():
        plt.plot(np.arange(num_out), np.array(prediction), 'ro', label='Predicted Future')
    plt.legend(loc='upper left')
    plt.show()

# Visualize the prediction
for x, y in val_multivariate.take(3): # Displaying first examples :3개의 배치의 각 첫번째 데이터셋 결과 시각화
    for i in range(1):
        plot_multi_step(x[i], y[i], multi_step_model.predict(np.expand_dims(x[i], axis=0))[0])