René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Compare agents1.py","articleBody":"\"\"\"\nCompare PPO, A2C, and DQN on MT5 trading environment.\n- Each agent is trained separately\n- Results (mean reward) are logged\n\"\"\"\n\nimport time\nimport numpy as np\nimport pandas as pd\nimport gym\nfrom gym import spaces\nimport MetaTrader5 as mt5\n\nfrom stable_baselines3 import PPO, A2C, DQN\nfrom stable_baselines3.common.vec_env import DummyVecEnv\nfrom stable_baselines3.common.evaluation import evaluate_policy\n\n# -------------------------\n# CONFIG\n# -------------------------\nSYMBOL = \"EURUSD\"\nTIMEFRAME = mt5.TIMEFRAME_M5\nLOOKBACK = 50\nTRAIN_TIMESTEPS = 20000\nN_BARS = 5000\nSEED = 42\n# -------------------------\n\n# -------------------------\n# MT5 Connection\n# -------------------------\ndef mt5_connect():\n    if not mt5.initialize():\n        raise RuntimeError(f\"MT5 init failed: {mt5.last_error()}\")\n    if not mt5.symbol_select(SYMBOL, True):\n        raise RuntimeError(f\"Could not select {SYMBOL}\")\n\ndef mt5_shutdown():\n    mt5.shutdown()\n\ndef fetch_bars(symbol, timeframe, n_bars):\n    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n_bars)\n    if rates is None:\n        raise RuntimeError(f\"Failed to fetch data: {mt5.last_error()}\")\n    df = pd.DataFrame(rates)\n    df['time'] = pd.to_datetime(df['time'], unit='s')\n    return df\n\n# -------------------------\n# Custom Gym Env\n# -------------------------\nclass MT5TradingEnv(gym.Env):\n    def __init__(self, df, lookback=LOOKBACK):\n        super().__init__()\n        self.df = df.reset_index(drop=True)\n        self.lookback = lookback\n        self.ptr = lookback\n        self.position = 0\n        self.entry_price = 0\n        self.observation_space = spaces.Box(low=-np.inf, high=np.inf,\n                                            shape=(lookback+1,), dtype=np.float32)\n        self.action_space = spaces.Discrete(3)  # 0=hold, 1=buy, 2=sell\n\n    def _get_obs(self):\n        closes = self.df.loc[self.ptr-self.lookback:self.ptr-1, \"close\"].values.astype(np.float32)\n        norm = closes / (closes[-1] + 1e-9) - 1.0\n        return np.concatenate([norm, [float(self.position)]], axis=0)\n\n    def reset(self):\n        self.ptr = self.lookback\n        self.position = 0\n        self.entry_price = 0\n        return self._get_obs()\n\n    def step(self, action):\n        done, reward = False, 0\n        price = float(self.df.loc[self.ptr, \"close\"])\n        if action == 1:  # buy\n            if self.position == 0:\n                self.position, self.entry_price = 1, price\n            elif self.position == -1:\n                reward += (self.entry_price - price)\n                self.position, self.entry_price = 1, price\n        elif action == 2:  # sell\n            if self.position == 0:\n                self.position, self.entry_price = -1, price\n            elif self.position == 1:\n                reward += (price - self.entry_price)\n                self.position, self.entry_price = -1, price\n\n        self.ptr += 1\n        if self.ptr \u003e= len(self.df):\n            done = True\n        else:\n            next_price = float(self.df.loc[self.ptr, \"close\"])\n            if self.position == 1:\n                reward += (next_price - self.entry_price) * 0.1\n            elif self.position == -1:\n                reward += (self.entry_price - next_price) * 0.1\n\n        obs = self._get_obs() if not done else np.zeros(self.observation_space.shape, dtype=np.float32)\n        return obs, float(reward), done, {}\n\n# -------------------------\n# Training \u0026 Evaluation\n# -------------------------\ndef run_comparison():\n    mt5_connect()\n    df = fetch_bars(SYMBOL, TIMEFRAME, N_BARS)\n    mt5_shutdown()\n\n    results = {}\n    agents = {\n        \"PPO\": PPO,\n        \"A2C\": A2C,\n        \"DQN\": DQN\n    }\n\n    for name, algo in agents.items():\n        print(f\"\\n=== Training {name} ===\")\n        env = DummyVecEnv([lambda: MT5TradingEnv(df, lookback=LOOKBACK)])\n        model = algo(\"MlpPolicy\", env, verbose=0, seed=SEED)\n        model.learn(total_timesteps=TRAIN_TIMESTEPS)\n        mean_reward, std_reward = evaluate_policy(model, env, n_eval_episodes=5)\n        results[name] = (mean_reward, std_reward)\n        print(f\"{name} → mean reward: {mean_reward:.2f}, std: {std_reward:.2f}\")\n\n    print(\"\\n=== Summary ===\")\n    for k, v in results.items():\n        print(f\"{k}: mean {v[0]:.2f}, std {v[1]:.2f}\")\n\nif __name__ == \"__main__\":\n    run_comparison()","author":{"url":"https://github.com/vicks4u","@type":"Person","name":"vicks4u"},"datePublished":"2025-09-16T01:34:59.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":0},"url":"https://github.com/7/PastebinPython/issues/7"}