1. Learning Objectives

By the end of this lesson, you will be able to:

  • Understand the mathematical formulation of Markov Decision Processes (MDPs) and their application to trading.

  • Derive the Bellman equations for state-value and action-value functions.

  • Implement Q-learning and Deep Q-Networks (DQN) for discrete trading actions.

  • Implement Policy Gradient methods (REINFORCE, PPO) for continuous action spaces.

  • Design reward functions that align with financial objectives (Sharpe ratio, risk-adjusted return).

  • Apply reinforcement learning to optimal execution (VWAP, TWAP) and market making.

  • Understand the challenges of reinforcement learning in finance: non-stationarity, partial observability, and transaction costs.

  • Implement a complete RL trading agent with experience replay and target networks.

2. Markov Decision Processes (MDPs) – The Mathematical Foundation

An MDP is a mathematical framework for sequential decision-making under uncertainty. Trading is a natural MDP: states (market conditions), actions (buy/sell/hold), rewards (profit/loss), and transitions (price movements).

2.1 MDP Definition
An MDP is defined by the tuple (S, A, P, R, γ):

  • S: Set of states (market features, portfolio composition).

  • A: Set of actions (buy, sell, hold, or continuous quantities).

  • P(s' | s, a): Transition probability to state s' from state s after action a.

  • R(s, a): Reward function (profit, Sharpe ratio, risk-adjusted return).

  • γ ∈ [0, 1]: Discount factor (weights future rewards).

2.2 The Policy
A policy π(a | s) is a distribution over actions given a state. The goal is to find the optimal policy π* that maximises the expected cumulative discounted reward.

2.3 The Bellman Equations
State-Value Function (V): The expected cumulative reward starting from state s and following policy π.
V^π(s) = E_π[ Σ_{t=0}^{∞} γ^t R_t | S_0 = s ]

Action-Value Function (Q): The expected cumulative reward starting from state s, taking action a, and then following policy π.
Q^π(s, a) = E_π[ Σ_{t=0}^{∞} γ^t R_t | S_0 = s, A_0 = a ]

Bellman Expectation Equations:
V^π(s) = Σ_a π(a | s) [ R(s, a) + γ Σ_{s'} P(s' | s, a) V^π(s') ]
Q^π(s, a) = R(s, a) + γ Σ_{s'} P(s' | s, a) Σ_{a'} π(a' | s') Q^π(s', a')

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

3. Q-Learning – Off-Policy Temporal Difference Learning

Q-learning learns the optimal action-value function Q* directly, without requiring a transition model P.

3.1 The Q-Learning Update Rule
Q(S_t, A_t) ← Q(S_t, A_t) + α [ R_{t+1} + γ max_a Q(S_{t+1}, a) - Q(S_t, A_t) ]
where α is the learning rate.

3.2 Tabular Q-Learning Implementation

text
class TabularQLearning:
    def __init__(self, n_states, n_actions, learning_rate=0.1, discount=0.95, epsilon=0.1):
        self.n_states = n_states
        self.n_actions = n_actions
        self.lr = learning_rate
        self.gamma = discount
        self.epsilon = epsilon  # Exploration rate

        # Initialise Q-table
        self.Q = np.zeros((n_states, n_actions))

    def get_action(self, state):
        """
        Epsilon-greedy action selection.
        """
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)  # Explore
        else:
            return np.argmax(self.Q[state, :])  # Exploit

    def update(self, state, action, reward, next_state, done):
        """
        Update Q-value using Q-learning update rule.
        """
        best_next_action = np.argmax(self.Q[next_state, :])
        td_target = reward + self.gamma * self.Q[next_state, best_next_action]
        td_error = td_target - self.Q[state, action]
        self.Q[state, action] += self.lr * td_error

3.3 Q-Learning for Trading (Discrete Actions)

text
class TradingQLearning:
    def __init__(self, n_state_features, n_actions=3, lr=0.01, gamma=0.95, epsilon=0.1):
        self.n_actions = n_actions  # 0=Hold, 1=Buy, 2=Sell
        self.lr = lr
        self.gamma = gamma
        self.epsilon = epsilon

        # Use a neural network for Q-learning (DQN) for continuous states
        self.q_network = self.build_network(n_state_features, n_actions)
        self.target_network = self.build_network(n_state_features, n_actions)
        self.target_network.load_state_dict(self.q_network.state_dict())

    def build_network(self, input_dim, output_dim):
        return nn.Sequential(
            nn.Linear(input_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, output_dim)
        )

    def get_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)
        else:
            with torch.no_grad():
                q_values = self.q_network(torch.FloatTensor(state).unsqueeze(0))
                return torch.argmax(q_values).item()

4. Deep Q-Networks (DQN) – Scaling Q-Learning with Neural Networks

DQN uses a neural network as a function approximator for Q(s, a). It addresses the instability of non-linear function approximation with two key innovations: Experience Replay and Target Networks.

4.1 Experience Replay
Store transitions (s, a, r, s', done) in a replay buffer. Sample random mini-batches for training. This breaks the temporal correlation between consecutive samples.

4.2 Target Network
Use a separate target network Q_target to compute the TD target. The target network is updated periodically (every N steps) from the online network. This stabilises training.

4.3 DQN Implementation

text
import random
from collections import deque

class ReplayBuffer:
    def __init__(self, capacity=100000):
        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)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (np.array(states), np.array(actions), np.array(rewards),
                np.array(next_states), np.array(dones))

    def __len__(self):
        return len(self.buffer)

class DQNAgent:
    def __init__(self, state_dim, action_dim, lr=1e-4, gamma=0.99, epsilon=1.0,
                 epsilon_decay=0.995, epsilon_min=0.01, batch_size=64):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.epsilon_min = epsilon_min
        self.batch_size = batch_size

        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

        # Networks
        self.q_network = self.build_network().to(self.device)
        self.target_network = self.build_network().to(self.device)
        self.target_network.load_state_dict(self.q_network.state_dict())
        self.target_network.eval()

        self.optimizer = optim.Adam(self.q_network.parameters(), lr=lr)
        self.replay_buffer = ReplayBuffer()

    def build_network(self):
        return nn.Sequential(
            nn.Linear(self.state_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, self.action_dim)
        )

    def get_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.action_dim)

        state = torch.FloatTensor(state).unsqueeze(0).to(self.device)
        with torch.no_grad():
            q_values = self.q_network(state)
        return torch.argmax(q_values).item()

    def update(self):
        if len(self.replay_buffer) < self.batch_size:
            return None

        # Sample batch
        states, actions, rewards, next_states, dones = self.replay_buffer.sample(self.batch_size)

        # Convert to tensors
        states = torch.FloatTensor(states).to(self.device)
        actions = torch.LongTensor(actions).to(self.device)
        rewards = torch.FloatTensor(rewards).to(self.device)
        next_states = torch.FloatTensor(next_states).to(self.device)
        dones = torch.FloatTensor(dones).to(self.device)

        # Current Q-values
        current_q = self.q_network(states).gather(1, actions.unsqueeze(1)).squeeze(1)

        # Target Q-values (using target network)
        with torch.no_grad():
            next_q = self.target_network(next_states).max(1)[0]
            target_q = rewards + (1 - dones) * self.gamma * next_q

        # Loss
        loss = nn.MSELoss()(current_q, target_q)

        # Backpropagation
        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)

        return loss.item()

    def update_target_network(self):
        self.target_network.load_state_dict(self.q_network.state_dict())

