Lesson 5.6: Reinforcement Learning in Finance: Q-Learning, Policy Gradients, and Execution Algorithms

Introduction: From Prediction to Optimal Action

Throughout previous lessons, we explored deep learning architectures designed to predict future market states, such as time-series pricing trends, volatility indices, and macroeconomic credit risks. However, prediction alone does not constitute an execution strategy. In real-world financial markets, an institution must make sequential decisions—such as whether to buy, sell, or hold, how to split a massive multi-million-dollar order across multiple exchanges without moving the market, or how to dynamically rebalance a portfolio in real-time.

To solve sequential decision-making under uncertainty, quantitative finance leverages Reinforcement Learning (RL). Unlike supervised learning that relies on static historical labels, reinforcement learning trains an autonomous agent to interact with an environment and maximize a cumulative reward function through trial and error. This lesson deconstructs Markov Decision Processes, Q-learning, Deep Q-Networks (DQN), policy gradients, and optimal execution algorithms like Almgren-Chriss.

Part 1: Markov Decision Processes (MDP) and Reinforcement Learning Foundations

Every financial reinforcement learning problem is mathematically formulated as a Markov Decision Process (MDP), defined by the tuple (S, A, P, R, γ):

State Space (S): The complete set of market conditions observed by the agent at time t (e.g., current asset price, order book depth, portfolio cash balance, technical indicators).
Action Space (A): The set of all possible actions the agent can take (e.g., buy 100 shares, sell 50 shares, hold position).
Transition Probability (P): The probability of transitioning from state s to state s’ given action a.
Reward Function (R): The immediate financial feedback received by the agent after taking action a in state s (e.g., net realized profit, Sharpe ratio increment, or penalty for excessive transaction costs).
Discount Factor (γ): A value between 0 and 1 that determines the present value of future rewards.

Part 2: Value-Based Methods (Q-Learning and DQN)

1. The Q-Function

In value-based reinforcement learning, the agent learns a Q-value function Q(s, a), which represents the expected cumulative discounted reward of taking action a in state s and following an optimal policy thereafter.

Bellman Equation: The optimal Q-value satisfies the recursive Bellman optimality equation:

Q(s, a) = E[R(s, a) + γ × max_a’ Q(s’, a’)]

2. Deep Q-Networks (DQN)

When state spaces are massive (such as high-frequency limit order books), a traditional Q-table cannot store every state-action pair. Deep Q-Networks (DQN) replace the table with a deep neural network that approximates the Q-function.

Financial Application: DQNs are used to execute automated market-making and optimal trade entry, learning to quote bid-ask spreads dynamically to capture inventory profits while avoiding adverse selection.

Part 3: Policy Gradient Methods and Actor-Critic Architectures

While value-based methods excel at discrete action spaces, continuous financial tasks (such as determining the exact percentage of capital to allocate across 500 different stocks) require Policy Gradient methods.

1. Direct Policy Optimization

Instead of calculating value functions, policy gradient methods parameterize a policy π_θ(a|s) using a neural network with weights θ and optimize the weights directly via gradient ascent to maximize expected cumulative rewards.

2. Actor-Critic Architectures

State-of-the-art financial RL models utilize Actor-Critic frameworks:

The Actor: Proposes and executes trading actions based on the current market state.
The Critic: Evaluates the actions taken by the actor, computing a value function (such as the Advantage function) to guide the actor toward more profitable strategies during training.

Part 4: Optimal Execution Algorithms (The Almgren-Chriss Framework)

Executing large institutional orders requires splitting blocks over time to minimize market impact. The industry benchmark is the Almgren-Chriss Framework.

1. Temporary and Permanent Market Impact

Permanent Impact: The lasting shift in asset price caused by acquiring a large position, proportional to the total trading volume.
Temporary Impact: The short-term price distortion caused by executing a trade rapidly within a specific time window, increasing quadratically with execution speed.

2. The Risk-Reward Trade-Off

The Almgren-Chriss algorithm solves an optimal execution trajectory by balancing two competing forces: Execution Cost (trading slowly to minimize temporary market impact) versus Timing Risk (trading quickly to avoid adverse market price drift). The model yields an optimal deterministic or stochastic trading schedule that minimizes expected cost subject to a tolerable variance risk limit.

1. Markov Decision Process Mathematical Formulation

MDP Mathematical Definition:

