본문 바로가기
딥러닝 with Python

[딥러닝 with Python] 시계열 데이터 분석을 위한 tsai 라이브러리 활용하기: tsai.data.preparation (데이터 준비 및 전처리)

by CodeCrafter 2025. 7. 20.
반응형

 

지난 시간에 tsai.data.external을 통해서 데이터를 쉽게 간편하게 로드하는 방법에 대해서 알아보았습니다.

[딥러닝 with Python] 시계열 데이터 분석을 위한 tsai 라이브러리 활용하기: tsai.data.external (데이터 준비)

 

[딥러닝 with Python] 시계열 데이터 분석을 위한 tsai 라이브러리 활용하기: tsai.data.external (데이터

이번에는 지난 포스팅에서 tsai.data.core 에 이어 [딥러닝 with Python] 시계열 데이터 분석을 위한 tsai 라이브러리 활용하기: tsai.data.core [딥러닝 with Python] 시계열 데이터 분석을 위한 tsai 라이브러리

jaylala.tistory.com

 

 

이번 시간에는 시계열 데이터를 딥러닝 모델에 효과적으로 입력하기 위해 raw data를 적절한 형태로 가공하는 과정에 대해서 알아보겠습니다.

 

tsai 라이브러리의 data.preparation 모듈은 이러한 데이터 준비 작업을 자동화하고 간소화하는 강력한 함수들을 제공하는데요.

오늘은 이 모듈의 핵심 기능들을 지난번과 마찬가지로 Google의 Colab을 활용해 누구나 실습이 가능한 환경에서 알아보도록하겠습니다.

 

1. tsai 라이브러리의 data.preparation 알아보기

1) 라이브러리 임포트

 

먼저 구글 코랩환경에서 드라이브를 연결해주고, tsai 라이브러리를 다운로드 받은 뒤 해당 모듈들을 활용 가능하도록 sys.path를 활용해 다운로드 받은 경로를 활용가능하게끔 추가해줍니다.

from google.colab import drive
drive.mount('/content/drive')

 

import os
import sys

target_dir = '/content/drive/MyDrive/tsai'

if not os.path.exists(target_dir):
    os.makedirs(target_dir)

%cd {target_dir}

if not os.path.exists(os.path.join(target_dir, '.git')):
    !git clone https://github.com/timeseriesAI/tsai.git .

tsai_library_root = target_dir
if tsai_library_root not in sys.path:
    sys.path.append(tsai_library_root)

 


# Python이 해당 경로에서 모듈을 찾을 수 있도록 sys.path에 추가합니다.
import sys
tsai_path = '/content/drive/MyDrive/tsai'
if tsai_path not in sys.path:
    sys.path.append(tsai_path)

 

 

이제 실습에 필요한 라이브러리들을 임포트 해줍니다.

# 필요한 라이브러리 임포트
import numpy as np
import pandas as pd
import torch
import matplotlib.pyplot as plt
from fastcore.all import * # L 객체 사용을 위해
from tsai.data.preparation import *
from tsai.data.validation import get_forecasting_splits # prepare_forecasting_data 예시를 위해

# 경고 메시지 무시 설정 (선택 사항)
import warnings
warnings.filterwarnings('ignore')

 

 

 

2) apply_sliding_window: 시계열 슬라이딩 윈도우 적용

 

시계열 데이터를 딥러닝 모델에 입력을 할 때, 고정된 길이의 과거 데이터를 윈도우(window)로 잘라내고 미래 값을 예측하는 방식으로 변환하는 것은 일반적입니다.

 

tsai에서는 apply_sliding_window 함수가 이 과정을 자동화해줍니다. 

print("--- apply_sliding_window 예시 ---")
# 예시 데이터 생성
data = np.arange(20).reshape(-1,1).repeat(3, 1) * np.array([1, 10, 100])
df = pd.DataFrame(data, columns=['feat_1', 'feat_2', 'feat_3'])
print("원본 DataFrame:")
display(df.head())

