DQN Implementation Guide

import gym from gym import spaces import numpy as np import pandas as pd class TradingEnvironment(gym.Env): def __init__(self, data, initial_balance=100000, transaction_cost=0.001): super(TradingEnvironment, self).__init__() self.data = data self.initial_balance = initial_balance self.transaction_cost = transaction_cost # Define action and observation space self.action_space = spaces.Discrete(3) # Buy, Hold, Sell self.observation_space = spaces.Box( low=-np.inf, high=np.inf, shape=(len(data.columns),), dtype=np.float32 ) self.reset() def reset(self): self.current_step = 0 self.balance = self.initial_balance self.shares = 0 self.portfolio_value = self.initial_balance return self._get_observation() def step(self, action): # Execute action self._execute_action(action) # Move to next step self.current_step += 1 # Calculate reward reward = self._calculate_reward() # Check if done done = self.current_step >= len(self.data) - 1 # Get next observation next_obs = self._get_observation() return next_obs, reward, done, {} def _execute_action(self, action): current_price = self.data.iloc[self.current_step]['close'] if action == 0: # Buy if self.balance > current_price: shares_to_buy = self.balance // current_price cost = shares_to_buy * current_price * (1 + self.transaction_cost) self.balance -= cost self.shares += shares_to_buy elif action == 2: # Sell if self.shares > 0: revenue = self.shares * current_price * (1 - self.transaction_cost) self.balance += revenue self.shares = 0 def _calculate_reward(self): current_price = self.data.iloc[self.current_step]['close'] self.portfolio_value = self.balance + self.shares * current_price # Calculate return if self.current_step == 0: return 0 prev_value = self._get_portfolio_value(self.current_step - 1) current_return = (self.portfolio_value - prev_value) / prev_value # Risk penalty risk_penalty = -0.01 * abs(current_return) if abs(current_return) > 0.05 else 0 return current_return + risk_penalty def _get_observation(self): return self.data.iloc[self.current_step].values.astype(np.float32)

import torch import torch.nn as nn import torch.nn.functional as F class DQNNetwork(nn.Module): def __init__(self, input_size, hidden_sizes=[128, 64], output_size=3): super(DQNNetwork, self).__init__() # Input layer self.input_layer = nn.Linear(input_size, hidden_sizes[0]) # Hidden layers self.hidden_layers = nn.ModuleList() for i in range(len(hidden_sizes) - 1): self.hidden_layers.append( nn.Linear(hidden_sizes[i], hidden_sizes[i + 1]) ) # Output layer self.output_layer = nn.Linear(hidden_sizes[-1], output_size) # Dropout for regularization self.dropout = nn.Dropout(0.2) def forward(self, x): # Input layer with ReLU activation x = F.relu(self.input_layer(x)) x = self.dropout(x) # Hidden layers for layer in self.hidden_layers: x = F.relu(layer(x)) x = self.dropout(x) # Output layer (no activation for Q-values) q_values = self.output_layer(x) return q_values # Dueling DQN Architecture class DuelingDQN(nn.Module): def __init__(self, input_size, hidden_sizes=[128, 64], output_size=3): super(DuelingDQN, self).__init__() # Shared feature layers self.feature_layer = nn.Linear(input_size, hidden_sizes[0]) # Value stream self.value_stream = nn.Sequential( nn.Linear(hidden_sizes[0], hidden_sizes[1]), nn.ReLU(), nn.Linear(hidden_sizes[1], 1) ) # Advantage stream self.advantage_stream = nn.Sequential( nn.Linear(hidden_sizes[0], hidden_sizes[1]), nn.ReLU(), nn.Linear(hidden_sizes[1], output_size) ) def forward(self, x): features = F.relu(self.feature_layer(x)) value = self.value_stream(features) advantage = self.advantage_stream(features) # Combine value and advantage q_values = value + advantage - advantage.mean(dim=1, keepdim=True) return q_values