MDP = (S, A, P, R, γ)

Where:
- S: State space (finite or continuous)
- A: Action space (finite or continuous)
- P: Transition probability P(s' | s, a)
- R: Reward function R(s, a, s')
- γ: Discount factor [0, 1]

Policy π: S → A (mapping states to actions)

Value Functions:
V^π(s) = E[Σ_{t=0}^∞ γ^t R(s_t, a_t, s_{t+1}) | s_0 = s, π]

Q^π(s, a) = E[Σ_{t=0}^∞ γ^t R(s_t, a_t, s_{t+1}) | s_0 = s, a_0 = a, π]

Optimal Value Functions:
V*(s) = max_π V^π(s)
Q*(s, a) = max_π Q^π(s, a)

Bellman Optimality Equation:
V*(s) = max_a Σ_{s'} P(s' | s, a) [R(s, a, s') + γ V*(s')]

Q*(s, a) = Σ_{s'} P(s' | s, a) [R(s, a, s') + γ max_a' Q*(s', a')]

2. Q-Learning Deep-Dive

Q-Learning Update Rule:

Q(s, a) ← Q(s, a) + α [r + γ × max_a' Q(s', a') - Q(s, a)]

Where:
- α: Learning rate (0 < α ≤ 1)
- r: Immediate reward
- γ: Discount factor
- s': Next state

Q-Learning Implementation:

import numpy as np
from collections import defaultdict

class QLearning:
    """
    Q-Learning algorithm for financial trading
    """
    def __init__(self, actions, alpha=0.1, gamma=0.95, epsilon=0.1):
        """
        Parameters:
        - actions: List of possible actions
        - alpha: Learning rate
        - gamma: Discount factor
        - epsilon: Exploration rate
        """
        self.actions = actions
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.q_table = defaultdict(lambda: np.zeros(len(actions)))
        
        # Store episode data
        self.episode_states = []
        self.episode_actions = []
        self.episode_rewards = []
    
    def get_action(self, state):
        """
        Epsilon-greedy action selection
        """
        if np.random.random() < self.epsilon:
            # Exploration: random action
            return np.random.choice(self.actions)
        else:
            # Exploitation: best action
            q_values = self.q_table[state]
            return self.actions[np.argmax(q_values)]
    
    def update(self, state, action, reward, next_state):
        """
        Update Q-value using Q-learning update rule
        """
        current_q = self.q_table[state][self.actions.index(action)]
        max_next_q = np.max(self.q_table[next_state])
        
        # Q-learning update
        new_q = current_q + self.alpha * (reward + self.gamma * max_next_q - current_q)
        self.q_table[state][self.actions.index(action)] = new_q
    
    def train_episode(self, env):
        """
        Train for one episode
        """
        state = env.reset()
        done = False
        
        while not done:
            action = self.get_action(state)
            next_state, reward, done = env.step(action)
            self.update(state, action, reward, next_state)
            state = next_state

3. Deep Q-Network (DQN) Implementation

import torch
import torch.nn as nn
import torch.nn.functional as F
from collections import deque
import random

class DQNetwork(nn.Module):
    """
    Deep Q-Network for financial trading
    """
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

class ReplayBuffer:
    """
    Experience replay buffer for DQN
    """
    def __init__(self, capacity=10000):
        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)
        state, action, reward, next_state, done = map(np.stack, zip(*batch))
        return state, action, reward, next_state, done
    
    def __len__(self):
        return len(self.buffer)

class DQNAgent:
    """
    DQN Agent for reinforcement learning trading
    """
    def __init__(self, state_dim, action_dim, lr=0.001, gamma=0.99, epsilon=1.0, epsilon_decay=0.995, epsilon_min=0.01):
        self.state_dim = state_dim
        self.action_dim = action_dim
        
        # Networks
        self.q_network = DQNetwork(state_dim, action_dim)
        self.target_network = DQNetwork(state_dim, action_dim)
        self.target_network.load_state_dict(self.q_network.state_dict())
        
        # Optimizer
        self.optimizer = torch.optim.Adam(self.q_network.parameters(), lr=lr)
        
        # Hyperparameters
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.epsilon_min = epsilon_min
        
        # Replay buffer
        self.memory = ReplayBuffer()
        
        # Training stats
        self.loss_history = []
    
    def get_action(self, state, train=True):
        """
        Get action using epsilon-greedy policy
        """
        if train and np.random.random() < self.epsilon:
            return np.random.randint(self.action_dim)
        
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        with torch.no_grad():
            q_values = self.q_network(state_tensor)
        return torch.argmax(q_values).item()
    
    def remember(self, state, action, reward, next_state, done):
        """
        Store experience in replay buffer
        """
        self.memory.push(state, action, reward, next_state, done)
    
    def train_step(self, batch_size=64):
        """
        Train the Q-network on a batch from replay buffer
        """
        if len(self.memory) < batch_size:
            return
        
        # Sample batch
        states, actions, rewards, next_states, dones = self.memory.sample(batch_size)
        
        # Convert to tensors
        states = torch.FloatTensor(states)
        actions = torch.LongTensor(actions).unsqueeze(1)
        rewards = torch.FloatTensor(rewards).unsqueeze(1)
        next_states = torch.FloatTensor(next_states)
        dones = torch.FloatTensor(dones).unsqueeze(1)
        
        # Current Q-values
        current_q = self.q_network(states).gather(1, actions)
        
        # Target Q-values
        with torch.no_grad():
            next_q = self.target_network(next_states).max(1, keepdim=True)[0]
            target_q = rewards + (1 - dones) * self.gamma * next_q
        
        # Loss
        loss = F.mse_loss(current_q, target_q)
        self.loss_history.append(loss.item())
        
        # Update network
        self.optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.q_network.parameters(), 1.0)
        self.optimizer.step()
        
        # Decay epsilon
        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)
    
    def update_target_network(self):
        """
        Update target network with Q-network weights
        """
        self.target_network.load_state_dict(self.q_network.state_dict())

4. Policy Gradient Methods

class PolicyGradient:
    """
    Policy Gradient (REINFORCE) for financial trading
    """
    def __init__(self, state_dim, action_dim, lr=0.001, gamma=0.99):
        self.policy_network = DQNetwork(state_dim, action_dim)
        self.optimizer = torch.optim.Adam(self.policy_network.parameters(), lr=lr)
        self.gamma = gamma
        
        # Store episode data
        self.states = []
        self.actions = []
        self.rewards = []
        self.log_probs = []
    
    def get_action(self, state):
        """
        Sample action from policy distribution
        """
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        logits = self.policy_network(state_tensor)
        probs = F.softmax(logits, dim=1)
        
        # Sample action
        dist = torch.distributions.Categorical(probs)
        action = dist.sample()
        log_prob = dist.log_prob(action)
        
        # Store for training
        self.states.append(state)
        self.actions.append(action.item())
        self.log_probs.append(log_prob)
        
        return action.item()
    
    def store_reward(self, reward):
        """
        Store reward for the episode
        """
        self.rewards.append(reward)
    
    def train_episode(self):
        """
        Train policy on completed episode
        """
        # Calculate discounted returns
        returns = []
        R = 0
        for r in reversed(self.rewards):
            R = r + self.gamma * R
            returns.insert(0, R)
        
        # Normalize returns
        returns = torch.FloatTensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)
        
        # Policy gradient loss
        loss = 0
        for log_prob, R in zip(self.log_probs, returns):
            loss += -log_prob * R
        
        # Update policy
        self.optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.policy_network.parameters(), 1.0)
        self.optimizer.step()
        
        # Clear episode data
        self.states = []
        self.actions = []
        self.rewards = []
        self.log_probs = []
        
        return loss.item()