5. Policy Gradient Methods – Directly Optimising the Policy

Policy gradient methods directly optimise the policy π(a | s) by climbing the gradient of the expected cumulative reward.

5.1 The Policy Gradient Theorem
∇J(θ) = E_π[ Σ_{t=0}^{∞} ∇_θ ln π_θ(a_t | s_t) * Q^π(s_t, a_t) ]
The gradient of the expected reward is the expected sum of the gradient of the log policy times the action-value function.

5.2 REINFORCE (Monte Carlo Policy Gradient)
REINFORCE uses the complete return G_t as an unbiased estimate of Q^π(s_t, a_t).
∇J(θ) ≈ (1/N) Σ_{i=1}^{N} Σ_{t=0}^{T-1} ∇_θ ln π_θ(a_{i,t} | s_{i,t}) * G_{i,t}

5.3 REINFORCE Implementation

text
class REINFORCEAgent:
    def __init__(self, state_dim, action_dim, hidden_dim=128, lr=1e-3, gamma=0.99):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.gamma = gamma

        self.policy_network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )
        self.optimizer = optim.Adam(self.policy_network.parameters(), lr=lr)

    def get_action(self, state):
        state = torch.FloatTensor(state).unsqueeze(0)
        probs = self.policy_network(state)
        action = torch.multinomial(probs, 1).item()
        return action

    def compute_returns(self, rewards):
        """
        Compute discounted returns (Monte Carlo returns).
        """
        returns = []
        G = 0
        for r in reversed(rewards):
            G = r + self.gamma * G
            returns.insert(0, G)
        return np.array(returns)

    def update(self, states, actions, rewards):
        """
        Update policy using REINFORCE.
        """
        states = torch.FloatTensor(np.array(states))
        actions = torch.LongTensor(actions)
        returns = torch.FloatTensor(self.compute_returns(rewards))

        # Normalise returns (reduces variance)
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)

        # Compute policy gradient
        probs = self.policy_network(states)
        log_probs = torch.log(probs.gather(1, actions.unsqueeze(1))).squeeze(1)

        loss = -(log_probs * returns).mean()

        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        return loss.item()

