如何基于gym-anytrading和stable-baselines3训练多回合强化学习智能体

1. 依赖安装 #

先安装所需库：

pip install gymnasium gym-anytrading stable-baselines3 numpy pandas

2. 自定义多回合交易环境 #

默认gym-anytrading环境是单回合模式，我们需要自定义环境实现每回合固定200时间步且支持1000个独立回合的逻辑。核心是通过reset方法随机/顺序切换200步数据切片：

import gymnasium as gym
import gym_anytrading
from gym_anytrading.envs import StocksEnv
from stable_baselines3 import PPO
import numpy as np
import pandas as pd

# 生成/加载历史交易数据（实际替换为真实行情数据）
def generate_trading_data(total_timesteps=200*1000):
    dates = pd.date_range(start='2020-01-01', periods=total_timesteps, freq='H')
    prices = np.cumsum(np.random.randn(total_timesteps)) + 100  # 模拟随机游走价格
    return pd.DataFrame({'Close': prices}, index=dates)

# 加载足够支撑1000个200步回合的数据
full_data = generate_trading_data()

# 自定义多回合交易环境
class MultiEpisodeTradeEnv(StocksEnv):
    def __init__(self, df, window_size=20, episode_steps=200):
        self.episode_steps = episode_steps
        self.total_data_len = len(df)
        # 校验数据量是否足够
        assert self.total_data_len >= self.episode_steps * 1000, "数据量不足，无法生成1000个独立回合"
        self.base_df = df.copy()  # 保存完整数据，避免重置时被修改
        super().__init__(df=df, window_size=window_size)
    
    def reset(self, seed=None, options=None):
        # 随机选择不重叠的200步数据切片（也可改为顺序切片）
        start_idx = np.random.randint(0, self.total_data_len - self.episode_steps)
        self.df = self.base_df.iloc[start_idx:start_idx+self.episode_steps]
        # 调用父类方法初始化环境状态
        return super().reset(seed=seed, options=options)

3. 训练多回合RL智能体 #

用PPO算法为例，设置训练总时间步覆盖1000个完整回合（200*1000=200000步）：

# 初始化自定义环境
env = MultiEpisodeTradeEnv(df=full_data, window_size=20, episode_steps=200)

# 初始化PPO智能体，n_steps设为单回合步数提升训练效率
model = PPO(
    "MlpPolicy", 
    env, 
    verbose=1, 
    learning_rate=3e-4, 
    n_steps=200,
    batch_size=64
)

# 开始训练
model.learn(total_timesteps=200000)

# 保存训练好的模型
model.save("multi_episode_trader_ppo")

4. 批量运行1000个独立回合 #

加载训练好的模型，循环运行1000次并统计结果：

# 加载模型
model = PPO.load("multi_episode_trader_ppo")

# 初始化统计变量
episode_rewards = []
episode_durations = []

# 运行1000个回合
for ep in range(1000):
    obs, info = env.reset()
    total_reward = 0
    steps = 0
    done = False
    
    while not done:
        action, _states = model.predict(obs, deterministic=True)  # 测试用确定性预测
        obs, reward, terminated, truncated, info = env.step(action)
        total_reward += reward
        steps += 1
        done = terminated or truncated
    
    episode_rewards.append(total_reward)
    episode_durations.append(steps)
    
    # 每100回合输出一次中间结果
    if (ep+1) % 100 == 0:
        print(f"完成第{ep+1}回合，最近100回合平均奖励：{np.mean(episode_rewards[-100:]):.2f}")

# 输出最终统计
print(f"\n1000回合统计结果：")
print(f"平均奖励：{np.mean(episode_rewards):.2f}，奖励标准差：{np.std(episode_rewards):.2f}")
print(f"平均回合步数：{np.mean(episode_durations):.2f}")

关键注意事项 #

• 数据独立性：确保每个回合的数据集不重叠（或随机采样无重叠切片），避免训练时的数据泄露
• 环境重置逻辑：自定义reset方法必须正确切换数据切片，保证每个回合的初始状态完全独立
• 训练参数匹配：将PPO的n_steps设为单回合步数（200），让算法的更新批次刚好覆盖一个完整回合，提升训练稳定性
• 可复现性：测试时使用deterministic=True，保证相同输入下输出一致

内容的提问来源于stack exchange，提问作者user9085964