# 모든 특성을 사용하여 윈도우 생성 (window_len=8, horizon=1)
window_len = 8
horizon = 1
X, y = apply_sliding_window(df, window_len, horizon=horizon, x_vars=None, y_vars=None)

print(f"\nX.shape: {X.shape}, y.shape: {y.shape}")
print(f"첫 번째 X 윈도우:\n{X[0]}")
print(f"첫 번째 y 값:\n{y[0]}")

# 시각화: 첫 번째 윈도우의 'feat_1'
plt.figure(figsize=(10, 4))
plt.plot(X[0, 0, :], label='feat_1_window')
# plt.scatter 대신 plt.plot 사용
plt.plot(window_len - 1 + horizon, y[0, 0], 'o', color='red', markersize=8, zorder=5, label='Next feat_1 (y)')
plt.title(f'Sliding Window Example (window_len={window_len}, horizon={horizon})')
plt.xlabel('Time Step in Window')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.show()

# 특정 특성(y_vars=0)을 타겟으로 사용하는 경우
print("\n--- 특정 특성(feat_1)을 타겟으로 사용하는 예시 ---")
X_single_target, y_single_target = apply_sliding_window(df, window_len, horizon=horizon, x_vars=None, y_vars=0)
print(f"X_single_target.shape: {X_single_target.shape}, y_single_target.shape: {y_single_target.shape}")
print(f"첫 번째 y 값 (feat_1만 타겟):\n{y_single_target[0]}")

plt.figure(figsize=(10, 4))
plt.plot(X_single_target[0, 0, :], label='feat_1_window')
# plt.scatter 대신 plt.plot 사용
plt.plot(window_len - 1 + horizon, y_single_target[0], 'o', color='green', markersize=8, zorder=5, label='Next feat_1 (y)')
plt.title(f'Sliding Window Example (Target: feat_1, window_len={window_len}, horizon={horizon})')
plt.xlabel('Time Step in Window')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.show()

 

 

 

3) df2Xy: DataFrame을 3D Numpy 배열로 변환

df2Xy 함수는 Pandas Dataframe을 tsai에서 사용하기 적합한 3D Numpy 배열(X)과 해당 레이블(y)로 변환하는데 사용됩니다. 이는 특히 여러 시계열 샘플이 하나의 DataFrame에 섞여 있을 때 유용한데요

 

print("\n--- df2Xy 예시 ---")
# 예시 DataFrame 생성
df_samples = pd.DataFrame()
df_samples['sample_id'] = np.array([1,1,1,2,2,2,3,3,3])
df_samples['var1'] = df_samples['sample_id'] * 10 + df_samples.index.values
df_samples['var2'] = df_samples['sample_id'] * 100 + df_samples.index.values
print("원본 DataFrame:")
display(df_samples)

# 'sample_id' 기준으로 각 시계열 샘플을 추출
X_df, y_df = df2Xy(df_samples, sample_col='sample_id', steps_in_rows=True, data_cols=['var1', 'var2'], target_col=None)

print(f"\nX_df shape: {X_df.shape}")
print(f"첫 번째 샘플 X:\n{X_df[0]}")
print(f"첫 번째 샘플 y (여기서는 None): {y_df}") # target_col이 None이라 y는 None

# 시각화: 각 sample_id별 시계열
plt.figure(figsize=(12, 5))
for i in range(X_df.shape[0]):
    plt.plot(X_df[i, 0, :], label=f'Sample {i+1} - var1')
    plt.plot(X_df[i, 1, :], linestyle='--', label=f'Sample {i+1} - var2')
plt.title('DataFrame to 3D NumPy Array (df2Xy)')
plt.xlabel('Time Step')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.show()
 

 

 

4)  add_missing_value_cols: 결측치 마스크 특성 추가

 

시계열 데이터에는 종종 결측치가 포함이 됩니다. add_missing_value_cols 함수는 특정 column의 결측치 여부를 나타내는 새로운 바이너리 특성(0 또는 1)을 DataFrame에 추가해주는 함수입니다.

 