6. Proximal Policy Optimisation (PPO) – The Industry Standard

PPO is a state-of-the-art policy gradient method that constrains policy updates to prevent destructive large changes.

6.1 The PPO Objective
L^{CLIP}(θ) = E_t[ min( r_t(θ) * A_t, clip(r_t(θ), 1-ε, 1+ε) * A_t ) ]
where:

  • r_t(θ) = π_θ(a_t | s_t) / π_{θ_old}(a_t | s_t) is the probability ratio.

  • A_t is the advantage function (how much better is this action than average).

  • ε is the clipping hyperparameter (typically 0.1-0.2).

6.2 Advantage Estimation (GAE)
A_t = δ_t + γλ δ_{t+1} + (γλ)^2 δ_{t+2} + ...
δ_t = R_t + γ V(s_{t+1}) - V(s_t)
where λ is the GAE parameter (typically 0.95).

6.3 PPO Implementation (Simplified)

text
class PPOAgent:
    def __init__(self, state_dim, action_dim, hidden_dim=128, lr=3e-4,
                 gamma=0.99, lam=0.95, epsilon=0.2, epochs=10):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.gamma = gamma
        self.lam = lam
        self.epsilon = epsilon
        self.epochs = epochs

        # Actor (policy) and Critic (value function)
        self.actor = self.build_actor(state_dim, action_dim, hidden_dim)
        self.critic = self.build_critic(state_dim, hidden_dim)
        self.optimizer = optim.Adam(
            list(self.actor.parameters()) + list(self.critic.parameters()),
            lr=lr
        )

    def build_actor(self, state_dim, action_dim, hidden_dim):
        return nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )

    def build_critic(self, state_dim, hidden_dim):
        return nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, 1)
        )

    def get_action(self, state):
        state = torch.FloatTensor(state).unsqueeze(0)
        probs = self.actor(state)
        action = torch.multinomial(probs, 1).item()
        log_prob = torch.log(probs[0, action])
        return action, log_prob.item()

    def compute_gae(self, rewards, values, dones):
        """
        Compute Generalised Advantage Estimation (GAE).
        """
        advantages = []
        gae = 0
        for t in reversed(range(len(rewards))):
            next_value = 0 if dones[t] else values[t+1]
            delta = rewards[t] + self.gamma * next_value - values[t]
            gae = delta + self.gamma * self.lam * (1 - dones[t]) * gae
            advantages.insert(0, gae)
        return np.array(advantages)

    def update(self, states, actions, log_probs, rewards, dones, values):
        """
        PPO update with clipping.
        """
        states = torch.FloatTensor(np.array(states))
        actions = torch.LongTensor(actions)
        old_log_probs = torch.FloatTensor(log_probs)

        # Compute advantages
        advantages = self.compute_gae(rewards, values, dones)
        advantages = torch.FloatTensor(advantages)

        # Normalise advantages
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

        # PPO update for multiple epochs
        for _ in range(self.epochs):
            # New log probabilities
            probs = self.actor(states)
            new_log_probs = torch.log(probs.gather(1, actions.unsqueeze(1))).squeeze(1)

            # Probability ratio
            ratio = torch.exp(new_log_probs - old_log_probs)

            # Surrogate loss
            surr1 = ratio * advantages
            surr2 = torch.clamp(ratio, 1 - self.epsilon, 1 + self.epsilon) * advantages
            actor_loss = -torch.min(surr1, surr2).mean()

            # Critic loss (value function)
            values_pred = self.critic(states).squeeze(1)
            critic_loss = nn.MSELoss()(values_pred, torch.FloatTensor(rewards))

            # Total loss
            loss = actor_loss + 0.5 * critic_loss

            self.optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(
                list(self.actor.parameters()) + list(self.critic.parameters()),
                0.5
            )
            self.optimizer.step()

        return loss.item()

