Reinforcement Learning

Reinforcement Learning Patterns

Custom Gymnasium Environment

import gymnasium as gym
from gymnasium import spaces
import numpy as np

class TradingEnv(gym.Env):
    metadata = {"render_modes": ["human"]}

    def __init__(self, prices: np.ndarray, window: int = 20):
        super().__init__()
        self.prices = prices
        self.window = window
        # Observation: last `window` returns + position (0/1)
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf, shape=(window + 1,), dtype=np.float32
        )
        # Action: 0=hold, 1=buy, 2=sell
        self.action_space = spaces.Discrete(3)

    def reset(self, *, seed=None, options=None):
        super().reset(seed=seed)
        self.t = self.window
        self.position = 0
        self.portfolio = 1.0
        return self._obs(), {}

    def step(self, action):
        price_return = (self.prices[self.t] - self.prices[self.t - 1]) / self.prices[self.t - 1]
        reward = 0.0
        if action == 1 and self.position == 0:
            self.position = 1
        elif action == 2 and self.position == 1:
            self.position = 0
            reward = price_return - 0.001   # transaction cost
        if self.position == 1:
            reward = price_return
        self.t += 1
        done = self.t >= len(self.prices) - 1
        return self._obs(), reward, done, False, {}

    def _obs(self):
        returns = np.diff(self.prices[self.t - self.window:self.t + 1]) / self.prices[self.t - self.window:self.t]
        return np.append(returns, self.position).astype(np.float32)

Stable-Baselines3 Training

from stable_baselines3 import PPO, DQN, SAC
from stable_baselines3.common.env_util import make_vec_env
from stable_baselines3.common.callbacks import EvalCallback, StopTrainingOnRewardThreshold

# Vectorized envs for parallel collection
vec_env = make_vec_env(lambda: TradingEnv(prices), n_envs=4)

model = PPO(
    policy="MlpPolicy",
    env=vec_env,
    learning_rate=3e-4,
    n_steps=2048,
    batch_size=64,
    n_epochs=10,
    gamma=0.99,
    gae_lambda=0.95,
    clip_range=0.2,
    ent_coef=0.01,        # encourage exploration
    tensorboard_log="tb_logs/",
    verbose=1,
)

eval_env = TradingEnv(val_prices)
stop_callback = StopTrainingOnRewardThreshold(reward_threshold=0.5, verbose=1)
eval_callback = EvalCallback(
    eval_env,
    callback_on_new_best=stop_callback,
    eval_freq=10_000,
    n_eval_episodes=20,
    best_model_save_path="models/",
)

model.learn(total_timesteps=1_000_000, callback=eval_callback)
model.save("trading_ppo")

Reward Shaping

class ShapedTradingEnv(TradingEnv):
    def step(self, action):
        obs, reward, done, truncated, info = super().step(action)
        # Potential-based shaping: F(s,a,s') = gamma*phi(s') - phi(s)
        phi_next = self.portfolio - 1.0
        phi_curr = getattr(self, "_phi_prev", 0.0)
        shaping = 0.99 * phi_next - phi_curr
        self._phi_prev = phi_next
        return obs, reward + 0.1 * shaping, done, truncated, info

Evaluation & TensorBoard Logging

from stable_baselines3.common.evaluation import evaluate_policy

mean_reward, std_reward = evaluate_policy(
    model, eval_env, n_eval_episodes=100, deterministic=True
)
print(f"Mean reward: {mean_reward:.2f} ± {std_reward:.2f}")

# Launch TensorBoard:
# tensorboard --logdir tb_logs/

Algorithm Selection Guide

Key Patterns

• Always normalize observations — use VecNormalize wrapper from SB3
• Entropy coefficient (ent_coef) prevents premature convergence
• Use n_envs=4-8 parallel envs for PPO; single env for SAC/DQN
• Reward scale matters: keep rewards in [-1, 1] range for stable training
• Evaluate with deterministic=True but train with stochastic policy