이번에는 지난 포스팅에서 알아본 개념을 바탕으로 파이썬 코드 구현을 해보도록 하겠습니다.
[개념정리] ROCKET(RandOm Convolutional KErnel Transform) (시계열 특징 추출 / 시계열 분류) (1/2)
ROCKET은 sklearn 과 tsai 라는 두 라이브러리에서 쉽게 구현이 가능하며,
이 중 tsai의 경우 gpu 연산을 지원하기에 딥러닝 모델에도 쉽게 적용할 수가 있어서 이번 코드 구현은 tsai를 통해서 해보겠습니다.
또한, 범용적인 구현을 위해 Google의 Colab 환경에서 무료로 제공하는 GPU인 T4를 가지고 구현했습니다.
1. ROCKET을 통한 시계열 분류
1) 이번 구현에 활용할 데이터는 UCR datasetd에 있는 "FigerMovements" 입니다.
해당 데이터에 대한 전반적인 설명은 아래와 같습니다
이는 BCI 2 라는 대회를 위해 제공된 데이터셋으로, 좌우 손가락 움직임을 예측하기 위한 EEG 데이터입니다.
(EEG: Electroencephalography / 뇌파측정)
실험 간, 피실험자가 자율적으로 여러가지 키를 입력하는 동안 28개의 EEG 채널에서 나오는 정보를 기록한 것으로
이때, 좌 / 우 손 움직임을 분류(0: 왼손 , 1: 오른속)하는 Binary Classification이 되겠습니다.
2-1. ROCKET + Ridge Classifer
이제 코랩 환경에서 T4 GPU를 활용해 ROCKET을 통한 분류를 구현해보겠습니다. 먼저, Ridge Classifier를 활용한 분류를 해보겠습니다.
1) 먼저, 실험에 필요한 library인 tsai 라이브러리를 설치해줍니다.
2) 이제 데이터셋을 불러와줍니다. tsai에 내장되어있는 get_UCR_data 함수로 쉽게 불러올 수 있습니다.
# tsai와 관련 라이브러리를 불러옵니다
from tsai.all import *
from sklearn.linear_model import RidgeClassifierCV
# UCR 데이터 중 'FordA' 데이터셋을 가져옵니다 (학습/검증 데이터로 나누기)
X_train , y_train , X_valid , y_valid = get_UCR_data ( 'FingerMovements' , split_data= True )
print ( "X_trian shape : " , format ( X_train.shape ))
print ( "y_train shape : " , format ( y_train.shape ))
print ( "X_valid shape : " , format ( X_valid.shape ))
print ( "y_valid shape : " , format ( y_valid.shape ))
위와 같이 데이터의 shape도 확인할 수 있으며,
X_train에는 316개의sample와 28개의 채널, 그리고 50개의 sequence length를 확인할 수 있습니다.
X_valid에는 100개의 sample이 있고 나머지는 동일한 것을 확인할 수 있습니다.
각 y는 대응하는 X와 동일한 샘플을 가지고 있음을 알 수 있습니다.
3) 이제 해당 데이터들을 tensor의 형태로 바꾼 뒤 gpu에서 처리하도록 설정해주고, ROCKET을 통해 특성을 추출한 뒤 Ridge Classifier로 결과를 도출해줍니다.
# numpy 배열을 torch.Tensor로 변환하고 GPU로 이동
device = torch.device ( 'cuda' if torch.cuda.is_available () else 'cpu' )
X_train_tensor = torch.tensor ( X_train , dtype=torch.float32 ) .to ( device )
X_valid_tensor = torch.tensor ( X_valid , dtype=torch.float32 ) .to ( device )
c_in = X_train.shape [ 1 ] # 특성의 수 (특징 수)
seq_len = X_train.shape [ 2 ] # 시계열 길이 (시간 스텝)
# ROCKET 모델을 설정하고 GPU로 이동
rocket = ROCKET ( c_in=c_in , seq_len=seq_len ) .to ( device )
# 학습 데이터를 사용하여 ROCKET 특징을 추출합니다
X_train_features = rocket ( X_train_tensor ) # forward 메서드를 사용
X_valid_features = rocket ( X_valid_tensor ) # forward 메서드를 사용
# 특징 벡터를 다시 CPU로 이동하고 numpy로 변환
X_train_features = X_train_features.detach () .cpu () .numpy ()
X_valid_features = X_valid_features.detach () .cpu () .numpy ()
# RidgeClassifier를 사용해 분류 모델을 학습합니다
model = RidgeClassifierCV ( alphas=np.logspace ( -3 , 3 , 10 ))
model.fit ( X_train_features , y_train )
# 검증 데이터에서 성능을 평가합니다
accuracy = model.score ( X_valid_features , y_valid )
print ( f "Validation Accuracy: { accuracy } " )
ROCKET 패키지에서는 데이터의 특성의 수 (채널 수 = 28), 시계열의 길이(= 50)을 입력해주어야합니다.
분류에 활용한 ridge classifier는 sklearn에서 가져왔으며, 이는 cpu에서 작동을 하기에 다음과 같이 처리된 특징을 다시 cpu로 이동시키고 numpy로 변환한 뒤 분류를 해주었습니다.
결과는 아래와 같이 59%의 정확도가 도출되었습니다.
2-2. ROCKET + MLP
이번에는 동일한 ROCKET을 활용하되 Classifier 부분만 간단한 MLP로 분류하는 것으로 바꾸어보았습니다.
# 필요한 라이브러리 불러오기
from tsai.all import *
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.preprocessing import LabelEncoder
from sklearn.model_selection import train_test_split
# UCR 데이터 중 'FingerMovements' 데이터셋을 가져옵니다 (학습/검증 데이터로 나누기)
X_train , y_train , X_test , y_test = get_UCR_data ( 'FingerMovements' , split_data= True )
# 라벨을 정수형으로 인코딩 (문자열 -> 정수)
le = LabelEncoder ()
y_train = le.fit_transform ( y_train )
y_test = le.transform ( y_test )
# Train/Val/Test로 데이터셋을 나눔 (stratify를 사용하여 클래스 비율 유지)
X_train , X_valid , y_train , y_valid = train_test_split ( X_train , y_train , test_size= 0.25 , stratify=y_train , random_state= 42 )
# c_in (입력 채널 수)와 seq_len (시계열 길이)를 정의합니다
c_in = X_train.shape [ 1 ] # 특성의 수 (특징 수)
seq_len = X_train.shape [ 2 ] # 시계열 길이 (시간 스텝)
# numpy 배열을 torch.Tensor로 변환하고 GPU로 이동
device = torch.device ( 'cuda' if torch.cuda.is_available () else 'cpu' )
X_train_tensor = torch.tensor ( X_train , dtype=torch.float32 ) .to ( device )
X_valid_tensor = torch.tensor ( X_valid , dtype=torch.float32 ) .to ( device )
X_test_tensor = torch.tensor ( X_test , dtype=torch.float32 ) .to ( device )
y_train_tensor = torch.tensor ( y_train , dtype=torch. long ) .to ( device )
y_valid_tensor = torch.tensor ( y_valid , dtype=torch. long ) .to ( device )
y_test_tensor = torch.tensor ( y_test , dtype=torch. long ) .to ( device )
# ROCKET 모델을 설정하고 GPU로 이동
rocket = ROCKET ( c_in=c_in , seq_len=seq_len ) .to ( device )
# 학습 데이터를 사용하여 ROCKET 특징을 추출합니다
X_train_features = rocket ( X_train_tensor ) .to ( device )
X_valid_features = rocket ( X_valid_tensor ) .to ( device )
X_test_features = rocket ( X_test_tensor ) .to ( device )
# MLP Classifier 정의 (다층 퍼셉트론)
class MLPClassifier ( nn . Module ) :
def __init__ ( self , input_size , num_classes ) :
super ( MLPClassifier , self ) . __init__ ()
self .fc1 = nn.Linear ( input_size , 128 ) # 첫 번째 레이어 (128개의 뉴런)
self .fc2 = nn.Linear ( 128 , 64 ) # 두 번째 레이어 (64개의 뉴런)
self .fc3 = nn.Linear ( 64 , num_classes ) # 출력 레이어 (클래스 수에 맞게 설정)
self .relu = nn.ReLU () # 활성화 함수로 ReLU 사용
self .dropout = nn.Dropout ( 0.5 ) # 드롭아웃으로 과적합 방지
def forward ( self , x ) :
x = self .relu ( self .fc1 ( x ))
x = self .dropout ( x )
x = self .relu ( self .fc2 ( x ))
x = self .dropout ( x )
x = self .fc3 ( x )
return x
# MLP Classifier 초기화
input_size = X_train_features.shape [ 1 ] # ROCKET 출력의 차원 수 (입력 크기)
num_classes = len ( torch.unique ( y_train_tensor )) # 출력 클래스의 수
mlp_model = MLPClassifier ( input_size , num_classes ) .to ( device )
# 손실 함수 및 최적화 함수 설정
criterion = nn.CrossEntropyLoss ()
optimizer = optim.Adam ( mlp_model.parameters (), lr= 0.001 )
# Early Stopping 관련 설정
early_stopping_patience = 20 # Early stopping을 적용할 patience
best_val_loss = float ( 'inf' )
patience_counter = 0
# 학습 파이프라인 정의
num_epochs = 200
for epoch in range ( num_epochs ):
# 학습 모드
mlp_model.train ()
optimizer.zero_grad ()
# MLP 모델에 학습 데이터 전달
outputs = mlp_model ( X_train_features )
loss = criterion ( outputs , y_train_tensor )
# 역전파 및 최적화
loss.backward ()
optimizer.step ()
# 검증 데이터 평가
mlp_model. eval ()
with torch.no_grad ():
valid_outputs = mlp_model ( X_valid_features )
valid_loss = criterion ( valid_outputs , y_valid_tensor )
valid_preds = torch.argmax ( valid_outputs , dim= 1 )
val_accuracy = ( valid_preds == y_valid_tensor ) . float () .mean () .item ()
print ( f "Epoch [ { epoch+ 1 } / { num_epochs } ], Loss: { loss.item () :.4f } , Validation Accuracy: { val_accuracy :.4f } , Validation Loss: { valid_loss.item () :.4f } " )
# Early Stopping 체크
if valid_loss.item () < best_val_loss :
best_val_loss = valid_loss.item ()
patience_counter = 0 # 검증 손실이 개선되면 patience를 초기화
best_model_state = mlp_model.state_dict () # 모델 상태 저장
else :
patience_counter += 1
if patience_counter >= early_stopping_patience :
print ( "Early stopping 적용" )
break
# 최종 검증 정확도 출력
print ( f "Final Validation Accuracy: { val_accuracy :.4f } " )
# 최적의 모델 상태로 복원
mlp_model.load_state_dict ( best_model_state )
# 테스트 데이터에서 성능 평가
mlp_model. eval ()
with torch.no_grad ():
test_outputs = mlp_model ( X_test_features )
test_preds = torch.argmax ( test_outputs , dim= 1 )
test_accuracy = ( test_preds == y_test_tensor ) . float () .mean () .item ()
print ( f "Final Test Accuracy: { test_accuracy :.4f } " )
MLP와 같은 DL 방식에서는 학습을 중단할 시점을 정해야 하며, 이를 위해 train 데이터를 train과 validation으로 stratify를 적용한 split을 하였습니다. test 데이터는 그대로 활용했습니다.
학습을 위해 Early stopping (patience=20) , 학습 epochs 는 200, 그리고 ReLU활성화 함수와 Dropout을 적용한 간단한 3층의 MLP Classifier를 만들었고, 옵티마이저는 Adam을 활용하며 학습률은 0.001로 정했고, Loss는 CrossEntropy를 활용했습니다.
결과는 아래와 같이 50%의 분류 결과를 도출했습니다.
데이터에 더 적합한 MLP Classifier 만들어서 Ridge보다 좋은 성능을 낼 수도 있을것 같았지만, 복잡하게 하더라도 Ridge Classifier의 성능보다 좋은 결과를 내기는 어려웠습니다.
최초 논문에서도 Ridge Classifier를 제안했었는데, Ridge의 L2 Regularization term을 활용이 ROCKET을 활용한 Prediction에서는 꽤나 좋은 성과를 내는것으로 보입니다.
댓글