7. Reward Engineering for Financial RL

The reward function is the most critical design choice. It must align with the financial objective.

7.1 Simple Profit/Loss Reward
R_t = P_{t+1} * (S_{t+1} - S_t) / S_t (position times price change)

7.2 Sharpe Ratio Reward

text
def sharpe_reward(returns, window=20):
    """
    Reward based on rolling Sharpe ratio.
    """
    if len(returns) < window:
        return np.mean(returns)

    recent_returns = returns[-window:]
    sharpe = np.mean(recent_returns) / (np.std(recent_returns) + 1e-8)
    return sharpe

7.3 Risk-Adjusted Reward

text
def risk_adjusted_reward(returns, risk_free=0.0, risk_penalty=1.0):
    """
    Reward = Return - risk_penalty * Risk (standard deviation)
    """
    ret = np.mean(returns)
    risk = np.std(returns)
    return ret - risk_penalty * risk

7.4 Sparse Reward – Only Reward at Episode End

text
def terminal_reward(portfolio_value, initial_capital):
    """
    Only reward at the end of the episode.
    """
    return (portfolio_value / initial_capital) - 1

7.5 Multi-Objective Reward

text
def multi_objective_reward(portfolio_value, initial_capital, holdings, returns):
    """
    Combine multiple objectives: return, drawdown, turnover.
    """
    profit = (portfolio_value / initial_capital) - 1

    # Drawdown penalty
    peak_value = max(initial_capital, portfolio_value)
    drawdown = (peak_value - portfolio_value) / peak_value
    drawdown_penalty = -5 * drawdown

    # Turnover penalty (transaction costs)
    turnover = np.abs(np.diff(holdings)).sum()
    turnover_penalty = -0.001 * turnover  # 10 bps per trade

    return profit + drawdown_penalty + turnover_penalty

8. Trading Environment Design

8.1 Custom Gym Environment

text
import gym
from gym import spaces

