강화학습 트레이딩이란?
강화학습(Reinforcement Learning, RL)은 에이전트가 환경과 상호작용하며 보상을 최대화하는 행동을 스스로 학습하는 머신러닝 기법입니다. 자동매매에 적용하면 에이전트가 시장 상태를 관찰하고, 매수·매도·홀드를 결정하며, 수익이라는 보상을 최대화하도록 훈련됩니다.
기존 규칙 기반 전략과의 차이점:
- 규칙 기반: “RSI 30 이하면 매수” → 사람이 규칙 설계
- 강화학습: 에이전트가 데이터로부터 최적 전략을 스스로 발견
이 글에서는 OpenAI Gym 환경을 만들고, DQN과 PPO로 트레이딩 에이전트를 훈련하는 전체 과정을 다룹니다.
트레이딩 환경 구축: Gym 커스텀 환경
강화학습의 첫 단계는 환경(Environment) 정의입니다. 에이전트가 상호작용할 시장 시뮬레이터를 Gymnasium(OpenAI Gym) 인터페이스로 구현합니다.
import gymnasium as gym
from gymnasium import spaces
import numpy as np
class TradingEnv(gym.Env):
"""강화학습 트레이딩 환경"""
metadata = {'render_modes': ['human']}
def __init__(self, prices, features, initial_balance=100000,
commission=0.001, window_size=20):
"""
prices: 종가 배열
features: 관측에 사용할 기술지표 (n_steps, n_features)
commission: 거래 수수료율
window_size: 관측 윈도우 크기
"""
super().__init__()
self.prices = prices
self.features = features
self.initial_balance = initial_balance
self.commission = commission
self.window_size = window_size
# 행동 공간: 0=홀드, 1=매수, 2=매도
self.action_space = spaces.Discrete(3)
# 관측 공간: 기술지표 윈도우 + 포지션 정보
n_features = features.shape[1]
obs_shape = window_size * n_features + 3 # +잔고, 보유량, 미실현손익
self.observation_space = spaces.Box(
low=-np.inf, high=np.inf, shape=(obs_shape,), dtype=np.float32
)
def reset(self, seed=None, options=None):
super().reset(seed=seed)
self.current_step = self.window_size
self.balance = self.initial_balance
self.shares = 0
self.total_trades = 0
self.entry_price = 0
return self._get_observation(), {}
def _get_observation(self):
"""현재 관측 상태 구성"""
start = self.current_step - self.window_size
end = self.current_step
window = self.features[start:end].flatten()
# 포지션 정보 정규화
price = self.prices[self.current_step]
position_info = np.array([
self.balance / self.initial_balance,
self.shares * price / self.initial_balance,
(price / self.entry_price - 1) if self.entry_price > 0 else 0
], dtype=np.float32)
return np.concatenate([window, position_info]).astype(np.float32)
def step(self, action):
price = self.prices[self.current_step]
reward = 0
done = False
if action == 1 and self.shares == 0: # 매수
max_shares = int(self.balance / (price * (1 + self.commission)))
if max_shares > 0:
cost = max_shares * price * (1 + self.commission)
self.balance -= cost
self.shares = max_shares
self.entry_price = price
self.total_trades += 1
elif action == 2 and self.shares > 0: # 매도
revenue = self.shares * price * (1 - self.commission)
self.balance += revenue
pnl_pct = (price - self.entry_price) / self.entry_price
reward = pnl_pct * 100 # 수익률 기반 보상
self.shares = 0
self.entry_price = 0
self.total_trades += 1
# 다음 스텝
self.current_step += 1
if self.current_step >= len(self.prices) - 1:
done = True
# 잔여 포지션 강제 청산
if self.shares > 0:
self.balance += self.shares * self.prices[self.current_step]
self.shares = 0
# 포트폴리오 가치
portfolio = self.balance + self.shares * self.prices[self.current_step]
total_return = (portfolio - self.initial_balance) / self.initial_balance
obs = self._get_observation()
info = {'portfolio': portfolio, 'total_return': total_return,
'trades': self.total_trades}
return obs, reward, done, False, info기술지표 피처 엔지니어링
에이전트에게 제공할 관측 데이터의 품질이 학습 성능을 좌우합니다. 원시 가격 대신 정규화된 기술지표를 사용합니다.
import pandas as pd
def create_features(df):
"""트레이딩 환경용 기술지표 생성"""
close = df['close']
features = pd.DataFrame(index=df.index)
# 수익률 (정규화된 가격 변화)
features['return_1'] = close.pct_change(1)
features['return_5'] = close.pct_change(5)
features['return_20'] = close.pct_change(20)
# 이동평균 대비 위치
features['ma_ratio_10'] = close / close.rolling(10).mean() - 1
features['ma_ratio_50'] = close / close.rolling(50).mean() - 1
# RSI (0~1 정규화)
delta = close.diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / (loss + 1e-10)
features['rsi'] = rs / (1 + rs)
# 변동성
features['volatility'] = close.pct_change().rolling(20).std()
# 거래량 변화
if 'volume' in df.columns:
features['volume_ratio'] = df['volume'] / df['volume'].rolling(20).mean()
# 볼린저 밴드 위치
ma20 = close.rolling(20).mean()
std20 = close.rolling(20).std()
features['bb_position'] = (close - ma20) / (2 * std20 + 1e-10)
features = features.dropna()
return features.values.astype(np.float32), features.indexDQN 에이전트: 가치 기반 학습
DQN(Deep Q-Network)은 각 행동의 가치(Q값)를 신경망으로 근사합니다. 트레이딩에서는 매수·매도·홀드 각각의 기대 수익을 학습합니다.
import torch
import torch.nn as nn
from collections import deque
import random
class QNetwork(nn.Module):
"""DQN 가치 네트워크"""
def __init__(self, obs_size, n_actions=3):
super().__init__()
self.net = nn.Sequential(
nn.Linear(obs_size, 256),
nn.ReLU(),
nn.LayerNorm(256),
nn.Linear(256, 128),
nn.ReLU(),
nn.LayerNorm(128),
nn.Linear(128, 64),
nn.ReLU(),
nn.Linear(64, n_actions)
)
def forward(self, x):
return self.net(x)
class DQNTrader:
"""DQN 트레이딩 에이전트"""
def __init__(self, obs_size, lr=1e-4, gamma=0.99,
epsilon_start=1.0, epsilon_end=0.01, epsilon_decay=0.995,
buffer_size=50000, batch_size=64):
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.q_net = QNetwork(obs_size).to(self.device)
self.target_net = QNetwork(obs_size).to(self.device)
self.target_net.load_state_dict(self.q_net.state_dict())
self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=lr)
self.gamma = gamma
self.epsilon = epsilon_start
self.epsilon_end = epsilon_end
self.epsilon_decay = epsilon_decay
self.batch_size = batch_size
self.replay_buffer = deque(maxlen=buffer_size)
def select_action(self, state):
"""ε-greedy 행동 선택"""
if random.random() < self.epsilon:
return random.randint(0, 2)
with torch.no_grad():
state_t = torch.FloatTensor(state).unsqueeze(0).to(self.device)
q_values = self.q_net(state_t)
return q_values.argmax(dim=1).item()
def store(self, state, action, reward, next_state, done):
self.replay_buffer.append((state, action, reward, next_state, done))
def train_step(self):
"""미니배치 학습"""
if len(self.replay_buffer) < self.batch_size:
return 0
batch = random.sample(self.replay_buffer, self.batch_size)
states, actions, rewards, next_states, dones = zip(*batch)
states = torch.FloatTensor(np.array(states)).to(self.device)
actions = torch.LongTensor(actions).to(self.device)
rewards = torch.FloatTensor(rewards).to(self.device)
next_states = torch.FloatTensor(np.array(next_states)).to(self.device)
dones = torch.FloatTensor(dones).to(self.device)
# 현재 Q값
q_values = self.q_net(states).gather(1, actions.unsqueeze(1)).squeeze()
# 타겟 Q값 (Double DQN)
with torch.no_grad():
next_actions = self.q_net(next_states).argmax(dim=1)
next_q = self.target_net(next_states).gather(
1, next_actions.unsqueeze(1)
).squeeze()
target = rewards + self.gamma * next_q * (1 - dones)
loss = nn.MSELoss()(q_values, target)
self.optimizer.zero_grad()
loss.backward()
nn.utils.clip_grad_norm_(self.q_net.parameters(), 1.0)
self.optimizer.step()
self.epsilon = max(self.epsilon_end,
self.epsilon * self.epsilon_decay)
return loss.item()
def update_target(self):
self.target_net.load_state_dict(self.q_net.state_dict())PPO 에이전트: 정책 기반 학습
PPO(Proximal Policy Optimization)는 현재 가장 널리 쓰이는 정책 기반 RL 알고리즘입니다. DQN보다 안정적인 학습이 가능하며, 연속적 행동 공간에도 확장할 수 있습니다.
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv
def train_ppo_trader(prices, features, total_timesteps=100000):
"""Stable-Baselines3 PPO로 트레이딩 에이전트 훈련"""
def make_env():
return TradingEnv(prices, features, window_size=20)
env = DummyVecEnv([make_env])
model = PPO(
'MlpPolicy',
env,
learning_rate=3e-4,
n_steps=2048,
batch_size=64,
n_epochs=10,
gamma=0.99,
gae_lambda=0.95,
clip_range=0.2,
ent_coef=0.01, # 탐색 장려
verbose=1,
policy_kwargs=dict(
net_arch=dict(pi=[256, 128], vf=[256, 128])
)
)
model.learn(total_timesteps=total_timesteps)
return model
def evaluate_agent(model, prices, features, window_size=20):
"""에이전트 성능 평가"""
env = TradingEnv(prices, features, window_size=window_size)
obs, _ = env.reset()
done = False
actions_taken = {0: 0, 1: 0, 2: 0}
while not done:
action, _ = model.predict(obs, deterministic=True)
obs, reward, done, _, info = env.step(action)
actions_taken[int(action)] += 1
print(f"총 수익률: {info['total_return']:.2%}")
print(f"최종 자산: {info['portfolio']:,.0f}")
print(f"총 거래: {info['trades']}회")
print(f"행동 분포: 홀드={actions_taken[0]}, 매수={actions_taken[1]}, 매도={actions_taken[2]}")
return info보상 함수 설계: 성패를 가르는 핵심
보상 함수(Reward Function) 설계가 강화학습 트레이딩의 가장 중요한 부분입니다. 단순 수익률만 사용하면 과도한 리스크를 감수하는 에이전트가 만들어집니다.
| 보상 함수 | 장점 | 단점 |
|---|---|---|
| 단순 PnL | 직관적 | 리스크 무시, 과적합 |
| 샤프 비율 | 리스크 대비 수익 최적화 | 계산 비용, 희소 보상 |
| 로그 수익률 | 복리 효과 반영 | 큰 손실에 민감 |
| 리스크 조정 보상 | MDD 패널티 포함 | 파라미터 튜닝 필요 |
class RiskAdjustedReward:
"""리스크 조정 보상 함수"""
def __init__(self, risk_penalty=0.5, trade_penalty=0.001,
holding_penalty=0.0001):
self.risk_penalty = risk_penalty
self.trade_penalty = trade_penalty
self.holding_penalty = holding_penalty
self.peak_portfolio = 0
self.returns_history = []
def calculate(self, portfolio_value, traded, has_position):
"""복합 보상 계산"""
# 기본 보상: 포트폴리오 수익률 변화
if len(self.returns_history) > 0:
ret = (portfolio_value - self.returns_history[-1])
/ self.returns_history[-1]
else:
ret = 0
self.returns_history.append(portfolio_value)
reward = ret * 100
# MDD 패널티
self.peak_portfolio = max(self.peak_portfolio, portfolio_value)
drawdown = (self.peak_portfolio - portfolio_value) / self.peak_portfolio
reward -= drawdown * self.risk_penalty
# 과다 거래 패널티
if traded:
reward -= self.trade_penalty
# 장기 미보유 패널티 (행동 유도)
if not has_position:
reward -= self.holding_penalty
return rewardMDD 관리 전략의 개념을 보상 함수에 직접 통합하면, 에이전트가 큰 낙폭을 회피하도록 학습합니다.
워크포워드 검증과 과적합 방지
강화학습은 특히 과적합 위험이 높습니다. 에이전트가 학습 데이터의 패턴을 암기하는 것을 방지해야 합니다.
def walk_forward_train(prices, features, n_splits=5,
train_ratio=0.7, total_timesteps=50000):
"""워크포워드 검증으로 강건성 평가"""
results = []
chunk_size = len(prices) // n_splits
for i in range(n_splits):
start = i * chunk_size
end = min(start + chunk_size, len(prices))
split = int((end - start) * train_ratio)
# 학습/검증 분리
train_prices = prices[start:start + split]
train_features = features[start:start + split]
test_prices = prices[start + split:end]
test_features = features[start + split:end]
# 학습
model = train_ppo_trader(train_prices, train_features,
total_timesteps=total_timesteps)
# 검증
info = evaluate_agent(model, test_prices, test_features)
results.append({
'fold': i + 1,
'train_period': f"{start}~{start+split}",
'test_return': info['total_return'],
'trades': info['trades']
})
print(f"Fold {i+1}: 수익률 {info['total_return']:.2%}")
avg_return = np.mean([r['test_return'] for r in results])
print(f"n평균 아웃샘플 수익률: {avg_return:.2%}")
return results백테스트 오버피팅 방지법의 원칙이 그대로 적용됩니다. 데이터 증강, 노이즈 주입, 앙상블 등의 기법도 함께 사용하세요.
실전 배포 시 주의사항
- 학습 데이터 양: 최소 2~3년 이상의 일봉 데이터, 고빈도면 수개월의 틱 데이터가 필요합니다.
- 시장 레짐 변화: 학습 기간과 다른 시장 환경에서는 성능이 급락할 수 있습니다. 주기적 재학습이 필수입니다.
- 행동 공간 설계: 이산(매수/매도/홀드) vs 연속(비율 조절) — 연속 행동이 더 유연하지만 학습이 어렵습니다.
- 시뮬레이션 현실성: 슬리피지, 수수료, 부분 체결을 환경에 반영하지 않으면 실전 괴리가 큽니다.
- 안전장치: 최대 손실 한도, 일일 거래 횟수 제한, 킬 스위치를 반드시 추가하세요.
강화학습 자동매매는 퀀트 트레이딩의 최전선입니다. 올바른 환경 설계, 보상 함수, 검증 프로세스를 갖추면 기존 규칙 기반 전략을 넘어서는 적응적 매매 시스템을 구축할 수 있습니다.