import random from collections import deque import torch class ExperienceReplay: def __init__(self, capacity=100000): self.buffer = deque(maxlen=capacity) self.capacity = capacity def push(self, state, action, reward, next_state, done): """Add experience to buffer""" experience = (state, action, reward, next_state, done) self.buffer.append(experience) def sample(self, batch_size): """Sample random batch from buffer""" batch = random.sample(self.buffer, batch_size) states = torch.FloatTensor([e[0] for e in batch]) actions = torch.LongTensor([e[1] for e in batch]) rewards = torch.FloatTensor([e[2] for e in batch]) next_states = torch.FloatTensor([e[3] for e in batch]) dones = torch.BoolTensor([e[4] for e in batch]) return states, actions, rewards, next_states, dones def __len__(self): return len(self.buffer) # Prioritized Experience Replay class PrioritizedReplay: def __init__(self, capacity=100000, alpha=0.6): self.capacity = capacity self.alpha = alpha self.buffer = [] self.priorities = [] self.position = 0 def push(self, state, action, reward, next_state, done): """Add experience with maximum priority""" max_priority = max(self.priorities) if self.priorities else 1.0 if len(self.buffer) < self.capacity: self.buffer.append(None) self.priorities.append(None) self.buffer[self.position] = (state, action, reward, next_state, done) self.priorities[self.position] = max_priority self.position = (self.position + 1) % self.capacity def sample(self, batch_size, beta=0.4): """Sample batch with prioritized sampling""" if len(self.buffer) == 0: return None, None, None # Calculate sampling probabilities priorities = np.array(self.priorities[:len(self.buffer)]) probabilities = priorities ** self.alpha probabilities /= probabilities.sum() # Sample indices indices = np.random.choice( len(self.buffer), batch_size, p=probabilities ) # Calculate importance sampling weights weights = (len(self.buffer) * probabilities[indices]) ** (-beta) weights /= weights.max() # Get experiences batch = [self.buffer[idx] for idx in indices] states = torch.FloatTensor([e[0] for e in batch]) actions = torch.LongTensor([e[1] for e in batch]) rewards = torch.FloatTensor([e[2] for e in batch]) next_states = torch.FloatTensor([e[3] for e in batch]) dones = torch.BoolTensor([e[4] for e in batch]) return (states, actions, rewards, next_states, dones), indices, weights def update_priorities(self, indices, td_errors): """Update priorities based on TD errors""" for idx, td_error in zip(indices, td_errors): priority = abs(td_error) + 1e-6 # Small constant for numerical stability self.priorities[idx] = priority

