Compare agents1.py

[vicks4u]

""" Compare PPO, A2C, and DQN on MT5 trading environment.

• Each agent is trained separately
• Results (mean reward) are logged """

import time import numpy as np import pandas as pd import gym from gym import spaces import MetaTrader5 as mt5

from stable_baselines3 import PPO, A2C, DQN from stable_baselines3.common.vec_env import DummyVecEnv from stable_baselines3.common.evaluation import evaluate_policy

-------------------------

CONFIG

-------------------------

SYMBOL = "EURUSD" TIMEFRAME = mt5.TIMEFRAME_M5 LOOKBACK = 50 TRAIN_TIMESTEPS = 20000 N_BARS = 5000 SEED = 42

-------------------------

-------------------------

MT5 Connection

-------------------------

def mt5_connect(): if not mt5.initialize(): raise RuntimeError(f"MT5 init failed: {mt5.last_error()}") if not mt5.symbol_select(SYMBOL, True): raise RuntimeError(f"Could not select {SYMBOL}")

def mt5_shutdown(): mt5.shutdown()

def fetch_bars(symbol, timeframe, n_bars): rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n_bars) if rates is None: raise RuntimeError(f"Failed to fetch data: {mt5.last_error()}") df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') return df

-------------------------

Custom Gym Env

-------------------------

class MT5TradingEnv(gym.Env): def init(self, df, lookback=LOOKBACK): super().init() self.df = df.reset_index(drop=True) self.lookback = lookback self.ptr = lookback self.position = 0 self.entry_price = 0 self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(lookback+1,), dtype=np.float32) self.action_space = spaces.Discrete(3) # 0=hold, 1=buy, 2=sell

def _get_obs(self):
    closes = self.df.loc[self.ptr-self.lookback:self.ptr-1, "close"].values.astype(np.float32)
    norm = closes / (closes[-1] + 1e-9) - 1.0
    return np.concatenate([norm, [float(self.position)]], axis=0)

def reset(self):
    self.ptr = self.lookback
    self.position = 0
    self.entry_price = 0
    return self._get_obs()

def step(self, action):
    done, reward = False, 0
    price = float(self.df.loc[self.ptr, "close"])
    if action == 1:  # buy
        if self.position == 0:
            self.position, self.entry_price = 1, price
        elif self.position == -1:
            reward += (self.entry_price - price)
            self.position, self.entry_price = 1, price
    elif action == 2:  # sell
        if self.position == 0:
            self.position, self.entry_price = -1, price
        elif self.position == 1:
            reward += (price - self.entry_price)
            self.position, self.entry_price = -1, price

    self.ptr += 1
    if self.ptr >= len(self.df):
        done = True
    else:
        next_price = float(self.df.loc[self.ptr, "close"])
        if self.position == 1:
            reward += (next_price - self.entry_price) * 0.1
        elif self.position == -1:
            reward += (self.entry_price - next_price) * 0.1

    obs = self._get_obs() if not done else np.zeros(self.observation_space.shape, dtype=np.float32)
    return obs, float(reward), done, {}

-------------------------

Training & Evaluation

-------------------------

def run_comparison(): mt5_connect() df = fetch_bars(SYMBOL, TIMEFRAME, N_BARS) mt5_shutdown()

results = {}
agents = {
    "PPO": PPO,
    "A2C": A2C,
    "DQN": DQN
}

for name, algo in agents.items():
    print(f"\n=== Training {name} ===")
    env = DummyVecEnv([lambda: MT5TradingEnv(df, lookback=LOOKBACK)])
    model = algo("MlpPolicy", env, verbose=0, seed=SEED)
    model.learn(total_timesteps=TRAIN_TIMESTEPS)
    mean_reward, std_reward = evaluate_policy(model, env, n_eval_episodes=5)
    results[name] = (mean_reward, std_reward)
    print(f"{name} → mean reward: {mean_reward:.2f}, std: {std_reward:.2f}")

print("\n=== Summary ===")
for k, v in results.items():
    print(f"{k}: mean {v[0]:.2f}, std {v[1]:.2f}")

if name == "main": run_comparison()