5. Actor-Critic Architecture

class ActorNetwork(nn.Module):
    """
    Actor network for continuous action spaces
    """
    def __init__(self, state_dim, action_dim, hidden_dim=128, max_action=1.0):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)
        self.max_action = max_action
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = torch.tanh(self.fc3(x)) * self.max_action
        return x

class CriticNetwork(nn.Module):
    """
    Critic network for value estimation
    """
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        self.fc1 = nn.Linear(state_dim + action_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, 1)
    
    def forward(self, state, action):
        x = torch.cat([state, action], dim=1)
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

class ActorCriticAgent:
    """
    Actor-Critic agent for continuous trading actions
    """
    def __init__(self, state_dim, action_dim, lr=0.001, gamma=0.99, tau=0.001):
        # Actor networks
        self.actor = ActorNetwork(state_dim, action_dim)
        self.target_actor = ActorNetwork(state_dim, action_dim)
        self.target_actor.load_state_dict(self.actor.state_dict())
        
        # Critic networks
        self.critic = CriticNetwork(state_dim, action_dim)
        self.target_critic = CriticNetwork(state_dim, action_dim)
        self.target_critic.load_state_dict(self.critic.state_dict())
        
        # Optimizers
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=lr)
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=lr)
        
        # Hyperparameters
        self.gamma = gamma
        self.tau = tau
        
        # Replay buffer
        self.memory = ReplayBuffer()
    
    def get_action(self, state, add_noise=True):
        """
        Get action from actor with optional exploration noise
        """
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        action = self.actor(state_tensor)
        
        if add_noise:
            noise = torch.normal(0, 0.1, size=action.shape)
            action = torch.clamp(action + noise, -1, 1)
        
        return action.detach().numpy().flatten()
    
    def remember(self, state, action, reward, next_state, done):
        self.memory.push(state, action, reward, next_state, done)
    
    def train_step(self, batch_size=64):
        """
        Train actor and critic on batch
        """
        if len(self.memory) < batch_size:
            return
        
        # Sample batch
        states, actions, rewards, next_states, dones = self.memory.sample(batch_size)
        
        # Convert to tensors
        states = torch.FloatTensor(states)
        actions = torch.FloatTensor(actions)
        rewards = torch.FloatTensor(rewards).unsqueeze(1)
        next_states = torch.FloatTensor(next_states)
        dones = torch.FloatTensor(dones).unsqueeze(1)
        
        # Update critic
        with torch.no_grad():
            next_actions = self.target_actor(next_states)
            target_q = self.target_critic(next_states, next_actions)
            target_q = rewards + (1 - dones) * self.gamma * target_q
        
        current_q = self.critic(states, actions)
        critic_loss = F.mse_loss(current_q, target_q)
        
        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()
        
        # Update actor
        actor_loss = -self.critic(states, self.actor(states)).mean()
        
        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()
        
        # Soft update target networks
        self.soft_update(self.target_actor, self.actor)
        self.soft_update(self.target_critic, self.critic)
        
        return {'critic_loss': critic_loss.item(), 'actor_loss': actor_loss.item()}
    
    def soft_update(self, target, source):
        """
        Soft update target network
        """
        for target_param, param in zip(target.parameters(), source.parameters()):
            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)