import torch.optim as optim from torch.utils.tensorboard import SummaryWriter class DQNTrainer: def __init__(self, env, model, target_model, replay_buffer, learning_rate=1e-4, gamma=0.99, epsilon=1.0, epsilon_decay=0.995, epsilon_min=0.01, target_update_freq=1000): self.env = env self.model = model self.target_model = target_model self.replay_buffer = replay_buffer # Hyperparameters self.learning_rate = learning_rate self.gamma = gamma self.epsilon = epsilon self.epsilon_decay = epsilon_decay self.epsilon_min = epsilon_min self.target_update_freq = target_update_freq # Optimizer and loss function self.optimizer = optim.Adam(model.parameters(), lr=learning_rate) self.criterion = nn.MSELoss() # Logging self.writer = SummaryWriter('runs/dqn_trading') self.episode_rewards = [] self.losses = [] def train(self, num_episodes=1000, batch_size=32): """Main training loop""" for episode in range(num_episodes): state = self.env.reset() episode_reward = 0 episode_loss = 0 step_count = 0 while True: # Select action using epsilon-greedy action = self.select_action(state) # Execute action in environment next_state, reward, done, _ = self.env.step(action) # Store experience self.replay_buffer.push(state, action, reward, next_state, done) # Train model if enough experiences if len(self.replay_buffer) > batch_size: loss = self.update_model(batch_size) episode_loss += loss # Update state and reward state = next_state episode_reward += reward step_count += 1 # Check if episode is done if done: break # Decay epsilon self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay) # Update target network if episode % self.target_update_freq == 0: self.target_model.load_state_dict(self.model.state_dict()) # Logging self.episode_rewards.append(episode_reward) self.losses.append(episode_loss / step_count if step_count > 0 else 0) self.writer.add_scalar('Episode/Reward', episode_reward, episode) self.writer.add_scalar('Episode/Loss', episode_loss / step_count, episode) self.writer.add_scalar('Episode/Epsilon', self.epsilon, episode) if episode % 100 == 0: print(f'Episode {episode}, Reward: {episode_reward:.2f}, ' f'Loss: {episode_loss/step_count:.4f}, Epsilon: {self.epsilon:.3f}') def select_action(self, state): """Select action using epsilon-greedy policy""" if np.random.random() < self.epsilon: return np.random.randint(0, self.env.action_space.n) with torch.no_grad(): state_tensor = torch.FloatTensor(state).unsqueeze(0) q_values = self.model(state_tensor) return q_values.argmax().item() def update_model(self, batch_size): """Update model using experience replay""" batch = self.replay_buffer.sample(batch_size) states, actions, rewards, next_states, dones = batch # Current Q-values current_q_values = self.model(states).gather(1, actions.unsqueeze(1)) # Next Q-values from target network with torch.no_grad(): next_q_values = self.target_model(next_states).max(1)[0] target_q_values = rewards + (self.gamma * next_q_values * ~dones) # Calculate loss loss = self.criterion(current_q_values.squeeze(), target_q_values) # Optimize self.optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0) self.optimizer.step() return loss.item() def save_model(self, path): """Save trained model""" torch.save({ 'model_state_dict': self.model.state_dict(), 'target_model_state_dict': self.target_model.state_dict(), 'optimizer_state_dict': self.optimizer.state_dict(), 'epsilon': self.epsilon, 'episode_rewards': self.episode_rewards }, path) def load_model(self, path): """Load trained model""" checkpoint = torch.load(path) self.model.load_state_dict(checkpoint['model_state_dict']) self.target_model.load_state_dict(checkpoint['target_model_state_dict']) self.optimizer.load_state_dict(checkpoint['optimizer_state_dict']) self.epsilon = checkpoint['epsilon'] self.episode_rewards = checkpoint['episode_rewards']

import optuna def objective(trial): """Optuna objective function for hyperparameter optimization""" # Sample hyperparameters learning_rate = trial.suggest_loguniform('learning_rate', 1e-5, 1e-2) gamma = trial.suggest_uniform('gamma', 0.9, 0.99) epsilon_decay = trial.suggest_uniform('epsilon_decay', 0.99, 0.999) target_update_freq = trial.suggest_categorical('target_update_freq', [100, 500, 1000, 2000]) hidden_size = trial.suggest_categorical('hidden_size', [64, 128, 256]) # Create model and trainer model = DQNNetwork(input_size=9, hidden_sizes=[hidden_size, hidden_size//2]) target_model = DQNNetwork(input_size=9, hidden_sizes=[hidden_size, hidden_size//2]) target_model.load_state_dict(model.state_dict()) replay_buffer = ExperienceReplay(capacity=100000) trainer = DQNTrainer( env=env, model=model, target_model=target_model, replay_buffer=replay_buffer, learning_rate=learning_rate, gamma=gamma, epsilon_decay=epsilon_decay, target_update_freq=target_update_freq ) # Train for limited episodes trainer.train(num_episodes=100) # Return average reward (objective to maximize) return np.mean(trainer.episode_rewards[-10:]) # Last 10 episodes # Run optimization study = optuna.create_study(direction='maximize') study.optimize(objective, n_trials=50) print("Best hyperparameters:") print(study.best_params) print(f"Best reward: {study.best_value}")

