트레이딩과 강화학습
주식 트레이딩은 강화학습의 자연스러운 적용 분야입니다. 트레이더(에이전트)가 시장(환경)에서 매수/매도/보유(행동)를 결정하고, 수익(보상)을 얻는 구조가 강화학습 프레임워크와 정확히 일치합니다.
주의사항
이 글은 강화학습을 배우기 위한 교육용 실습이며, 투자 조언이 아닙니다. 실제 금융 시장은 여기서 다루는 것보다 훨씬 복잡합니다.
여기서 만드는 것은 시장에서 수익을 내는 시스템이 아니라, MDP를 직접 정의하고 커스텀 Gymnasium 환경을 처음부터 구현해 보는 연습 과제입니다. 뒤에서 쓰는 가격 데이터도 실제 시세가 아니라 코드로 생성한 합성 데이터입니다. 그래서 이 환경에서 나오는 어떤 숫자도 실제 시장에 대해 알려 주는 바가 없습니다. 이 글에 성과 수치를 싣지 않은 것도 같은 이유입니다.
트레이딩을 예제로 고른 이유는 하나입니다. 상태와 행동과 보상과 에피소드 경계를 전부 스스로 정해야 하는 문제 중에서, 각 요소가 무엇에 대응하는지 직관적으로 이해하기 쉬운 축에 들기 때문입니다. 앞선 글들의 CartPole이나 Atari는 환경이 이미 주어져 있어서 이 설계 과정을 건너뛰게 됩니다. 여기서 익힐 기술은 재고 관리나 자원 스케줄링 같은 문제에 그대로 옮겨집니다.
트레이딩 기초 개념
기본 용어
- 매수 (Buy/Long): 주식을 구매하여 가격 상승에 베팅
- 매도 (Sell/Short): 보유 주식을 판매하여 포지션 청산
- 포지션 (Position): 현재 보유하고 있는 주식의 상태
- 수익률 (Return): 투자 대비 이익의 비율
- 수수료 (Commission): 거래 시 발생하는 비용
- 슬리피지 (Slippage): 주문 가격과 실제 체결 가격의 차이
강화학습으로 트레이딩을 모델링하기
| RL 요소 | 트레이딩 대응 |
|---|---|
| 상태 | 과거 가격 데이터, 기술적 지표, 현재 포지션 |
| 행동 | 매수, 매도, 보유 |
| 보상 | 실현 수익, 미실현 수익 변화량 |
| 에피소드 | 일정 기간의 트레이딩 세션 |
데이터 준비
가격 데이터 생성
실습을 위해 합성 데이터를 생성합니다. 실제 적용 시에는 Yahoo Finance 등에서 데이터를 가져올 수 있습니다.
import numpy as np
import pandas as pd
def generate_stock_data(n_days=1000, initial_price=100.0, volatility=0.02, seed=42):
"""합성 주가 데이터 생성 (기하 브라운 운동 모델)"""
np.random.seed(seed)
# 일별 수익률 생성
daily_returns = np.random.normal(0.0005, volatility, n_days)
# 가격 계산
prices = initial_price * np.cumprod(1 + daily_returns)
# OHLCV 데이터 생성
data = pd.DataFrame()
data['close'] = prices
data['open'] = prices * (1 + np.random.normal(0, 0.005, n_days))
data['high'] = np.maximum(data['open'], data['close']) * (1 + np.abs(np.random.normal(0, 0.01, n_days)))
data['low'] = np.minimum(data['open'], data['close']) * (1 - np.abs(np.random.normal(0, 0.01, n_days)))
data['volume'] = np.random.randint(100000, 1000000, n_days).astype(float)
return data
# 데이터 생성
stock_data = generate_stock_data(n_days=2000)
print(f"데이터 크기: {len(stock_data)}")
print(stock_data.head())
기술적 지표 계산
def add_technical_indicators(df):
"""기술적 지표 추가"""
# 이동평균
df['sma_10'] = df['close'].rolling(window=10).mean()
df['sma_30'] = df['close'].rolling(window=30).mean()
# 상대강도지수 (RSI)
delta = df['close'].diff()
gain = delta.where(delta > 0, 0).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / (loss + 1e-10)
df['rsi'] = 100 - (100 / (1 + rs))
# 볼린저 밴드
bb_mean = df['close'].rolling(window=20).mean()
bb_std = df['close'].rolling(window=20).std()
df['bb_upper'] = bb_mean + 2 * bb_std
df['bb_lower'] = bb_mean - 2 * bb_std
df['bb_position'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'] + 1e-10)
# MACD
ema12 = df['close'].ewm(span=12).mean()
ema26 = df['close'].ewm(span=26).mean()
df['macd'] = ema12 - ema26
df['macd_signal'] = df['macd'].ewm(span=9).mean()
# 수익률
df['returns'] = df['close'].pct_change()
df['returns_5d'] = df['close'].pct_change(5)
# NaN 제거
df.dropna(inplace=True)
df.reset_index(drop=True, inplace=True)
return df
stock_data = add_technical_indicators(stock_data)
print(f"지표 추가 후 데이터 크기: {len(stock_data)}")
print(f"특성 목록: {list(stock_data.columns)}")
트레이딩 환경 설계
Gymnasium 인터페이스를 따르는 커스텀 트레이딩 환경을 구현합니다.
import gymnasium as gym
from gymnasium import spaces
class StockTradingEnv(gym.Env):
"""주식 트레이딩 환경"""
metadata = {"render_modes": ["human"]}
def __init__(self, df, window_size=30, commission=0.001,
initial_balance=100000):
super().__init__()
self.df = df
self.window_size = window_size
self.commission = commission
self.initial_balance = initial_balance
# 특성 열 선택
self.feature_columns = [
'close', 'volume', 'sma_10', 'sma_30', 'rsi',
'bb_position', 'macd', 'macd_signal', 'returns', 'returns_5d'
]
self.n_features = len(self.feature_columns)
# 행동 공간: 0=보유, 1=매수, 2=매도
self.action_space = spaces.Discrete(3)
# 관찰 공간: 가격 윈도우 + 포지션 정보
obs_shape = self.window_size * self.n_features + 3 # +3: 포지션, 수익률, 보유비율
self.observation_space = spaces.Box(
low=-np.inf, high=np.inf, shape=(obs_shape,), dtype=np.float32
)
def _get_observation(self):
"""현재 관찰값 생성"""
# 가격 윈도우 데이터
start = self.current_step - self.window_size
end = self.current_step
window_data = self.df[self.feature_columns].iloc[start:end].values
# 정규화 (각 특성별로 윈도우 내 min-max)
for i in range(self.n_features):
col = window_data[:, i]
min_val = col.min()
max_val = col.max()
if max_val - min_val > 0:
window_data[:, i] = (col - min_val) / (max_val - min_val)
else:
window_data[:, i] = 0.0
flat_window = window_data.flatten()
# 포지션 정보
position_info = np.array([
1.0 if self.position > 0 else 0.0, # 포지션 보유 여부
self.unrealized_pnl / self.initial_balance, # 미실현 수익률
self.shares * self.current_price / self.total_value, # 주식 보유 비율
], dtype=np.float32)
return np.concatenate([flat_window, position_info]).astype(np.float32)
@property
def current_price(self):
return self.df['close'].iloc[self.current_step]
@property
def unrealized_pnl(self):
if self.position > 0:
return self.shares * (self.current_price - self.entry_price)
return 0.0
@property
def total_value(self):
return self.balance + self.shares * self.current_price
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.position = 0 # 0: 없음, 1: 롱
self.entry_price = 0.0
self.total_trades = 0
self.winning_trades = 0
self.trade_history = []
return self._get_observation(), {}
def step(self, action):
prev_total = self.total_value
reward = 0.0
trade_info = ""
current_price = self.current_price
if action == 1 and self.position == 0: # 매수
# 전체 잔고의 95%로 매수 (수수료 고려)
max_shares = int(self.balance * 0.95 / (current_price * (1 + self.commission)))
if max_shares > 0:
cost = max_shares * current_price * (1 + self.commission)
self.balance -= cost
self.shares = max_shares
self.position = 1
self.entry_price = current_price
trade_info = f"매수 {max_shares}주 @ {current_price:.2f}"
elif action == 2 and self.position == 1: # 매도
proceeds = self.shares * current_price * (1 - self.commission)
self.balance += proceeds
pnl = (current_price - self.entry_price) / self.entry_price
self.total_trades += 1
if pnl > 0:
self.winning_trades += 1
self.trade_history.append(pnl)
trade_info = f"매도 {self.shares}주 @ {current_price:.2f}, 수익률: {pnl:.2%}"
self.shares = 0
self.position = 0
self.entry_price = 0.0
# 다음 스텝으로 이동
self.current_step += 1
# 보상: 포트폴리오 가치 변화율
current_total = self.total_value
reward = (current_total - prev_total) / prev_total
# 종료 조건
terminated = self.current_step >= len(self.df) - 1
truncated = self.total_value < self.initial_balance * 0.5 # 50% 이상 손실
info = {
"total_value": self.total_value,
"balance": self.balance,
"position": self.position,
"total_trades": self.total_trades,
"trade_info": trade_info,
}
return self._get_observation(), reward, terminated, truncated, info
# 환경 테스트
env = StockTradingEnv(stock_data, window_size=30)
obs, info = env.reset()
print(f"관찰 차원: {obs.shape}")
print(f"초기 포트폴리오: {env.total_value:,.0f}원")
무작위 에이전트 기준선
def evaluate_random_agent(env, n_episodes=10):
"""무작위 에이전트 평가"""
results = []
for episode in range(n_episodes):
obs, _ = env.reset()
while True:
action = env.action_space.sample()
obs, reward, terminated, truncated, info = env.step(action)
if terminated or truncated:
break
final_value = info['total_value']
total_return = (final_value - env.initial_balance) / env.initial_balance
results.append({
'final_value': final_value,
'return': total_return,
'trades': info['total_trades'],
})
returns = [r['return'] for r in results]
print(f"=== 무작위 에이전트 ({n_episodes}회) ===")
print(f"평균 수익률: {np.mean(returns):.2%}")
print(f"최대 수익률: {np.max(returns):.2%}")
print(f"최소 수익률: {np.min(returns):.2%}")
print(f"평균 거래 횟수: {np.mean([r['trades'] for r in results]):.1f}")
return results
# random_results = evaluate_random_agent(env)
피드포워드 DQN 모델
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random
class TradingDQN(nn.Module):
"""트레이딩용 피드포워드 DQN"""
def __init__(self, obs_size, n_actions):
super().__init__()
self.net = nn.Sequential(
nn.Linear(obs_size, 256),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(256, 128),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(128, 64),
nn.ReLU(),
nn.Linear(64, n_actions),
)
def forward(self, x):
return self.net(x)
CNN 모델: 가격 차트를 이미지처럼 처리
가격 데이터의 시간적 패턴을 1D 합성곱으로 포착합니다.
class TradingCNN(nn.Module):
"""1D CNN 기반 트레이딩 모델"""
def __init__(self, window_size, n_features, n_actions):
super().__init__()
self.conv = nn.Sequential(
nn.Conv1d(n_features, 32, kernel_size=5, padding=2),
nn.ReLU(),
nn.Conv1d(32, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool1d(1), # 시간 차원을 1로 축소
)
# CNN 출력 + 포지션 정보 (3개)
self.fc = nn.Sequential(
nn.Linear(64 + 3, 64),
nn.ReLU(),
nn.Linear(64, n_actions),
)
self.window_size = window_size
self.n_features = n_features
def forward(self, x):
batch_size = x.shape[0]
# 윈도우 데이터와 포지션 정보 분리
window_data = x[:, :-3].view(batch_size, self.window_size, self.n_features)
position_info = x[:, -3:]
# CNN: (batch, features, time) 형태로 변환
window_data = window_data.transpose(1, 2)
conv_out = self.conv(window_data).squeeze(-1)
# 포지션 정보와 결합
combined = torch.cat([conv_out, position_info], dim=1)
return self.fc(combined)
트레이딩 에이전트 학습
class ReplayBuffer:
def __init__(self, capacity):
self.buffer = deque(maxlen=capacity)
def push(self, state, action, reward, next_state, done):
self.buffer.append((state, action, reward, next_state, done))
def sample(self, batch_size):
batch = random.sample(self.buffer, batch_size)
s, a, r, ns, d = zip(*batch)
return (np.array(s), np.array(a), np.array(r, dtype=np.float32),
np.array(ns), np.array(d, dtype=np.bool_))
def __len__(self):
return len(self.buffer)
def train_trading_agent(env, model_type="ff", n_episodes=500):
"""트레이딩 에이전트 학습"""
obs_size = env.observation_space.shape[0]
n_actions = env.action_space.n
device = torch.device("cpu")
if model_type == "cnn":
online_net = TradingCNN(env.window_size, env.n_features, n_actions).to(device)
target_net = TradingCNN(env.window_size, env.n_features, n_actions).to(device)
else:
online_net = TradingDQN(obs_size, n_actions).to(device)
target_net = TradingDQN(obs_size, n_actions).to(device)
target_net.load_state_dict(online_net.state_dict())
optimizer = optim.Adam(online_net.parameters(), lr=1e-4)
buffer = ReplayBuffer(50000)
epsilon = 1.0
epsilon_min = 0.05
epsilon_decay = 0.995
gamma = 0.99
batch_size = 64
target_update = 50
best_return = -float('inf')
returns_history = []
for episode in range(n_episodes):
obs, _ = env.reset()
total_reward = 0
while True:
# 엡실론-탐욕 행동 선택
if random.random() < epsilon:
action = env.action_space.sample()
else:
with torch.no_grad():
q = online_net(torch.tensor([obs], dtype=torch.float32).to(device))
action = q.argmax(dim=1).item()
next_obs, reward, terminated, truncated, info = env.step(action)
done = terminated or truncated
buffer.push(obs, action, reward, next_obs, done)
total_reward += reward
obs = next_obs
# 학습
if len(buffer) >= batch_size:
s, a, r, ns, d = buffer.sample(batch_size)
s_t = torch.tensor(s, dtype=torch.float32).to(device)
a_t = torch.tensor(a, dtype=torch.long).to(device)
r_t = torch.tensor(r, dtype=torch.float32).to(device)
ns_t = torch.tensor(ns, dtype=torch.float32).to(device)
d_t = torch.tensor(d, dtype=torch.bool).to(device)
current_q = online_net(s_t).gather(1, a_t.unsqueeze(1)).squeeze(1)
with torch.no_grad():
# Double DQN
best_a = online_net(ns_t).argmax(dim=1)
next_q = target_net(ns_t).gather(1, best_a.unsqueeze(1)).squeeze(1)
next_q[d_t] = 0.0
target_q = r_t + gamma * next_q
loss = nn.SmoothL1Loss()(current_q, target_q)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(online_net.parameters(), 1.0)
optimizer.step()
if done:
break
# 타겟 네트워크 업데이트
if episode % target_update == 0:
target_net.load_state_dict(online_net.state_dict())
epsilon = max(epsilon_min, epsilon * epsilon_decay)
# 결과 기록
episode_return = (env.total_value - env.initial_balance) / env.initial_balance
returns_history.append(episode_return)
if episode_return > best_return:
best_return = episode_return
torch.save(online_net.state_dict(), "best_trading_model.pth")
if episode % 50 == 0:
mean_return = np.mean(returns_history[-50:])
print(
f"에피소드 {episode}: "
f"수익률={episode_return:.2%}, "
f"평균 수익률={mean_return:.2%}, "
f"거래={info['total_trades']}, "
f"엡실론={epsilon:.3f}"
)
return online_net, returns_history
# 학습 실행
# trained_model, history = train_trading_agent(env, model_type="ff", n_episodes=500)
에이전트 평가 및 분석
def backtest_agent(env, net, n_episodes=5):
"""학습된 에이전트 백테스트"""
all_results = []
for episode in range(n_episodes):
obs, _ = env.reset()
portfolio_values = [env.total_value]
trade_log = []
while True:
with torch.no_grad():
q = net(torch.tensor([obs], dtype=torch.float32))
action = q.argmax(dim=1).item()
obs, reward, terminated, truncated, info = env.step(action)
portfolio_values.append(env.total_value)
if info.get('trade_info'):
trade_log.append(info['trade_info'])
if terminated or truncated:
break
total_return = (portfolio_values[-1] - portfolio_values[0]) / portfolio_values[0]
# 성과 지표 계산
daily_returns = np.diff(portfolio_values) / portfolio_values[:-1]
sharpe_ratio = np.mean(daily_returns) / (np.std(daily_returns) + 1e-10) * np.sqrt(252)
# 최대 낙폭 (MDD)
peak = np.maximum.accumulate(portfolio_values)
drawdown = (np.array(portfolio_values) - peak) / peak
max_drawdown = drawdown.min()
result = {
'total_return': total_return,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'total_trades': info['total_trades'],
'portfolio_values': portfolio_values,
}
all_results.append(result)
print(f"\n에피소드 {episode + 1}:")
print(f" 총 수익률: {total_return:.2%}")
print(f" 샤프 비율: {sharpe_ratio:.2f}")
print(f" 최대 낙폭: {max_drawdown:.2%}")
print(f" 총 거래 횟수: {info['total_trades']}")
# 전체 평균
print("\n=== 전체 백테스트 결과 ===")
avg_return = np.mean([r['total_return'] for r in all_results])
avg_sharpe = np.mean([r['sharpe_ratio'] for r in all_results])
avg_mdd = np.mean([r['max_drawdown'] for r in all_results])
print(f"평균 수익률: {avg_return:.2%}")
print(f"평균 샤프 비율: {avg_sharpe:.2f}")
print(f"평균 최대 낙폭: {avg_mdd:.2%}")
return all_results
# backtest_results = backtest_agent(env, trained_model)
바이앤홀드 전략과 비교
def compare_with_buy_and_hold(df, agent_results, initial_balance=100000):
"""바이앤홀드 전략과 RL 에이전트 비교"""
# 바이앤홀드: 처음에 매수하고 끝까지 보유
start_price = df['close'].iloc[30] # window_size 이후
end_price = df['close'].iloc[-1]
bnh_return = (end_price - start_price) / start_price
agent_return = np.mean([r['total_return'] for r in agent_results])
print("=== 전략 비교 ===")
print(f"바이앤홀드 수익률: {bnh_return:.2%}")
print(f"RL 에이전트 수익률: {agent_return:.2%}")
print(f"초과 수익률: {agent_return - bnh_return:.2%}")
# compare_with_buy_and_hold(stock_data, backtest_results)
보상 함수 설계의 중요성
보상 함수의 설계는 트레이딩 에이전트의 성능에 결정적입니다.
class ImprovedRewardEnv(StockTradingEnv):
"""개선된 보상 함수를 사용하는 트레이딩 환경"""
def _compute_reward(self, prev_total, action):
"""다양한 보상 설계 방식"""
current_total = self.total_value
# 1. 단순 포트폴리오 변화율
basic_reward = (current_total - prev_total) / prev_total
# 2. 리스크 조정 보상 (샤프 비율 유사)
# 수익률에서 변동성 페널티를 차감
volatility_penalty = 0.0
if len(self.trade_history) > 1:
recent_returns = self.trade_history[-10:]
volatility_penalty = np.std(recent_returns) * 0.1
# 3. 과도한 거래 페널티
trade_penalty = 0.0
if action != 0: # 보유가 아닌 경우
trade_penalty = -0.0001
# 4. 승률 보너스
win_bonus = 0.0
if self.total_trades > 10:
win_rate = self.winning_trades / self.total_trades
if win_rate > 0.5:
win_bonus = 0.0001
return basic_reward - volatility_penalty + trade_penalty + win_bonus
학습을 실제로 돌려 보면 무엇이 보이나
train_trading_agent를 그대로 실행하면 에피소드마다 한 줄씩 로그가 쌓입니다. 여기서 봐야 할 것은 수익률 숫자가 아니라 학습이 진행되고 있다는 신호가 있는지 여부입니다. 신호는 세 가지를 같이 봐야 판별됩니다.
첫째는 엡실론입니다. 코드의 epsilon_decay가 0.995이므로 에피소드마다 0.5퍼센트씩 줄어듭니다. 1.0에서 시작해 epsilon_min인 0.05에 닿으려면 대략 600 에피소드가 필요합니다. 그런데 기본 n_episodes는 500입니다. 즉 이 설정으로 끝까지 돌려도 에이전트는 마지막 순간까지 6퍼센트 남짓의 확률로 무작위 행동을 하고 있습니다. 학습 후반의 지표가 들쭉날쭉하다면 에이전트가 못 배운 것이 아니라 아직 탐험 중인 것일 수 있습니다. 평가는 반드시 엡실론을 끄고 따로 해야 합니다. backtest_agent가 argmax만 쓰는 이유가 이것입니다.
둘째는 거래 횟수입니다. info['total_trades']를 에피소드마다 찍어 보세요. 이 값이 0에 수렴하는지, 아니면 매 스텝 거래하는 수준으로 폭주하는지가 보상 함수가 제대로 작동하는지를 가장 빨리 알려 줍니다. 둘 다 흔한 실패 모드이고, 뒤에서 따로 다룹니다.
셋째는 손실값입니다. DQN의 TD 손실은 지도학습처럼 매끄럽게 내려가지 않습니다. 타겟 네트워크를 target_update 주기마다 갱신하므로 그 시점마다 손실이 튀는 것이 정상입니다. 문제가 있는 패턴은 따로 있습니다. 손실이 발산하거나, 반대로 아주 이른 시점에 0에 붙어 버리는 경우입니다. 후자는 대개 네트워크가 모든 상태에 같은 Q값을 내놓는 상태, 즉 붕괴한 상태입니다.
로그에 이 세 가지를 함께 남기도록 학습 루프를 조금 고쳐 두면 진단이 훨씬 쉬워집니다.
# train_trading_agent의 에피소드 루프 끝에 추가
if episode % 10 == 0:
print(
f"ep={episode:4d} "
f"eps={epsilon:.3f} "
f"trades={env.total_trades:3d} "
f"value={env.total_value:,.0f} "
f"loss={np.mean(recent_losses) if recent_losses else float('nan'):.5f}"
)
recent_losses = []
출력은 대략 이런 모양이 됩니다. 숫자 자체가 아니라 열들이 어떻게 함께 움직이는지를 보세요.
ep= 0 eps=1.000 trades= 87 value=... loss=0.01243
ep= 10 eps=0.951 trades= 74 value=... loss=0.00871
ep= 20 eps=0.905 trades= 61 value=... loss=0.00655
...
ep= 200 eps=0.367 trades= 9 value=... loss=0.00112
ep= 400 eps=0.135 trades= 2 value=... loss=0.00038
거래 횟수가 엡실론을 따라 단조롭게 줄어드는 이 패턴은 좋은 신호가 아닙니다. 에이전트가 전략을 배운 것이 아니라, 무작위 행동이 줄어든 만큼만 거래가 줄어든 것입니다. 즉 학습된 정책이 사실상 보유 일변도이고, 남은 거래는 전부 탐험 노이즈라는 뜻입니다. 다음 절의 첫 번째 항목이 정확히 이 상황입니다.
실패 사례와 진단 순서
에이전트가 아무것도 하지 않는다. 가장 흔한 결말입니다. 증상은 total_trades가 0이거나 한 자릿수이고, 포트폴리오 가치가 초기값에서 거의 움직이지 않는 것입니다. 원인은 보상 구조에 있습니다. commission이 0.001이므로 매수와 매도를 한 번씩 하면 왕복 0.2퍼센트가 확정 손실로 나갑니다. 반면 보유는 비용이 0입니다. 확실한 손실과 불확실한 이득 사이에서 Q러닝이 확실한 쪽을 고르는 것은 완벽하게 합리적인 행동입니다. 진단 순서는 이렇습니다. 먼저 commission=0.0으로 두고 다시 학습시켜 보세요. 거래가 살아난다면 원인이 수수료 대비 보상 스케일이라는 것이 확정됩니다. 그다음에 수수료를 되돌리고 보상 쪽 스케일을 키우거나, 보유 상태에 아주 작은 시간 페널티를 주는 식으로 균형을 다시 잡습니다. 수수료를 없앤 채로 두는 것은 답이 아닙니다. 그건 문제를 푼 게 아니라 문제를 지운 것입니다.
반대로 매 스텝 거래한다. 보상 스케일을 키우다 보면 쉽게 넘어가는 반대편 실패입니다. ImprovedRewardEnv에 trade_penalty가 들어 있는 이유가 이것입니다. 다만 페널티 상수를 손으로 맞추는 것은 임시방편에 가깝습니다. 더 견고한 방법은 행동 공간 자체를 바꾸는 것입니다. 예를 들어 매 스텝 매수와 매도를 허용하는 대신 목표 포지션 비중을 행동으로 두면, 같은 비중을 유지하는 행동이 자연스럽게 거래 없음이 되어 페널티 없이도 과잉 거래가 사라집니다.
같은 결과가 5번 반복된다. backtest_agent(env, net, n_episodes=5)를 돌리면 5개 에피소드의 지표가 완전히 동일하게 나옵니다. 버그처럼 보이지만 코드대로의 결과입니다. reset()이 current_step을 항상 window_size로 되돌리고, df도 매번 같고, 평가 정책은 argmax라 결정적입니다. 결정적 환경에 결정적 정책이면 궤적이 하나뿐입니다. 즉 이 백테스트의 표본 수는 5가 아니라 1입니다. 여기에 평균과 표준편차를 계산하는 것은 의미가 없습니다. 제대로 하려면 reset()에서 시작 지점을 무작위로 고르거나, 서로 다른 기간을 담은 여러 개의 df로 각각 평가해야 합니다.
학습 데이터에서만 잘한다. 위 문제와 짝을 이루는 더 근본적인 사안입니다. 현재 학습 루프는 500 에피소드 내내 같은 가격 경로 하나를 반복해서 지나갑니다. 강화학습에서 이것은 훈련 데이터 한 개로 학습시키는 것과 같습니다. 에이전트가 배우는 것은 일반화된 전략이 아니라 그 특정 경로의 특정 지점에서 무엇을 했어야 하는지에 대한 암기입니다. 데이터를 나눌 때는 반드시 시간순으로 자르세요. 시계열을 무작위로 섞어 나누면 미래 구간에서 학습하고 과거 구간에서 평가하는 상황이 생기고, 그 결과는 아무 의미가 없습니다.
지표를 전체 구간에서 미리 계산했다. 조용하고 위험한 함정입니다. sma_30이나 rsi 같은 지표를 pandas의 rolling으로 계산하면 과거만 참조하므로 안전합니다. 그러나 정규화 통계를 데이터프레임 전체의 평균과 표준편차로 잡는 순간, 학습 시점의 관찰값에 미래 정보가 섞여 들어갑니다. 이 글의 _get_observation이 윈도우 내부의 최솟값과 최댓값으로만 정규화하는 것은 이 문제를 피하기 위한 설계입니다. 전처리를 고칠 때 이 성질을 깨뜨리지 않도록 주의하세요. 증상은 학습 중 지표는 좋은데 새 구간에서 무너지는 형태로 나타나며, 원인을 찾기 어렵기로 유명합니다.
샤프 비율이 이상한 값으로 나온다. backtest_agent의 샤프 계산은 np.sqrt(252)를 곱합니다. 이 상수는 하루 단위 봉을 가정한 연율화 계수입니다. 시간 봉이나 분 봉 데이터를 넣으면서 이 값을 그대로 두면 결과가 무의미해집니다. 또 거래가 거의 없어 일별 수익률이 대부분 0이면 분모가 0에 가까워져 값이 폭주합니다. 코드의 1e-10은 0으로 나누는 것을 막을 뿐 값의 의미까지 지켜 주지는 않습니다. 지표를 읽기 전에 항상 total_trades를 먼저 보세요.
언제 이 접근을 쓰지 않나
강화학습으로 트레이딩을 다루는 예제는 교재에 자주 등장하지만, 실제로 이 문제 구조가 강화학습에 잘 맞는지는 따로 따져 볼 필요가 있습니다.
강화학습이 힘을 내는 전형적인 조건은 이렇습니다. 환경이 대체로 안정적이고, 시뮬레이터에서 값싸게 무한히 많은 경험을 만들 수 있으며, 행동이 다음 상태에 실제로 영향을 주고, 보상이 비교적 분명합니다. 게임과 로봇 제어가 여기에 딱 들어맞습니다.
시장 데이터는 이 조건을 대부분 만족하지 않습니다. 규칙 자체가 시간에 따라 변하므로 과거에서 배운 것이 미래에 성립한다는 보장이 없습니다. 경험은 값싸지 않습니다. 실제 데이터는 유한하고, 합성 데이터로 늘리면 그 합성 과정의 가정을 학습하게 됩니다. 신호 대 잡음비도 극단적으로 낮아서, 정책이 좋아서 얻은 결과인지 운이었는지를 구분하는 데 필요한 표본 수가 감당하기 어려운 수준입니다.
그래서 순서를 이렇게 잡는 편이 정직합니다. 예측 문제라면 지도학습으로 먼저 풀어 보세요. 다음 구간의 방향이나 변동성을 맞히는 문제는 강화학습 없이 다룰 수 있고, 검증도 훨씬 쉽습니다. 강화학습이 고유하게 기여하는 부분은 예측이 아니라 순차적 의사결정, 즉 포지션 크기 조절이나 주문 집행 같은 문제입니다. 예측 성능이 확보되지 않은 상태에서 강화학습을 얹으면, 잘 안 될 때 원인이 예측인지 정책인지 보상 설계인지 구분할 수 없게 됩니다.
한 가지 더. 이 글의 코드는 학습 목적에 맞춰 의도적으로 단순화되어 있습니다. 슬리피지가 없고, 주문은 항상 원하는 가격에 전량 체결되며, 시장 충격도 없고, 하루 안의 가격 움직임도 없습니다. 용어 절에서 슬리피지를 소개해 놓고 환경에는 넣지 않은 것이 그 예입니다. 이런 단순화를 하나씩 되돌리는 것이 좋은 후속 연습이고, 되돌릴 때마다 문제가 얼마나 어려워지는지를 체감하는 것이 이 실습의 진짜 소득입니다.
참고 자료
- Gymnasium: Making a Custom Environment (gymnasium.farama.org) —
reset과step의 반환 규약,observation_space와action_space정의 방식을 확인한 곳입니다. 2026-08-16 확인. - Gymnasium Env API (gymnasium.farama.org) —
step이 돌려주는 다섯 값과terminated와truncated의 구분에 대한 정의입니다. 2026-08-16 확인. - Human-level control through deep reinforcement learning (Nature, 2015) — 경험 재현 버퍼와 타겟 네트워크를 도입한 DQN 원 논문입니다.
- 이 시리즈의 앞선 글에서 DQN의 구성 요소를 다루었습니다. 여기서는 그 구조를 커스텀 환경에 옮기는 부분에 집중했습니다.
정리
- 문제 정의: 주식 트레이딩을 MDP로 모델링 (상태=시장 데이터, 행동=매수/매도/보유, 보상=수익)
- 데이터 준비: 가격 데이터에 기술적 지표를 추가하여 관찰 공간 구성
- 환경 설계: Gymnasium 인터페이스의 커스텀 트레이딩 환경 구현
- 모델: 피드포워드 DQN과 1D CNN 모델 모두 활용 가능
- 보상 설계: 단순 수익률 외에 리스크 조정, 거래 페널티 등 다양한 요소 고려
- 평가: 수익률, 샤프 비율, 최대 낙폭 등 다양한 성과 지표로 평가
다음 글에서는 가치 기반 방법에서 벗어나 정책을 직접 최적화하는 Policy Gradient 방법을 살펴보겠습니다.