René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Compare_agent.py","articleBody":"\"\"\"\nRL + MetaTrader5 trading bot template\n- Train with historical data (PPO from stable-baselines3)\n- Optionally execute trades via MetaTrader5 (set LIVE=True to enable)\nCAVEAT: This is an educational template. Backtest \u0026 paper-trade first.\n\"\"\"\n\nimport time\nimport numpy as np\nimport pandas as pd\nimport gym\nfrom gym import spaces\nimport MetaTrader5 as mt5\nfrom stable_baselines3 import PPO\nfrom stable_baselines3.common.vec_env import DummyVecEnv\nfrom stable_baselines3.common.callbacks import CheckpointCallback\n\n# -------------------------\n# USER CONFIG\n# -------------------------\nSYMBOL = \"EURUSD\"\nTIMEFRAME = mt5.TIMEFRAME_M5     # 5 minute bars\nLOOKBACK = 50                   # observation window (bars)\nSTART_POS = 0                   # for historical fetch offset\nLOT_SIZE = 0.01                 # trade lot size\nLIVE = False                    # \u003c-- Set to True only after full testing (demo account first!)\nMODEL_PATH = \"ppo_mt5_model\"\nTRAIN_TIMESTEPS = 20000         # adjust as you like\n# -------------------------\n\n# -------------------------\n# Helper: connect to MT5\n# -------------------------\ndef mt5_connect():\n    if not mt5.initialize():\n        raise RuntimeError(f\"MT5 initialize() failed, error={mt5.last_error()}\")\n    info = mt5.terminal_info()\n    if info is None:\n        raise RuntimeError(\"Failed to get terminal info after initialize()\")\n    print(\"MT5 terminal initialized:\", info.product)\n    # Ensure symbol is available\n    if not mt5.symbol_select(SYMBOL, True):\n        raise RuntimeError(f\"Failed to select symbol {SYMBOL}\")\n    return True\n\ndef mt5_shutdown():\n    mt5.shutdown()\n\n# -------------------------\n# Get historical OHLCV\n# -------------------------\ndef fetch_bars(symbol, timeframe, n_bars):\n    # copy_rates_from_pos returns numpy array with fields: time, open, high, low, close, tick_volume, ...\n    rates = mt5.copy_rates_from_pos(symbol, timeframe, START_POS, n_bars)\n    if rates is None:\n        raise RuntimeError(f\"Failed to fetch rates for {symbol}: {mt5.last_error()}\")\n    df = pd.DataFrame(rates)\n    df['time'] = pd.to_datetime(df['time'], unit='s')\n    return df\n\n# -------------------------\n# Simple trading Gym env\n# -------------------------\nclass MT5TradingEnv(gym.Env):\n    \"\"\"\n    Observation: last LOOKBACK closes normalized + current position (0/1/-1)\n    Actions: 0=hold, 1=buy (long), 2=sell (short/close long)\n    Reward: change in account equity approximated by price moves * position\n    NOTE: Simplified; this is a research template, not production-ready.\n    \"\"\"\n    def __init__(self, df: pd.DataFrame, lookback=LOOKBACK):\n        super(MT5TradingEnv, self).__init__()\n        self.df = df.reset_index(drop=True)\n        self.lookback = lookback\n        self.ptr = lookback  # current index in df\n        self.position = 0    # -1 short, 0 flat, 1 long\n        self.entry_price = 0.0\n        # Observations: lookback closes (normalized) + position\n        self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(lookback + 1,), dtype=np.float32)\n        # Actions: hold(0), buy(1), sell(2)\n        self.action_space = spaces.Discrete(3)\n\n    def _get_obs(self):\n        closes = self.df.loc[self.ptr - self.lookback:self.ptr - 1, \"close\"].values.astype(np.float32)\n        # normalize closes by dividing by last close\n        norm = closes / (closes[-1] + 1e-9) - 1.0\n        obs = np.concatenate([norm, np.array([float(self.position)])], axis=0)\n        return obs\n\n    def reset(self):\n        self.ptr = self.lookback\n        self.position = 0\n        self.entry_price = 0.0\n        return self._get_obs()\n\n    def step(self, action):\n        done = False\n        reward = 0.0\n        price = float(self.df.loc[self.ptr, \"close\"])\n        # Action logic\n        if action == 1:  # buy\n            if self.position == 0:\n                self.position = 1\n                self.entry_price = price\n            elif self.position == -1:\n                # close short and go long\n                reward += (self.entry_price - price)  # profit from short\n                self.position = 1\n                self.entry_price = price\n        elif action == 2:  # sell\n            if self.position == 0:\n                self.position = -1\n                self.entry_price = price\n            elif self.position == 1:\n                reward += (price - self.entry_price)  # profit from long\n                self.position = -1\n                self.entry_price = price\n        # Move pointer\n        self.ptr += 1\n        if self.ptr \u003e= len(self.df):\n            done = True\n        else:\n            # reward can also be shaped by unrealized pnl:\n            next_price = float(self.df.loc[self.ptr, \"close\"])\n            unrealized = 0.0\n            if self.position == 1:\n                unrealized = next_price - self.entry_price\n            elif self.position == -1:\n                unrealized = self.entry_price - next_price\n            # small per-step reward = unrealized PnL scaled\n            reward += unrealized * 0.1\n\n        obs = self._get_obs() if not done else np.zeros(self.observation_space.shape, dtype=np.float32)\n        info = {\"ptr\": self.ptr}\n        return obs, float(reward), done, info\n\n# -------------------------\n# Order helpers\n# -------------------------\ndef send_order(symbol, action, lot=LOT_SIZE, deviation=20):\n    \"\"\"\n    action: 1=buy, 2=sell\n    This function sends a ORDER_TYPE_BUY / ORDER_TYPE_SELL market order.\n    Basic error checking included. For production you need more robust code.\n    \"\"\"\n    price = mt5.symbol_info_tick(symbol).ask if action == 1 else mt5.symbol_info_tick(symbol).bid\n    request = {\n        \"action\": mt5.TRADE_ACTION_DEAL,\n        \"symbol\": symbol,\n        \"volume\": float(lot),\n        \"type\": mt5.ORDER_TYPE_BUY if action == 1 else mt5.ORDER_TYPE_SELL,\n        \"price\": float(price),\n        \"deviation\": deviation,\n        \"magic\": 234000,\n        \"comment\": \"RL-bot\",\n        \"type_filling\": mt5.ORDER_FILLING_IOC,\n    }\n    result = mt5.order_send(request)\n    return result\n\n# -------------------------\n# Main: training flow\n# -------------------------\ndef train_agent():\n    mt5_connect()\n    # fetch historical bars\n    n_bars = 5000\n    df = fetch_bars(SYMBOL, TIMEFRAME, n_bars)\n    print(f\"Fetched {len(df)} bars for {SYMBOL}\")\n    # Create env\n    env = DummyVecEnv([lambda: MT5TradingEnv(df, lookback=LOOKBACK)])\n    # model\n    model = PPO(\"MlpPolicy\", env, verbose=1)\n    # save checkpoints\n    cb = CheckpointCallback(save_freq=5000, save_path=\"./logs/\", name_prefix=\"ppo_mt5\")\n    model.learn(total_timesteps=TRAIN_TIMESTEPS, callback=cb)\n    model.save(MODEL_PATH)\n    mt5_shutdown()\n    print(\"Training complete, model saved to\", MODEL_PATH)\n\n# -------------------------\n# Real-time execution loop (paper/live)\n# -------------------------\ndef run_live_loop(model_path=MODEL_PATH, poll_seconds=5):\n    mt5_connect()\n    model = PPO.load(model_path)\n    print(\"Loaded model:\", model_path)\n    # We'll maintain a small in-memory buffer of recent bars\n    n_history = LOOKBACK + 10\n    df = fetch_bars(SYMBOL, TIMEFRAME, n_history)\n    # pointer is at last bar\n    while True:\n        try:\n            latest = fetch_bars(SYMBOL, TIMEFRAME, 1)\n            if latest['time'].iloc[-1] \u003e df['time'].iloc[-1]:\n                # append new bar\n                df = pd.concat([df, latest]).reset_index(drop=True)\n                if len(df) \u003e n_history:\n                    df = df.iloc[-n_history:].reset_index(drop=True)\n                # Build an env instance for this single-step decision\n                env = MT5TradingEnv(df, lookback=LOOKBACK)\n                obs = env.reset()\n                action, _states = model.predict(obs, deterministic=True)\n                print(f\"[{pd.to_datetime('now')}] Action: {action} | Price: {df['close'].iloc[-1]}\")\n                # Send order if LIVE\n                if LIVE:\n                    if int(action) == 1:\n                        res = send_order(SYMBOL, 1)\n                        print(\"Order send result:\", res)\n                    elif int(action) == 2:\n                        res = send_order(SYMBOL, 2)\n                        print(\"Order send result:\", res)\n                else:\n                    # paper-trade: just log what would happen\n                    print(\"LIVE=False -\u003e paper trade logged only\")\n            else:\n                # no new bar yet\n                pass\n            time.sleep(poll_seconds)\n        except KeyboardInterrupt:\n            print(\"Stopping live loop (KeyboardInterrupt).\")\n            break\n        except Exception as e:\n            print(\"Exception in live loop:\", str(e))\n            time.sleep(5)\n    mt5_shutdown()\n\n# -------------------------\n# If run as script\n# -------------------------\nif __name__ == \"__main__\":\n    import argparse\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--mode\", choices=[\"train\", \"run\"], default=\"train\")\n    args = parser.parse_args()\n    if args.mode == \"train\":\n        print(\"Starting training...\")\n        train_agent()\n    elif args.mode == \"run\":\n        print(\"Starting live/paper run loop...\")\n        run_live_loop()","author":{"url":"https://github.com/vicks4u","@type":"Person","name":"vicks4u"},"datePublished":"2025-09-16T01:33:48.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":0},"url":"https://github.com/6/PastebinPython/issues/6"}