Mohammad Aghdam
350 lines
16 KiB
Python
350 lines
16 KiB
Python
{
|
|
"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
|
|
}
|