PyTorch & Deep Learning
LSTM + Transformer로 주가 시계열을 학습하는 법
이 글은 AI_Finance(ALCHEMETRIC) 프로젝트에서 PyTorch를 사용해 LSTM과 Transformer 모델로 주가 등락을 예측하는 딥러닝 파이프라인을 다룬다. ML/DL을 처음 접하는 독자도 이해할 수 있도록, 핵심 개념을 비유와 함께 설명하고 실제 프로젝트 코드가 어떻게 동작하는지 보여준다.
목차
1. 왜 딥러닝? 전통 ML과 뭐가 다른가
ALCHEMETRIC 프로젝트에는 이미 Random Forest, XGBoost, LightGBM 같은 전통 머신러닝 모델이 돌아가고 있다. 그런데 왜 굳이 딥러닝을 추가했을까? 핵심은 "시간의 흐름"에 있다.
전통 ML: "오늘만 보는 투자자"
Random Forest나 XGBoost는 각 행(row)을 독립적으로 판단한다. 오늘 삼성전자의 RSI가 70이고 PER이 12이면 → 상승/하락을 예측한다. 어제, 그저께 데이터는 보지 않는다.
비유하면, 오늘 주가 숫자만 보고 "비싸다/싸다"를 판단하는 투자자와 같다.
딥러닝: "차트를 읽는 투자자"
LSTM이나 Transformer 같은 딥러닝 모델은 최근 20일치 데이터를 한꺼번에 입력받는다. "3일 전부터 거래량이 급증했고, RSI가 어제 70을 돌파했고, 오늘 골든크로스가 발생했다" — 이런 시간 흐름 속의 패턴을 학습할 수 있다.
비유하면, 차트의 모양(패턴)을 보고 판단하는 기술적 분석가와 같다.
Sequence — "순서가 있는 데이터"
주가는 대표적인 시계열(time series) 데이터다. "1일차 → 2일차 → 3일차..." 처럼 순서가 중요하다. 이렇게 순서가 있는 데이터를 Sequence(시퀀스)라고 부른다.
우리 프로젝트에서 딥러닝 모델에 입력되는 데이터의 형태는 다음과 같다.
# 1개 종목의 입력 텐서 shape
(1, 20, 75)
# 1 = 배치(종목 1개)
# 20 = 시퀀스 길이 (최근 20 거래일)
# 75 = 피처 수 (가격, 기술지표, 재무, 거시, 뉴스, 섹터, 대안 데이터)
2. PyTorch란?
PyTorch는 Meta(Facebook)가 만든 딥러닝 프레임워크다. 쉽게 말하면, 딥러닝 모델을 만들고 학습시키기 위한 도구 상자라고 생각하면 된다.
TensorFlow vs PyTorch — 왜 PyTorch?
딥러닝 프레임워크의 양대 산맥은 Google의 TensorFlow와 Meta의 PyTorch다. 우리 프로젝트에서 PyTorch를 선택한 이유는 다음과 같다.
| 항목 | TensorFlow | PyTorch |
|---|---|---|
| 코딩 스타일 | 선언적 (그래프 먼저 정의) | Pythonic (일반 Python처럼 코딩) |
| 디버깅 | 어려움 (그래프 모드) | 쉬움 (print로 바로 확인) |
| 연구/학계 | 감소 추세 | 압도적 1위 (논문의 80%+) |
| 프로덕션 | 강점 (TF Serving 등) | TorchServe로 충분 |
핵심 개념 3가지
PyTorch를 이해하려면 딱 3가지만 알면 된다.
1) Tensor — 데이터의 그릇
numpy 배열과 비슷하지만, GPU에서 고속 연산이 가능하고 자동 미분(autograd)을 지원한다. 주가 데이터를 담는 그릇이라고 생각하면 된다.
import torch
# numpy 배열을 텐서로 변환
X = torch.tensor(feature_array, dtype=torch.float32)
# shape: (1, 20, 75) = 1종목 x 20일 x 75피처
# GPU로 보내기
X = X.to("cuda") # 또는 "mps" (Mac), "cpu"
2) nn.Module — 신경망 레고 블록
모든 PyTorch 신경망은 nn.Module을 상속한다. 레고 블록처럼 작은 모듈을 조합해서 큰 모델을 만든다. 핵심은 forward() 메서드 — "데이터가 들어오면 어떤 연산을 거쳐 나갈지"를 정의한다.
class MyModel(nn.Module):
def __init__(self):
super().__init__()
self.layer1 = nn.Linear(75, 128) # 75개 입력 -> 128개 출력
self.layer2 = nn.Linear(128, 2) # 128개 -> 2개 (상승/하락)
def forward(self, x):
x = self.layer1(x)
x = torch.relu(x) # 활성화 함수
return self.layer2(x)
3) Autograd — 역전파 자동 계산
딥러닝 학습의 핵심은 "손실(loss)을 줄이기 위해 가중치(weight)를 얼마나 조정할지" 계산하는 것이다. 이 과정에서 미분이 필요한데, PyTorch의 Autograd가 자동으로 미분을 계산해준다. 개발자는 loss.backward() 한 줄만 호출하면 된다.
3. LSTM — 기억하는 신경망
RNN의 한계: 과거를 잊는다
시퀀스를 처리하는 가장 기본적인 신경망은 RNN(Recurrent Neural Network)이다. 데이터를 한 타임스텝씩 순서대로 읽으면서 이전 정보를 다음으로 전달한다. 하지만 시퀀스가 길어지면 기울기 소실(Vanishing Gradient) 문제가 발생한다 — 20일 전의 중요한 패턴이 학습 과정에서 사라져 버린다.
비유하면, 전화 통화로 이야기를 전달하는 "전화 게임"과 같다. 사람이 10명, 20명 늘어나면 원래 메시지가 왜곡되거나 사라지는 것처럼, RNN도 긴 시퀀스에서는 초기 정보를 잃어버린다.
LSTM: 3개의 게이트로 기억을 관리
LSTM(Long Short-Term Memory)은 이 문제를 해결하기 위해 3개의 게이트(gate)를 도입했다. 각 게이트는 "어떤 정보를 기억하고, 어떤 정보를 잊고, 어떤 정보를 출력할지"를 학습한다.
| 게이트 | 역할 | 비유 |
|---|---|---|
| Forget Gate | "이 정보 버릴까?" — 이전 기억 중 불필요한 것을 삭제 | 메모장에서 지우개로 필요 없는 내용 지우기 |
| Input Gate | "이 새 정보 저장할까?" — 새로 들어온 정보 중 중요한 것만 저장 | 새로운 내용 중 중요한 것에 밑줄 긋기 |
| Output Gate | "이 정보 출력할까?" — 현재 시점에 필요한 정보만 내보냄 | 발표할 때 필요한 내용만 골라서 말하기 |
Bidirectional LSTM: 양방향으로 읽기
일반 LSTM은 과거 → 미래 방향으로만 시퀀스를 읽는다. Bidirectional LSTM(양방향 LSTM)은 미래 → 과거 방향도 함께 읽어서, 양쪽 맥락을 모두 활용한다. 우리 프로젝트에서는 bidirectional=True를 기본 설정으로 사용한다.
비유하면, 소설을 처음부터 읽는 것과 끝에서부터 역순으로 읽는 것을 동시에 해서, 스토리를 더 깊이 이해하는 것과 같다.
프로젝트의 LSTM 아키텍처
입력 (batch, 20, 75)
|
BiLSTM (2 layer, hidden=128, bidirectional)
|
마지막 타임스텝 hidden 추출 -> (batch, 256)
|
Dropout(0.3)
|
Linear(256 -> 128) + ReLU
|
Dropout(0.3)
|
Linear(128 -> 2)
|
출력: 2 classes (하락=0, 상승=1)
실제 코드에서 이 구조가 어떻게 구현되는지 보자.
class LSTMClassifier(nn.Module):
def __init__(self, n_features, hidden_size=128, num_layers=2,
dropout=0.3, bidirectional=True):
super().__init__()
self.lstm = nn.LSTM(
input_size=n_features, # 75 피처
hidden_size=hidden_size, # 128
num_layers=num_layers, # 2층 쌓기
batch_first=True,
dropout=dropout,
bidirectional=bidirectional, # 양방향
)
fc_in = hidden_size * 2 # 양방향이면 128*2 = 256
self.classifier = nn.Sequential(
nn.Dropout(dropout),
nn.Linear(fc_in, hidden_size),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_size, 2), # 상승/하락 2 classes
)
def forward(self, x):
# x: (batch, 20, 75)
lstm_out, _ = self.lstm(x) # (batch, 20, 256)
last_hidden = lstm_out[:, -1, :] # (batch, 256) 마지막 날
return self.classifier(last_hidden) # (batch, 2)
4. Transformer — Attention이 전부다
2017년 Google의 논문 "Attention Is All You Need"에서 등장한 Transformer는, 원래 번역(NLP)을 위해 만들어졌지만, 지금은 이미지, 음성, 시계열 등 거의 모든 분야에서 사용된다. ChatGPT도 Transformer 기반이다.
Self-Attention: "어떤 날이 중요한가?"
LSTM은 데이터를 순서대로 하나씩 읽는다. Transformer의 Self-Attention은 전혀 다른 접근법을 취한다 — 모든 날이 동시에 다른 모든 날을 바라보며 "중요도 점수"를 매긴다.
예를 들어 20일치 데이터가 있을 때:
- 15일차(급등일)가 20일차(오늘)에 높은 점수 → "이 급등이 오늘에 영향을 줬다"
- 3일차(평범한 날)가 20일차에 낮은 점수 → "이 날은 크게 관계없다"
Query, Key, Value — Attention의 메커니즘
Self-Attention은 3가지 요소로 동작한다.
| 요소 | 역할 | 비유 |
|---|---|---|
| Query (Q) | "나는 어떤 정보가 필요한가?" | 도서관에서 검색어 입력 |
| Key (K) | "나는 이런 정보를 갖고 있다" | 책 표지의 키워드 |
| Value (V) | "내 실제 내용은 이거다" | 책의 본문 내용 |
Query와 Key를 비교해서 "관련도 점수"를 구하고, 그 점수로 Value를 가중합(weighted sum)한다. 관련도가 높은 날의 정보가 더 많이 반영된다.
Multi-Head Attention: 여러 관점에서 동시에
우리 프로젝트는 8개의 Head를 사용한다. 8명의 애널리스트가 각자 다른 관점으로 같은 차트를 분석하는 것과 같다.
- Head 1: "거래량 급증 패턴에 주목"
- Head 2: "RSI 과매수/과매도 반전 패턴에 주목"
- Head 3: "골든크로스/데드크로스 타이밍에 주목"
- ... 각 Head가 독립적으로 attention 계산
최종적으로 8명의 분석을 합쳐서 종합 판단을 내린다.
Positional Encoding: 순서 정보 주입
LSTM은 데이터를 순서대로 읽기 때문에 자연스럽게 "몇 번째 날인지" 안다. 하지만 Transformer는 모든 날을 동시에 보기 때문에, 순서 정보가 없다. "1일차인지 20일차인지" 구분을 못 한다.
그래서 Positional Encoding을 통해 각 타임스텝에 "위치 정보"를 더해준다. sin/cos 함수를 사용해서 각 위치마다 고유한 패턴을 만든다.
class PositionalEncoding(nn.Module):
def __init__(self, d_model, max_len=500, dropout=0.1):
super().__init__()
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len).unsqueeze(1)
div_term = torch.exp(
torch.arange(0, d_model, 2) * (-math.log(10000.0) / d_model)
)
pe[:, 0::2] = torch.sin(position * div_term) # 짝수 차원
pe[:, 1::2] = torch.cos(position * div_term) # 홀수 차원
self.register_buffer("pe", pe.unsqueeze(0))
def forward(self, x):
# x에 위치 정보를 더해줌
return self.dropout(x + self.pe[:, :x.size(1), :])
프로젝트의 Transformer 아키텍처
입력 (batch, 20, 75)
|
Linear(75 -> 128) # 피처 차원을 d_model로 투영
|
PositionalEncoding # sin/cos 위치 정보 추가
|
TransformerEncoder # 3 layer, 8 head, FFN=256
|
Mean Pooling # 20일의 평균 -> (batch, 128)
|
LayerNorm -> Linear(128->64) -> GELU -> Dropout
|
Linear(64 -> 2)
|
출력: 2 classes (하락=0, 상승=1)
실제 코드를 보자.
class TransformerClassifier(nn.Module):
def __init__(self, n_features, d_model=128, nhead=8,
num_layers=3, dim_feedforward=256, dropout=0.2):
super().__init__()
self.input_proj = nn.Linear(n_features, d_model) # 75->128
self.pos_encoder = PositionalEncoding(d_model, dropout=dropout)
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model, nhead=nhead, # 8개 head
dim_feedforward=dim_feedforward, # FFN 256
dropout=dropout, batch_first=True,
activation="gelu",
)
self.transformer_encoder = nn.TransformerEncoder(
encoder_layer, num_layers=num_layers # 3층 쌓기
)
self.classifier = nn.Sequential(
nn.LayerNorm(d_model),
nn.Linear(d_model, d_model // 2), # 128->64
nn.GELU(),
nn.Dropout(dropout),
nn.Linear(d_model // 2, 2), # 64->2
)
def forward(self, x):
x = self.input_proj(x) # (batch, 20, 128)
x = self.pos_encoder(x) # + 위치 정보
x = self.transformer_encoder(x) # Self-Attention 3번
x = x.mean(dim=1) # Mean Pooling
return self.classifier(x) # (batch, 2)
5. 학습 루프 — DL은 이렇게 배운다
ML vs DL 학습의 차이
전통 ML은 학습이 정말 간단하다.
# scikit-learn ML: 한 줄이면 끝
model = RandomForestClassifier()
model.fit(X_train, y_train) # 끝! 내부에서 알아서 다 해줌
딥러닝은 다르다. 개발자가 학습 과정을 직접 제어해야 한다.
# PyTorch DL: 학습 루프를 직접 작성
for epoch in range(100):
model.train()
for X_batch, y_batch in train_loader:
optimizer.zero_grad() # 1. 기울기 초기화
logits = model(X_batch) # 2. 예측 (forward)
loss = criterion(logits, y_batch) # 3. 손실 계산
loss.backward() # 4. 역전파 (backward)
clip_grad_norm_(model.parameters(), 1.0) # 5. 기울기 클리핑
optimizer.step() # 6. 가중치 업데이트
# 검증
model.eval()
val_f1 = evaluate(model, val_loader)
scheduler.step(val_f1)
if val_f1 > best_f1:
best_f1 = val_f1
torch.save(model.state_dict(), "best_model.pt")
왜 이렇게 복잡할까? 유연성 때문이다. 학습 과정의 모든 단계를 제어할 수 있기 때문에, mixed precision, gradient accumulation 등 다양한 최적화 기법을 적용할 수 있다.
핵심 구성요소 5가지
| 구성요소 | 프로젝트 설정 | 역할 |
|---|---|---|
| Optimizer | AdamW (weight_decay=1e-4) | 가중치를 얼마나, 어느 방향으로 업데이트할지 결정 |
| Loss Function | CrossEntropyLoss | 예측과 정답 사이의 "거리" 측정 (분류 문제용) |
| LR Scheduler | ReduceLROnPlateau | 성능이 정체되면 학습률을 절반으로 낮춤 |
| Early Stopping | patience=10 | 10 epoch 연속 개선 없으면 학습 중단 (과적합 방지) |
| Gradient Clipping | max_norm=1.0 | 기울기가 폭발적으로 커지는 것을 방지 |
6. 시퀀스 데이터 만들기
딥러닝 모델에 데이터를 넣으려면, FeatureStore의 플랫한 행(row) 데이터를 (batch, seq_len, n_features) 형태의 3D 텐서로 변환해야 한다. 이 과정이 시퀀스 생성이다.
슬라이딩 윈도우
각 종목(code)별로 날짜순 정렬 후, 20일씩 잘라서 카드(시퀀스)를 만든다.
# 엑셀 시트 (종목별 100일 데이터)
# Day 1 | RSI=45, MACD=0.2, ... (75개 피처)
# Day 2 | RSI=48, MACD=0.3, ...
# ... |
# Day 100| RSI=72, MACD=1.1, ...
# 슬라이딩 윈도우 (seq_len=20)
# 카드 1: Day 1~20 -> 타겟: Day 21의 상승/하락
# 카드 2: Day 2~21 -> 타겟: Day 22의 상승/하락
# 카드 3: Day 3~22 -> 타겟: Day 23의 상승/하락
# ...총 80장의 학습 카드 생성
실제 코드를 살펴보자.
def create_sequences_from_df(df, feature_columns, target_column, seq_len=20):
all_sequences = []
all_targets = []
for code, group in df.groupby("code"): # 종목별 그룹
group = group.sort_values("date") # 날짜순 정렬
feat_vals = group[feature_columns].values
tgt_vals = group[target_column].values
if len(group) < seq_len: # 데이터 부족하면 skip
continue
for i in range(seq_len, len(group)): # 슬라이딩 윈도우
seq = feat_vals[i - seq_len : i] # 20일치 피처
tgt = tgt_vals[i] # 마지막 날의 타겟
all_sequences.append(seq)
all_targets.append(int(tgt))
return np.array(all_sequences), np.array(all_targets)
# sequences shape: (N, 20, 75)
# targets shape: (N,)
생성된 시퀀스는 SequenceDataset 클래스로 감싸서 PyTorch의 DataLoader에 넣으면, 학습 시 배치 단위로 자동 공급된다.
class SequenceDataset(Dataset):
def __init__(self, sequences, targets):
self.X = torch.tensor(sequences, dtype=torch.float32)
self.y = torch.tensor(targets, dtype=torch.long)
def __getitem__(self, idx):
return self.X[idx], self.y[idx]
# DataLoader: 배치 단위로 셔플해서 공급
train_loader = DataLoader(
SequenceDataset(train_seqs, train_tgts),
batch_size=64, shuffle=True
)
7. LSTM vs Transformer 비교 정리
| 항목 | LSTM | Transformer |
|---|---|---|
| 구조 | 순서대로 하나씩 읽음 (순차적) | 모든 날을 동시에 봄 (병렬) |
| 기억 방식 | 셀 상태(메모장)에 기억 저장 | Attention으로 필요할 때 직접 참조 |
| 병렬 처리 | 불가 (순차 처리 필수) | 가능 (GPU에서 빠름) |
| 긴 시퀀스 | 기울기 소실 가능 (완화했지만 한계) | Attention으로 먼 과거도 직접 참조 |
| 순서 인식 | 구조 자체가 순서를 반영 | Positional Encoding 필요 |
| 학습 데이터 양 | 적은 데이터에서도 비교적 안정적 | 데이터가 많을수록 강점 |
| Pooling 전략 | 마지막 타임스텝 hidden 사용 | Mean Pooling (전체 평균) |
| 활성화 함수 | ReLU | GELU |
| 프로젝트 기본 lr | 1e-3 | 5e-4 |
8. 프로젝트 적용 (Alchemetric)
ALCHEMETRIC 프로젝트에서 딥러닝이 기존 ML 파이프라인과 어떻게 통합되는지 정리한다.
Strategy 패턴 — 자동 분기
기존 ModelTrainer.train()에 algorithm="lstm" 또는 algorithm="transformer"를 전달하면, 내부에서 is_deep_learning() 체크 후 자동으로 DeepLearningTrainer로 위임한다. 호출하는 쪽에서는 ML인지 DL인지 신경 쓸 필요가 없다.
# 학습: 호출 코드는 ML과 동일
trainer = ModelTrainer()
# ML 모델
trainer.train(market="KOSPI", algorithm="random_forest")
trainer.train(market="KOSPI", algorithm="xgboost")
# DL 모델 -> 자동으로 DeepLearningTrainer로 분기
trainer.train(market="KOSPI", algorithm="lstm")
trainer.train(market="KOSPI", algorithm="transformer")
분기 로직은 간단하다.
# ModelTrainer.train() 내부
from .ml_config_loader import is_deep_learning
if is_deep_learning(algorithm): # "lstm", "transformer" -> True
from .deep_learning import DeepLearningTrainer
return DeepLearningTrainer().train(...) # DL 경로
# 여기 도달하면 전통 ML
model = _get_classifier(algorithm, best_params)
model.fit(X_train, y_train) # ML 경로
예측도 마찬가지
예측 시에도 모델 파일 확장자로 자동 분기한다.
# Predictor._run_prediction() 내부
if ml_model.model_path.endswith(".pt"): # .pt 파일이면 DL
return DeepLearningPredictor().predict_single(...)
# .joblib 파일이면 ML
saved = joblib.load(ml_model.model_path)
model = saved["model"]
y_pred = model.predict(X_scaled)
.pt 파일 — 모델 + 전처리기 통째로 저장
ML 모델은 .joblib으로, DL 모델은 .pt로 저장된다. DL의 .pt 파일에는 모델 가중치뿐 아니라 추론에 필요한 모든 것이 포함된다.
save_data = {
"model_state_dict": model.state_dict(), # 학습된 가중치
"model_class": "lstm", # 어떤 아키텍처?
"model_params": params, # 하이퍼파라미터
"scaler": scaler, # RobustScaler 객체
"imputer": imputer, # SimpleImputer 객체
"feature_columns": features, # 피처 컬럼명 리스트
"n_features": n_features, # 피처 수
"seq_len": seq_len, # 시퀀스 길이
}
torch.save(save_data, "lstm_class_kospi_1d_v1.pt")
이렇게 하면 예측 시 별도로 scaler나 imputer를 찾을 필요 없이, .pt 파일 하나만 로드하면 된다.
GPU / MPS 자동 감지
학습과 추론에서 사용할 하드웨어를 자동으로 감지한다.
def _get_device() -> torch.device:
if torch.cuda.is_available(): # NVIDIA GPU
return torch.device("cuda")
if torch.backends.mps.is_available(): # Apple Silicon (M1/M2)
return torch.device("mps")
return torch.device("cpu") # 최후의 보루
전체 파이프라인 요약
[ 학습 파이프라인 ]
FeatureStore(DB)
|
_load_data() # DB에서 피처 로드
|
_filter_features() # NaN 비율 높은 피처 제외
|
_split_data() # 시계열 기반 train/val 분할 (85:15)
|
SimpleImputer # NaN -> 중앙값 대체
|
RobustScaler # 피처 스케일링
|
create_sequences() # 슬라이딩 윈도우 (종목별 20일)
|
Optuna 튜닝 # 하이퍼파라미터 최적화 (20 trials)
|
_train_loop() # epoch 루프 + early stopping
|
torch.save() # .pt 파일 저장 (모델+scaler+imputer)
|
DB 저장 # ml_model 테이블에 메타데이터 기록
[ 예측 파이프라인 ]
torch.load("model.pt") # 모델+scaler+imputer 로드
|
최근 20일 피처 조회 # DB에서 seq_len개 행 가져오기
|
imputer.transform() # NaN 처리
|
scaler.transform() # 스케일링
|
model(X) # 추론 -> logits (2개 클래스)
|
softmax -> signal # 확률 -> BUY/SELL/HOLD 시그널
ALCHEMETRIC — AI-Powered Stock Prediction Platform
Phase 8A: Deep Learning (LSTM + Transformer) / PyTorch Pipeline
'Project > 🚀 AI Finance 2026' 카테고리의 다른 글
| #27 [모듈4-3] CUDA, MPS, CPU - PyTorch 디바이스 감지와 혼합 정밀도 학습 (0) | 2026.03.19 |
|---|---|
| #26 [모듈4-2] DL Early Stopping & Overfitting (0) | 2026.03.18 |
| #24 [모듈4] Deep Learning (0) | 2026.03.18 |
| #23 [모듈3-2] Optuna — 하이퍼파라미터 자동 튜닝의 모든 것 (0) | 2026.03.16 |
| #22 [모듈3-1] scikit-learn, XGBoost, LightGBM — 3대 ML 라이브러리 (0) | 2026.03.13 |