6. Almgren-Chriss Optimal Execution

class AlmgrenChriss:
    """
    Almgren-Chriss optimal execution framework
    """
    def __init__(self, total_shares, time_horizon, price, volatility, market_impact):
        """
        Parameters:
        - total_shares: Total shares to execute
        - time_horizon: Trading horizon (in days)
        - price: Current price
        - volatility: Price volatility (annualized)
        - market_impact: Market impact parameters
        """
        self.total_shares = total_shares
        self.time_horizon = time_horizon
        self.price = price
        self.volatility = volatility
        self.market_impact = market_impact
        
        # Calculate optimal trajectory
        self.optimal_trajectory = self.solve_optimal_trajectory()
    
    def solve_optimal_trajectory(self):
        """
        Solve optimal execution trajectory
        """
        # Time discretization
        n_periods = self.time_horizon * 252  # Daily periods
        dt = 1 / 252  # Time step
        
        # Parameters
        eta = self.market_impact['temporary']
        gamma = self.market_impact['permanent']
        sigma = self.volatility
        X = self.total_shares
        T = self.time_horizon
        
        # Optimal trading speed (Almgren-Chriss formula)
        # v_t = (k * X / T) * (sinh(k * (T - t)) / sinh(k * T))
        k = np.sqrt(eta / (gamma * sigma**2))
        
        # Calculate optimal speed at each time
        speeds = []
        positions = []
        
        current_position = X
        for t in np.linspace(0, T, n_periods):
            # Optimal speed
            v_t = (k * X / T) * (np.sinh(k * (T - t)) / np.sinh(k * T))
            speeds.append(v_t)
            
            # Update position
            current_position -= v_t * dt
            positions.append(current_position)
        
        return {
            'positions': positions,
            'speeds': speeds
        }
    
    def calculate_expected_cost(self, trajectory):
        """
        Calculate expected execution cost
        """
        # Implementation shortfall
        total_cost = 0
        
        for t in range(len(trajectory['positions'])):
            # Temporary impact cost
            speed = trajectory['speeds'][t]
            temp_impact = self.market_impact['temporary'] * speed
            
            # Permanent impact cost
            permanent_impact = self.market_impact['permanent'] * (self.total_shares - trajectory['positions'][t])
            
            total_cost += (temp_impact + permanent_impact) * speed / self.total_shares
        
        return total_cost
    
    def calculate_var_cost(self, trajectory):
        """
        Calculate VaR of execution cost
        """
        # Variance of execution cost
        sigma = self.volatility
        total_shares = self.total_shares
        
        # Variance from Almgren-Chriss
        var_cost = 0
        for t in range(len(trajectory['positions'])):
            speed = trajectory['speeds'][t]
            # Approximate variance
            var_cost += (speed * sigma)**2
        
        # 95% VaR
        var_95 = 1.645 * np.sqrt(var_cost)
        
        return var_95
    
    def plot_trajectory(self):
        """
        Visualize optimal execution trajectory
        """
        import matplotlib.pyplot as plt
        
        time_points = np.linspace(0, self.time_horizon, len(self.optimal_trajectory['positions']))
        
        fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))
        
        # Position trajectory
        ax1.plot(time_points, self.optimal_trajectory['positions'] / self.total_shares)
        ax1.set_xlabel('Time (days)')
        ax1.set_ylabel('Remaining Position (%)')
        ax1.set_title('Optimal Execution Trajectory')
        ax1.grid(True)
        
        # Trading speed
        ax2.plot(time_points, self.optimal_trajectory['speeds'])
        ax2.set_xlabel('Time (days)')
        ax2.set_ylabel('Trading Speed (shares/day)')
        ax2.set_title('Optimal Trading Speed')
        ax2.grid(True)
        
        plt.tight_layout()
        plt.show()

