{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "1efd5b49",
   "metadata": {},
   "source": [
    "# 4) Live Demo — Dry Run Loop (MT5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d3cefae7",
   "metadata": {},
   "source": [
    "This replaces the minimal print with a richer, FinRL-style trace:\n",
    "- Model action + BUY/SELL/HOLD flags\n",
    "- Existing positions check, close-opposite logic\n",
    "- Place market order with result log\n",
    "- Sleep until next bar boundary (derived from `TIMEFRAME`)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b638271d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/usr/bin/env python3\n",
    "# -*- coding: utf-8 -*-\n",
    "\"\"\"\n",
    "Live trading runner (MT5) with verbose logs (ASCII-safe) + market-hours check.\n",
    "\"\"\"\n",
    "import os, sys, time, signal, argparse, logging\n",
    "from logging.handlers import RotatingFileHandler\n",
    "from datetime import datetime, timedelta, timezone\n",
    "from pathlib import Path\n",
    "from typing import Optional, Dict\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "try:\n",
    "    from dotenv import load_dotenv\n",
    "except Exception:\n",
    "    def load_dotenv(*args, **kwargs):\n",
    "        return False\n",
    "\n",
    "try:\n",
    "    import MetaTrader5 as mt5\n",
    "except Exception:\n",
    "    mt5 = None\n",
    "\n",
    "from stable_baselines3 import PPO\n",
    "import features\n",
    "from adapters import broker\n",
    "\n",
    "STOP = False\n",
    "\n",
    "def handle_exit(signum, frame):\n",
    "    global STOP\n",
    "    STOP = True\n",
    "    logging.getLogger(\"live\").info(\"Received signal %s, shutting down...\", signum)\n",
    "\n",
    "for sig in (signal.SIGINT, signal.SIGTERM):\n",
    "    try:\n",
    "        signal.signal(sig, handle_exit)\n",
    "    except Exception:\n",
    "        pass\n",
    "\n",
    "def setup_logging(log_path: Optional[Path] = None, level: int = logging.INFO) -> logging.Logger:\n",
    "    logger = logging.getLogger(\"live\")\n",
    "    logger.setLevel(level)\n",
    "    fmt = logging.Formatter(\"%(asctime)s | %(levelname)s | %(message)s\")\n",
    "    sh = logging.StreamHandler(sys.stdout); sh.setFormatter(fmt); logger.addHandler(sh)\n",
    "    if log_path:\n",
    "        log_path.parent.mkdir(parents=True, exist_ok=True)\n",
    "        fh = RotatingFileHandler(log_path, maxBytes=2000000, backupCount=3, encoding=\"utf-8\")\n",
    "        fh.setFormatter(fmt); logger.addHandler(fh)\n",
    "    return logger\n",
    "\n",
    "def minutes_per_bar(tf: str) -> int:\n",
    "    tf = str(tf).upper()\n",
    "    if tf.startswith(\"M\"): return int(tf[1:])\n",
    "    if tf.startswith(\"H\"): return int(tf[1:]) * 60\n",
    "    if tf in (\"D1\", \"1D\"): return 1440\n",
    "    return 15\n",
    "\n",
    "def seconds_to_next_bar(tf: str) -> float:\n",
    "    m = minutes_per_bar(tf)\n",
    "    now = datetime.now(timezone.utc)\n",
    "    base = now.replace(second=0, microsecond=0)\n",
    "    mins = now.minute\n",
    "    next_min = ((mins // m) + 1) * m\n",
    "    next_bar = base + timedelta(minutes=(next_min - mins))\n",
    "    delta = (next_bar - now).total_seconds()\n",
    "    if delta < 5: delta += m * 60\n",
    "    return float(delta)\n",
    "\n",
    "def scalar_action(action) -> int:\n",
    "    return int(np.asarray(action).reshape(-1)[0])\n",
    "\n",
    "def action_to_signals(a: int) -> Dict[str, bool]:\n",
    "    return {\"buy\": a == 2, \"sell\": a == 0, \"hold\": a == 1}\n",
    "\n",
    "def print_header(logger: logging.Logger, symbol: str, a: int, sig: Dict[str, bool]) -> None:\n",
    "    logger.info(\"Model action: %s\", a)\n",
    "    logger.info(\"-\" * 66)\n",
    "    logger.info(\"Date: %s, SYMBOL: %s, BUY SIGNAL: %s, SELL SIGNAL: %s, HOLD SIGNAL: %s\",\n",
    "        datetime.utcnow().strftime(\"%Y-%m-%d %H:%M:%S\"), symbol, sig[\"buy\"], sig[\"sell\"], sig[\"hold\"])\n",
    "\n",
    "def positions_snapshot(logger: logging.Logger, symbol: str) -> None:\n",
    "    try:\n",
    "        pos = broker.current_positions(symbol)\n",
    "        if pos is None or len(pos) == 0:\n",
    "            logger.info(\"No open positions.\")\n",
    "            return\n",
    "        n_long = int((pos[\"type\"] == 0).sum())\n",
    "        n_short = int((pos[\"type\"] == 1).sum())\n",
    "        vol_total = float(pos[\"volume\"].sum())\n",
    "        flt_pnl = float(pos[\"profit\"].sum())\n",
    "        logger.info(\"Positions snapshot | longs=%d shorts=%d total_vol=%.3f floatingPnL=%.2f\",\n",
    "                    n_long, n_short, vol_total, flt_pnl)\n",
    "    except Exception as e:\n",
    "        logger.warning(\"Snapshot failed: %s\", e)\n",
    "\n",
    "def has_same_side_position(symbol: str, want_buy: bool, want_sell: bool) -> bool:\n",
    "    try:\n",
    "        pos_now = broker.current_positions(symbol)\n",
    "        if pos_now is None or len(pos_now) == 0: return False\n",
    "        if want_buy and (pos_now[\"type\"] == 0).any(): return True\n",
    "        if want_sell and (pos_now[\"type\"] == 1).any(): return True\n",
    "    except Exception:\n",
    "        return False\n",
    "    return False\n",
    "\n",
    "def close_opposite_positions_if_any(logger: logging.Logger, symbol: str, sig: Dict[str, bool], dry_run: bool) -> bool:\n",
    "    try:\n",
    "        pos = broker.current_positions(symbol)\n",
    "    except Exception as e:\n",
    "        logger.warning(\"Could not fetch positions: %s\", e)\n",
    "        return False\n",
    "    if pos is None or len(pos) == 0:\n",
    "        return False\n",
    "    if sig[\"buy\"]:\n",
    "        to_close = pos[pos[\"type\"] == 1]\n",
    "        side_text = \"sell\"\n",
    "    elif sig[\"sell\"]:\n",
    "        to_close = pos[pos[\"type\"] == 0]\n",
    "        side_text = \"buy\"\n",
    "    else:\n",
    "        return False\n",
    "    if len(to_close) == 0:\n",
    "        return False\n",
    "    if dry_run:\n",
    "        tickets = \", \".join(map(lambda x: str(int(x)), to_close[\"ticket\"]))\n",
    "        logger.info(\"DRY_RUN=True -> Would close %d %s position(s): %s\", len(to_close), side_text, tickets)\n",
    "        return False\n",
    "    logger.info(\"Existing %s positions found. Attempting to close...\", side_text)\n",
    "    any_closed = False\n",
    "    for _, row in to_close.iterrows():\n",
    "        ticket = int(row[\"ticket\"])\n",
    "        res = broker.close_position(ticket)\n",
    "        if res.get(\"ok\"):\n",
    "            logger.info(\"Successfully closed position %s for %s\", ticket, symbol); any_closed = True\n",
    "        else:\n",
    "            logger.warning(\"Failed to close position %s: %s\", ticket, res)\n",
    "    return any_closed\n",
    "\n",
    "def place_signal_order(logger: logging.Logger, symbol: str, sig: Dict[str, bool], vol: float, dry_run: bool, comment: str):\n",
    "    if sig[\"hold\"]:\n",
    "        logger.info(\"Hold signal detected. No actions taken.\"); return None\n",
    "    side = \"buy\" if sig[\"buy\"] else \"sell\"\n",
    "    if has_same_side_position(symbol, sig[\"buy\"], sig[\"sell\"]):\n",
    "        logger.info(\"Same-side position already exists. Skipping new %s order.\", side)\n",
    "        return None\n",
    "    if dry_run:\n",
    "        logger.info(\"DRY_RUN=True -> Skipping order. Would place %s %.3f on %s\", side.upper(), vol, symbol)\n",
    "        return {\"ok\": True, \"dry_run\": True, \"side\": side, \"volume\": vol}\n",
    "    logger.info(\"%s positions closed (if any). Placing new %s order.\", \"Buy\" if side==\"buy\" else \"Sell\", side)\n",
    "    res = broker.place_market_order(symbol, side=side, volume=vol, comment=comment)\n",
    "    detail = res.get(\"result\", {}) if isinstance(res, dict) else {}\n",
    "    price = detail.get(\"price\") if isinstance(detail, dict) else None\n",
    "    order = detail.get(\"order\") if isinstance(detail, dict) else None\n",
    "    deal  = detail.get(\"deal\")  if isinstance(detail, dict) else None\n",
    "    logger.info(\"Order result: ok=%s, retcode=%s, order=%s, deal=%s, price=%s\",\n",
    "                res.get(\"ok\"), res.get(\"retcode\"), order, deal, price)\n",
    "    return res\n",
    "\n",
    "def print_bar_context(logger: logging.Logger, last_row: pd.Series):\n",
    "    fields = []\n",
    "    if \"close\" in last_row: fields.append(\"close=%.5f\" % float(last_row[\"close\"]))\n",
    "    if \"ma_fast\" in last_row: fields.append(\"ma_fast=%.5f\" % float(last_row[\"ma_fast\"]))\n",
    "    if \"ma_slow\" in last_row: fields.append(\"ma_slow=%.5f\" % float(last_row[\"ma_slow\"]))\n",
    "    if \"rsi\" in last_row: fields.append(\"rsi=%.1f\" % float(last_row[\"rsi\"]))\n",
    "    if fields:\n",
    "        logger.info(\"Context: \" + \" | \".join(fields))\n",
    "\n",
    "def is_market_open(symbol: str, tf: str, logger: logging.Logger, allow_weekend: bool) -> bool:\n",
    "    try:\n",
    "        now = datetime.now(timezone.utc)\n",
    "        dow = now.weekday()\n",
    "        if not allow_weekend and dow in (5, 6):\n",
    "            logger.info(\"Market likely closed (weekend).\")\n",
    "            return False\n",
    "        if mt5 is None:\n",
    "            return True\n",
    "        tick = mt5.symbol_info_tick(symbol)\n",
    "        if tick is None:\n",
    "            logger.info(\"No tick available for %s.\", symbol)\n",
    "            return False\n",
    "        tick_ts = datetime.fromtimestamp(tick.time, tz=timezone.utc)\n",
    "        age_sec = (now - tick_ts).total_seconds()\n",
    "        bar_min = minutes_per_bar(tf)\n",
    "        thresh = max(bar_min * 120, 1800)\n",
    "        if age_sec > thresh:\n",
    "            logger.info(\"Market likely closed (last tick %.1f min ago).\", age_sec / 60.0)\n",
    "            return False\n",
    "        return True\n",
    "    except Exception as e:\n",
    "        logger.warning(\"Market check failed: %s\", e)\n",
    "        return True\n",
    "\n",
    "def main():\n",
    "    load_dotenv()\n",
    "    p = argparse.ArgumentParser(description=\"Live DRL trading bot (MT5)\")\n",
    "    p.add_argument(\"--symbol\", default=os.getenv(\"TRAINING_SYMBOL\",\"EURUSD\"))\n",
    "    p.add_argument(\"--timeframe\", default=os.getenv(\"TIMEFRAME\",\"M15\"))\n",
    "    p.add_argument(\"--model\", default=None)\n",
    "    p.add_argument(\"--features\", default=\"notebooks/models/selected_features.json\")\n",
    "    p.add_argument(\"--volume\", type=float, default=float(os.getenv(\"VOLUME\",\"0.01\")))\n",
    "    p.add_argument(\"--live\", action=\"store_true\")\n",
    "    p.add_argument(\"--dry-run\", action=\"store_true\")\n",
    "    p.add_argument(\"--order-comment\", default=os.getenv(\"ORDER_COMMENT\",\"DRL-Live\"))\n",
    "    p.add_argument(\"--log-file\", default=\"logs/live_bot.log\")\n",
    "    p.add_argument(\"--skip-market-check\", action=\"store_true\", help=\"Disable market closed guard\")\n",
    "    args = p.parse_args()\n",
    "\n",
    "    symbol = args.symbol; tf = args.timeframe\n",
    "    dry_run = not args.live or args.dry_run\n",
    "    vol = max(args.volume, 0.0); comment = args.order_comment\n",
    "    allow_weekend = os.getenv(\"WEEKEND_TRADING\",\"false\").lower() in (\"1\",\"true\",\"yes\")\n",
    "\n",
    "    logger = setup_logging(Path(args.log_file))\n",
    "    logger.info(\"Connecting to MT5...\")\n",
    "    ok = broker.open_session(\n",
    "        login=int(os.getenv(\"MT5_LOGIN\",\"0\")) or None,\n",
    "        password=os.getenv(\"MT5_PASSWORD\"),\n",
    "        server=os.getenv(\"MT5_SERVER\"),\n",
    "        path=os.getenv(\"MT5_PATH\"),\n",
    "    )\n",
    "    logger.info(\"MT5 connected: %s\", ok)\n",
    "\n",
    "    model_path = args.model or f\"notebooks/models/ppo_{symbol}_{tf}.zip\"\n",
    "    mp = Path(model_path)\n",
    "    if not mp.exists():\n",
    "        alt = Path(\"notebooks\") / \"notebooks/models\" / Path(f\"ppo_{symbol}_{tf}.zip\")\n",
    "        if alt.exists():\n",
    "            mp = alt\n",
    "    if not mp.exists():\n",
    "        logger.error(\"Model not found: %s\", model_path)\n",
    "        logger.error(\"Also checked notebooks/models/ppo_%s_%s.zip\", symbol, tf)\n",
    "        sys.exit(2)\n",
    "\n",
    "    try:\n",
    "        import json\n",
    "        with open(args.features, \"r\", encoding=\"utf-8\") as f:\n",
    "            feature_cols = json.load(f)\n",
    "    except Exception as e:\n",
    "        logger.error(\"Failed to load features from %s: %s\", args.features, e); sys.exit(2)\n",
    "\n",
    "    logger.info(\"Loading model: %s\", mp)\n",
    "    model = PPO.load(mp.as_posix())\n",
    "\n",
    "    logger.info(\"Starting live loop | symbol=%s timeframe=%s dry_run=%s volume=%.3f\", symbol, tf, dry_run, vol)\n",
    "    try:\n",
    "        while not STOP:\n",
    "            try:\n",
    "                open_ok = True if args.skip_market_check else is_market_open(symbol, tf, logger, allow_weekend)\n",
    "                if not open_ok:\n",
    "                    secs = min(900, seconds_to_next_bar(tf))\n",
    "                    logger.info(\"Sleeping %.0f seconds until next check (market closed).\", secs)\n",
    "                    time.sleep(secs)\n",
    "                    continue\n",
    "\n",
    "                df = broker.fetch_last_n(symbol, tf, n=500)\n",
    "                if df is None or len(df) < 50:\n",
    "                    logger.warning(\"Insufficient bars fetched; sleeping 5s\"); time.sleep(5); continue\n",
    "\n",
    "                df_feat = features.add_indicators(df.copy())\n",
    "                last_row = df_feat.iloc[-1]\n",
    "                print_bar_context(logger, last_row)\n",
    "\n",
    "                obs = last_row[list(feature_cols)].astype(np.float32).values\n",
    "                if np.isnan(obs).any() or np.isinf(obs).any():\n",
    "                    logger.warning(\"Obs has NaN/Inf; skipping this tick\"); time.sleep(5); continue\n",
    "\n",
    "                action, _ = model.predict(obs, deterministic=True)\n",
    "                a = scalar_action(action); sig = action_to_signals(a)\n",
    "                print_header(logger, symbol, a, sig)\n",
    "\n",
    "                positions_snapshot(logger, symbol)\n",
    "                _ = close_opposite_positions_if_any(logger, symbol, sig, dry_run)\n",
    "\n",
    "                if sig[\"hold\"]:\n",
    "                    logger.info(\"Appropriate position already exists or HOLD signal. No new order.\")\n",
    "                else:\n",
    "                    _ = place_signal_order(logger, symbol, sig, vol, dry_run, comment)\n",
    "\n",
    "                secs = seconds_to_next_bar(tf)\n",
    "                logger.info(\"Waiting for new signals...\")\n",
    "                logger.info(\"Calculated sleep time: %.3f seconds\", secs)\n",
    "                time.sleep(secs)\n",
    "\n",
    "            except Exception as e:\n",
    "                logger.exception(\"Loop error: %s\", e); time.sleep(5)\n",
    "    finally:\n",
    "        try: broker.close_session()\n",
    "        except Exception: pass\n",
    "        logger.info(\"Shutdown complete.\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "drl",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