class TradingEnv(gym.Env):
    def __init__(self, data, initial_capital=100000, transaction_cost=0.001, max_position=1.0):
        super(TradingEnv, self).__init__()

        self.data = data
        self.initial_capital = initial_capital
        self.transaction_cost = transaction_cost
        self.max_position = max_position

        # Action space: continuous (position size) or discrete (buy/sell/hold)
        # Here we use discrete: -1 (sell), 0 (hold), 1 (buy)
        self.action_space = spaces.Discrete(3)

        # Observation space: price features, portfolio state
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf, shape=(len(data.columns) + 2,)
        )

        self.reset()

    def reset(self):
        self.current_step = 0
        self.capital = self.initial_capital
        self.position = 0.0
        self.portfolio_value = self.initial_capital
        self.returns = []
        return self._get_observation()

    def _get_observation(self):
        """
        Construct observation vector.
        """
        features = self.data.iloc[self.current_step].values
        portfolio_state = np.array([self.position, self.portfolio_value / self.initial_capital])
        return np.concatenate([features, portfolio_state])

    def step(self, action):
        """
        Execute one step in the environment.
        """
        # Current price
        current_price = self.data.iloc[self.current_step]['Close']

        # Next price
        next_step = min(self.current_step + 1, len(self.data) - 1)
        next_price = self.data.iloc[next_step]['Close']

        # Convert action: 0=sell, 1=hold, 2=buy
        # In this simple version, action determines position change
        action_scale = {0: -0.2, 1: 0, 2: 0.2}  # Sell 20%, Hold, Buy 20%
        target_position = np.clip(self.position + action_scale[action], -1, 1)

        # Transaction cost
        trade_size = abs(target_position - self.position)
        cost = trade_size * self.transaction_cost * self.portfolio_value

        # Execute trade
        self.position = target_position

        # Update portfolio value
        price_change = (next_price / current_price) - 1
        self.portfolio_value = self.portfolio_value * (1 + self.position * price_change) - cost

        # Reward
        reward = (self.portfolio_value - self.initial_capital) / self.initial_capital

        # Update step
        self.current_step = next_step
        done = self.current_step >= len(self.data) - 1

        # Additional reward: Sharpe ratio at the end
        if done:
            reward += self._compute_sharpe_reward()

        return self._get_observation(), reward, done, {}

    def _compute_sharpe_reward(self):
        """
        Compute Sharpe ratio reward at the end of the episode.
        """
        if len(self.returns) > 0:
            return np.mean(self.returns) / (np.std(self.returns) + 1e-8)
        return 0

9. Challenges and Solutions in Financial RL

 
 
Challenge Description Solution
Non-Stationarity Market dynamics change over time. Use online learning, continual adaptation, regime detection.
Partial Observability We cannot observe all market drivers. Use recurrent policies (LSTM policies) that maintain internal state.
Transaction Costs Trading costs erode profits. Include transaction costs in the reward function.
Sparse Rewards Profit only at the end of trades. Use dense rewards (paper profits, risk metrics).
Sample Inefficiency RL requires many episodes. Use offline RL with historical data (batch RL).
Overfitting RL agents can overfit to historical patterns. Use robust cross-validation, adversarial training.
Exploration Need to explore to find profitable strategies. Use epsilon-greedy, entropy bonus in policy gradient.

10. Summary for the AI Practitioner

  1. MDPs formalise the trading problem. States are market features; actions are trading decisions; rewards are financial metrics.

  2. Q-Learning is for discrete action spaces (buy/sell/hold). Use DQN with Experience Replay and Target Networks for continuous states.

  3. Policy Gradient methods (REINFORCE, PPO) are for continuous action spaces (position sizing).

  4. PPO is the industry standard. It clips policy updates to prevent destructive large changes.

  5. Reward engineering is critical. Use risk-adjusted rewards (Sharpe, drawdown penalty) not just raw profit.

  6. Environmental design should include transaction costs, slippage, and realistic market impact.

  7. Challenges: Non-stationarity requires continual learning and regime detection. Sample inefficiency requires offline RL or large replay buffers.

  8. Production deployment: RL agents must be monitored for drift and retrained periodically.

 

In Lessons 3.7 and 3.8, we will cover Model Interpretability and Explainability (SHAP, LIME, Integrated Gradients) and Production Pipelines and Deployment (Kubernetes, CI/CD, model monitoring, and regulatory compliance).

 
 
PreviousNext