print("\n--- add_missing_value_cols 예시 ---")
# 결측치가 있는 데이터프레임 생성
data_missing = np.random.randn(10, 2)
mask = data_missing > 0.8 # 임의로 결측치 생성
data_missing[mask] = np.nan
df_missing = pd.DataFrame(data_missing, columns=['A', 'B'])
print("원본 DataFrame (결측치 포함):")
display(df_missing)

# 결측치 마스크 컬럼 추가
df_with_missing_cols = add_missing_value_cols(df_missing, cols=None, dtype=float)
print("\n결측치 마스크 컬럼이 추가된 DataFrame:")
display(df_with_missing_cols)

# 시각화: 결측치 패턴 (히트맵으로)
plt.figure(figsize=(8, 6))
plt.imshow(df_with_missing_cols[['missing_A', 'missing_B']].T, cmap='Greys', aspect='auto')
plt.yticks([0, 1], ['missing_A', 'missing_B'])
plt.xticks(np.arange(len(df_with_missing_cols)), np.arange(len(df_with_missing_cols)))
plt.title('Missing Value Pattern (1 = Missing)')
plt.xlabel('Row Index')
plt.ylabel('Feature')
plt.colorbar(label='Missing (1) / Not Missing (0)')
plt.show()

 

 

5) add_missing_timestamps: 누락된 타임스탬프 채우기

 

불규칙한 시계열 데이터에서 누락된 타임스탬프를 채우는 것은 일반적인 전처리 과정입니다. add_missing_timestamps 함수는 지정된 빈도에 따라 누락된 날짜/시간을 채우고, 해당 값들을 fill_value로 채워줍니다.

 

print("\n--- add_missing_timestamps 예시 ---")
# 누락된 날짜를 가진 데이터프레임 생성
dates = pd.date_range('2021-05-01', '2021-05-07').values
data_ts = np.zeros((len(dates), 3))
data_ts[:, 0] = dates
data_ts[:, 1] = np.random.rand(len(dates))
data_ts[:, 2] = np.random.rand(len(dates))
cols_ts = ['date', 'feature1', 'feature2']
date_df = pd.DataFrame(data_ts, columns=cols_ts).astype({'date': 'datetime64[ns]', 'feature1': float, 'feature2': float})
date_df_with_missing_dates = date_df.drop([1,3]).reset_index(drop=True) # 2021-05-02, 2021-05-04 누락
print("원본 DataFrame (누락된 날짜 포함):")
display(date_df_with_missing_dates)

# 누락된 타임스탬프 채우기 (그룹 없음)
output_df_filled = add_missing_timestamps(date_df_with_missing_dates.copy(),
                                          'date',
                                          unique_id_cols=None,
                                          fill_value=np.nan,
                                          range_by_group=False, # 전체 범위 기준으로 채움
                                          freq='D') # 일별 빈도
print("\n누락된 타임스탬프가 채워진 DataFrame (그룹 없음):")
display(output_df_filled)

# 시각화: 누락된 타임스탬프 채우기 (feature1)
plt.figure(figsize=(10, 5))
plt.plot(date_df_with_missing_dates['date'], date_df_with_missing_dates['feature1'], 'o-', label='Original Data')
plt.plot(output_df_filled['date'], output_df_filled['feature1'], 'x--', label='Filled Data')
plt.title('Filling Missing Timestamps')
plt.xlabel('Date')
plt.ylabel('Feature1 Value')
plt.legend()
plt.grid(True)
plt.show()

 

 

6) time_encoding: 시간 특성 인코딩(Sin/Cos 변환)

시간 정보는 모델이 패턴 학습을 하는데 중요한 요소입니다. time_encoding 함수는 주기적인 시간 특성(ex. 요일, 월)을 sine/cosine 변환을 사용해 연속적인 값으로 인코딩해주는데요.