7. Reinforcement Learning Trading Environment

class TradingEnvironment:
    """
    Reinforcement learning trading environment
    """
    def __init__(self, data, initial_capital=100000, transaction_cost=0.001):
        """
        Parameters:
        - data: Historical price data
        - initial_capital: Starting capital
        - transaction_cost: Cost per trade
        """
        self.data = data
        self.initial_capital = initial_capital
        self.transaction_cost = transaction_cost
        
        self.reset()
    
    def reset(self):
        """
        Reset environment to initial state
        """
        self.capital = self.initial_capital
        self.position = 0  # Number of shares held
        self.time_step = 0
        self.portfolio_values = [self.initial_capital]
        
        return self.get_state()
    
    def get_state(self):
        """
        Get current state representation
        """
        # Current price
        price = self.data['close'].iloc[self.time_step]
        
        # Technical indicators
        returns = self.data['close'].pct_change().iloc[self.time_step - 10:self.time_step]
        volatility = returns.std() if len(returns) > 1 else 0
        
        # Portfolio state
        portfolio_value = self.capital + self.position * price
        
        state = np.array([
            price / 100,  # Normalized price
            self.position / 1000,  # Normalized position
            portfolio_value / self.initial_capital,  # Normalized portfolio
            volatility * 100,  # Volatility (%)
            self.time_step / len(self.data)  # Time progress
        ])
        
        return state
    
    def step(self, action):
        """
        Take action in environment
        """
        # Action: -1 (sell), 0 (hold), 1 (buy)
        price = self.data['close'].iloc[self.time_step]
        
        # Execute trade if action is not 0
        if action != 0:
            # Calculate trade size (10% of capital)
            trade_value = 0.1 * self.capital
            
            # Number of shares to trade
            shares_to_trade = trade_value / price
            
            # Apply transaction cost
            cost = trade_value * self.transaction_cost
            
            # Update position and capital
            self.position += action * shares_to_trade
            self.capital -= action * trade_value + cost
        
        # Move to next time step
        self.time_step += 1
        
        # Calculate reward
        new_price = self.data['close'].iloc[self.time_step - 1]
        new_portfolio_value = self.capital + self.position * new_price
        reward = new_portfolio_value - self.portfolio_values[-1]
        
        # Update portfolio value
        self.portfolio_values.append(new_portfolio_value)
        
        # Check if done
        done = self.time_step >= len(self.data) - 1
        
        next_state = self.get_state() if not done else None
        
        return next_state, reward, done

Certificate in FinTech Lesson 5.6: Reinforcement Learning in Finance: Q-Learning, Policy Gradients, and Execution Algorithms

Certificate in FinTech