This tutorial explores advanced applications of Stable-Baselines3 In reinforcement learning. Our custom-built trading environment integrates multiple algorithms, such as A2C and PPO, while we develop our training callbacks to track performance. We train, analyze, and compare agent performance as we move forward. This allows us to evaluate algorithmic efficiency and learning curves. Visit the FULL CODES here.
!pip Install Stable-Baselines3[extra] gymnasium pygame
Numpy can be imported as a np
Gymnasium imported as gym
Import gymnasium spaces
Import matplotlib.pyplot into plt
From stable_baselines3 you can import A2C PPO DQN SAC
from stable_baselines3.common.env_checker import check_env
from stable_baselines3.common.callbacks import BaseCallback
from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize
from stable_baselines3.common.evaluation import evaluate_policy
from stable_baselines3.common.monitor import Monitor
Buy a torch
Class TradingEnv (gym.Env).
def __init__(self, max_steps=200):
Supermarkets are a great way to buy goods and services.().__init__()
self.max_steps = max_steps
self.action_space = spaces.Discrete(3)
self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(5,), dtype=np.float32)
self.reset()
Def reset(self; seed = None, options=None);
Supermarkets are a great way to buy goods and services.().reset(seed=seed)
self.current_step = 0
self.balance = 1000.0
self.shares = 0
self.price = 100.0
self.price_history = [self.price]
Self-returning Obs(), {}
def _get_obs(self):
price_trend = np.mean(self.price_history[-5:]) if len(self.price_history) >= 5 else self.price
Return np.array ([
self.balance / 1000.0,
self.shares / 10.0,
self.price / 100.0,
price_trend / 100.0,
self.current_step / self.max_steps
], dtype=np.float32)
def step(self, action):
self.current_step += 1
trend = 0.001 * np.sin(self.current_step / 20)
self.price *= (1 + trend + np.random.normal(0, 0.02))
self.price = np.clip(self.price, 50, 200)
self.price_history.append(self.price)
Rewards = 0.
if action == 1 and self.balance >= self.price:
shares_to_buy = int(self.balance / self.price)
cost = shares_to_buy * self.price
Self.balance = cost
self.shares += shares_to_buy
Rewards = 0.01
elif action == 2 and self.shares > 0:
revenue = self.shares * self.price
Self-balance = revenue
self.shares = 0
Rewards = 0.01
portfolio_value = self.balance + self.shares * self.price
reward += (portfolio_value - 1000) / 1000
terminated = self.current_step >= self.max_steps
Truncated = False
Self-returning Obs(), reward, terminated, truncated, {"portfolio": portfolio_value}
def render(self):
print(f"Step: {self.current_step}, Balance: ${self.balance:.2f}, Shares: {self.shares}, Price: ${self.price:.2f}")TradingEnvs are created to teach agents how to take buy, hold, and sell decisions using simulated price changes. The reward system is implemented, the action and observation spaces are defined, as well as the market environment with its fluctuating noise and trends. Click here to see the FULL CODES here.
class ProgressCallback(BaseCallback):
def __init__(self, check_freq=1000, verbose=1):
Supermarkets are a great way to buy goods and services.().__init__(verbose)
self.check_freq = check_freq
self.rewards = []
def _on_step(self):
if self.n_calls % self.check_freq == 0:
mean_reward = np.mean([ep_info["r"] for ep_info in self.model.ep_info_buffer])
self.rewards.append(mean_reward)
if self.verbose:
print(f"Steps: {self.n_calls}, Mean Reward: {mean_reward:.2f}")
Return True
print("=" * 60)
print("Setting up custom trading environment...")
TradingEnv()
check_env(env, warn=True)
print("✓ Environment validation passed!")
env = monitor(env).
vec_env = DummyVecEnv([lambda: env])
vec_env = VecNormalize(vec_env, norm_obs=True, norm_reward=True)We create a ProgressCallback here to track training progress at regular intervals and log mean rewards. Then, we validate the custom environment with Stable-Baselines3’s built-in Checker. We wrap it to monitor and normalize it, then prepare it for multiple algorithm training. Click here to view theThe algorithm = FULL CODES here.
print("n" + "=" * 60)
print("Training multiple RL algorithms...")
algorithms = {
"PPO": PPO("MlpPolicy", vec_env, verbose=0, learning_rate=3e-4, n_steps=2048),
"A2C": A2C("MlpPolicy", vec_env, verbose=0, learning_rate=7e-4),
}
results = {}
Name model item in algorithmic items():
print(f"nTraining {name}...")
callback = ProgressCallback(check_freq=2000, verbose=0)
model.learn(total_timesteps=50000, callback=callback, progress_bar=True)
The following are results of the search:[name] = {"model": model, "rewards": callback.rewards}
print(f"✓ {name} training complete!")
print("n" + "=" * 60)
print("Evaluating trained models...")
eval_env = Monitor(TradingEnv())
Results.items for Name():
mean_reward, std_reward = evaluate_policy(data["model"], eval_env, n_eval_episodes=20, deterministic=True)
The following are results of the search:[name]["eval_mean"] = mean_reward
The following are results of the search:[name]["eval_std"] = std_reward
print(f"{name}: Mean Reward = {mean_reward:.2f} +/- {std_reward:.2f}")On our trading platform, we train and test two different reinforcement-learning algorithms: PPO and A2C. The agents’ performance is logged, mean rewards are captured, and the efficiency of each agent in learning profitable trading strategies by exploring and exploiting consistently and repeatedly, through exploration, is compared. Visit the FULL CODES here.
print("n" + "=" * 60)
print("Generating visualizations...")
Figure, Axis = plt.subplots(2), figsize=(14.10)
ax = Axes[0, 0]
Name, data and results.items():
ax.plot(data["rewards"], label=name, linewidth=2)
ax.set_xlabel("Training Checkpoints (x1000 steps)")
ax.set_ylabel("Mean Episode Reward")
ax.set_title("Training Progress Comparison")
ax.legend()
ax.grid(True, alpha=0.3)
Axes[0, 1]
List(results.keys) = names())
means = [results[n]["eval_mean"] For n in name]
The stds are = [results[n]["eval_std"] For n in name]
ax.bar(names, means, yerr=stds, capsize=10, alpha=0.7, color=['#1f77b4', '#ff7f0e'])
ax.set_ylabel("Mean Reward")
ax.set_title("Evaluation Performance (20 episodes)")
ax.grid(True, alpha=0.3, axis="y")
Axes = ax[1, 0]
best_model = max(results.items()", key=lambda" x:[1]["eval_mean"])[1]["model"]
obs = eval_env.reset()[0]
portfolio_values = [1000]
For _, in the range (200):
action, _ = best_model.predict(obs, deterministic=True)
obs, reward, done, truncated, info = eval_env.step(action)
portfolio_values.append(info.get("portfolio", portfolio_values[-1]))
if done:
Breaking News
ax.plot(portfolio_values, linewidth=2, color="green")
ax.axhline(y=1000, color="red", linestyle="--", label="Initial Value")
ax.set_xlabel("Steps")
ax.set_ylabel("Portfolio Value ($)")
ax.set_title(f"Best Model ({max(results.items(), key=lambda x: x[1]['eval_mean'])[0]}) Episode")
ax.legend()
ax.grid(True, alpha=0.3)Visualize our results for training by plotting the learning curves, scores and trajectory of portfolios. The agent’s behavior is also analyzed to determine how it translates into the portfolio, helping us understand model behavior. See the FULL CODES here.
ax = Axes[1, 1]
obs = eval_env.reset()[0]
Action = []
For _, in the range (200):
action, _ = best_model.predict(obs, deterministic=True)
actions.append(action)
obs, _, done, truncated, _ = eval_env.step(action)
if done:
Breaking News
action_names = ['Hold', 'Buy', 'Sell']
action_counts = [actions.count(i) for i in range(3)]
ax.pie(action_counts, labels=action_names, autopct="%1.1f%%", startangle=90, colors=['#ff9999', '#66b3ff', '#99ff99'])
ax.set_title("Action Distribution (Best Model)")
plt.tight_layout()
plt.savefig('sb3_advanced_results.png', dpi=150, bbox_inches="tight")
print("✓ Visualizations saved as 'sb3_advanced_results.png'")
plt.show()
print("n" + "=" * 60)
print("Saving and loading models...")
best_name = max(results.items()", key=lambda" x:[1]["eval_mean"])[0]
best_model = results[best_name]["model"]
best_model.save(f"best_trading_model_{best_name}")
vec_env.save("vec_normalize.pkl")
loaded_model = PPO.load(f"best_trading_model_{best_name}")
print(f"✓ Best model ({best_name}) saved and loaded successfully!")
print("n" + "=" * 60)
print("TUTORIAL COMPLETE!")
print(f"Best performing algorithm: {best_name}")
print(f"Final evaluation score: {results[best_name]['eval_mean']:.2f}")
print("=" * 60)Finally, to fully understand trading behavior of our best agent we can visualize its action distribution. This allows us to save and reuse the model with the highest performance. The tutorial concludes with an overview of the performance results and the insights gained. We show how to load the model, verify the algorithm and confirm it.
As a conclusion, using Stable-Baselines3, we created, trained and then compared multiple reinforcement agents within a trading simulation. We can observe the way each algorithm responds to changes in market conditions, track their progress, and determine which strategy is most effective. This practical implementation helps to improve our understanding of RL and shows just how flexible, efficient and scalable Stable Baselines3 can prove for tasks that are complex and domain specific, such as financial modelling.
Take a look at the FULL CODES here. Please feel free to browse our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter Join our Facebook group! 100k+ ML SubReddit Subscribe now our Newsletter. Wait! Are you using Telegram? now you can join us on telegram as well.
Asif Razzaq serves as the CEO at Marktechpost Media Inc. As an entrepreneur, Asif has a passion for harnessing Artificial Intelligence to benefit society. Marktechpost was his most recent venture. This platform, which focuses on machine learning and deep-learning news, is accessible to a broad audience and offers a technical and unbiased coverage. This platform has over 2,000,000 monthly views which shows its popularity.
