トレーディングと強化学習
株式トレーディングは強化学習の自然な応用分野です。トレーダー（エージェント）が市場（環境）で売買/保有（行動）を決定し、利益（報酬）を得る構造は強化学習フレームワークと正確に一致します。
注意事項
この記事は強化学習を学ぶための教育用の実習であり、投資助言ではありません。実際の金融市場はここで扱うものよりはるかに複雑です。
ここで作るのは市場で利益を出すシステムではなく、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)
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()
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)
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)
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)
self.action_space = spaces.Discrete(3)
obs_shape = self.window_size * self.n_features + 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
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
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:
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
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),
)
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:]
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をそのまま実行すると、エピソードごとに1行ずつログが積み上がります。ここで見るべきなのは収益率の数字ではなく、学習が進んでいるという信号があるかどうかです。信号は3つを一緒に見て初めて判別できます。
一つ目はイプシロンです。コードの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値を出す状態、つまり崩壊した状態です。
ログにこの3つを一緒に残すよう学習ループを少し直しておくと、診断がずっと楽になります。
# 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エピソードのあいだ、同じ価格経路を一つだけ繰り返し通り抜けます。強化学習でこれは、訓練データ1個で学習させるのと同じです。エージェントが学ぶのは一般化された戦略ではなく、その特定の経路の特定の地点で何をすべきだったかという丸暗記です。データを分けるときは、必ず時間順に切ってください。時系列を無作為に混ぜて分けると、未来の区間で学習して過去の区間で評価する状況が生まれ、その結果には何の意味もありません。
指標を全区間で先に計算してしまった。静かで危険な落とし穴です。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が返す5つの値と、terminatedとtruncatedの区別についての定義です。2026-08-16確認。 - Human-level control through deep reinforcement learning (Nature, 2015) — 経験再生バッファとターゲットネットワークを導入したDQNの原論文です。
- このシリーズの以前の記事でDQNの構成要素を扱いました。ここではその構造をカスタム環境に移す部分に集中しました。
まとめ
- 問題定義: 株式トレーディングをMDPとしてモデル化（状態=市場データ、行動=売買/保有、報酬=利益）
- データ準備: 価格データにテクニカル指標を追加して観測空間を構成
- 環境設計: Gymnasiumインターフェースのカスタムトレーディング環境を実装
- モデル: フィードフォワードDQNと1D CNNモデルの両方が活用可能
- 報酬設計: 単純な収益率以外にリスク調整、取引ペナルティなど多様な要素を考慮
- 評価: 収益率、シャープレシオ、最大ドローダウンなど多様な成果指標で評価
次の記事では価値ベースの方法から離れて方策を直接最適化するPolicy Gradient方法を見ていきます。