def evaluate_model(model, env, num_episodes=100): """Evaluate trained model on test environment""" model.eval() episode_rewards = [] episode_returns = [] episode_sharpe_ratios = [] with torch.no_grad(): for episode in range(num_episodes): state = env.reset() episode_reward = 0 returns = [] while True: # Select action greedily (no exploration) state_tensor = torch.FloatTensor(state).unsqueeze(0) q_values = model(state_tensor) action = q_values.argmax().item() # Execute action next_state, reward, done, info = env.step(action) episode_reward += reward returns.append(reward) state = next_state if done: break episode_rewards.append(episode_reward) # Calculate Sharpe ratio if len(returns) > 1: sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) episode_sharpe_ratios.append(sharpe) episode_returns.append(sum(returns)) return { 'mean_reward': np.mean(episode_rewards), 'std_reward': np.std(episode_rewards), 'mean_return': np.mean(episode_returns), 'mean_sharpe': np.mean(episode_sharpe_ratios), 'win_rate': sum(1 for r in episode_returns if r > 0) / len(episode_returns) } # Usage results = evaluate_model(trained_model, test_env) print(f"Mean Reward: {results['mean_reward']:.2f}") print(f"Mean Return: {results['mean_return']:.2%}") print(f"Sharpe Ratio: {results['mean_sharpe']:.2f}") print(f"Win Rate: {results['win_rate']:.2%}")

# Learning rate scheduling scheduler = optim.lr_scheduler.ReduceLROnPlateau( optimizer, mode='max', factor=0.5, patience=100, verbose=True ) # Double DQN implementation def double_dqn_loss(current_model, target_model, states, actions, rewards, next_states, dones, gamma): """Calculate Double DQN loss""" # Current Q-values current_q_values = current_model(states).gather(1, actions.unsqueeze(1)) # Next actions from current model with torch.no_grad(): next_actions = current_model(next_states).argmax(1) next_q_values = target_model(next_states).gather(1, next_actions.unsqueeze(1)) target_q_values = rewards.unsqueeze(1) + (gamma * next_q_values * ~dones.unsqueeze(1)) loss = F.mse_loss(current_q_values, target_q_values) return loss # Noisy Networks for exploration class NoisyLinear(nn.Module): def __init__(self, in_features, out_features, std_init=0.5): super(NoisyLinear, self).__init__() self.in_features = in_features self.out_features = out_features self.std_init = std_init self.weight_mu = nn.Parameter(torch.empty(out_features, in_features)) self.weight_sigma = nn.Parameter(torch.empty(out_features, in_features)) self.bias_mu = nn.Parameter(torch.empty(out_features)) self.bias_sigma = nn.Parameter(torch.empty(out_features)) self.register_buffer('weight_epsilon', torch.empty(out_features, in_features)) self.register_buffer('bias_epsilon', torch.empty(out_features)) self.reset_parameters() self.reset_noise() def reset_parameters(self): mu_range = 1 / np.sqrt(self.in_features) self.weight_mu.data.uniform_(-mu_range, mu_range) self.weight_sigma.data.fill_(self.std_init / np.sqrt(self.in_features)) self.bias_mu.data.uniform_(-mu_range, mu_range) self.bias_sigma.data.fill_(self.std_init / np.sqrt(self.out_features)) def reset_noise(self): epsilon_in = self._scale_noise(self.in_features) epsilon_out = self._scale_noise(self.out_features) self.weight_epsilon.copy_(epsilon_out.ger(epsilon_in)) self.bias_epsilon.copy_(epsilon_out) def _scale_noise(self, size): x = torch.randn(size, device=self.weight_mu.device) return x.sign().mul_(x.abs().sqrt_()) def forward(self, input): if self.training: weight = self.weight_mu + self.weight_sigma * self.weight_epsilon bias = self.bias_mu + self.bias_sigma * self.bias_epsilon else: weight = self.weight_mu bias = self.bias_mu return F.linear(input, weight, bias)