저번에 완성한 코드를 통해 시퀀스 데이터 자동으로 수집할 수 있게 되었다. 한 시간 넘게 돌리며 수집한 데이터는 다음과 같다. 오늘은 평소보다 차분한 시장이라 급등, 급락 클래스의 데이터 샘플이 충분히 안 모인것 같다.
시퀀스 데이터의 제일 끝 값은 (0: 매도, 1: 관망, 2: 매수)로 label 되어있다. 수많은 클래스 1(관망)의 데이터 속에서 2(매수)와 0(매도)이 섞인걸 포착했다. 해당 시간대에 급등한 KRW-DOOD와 급락한 KRW-UNI의 데이터임을 확인했다.
이제 모은 데이터를 학습할 기본 모델을 구성하였다. 시계열 데이터 국룰 모델인 lstm과 초단기 패턴이니만큼 국소적인 패턴인식에 강한 CNN을 합친 모델을 기본 골격으로 설정했다. 클래스별로 균형 잡힌 양질의 데이터가 아님을 감안하여 일단은 50 에포크만 돌려봤다.
모은 시퀀스 데이터의 0(매도), 1(관망), 2(매수) 클래스 비율이 약 5 : 90 : 5인 것을 감안했을 때 모델이 1로만 찍어도 90% 정확도를 달성할 수 있으므로 딱히 의미있는 결과는 아니다. 의미있는 baseline 설정을 위해 클래스 불균형을 반영하도록 손실 함수에서 클래스 가중치를 두고 지표도 Macro F1으로 변경했다.
# Filename: train_cnn_lstm_baseline.py
import logging
import os
from matplotlib import pyplot as plt
import pandas as pd
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import f1_score, confusion_matrix
import seaborn as sns
# 하이퍼파라미터
SEQ_LEN = 10 # 시퀀스 길이 (CSV 시퀀스 길이)
PER_STEP_FEATURE = 8 # interval 당 featrue 개수
BATCH_SIZE = 32
EPOCHS = 50
LEARNING_RATE = 1e-3
CSV_PATH = "dataset/sequence_dataset.csv"
# 맥북이라 mts
DEVICE = torch.device("mps") if torch.backends.mps.is_available() else torch.device("cpu")
print("Using device:", DEVICE)
# dataset 정의
class SequenceDataset(Dataset):
def __init__(self, X, y):
self.X = torch.tensor(X, dtype=torch.float32)
self.y = torch.tensor(y, dtype=torch.long) # 분류라 long
def __len__(self):
return len(self.X)
def __getitem__(self, idx):
return self.X[idx], self.y[idx]
# CNN-LSTM 모델 정의
class CNNLSTM(nn.Module):
def __init__(self, per_step_feature, num_classes=3):
super().__init__()
# 1D CNN
self.conv1 = nn.Conv1d(in_channels=per_step_feature, out_channels=32, kernel_size=3, padding=1)
self.relu = nn.ReLU()
self.pool = nn.MaxPool1d(kernel_size=2)
# LSTM
self.lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# FC
self.fc = nn.Linear(64, num_classes)
def forward(self, x):
# x: (batch, seq_len=10, feature=8)
x = x.permute(0, 2, 1) # (batch, 8, 10)
x = self.pool(self.relu(self.conv1(x))) # (batch, 32, 5)
x = x.permute(0, 2, 1) # (batch, 5, 32)
x, _ = self.lstm(x)
x = x[:, -1, :] # last timestep
return self.fc(x)
# CSV 데이터 로드
df = pd.read_csv(CSV_PATH)
print("CSV shape:", df.shape)
# 라벨 분리
X = df.drop(columns=['label']).values # (N, 80)
y = df['label'].values # (N,)
# CSV는 이미 시퀀스 형태로 저장되어 있으므로 reshape만 수행
num_samples = len(X) # N = 1039
X = X.reshape(num_samples, SEQ_LEN, PER_STEP_FEATURE) # (N, 10, 8)
print("X shape:", X.shape)
print("y shape:", y.shape)
print("Label distribution:", np.bincount(y)) # 클래스 분포 확인
# 표준화 생략
# train/test split
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
train_dataset = SequenceDataset(X_train, y_train)
val_dataset = SequenceDataset(X_val, y_val)
train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False)
# 모델 학습
# 손실함수에 class weights 도입
class_counts = np.bincount(y_train)
class_weights = 1.0 / class_counts
class_weights = torch.tensor(class_weights, dtype=torch.float32).to(DEVICE)
criterion = nn.CrossEntropyLoss(weight=class_weights)
model = CNNLSTM(PER_STEP_FEATURE).to(DEVICE)
optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE)
# 시각화를 위해 로그 저장
train_losses = []
val_losses = []
# Macro F1을 성능 지표로 설정
train_macro_f1s = []
val_macro_f1s = []
for epoch in range(EPOCHS):
model.train()
total_train_loss = 0.0
all_train_preds = []
all_train_labels = []
for X_batch, y_batch in train_loader:
X_batch = X_batch.to(DEVICE)
y_batch = y_batch.to(DEVICE)
optimizer.zero_grad()
out = model(X_batch)
loss = criterion(out, y_batch)
loss.backward()
optimizer.step()
total_train_loss += loss.item() * X_batch.size(0)
preds = torch.argmax(out, dim=1)
all_train_preds.extend(preds.cpu().numpy())
all_train_labels.extend(y_batch.cpu().numpy())
avg_train_loss = total_train_loss / len(train_loader.dataset)
train_losses.append(avg_train_loss)
train_macro_f1 = f1_score(all_train_labels, all_train_preds, average='macro')
train_macro_f1s.append(train_macro_f1)
# 평가
model.eval()
all_val_preds = []
all_val_labels = []
total_val_loss = 0.0
with torch.no_grad():
for X_batch, y_batch in val_loader:
X_batch = X_batch.to(DEVICE)
y_batch = y_batch.to(DEVICE)
out = model(X_batch)
loss = criterion(out, y_batch)
total_val_loss += loss.item() * X_batch.size(0)
preds = torch.argmax(out, dim=1)
all_val_preds.extend(preds.cpu().numpy())
all_val_labels.extend(y_batch.cpu().numpy())
avg_val_loss = total_val_loss / len(val_loader.dataset)
val_losses.append(avg_val_loss)
val_macro_f1 = f1_score(all_val_labels, all_val_preds, average='macro')
val_macro_f1s.append(val_macro_f1)
print(f"Epoch {epoch+1}/{EPOCHS} | Train Loss: {avg_train_loss:.4f} | Val Loss: {avg_val_loss:.4f} | Train Macro F1: {train_macro_f1:.4f} | Val Macro F1: {val_macro_f1:.4f}")
# 시각화
# loss 비교 그래프
plt.figure(figsize=(8, 5))
plt.plot(train_losses, label="Train Loss")
plt.plot(val_losses, label="Val Loss")
plt.xlabel("Epoch")
plt.ylabel("Loss")
plt.title("Train vs Validation Loss")
plt.legend()
plt.grid()
plt.savefig("fig/train_val_loss.png")
plt.show()
# Macro F1 비교 그래프
plt.figure(figsize=(8,5))
plt.plot(train_macro_f1s, label="Train Macro F1")
plt.plot(val_macro_f1s, label="Val Macro F1")
plt.xlabel("Epoch")
plt.ylabel("Macro F1")
plt.title("Macro F1: Train vs Val")
plt.legend()
plt.grid(True)
plt.savefig("fig/train_val_macro_f1.png")
plt.show()
# Confusion Matrix Heatmap (last epoch)
cm = confusion_matrix(all_val_labels, all_val_preds)
plt.figure(figsize=(5, 4))
sns.heatmap(cm, annot=True, fmt="d", cmap="Blues")
plt.xlabel("Predicted")
plt.ylabel("True")
plt.title("Confusion Matrix")
plt.savefig("fig/confusion_matrix.png")
plt.show()
# 모델 저장
os.makedirs("models", exist_ok=True)
torch.save(model.state_dict(), "models/cnn_lstm_model.pth")
print("Model saved to models/cnn_lstm_model.pth")
bincount로 클래스 분포를 확인하고 data split시에 startify로 얼마 없는 소수 클래스가 훈련, 검증 데이터셋에 골고루 포진되도록 구성했다. 이런 불균형한 데이터셋에서 중요한 클래스별 recall과 precision을 직관적으로 확인할 수 있는 confusion matrix도 출력해봤다.
confusion matrix를 보면 확실히 클래스 1로 예측하는 경향이 압도적이다. 그러나 이런 불균형 속에서도 0이나 2의 클래스를 조금이나마 예측을 하는 것으로 보아 feature의 특정 패턴이 클래스 분류에 있어서 나름 의미있게 작용한다는 점은 희망적이다. 굳이 분석하자면 데이터 수가 비슷했던 클래스 0의 recall(6/11)이 클래스 2의 recall(4/12)보다 확연히 크게 나타나는 것을 보면 급락이 급등보다 더 특징있게 드러난다고도 볼 수 있다. (대부분 찬찬히 오르고 급격하게 떨어져서 그런가)
사실상 모델의 목적인 클래스 2의 감지가 제일 암울한것으로 보인다. 2보다도 1로 더 많이 예측하는 것을 보면 threshold에 따라 1과 2의 분류가 크게 영향을 받을 것으로도 볼 수 있겠다.
이렇게 부실하고 적은 데이터로 학습한 모델을 튜닝하는 게 의미가 없을 것 같긴하지만 일단 튜닝을 진행했다. 여느 때와 같이 일반화 능력을 올리는 튜닝부터 진행하려했다. 그러나 모델 목적상 희귀하지만 분명하게 나타나는 몇 개의 단기 패턴에 민감하게 반응해야하는데 과도하게 일반화 능력을 추구할 경우 여러 급등 패턴들이 애매하게 섞여 어떤 패턴도 감지하지 못할 가능성도 있었다. 물론 8개의 상대값들 각각의 특성과 패턴을 제대로 연구해서(특성 공학) 일반화와 민감도의 적절한 선을 찾아 튜닝할 수도 있겠지만 그건 너무 품이 많이 드므로 일단은 일반화능력보단 Macro F1을 늘리는 방향을 추구하기로 했다.
따라서 weight decay를 나름 작게 적용하고 스케쥴러를 넣어보려 했다. 그 전에 feature가 모두 상대값이라 별 의미없을 것 같아서 건너 뛴 scaler 표준화를 적용해보았더니 바로 Macro F1이 0.55까지 올랐다. 실시간 모델 입력 시에도 학습때와 유사한 스케일로 넣어줘야하므로 scaler를 따로 저장해두는 코드도 추가했다.
# 표준화 추가
scaler = StandardScaler()
X_2d = X.reshape(-1, PER_STEP_FEATURE) # (num_samples * SEQ_LEN, PER_STEP_FEATURE)
X_2d = scaler.fit_transform(X_2d)
# scaler 저장
os.makedirs("models", exist_ok=True)
joblib.dump(scaler, "models/feature_scaler.pkl")
print("Scaler saved to models/feature_scaler.pkl")
feature 중에 거래량 변화율인 volume_ratio라는 컬럼이 있는데 종목별로 거래량이 천차만별이다 보니 이 때문인가 싶어서 확인해봤다. 실제로 종목에 따라 100배까지 차이나는 경향을 확인할 수 있었고 다른 컬럼에서도 상대값인데 반해 스케일 차이가 큰 경우를 꽤 확인할 수 있었다.
scaler를 통해 Macro F1을 55 수준까지 끌어올린 후 lr=1e-5로 나름 작게 weight decay를 넣고 step LR 스케쥴러를 적용해보았다. Macro F1은 62를 넘겼고 확실히 이전에 비해 안정적으로 수렴하는 모양새를 보인다. step LR의 gamma 값은 0.5일때 가장 높은 성능을 보였고 cosine LR 적용 시에 오히려 적용 전보다 줄어드는 것도 확인했다.
물론 적용 가능한 더 나은 기법이 여럿 존재하겠지만 지금 이 저질 데이터로는 의미가 없을 것 같다. 몇 개월 전부터 시작된 코인 시장의 침체가 끝나고 활력이 돌 때.. 다시 시장이 요동치고 거래량이 터지면 그때 양질의 데이터를 수집해서 제대로 돌려봐야지..
'dev > ai' 카테고리의 다른 글
| Codex Skill 등록하기 - bcpprm 사례로 익히는 워크플로 자동화 가이드 (1) | 2026.05.23 |
|---|---|
| Upbit 웹소켓을 이용한 급등 코인의 움직임 예측 모델 개발 (5) (1) | 2025.12.22 |
| Upbit 웹소켓을 이용한 급등 코인의 움직임 예측 모델 개발 (3) (0) | 2025.12.22 |
| Upbit 웹소켓을 이용한 급등 코인의 움직임 예측 모델 개발 (2) (0) | 2025.12.21 |
| Upbit 웹소켓을 이용한 급등 코인의 움직임 예측 모델 개발 (1) (0) | 2025.12.20 |