이는 모델이 주기성을 더 잘 파악하도록 도와주는 역할을 합니다.

 

 

7) SlidingWindow 및 SlidingWindowPanel

SlidingWindow 클래스는 apply_sliding_window보다 더 많은 유연성을 제공해주는데요

SlidingWindowPanel은 여러 독립적인 시계열(ex. 여러 장치 또는 지역)을 포함하는 DataFrame에 슬라이딩 윈도우를 적용할 때 사용합니다.

 

print("\n--- SlidingWindow 예시 ---")
# 1D 배열에 SlidingWindow 적용
wl_sw = 5
t_sw = np.arange(10)
print(f"입력 배열: {t_sw}")
X_sw, y_sw = SlidingWindow(wl_sw)(t_sw)
print(f"X_sw (윈도우):\n{X_sw}")
print(f"y_sw (타겟):\n{y_sw}")

# 시각화: 생성된 윈도우 (처음 3개)
plt.figure(figsize=(12, 5))
for i in range(min(3, len(X_sw))):
    plt.plot(X_sw[i, 0, :], label=f'Window {i+1} X')
    plt.scatter(wl_sw - 1 + 1, y_sw[i], color='red', s=100, zorder=5, label=f'Window {i+1} Y (Next Value)' if i==0 else "")
plt.title(f'SlidingWindow on 1D Array (Window Length: {wl_sw})')
plt.xlabel('Time Step in Window')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.show()

print("\n--- SlidingWindowPanel 예시 ---")
# 여러 장치/지역 데이터를 포함하는 DataFrame 생성
samples = 100
n_vars = 3
t_panel = (torch.stack(n_vars * [torch.arange(samples)]).T * torch.tensor([1, 10, 100]))
df_panel = pd.DataFrame(t_panel.numpy(), columns=[f'var_{i}' for i in range(n_vars)])
df_panel['time'] = np.arange(len(t_panel))
df_panel['device'] = 0
df_panel['target'] = np.random.randint(0, 2, len(df_panel))

# 두 번째 장치 데이터
df_panel2 = df_panel.copy()
df_panel2['device'] = 1
df_panel2['target'] = np.random.randint(0, 2, len(df_panel2)) + 1 # 다른 타겟

df_combined = pd.concat([df_panel, df_panel2], ignore_index=True)
df_combined = df_combined.sample(frac=1).reset_index(drop=True) # 순서 섞기
print("SlidingWindowPanel 입력 DataFrame (일부):")
display(df_combined.head())

# device를 기준으로 슬라이딩 윈도우 적용
X_panel, y_panel = SlidingWindowPanel(window_len=5, unique_id_cols=['device'], stride=1,
                                     get_x=df_combined.columns[:n_vars], get_y=['target'],
                                     horizon=0, seq_first=True, sort_by=['time'], ascending=True)(df_combined)

print(f"\nX_panel shape: {X_panel.shape}, y_panel shape: {y_panel.shape}")

# 시각화: 각 device별 첫 번째 윈도우 (X)
# 편의상 두 개의 디바이스 중 각 첫 번째 윈도우만 시각화합니다.
device_0_first_X = X_panel[0]
device_1_first_X = X_panel[samples-4] # 대략적인 위치 (window_len-1만큼 덜 생성)

plt.figure(figsize=(12, 5))
plt.plot(device_0_first_X[0, :], label='Device 0 - Var 0 (First Window)')
plt.plot(device_0_first_X[1, :], linestyle='--', label='Device 0 - Var 1 (First Window)')
plt.plot(device_1_first_X[0, :], label='Device 1 - Var 0 (First Window)', color='purple')
plt.plot(device_1_first_X[1, :], linestyle='--', label='Device 1 - Var 1 (First Window)', color='orange')
plt.title('SlidingWindowPanel Output (First Window per Device)')
plt.xlabel('Time Step in Window')
plt.ylabel('Value')
plt.legend()
plt.grid(True)
plt.show()

 

반응형

댓글