Deep-Reinforcement-Learning-MT5-Bot/live_bot.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Live trading runner (MT5) with verbose logs (ASCII-safe) + market-hours check.
"""
import os, sys, time, signal, argparse, logging
from logging.handlers import RotatingFileHandler
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional, Dict

import numpy as np
import pandas as pd

try:
    from dotenv import load_dotenv
except Exception:
    def load_dotenv(*args, **kwargs):
        return False

try:
    import MetaTrader5 as mt5
except Exception:
    mt5 = None

from stable_baselines3 import PPO
import features
from adapters import broker

STOP = False

def handle_exit(signum, frame):
    global STOP
    STOP = True
    logging.getLogger("live").info("Received signal %s, shutting down...", signum)

for sig in (signal.SIGINT, signal.SIGTERM):
    try:
        signal.signal(sig, handle_exit)
    except Exception:
        pass

def setup_logging(log_path: Optional[Path] = None, level: int = logging.INFO) -> logging.Logger:
    logger = logging.getLogger("live")
    logger.setLevel(level)
    fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
    sh = logging.StreamHandler(sys.stdout); sh.setFormatter(fmt); logger.addHandler(sh)
    if log_path:
        log_path.parent.mkdir(parents=True, exist_ok=True)
        fh = RotatingFileHandler(log_path, maxBytes=2000000, backupCount=3, encoding="utf-8")
        fh.setFormatter(fmt); logger.addHandler(fh)
    return logger

def minutes_per_bar(tf: str) -> int:
    tf = str(tf).upper()
    if tf.startswith("M"): return int(tf[1:])
    if tf.startswith("H"): return int(tf[1:]) * 60
    if tf in ("D1", "1D"): return 1440
    return 15

def seconds_to_next_bar(tf: str) -> float:
    m = minutes_per_bar(tf)
    now = datetime.now(timezone.utc)
    base = now.replace(second=0, microsecond=0)
    mins = now.minute
    next_min = ((mins // m) + 1) * m
    next_bar = base + timedelta(minutes=(next_min - mins))
    delta = (next_bar - now).total_seconds()
    if delta < 5: delta += m * 60
    return float(delta)

def scalar_action(action) -> int:
    return int(np.asarray(action).reshape(-1)[0])

def action_to_signals(a: int) -> Dict[str, bool]:
    return {"buy": a == 2, "sell": a == 0, "hold": a == 1}

def print_header(logger: logging.Logger, symbol: str, a: int, sig: Dict[str, bool]) -> None:
    logger.info("Model action: %s", a)
    logger.info("-" * 66)
    logger.info("Date: %s, SYMBOL: %s, BUY SIGNAL: %s, SELL SIGNAL: %s, HOLD SIGNAL: %s",
        datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"), symbol, sig["buy"], sig["sell"], sig["hold"])

def positions_snapshot(logger: logging.Logger, symbol: str) -> None:
    try:
        pos = broker.current_positions(symbol)
        if pos is None or len(pos) == 0:
            logger.info("No open positions.")
            return
        n_long = int((pos["type"] == 0).sum())
        n_short = int((pos["type"] == 1).sum())
        vol_total = float(pos["volume"].sum())
        flt_pnl = float(pos["profit"].sum())
        logger.info("Positions snapshot | longs=%d shorts=%d total_vol=%.3f floatingPnL=%.2f",
                    n_long, n_short, vol_total, flt_pnl)
    except Exception as e:
        logger.warning("Snapshot failed: %s", e)

def has_same_side_position(symbol: str, want_buy: bool, want_sell: bool) -> bool:
    try:
        pos_now = broker.current_positions(symbol)
        if pos_now is None or len(pos_now) == 0: return False
        if want_buy and (pos_now["type"] == 0).any(): return True
        if want_sell and (pos_now["type"] == 1).any(): return True
    except Exception:
        return False
    return False

def close_opposite_positions_if_any(logger: logging.Logger, symbol: str, sig: Dict[str, bool], dry_run: bool) -> bool:
    try:
        pos = broker.current_positions(symbol)
    except Exception as e:
        logger.warning("Could not fetch positions: %s", e)
        return False
    if pos is None or len(pos) == 0:
        return False
    if sig["buy"]:
        to_close = pos[pos["type"] == 1]
        side_text = "sell"
    elif sig["sell"]:
        to_close = pos[pos["type"] == 0]
        side_text = "buy"
    else:
        return False
    if len(to_close) == 0:
        return False
    if dry_run:
        tickets = ", ".join(map(lambda x: str(int(x)), to_close["ticket"]))
        logger.info("DRY_RUN=True -> Would close %d %s position(s): %s", len(to_close), side_text, tickets)
        return False
    logger.info("Existing %s positions found. Attempting to close...", side_text)
    any_closed = False
    for _, row in to_close.iterrows():
        ticket = int(row["ticket"])
        res = broker.close_position(ticket)
        if res.get("ok"):
            logger.info("Successfully closed position %s for %s", ticket, symbol); any_closed = True
        else:
            logger.warning("Failed to close position %s: %s", ticket, res)
    return any_closed

def place_signal_order(logger: logging.Logger, symbol: str, sig: Dict[str, bool], vol: float, dry_run: bool, comment: str):
    if sig["hold"]:
        logger.info("Hold signal detected. No actions taken."); return None
    side = "buy" if sig["buy"] else "sell"
    if has_same_side_position(symbol, sig["buy"], sig["sell"]):
        logger.info("Same-side position already exists. Skipping new %s order.", side)
        return None
    if dry_run:
        logger.info("DRY_RUN=True -> Skipping order. Would place %s %.3f on %s", side.upper(), vol, symbol)
        return {"ok": True, "dry_run": True, "side": side, "volume": vol}
    logger.info("%s positions closed (if any). Placing new %s order.", "Buy" if side=="buy" else "Sell", side)
    res = broker.place_market_order(symbol, side=side, volume=vol, comment=comment)
    detail = res.get("result", {}) if isinstance(res, dict) else {}
    price = detail.get("price") if isinstance(detail, dict) else None
    order = detail.get("order") if isinstance(detail, dict) else None
    deal  = detail.get("deal")  if isinstance(detail, dict) else None
    logger.info("Order result: ok=%s, retcode=%s, order=%s, deal=%s, price=%s",
                res.get("ok"), res.get("retcode"), order, deal, price)
    return res

def print_bar_context(logger: logging.Logger, last_row: pd.Series):
    fields = []
    if "close" in last_row: fields.append("close=%.5f" % float(last_row["close"]))
    if "ma_fast" in last_row: fields.append("ma_fast=%.5f" % float(last_row["ma_fast"]))
    if "ma_slow" in last_row: fields.append("ma_slow=%.5f" % float(last_row["ma_slow"]))
    if "rsi" in last_row: fields.append("rsi=%.1f" % float(last_row["rsi"]))
    if fields:
        logger.info("Context: " + " | ".join(fields))

def is_market_open(symbol: str, tf: str, logger: logging.Logger, allow_weekend: bool) -> bool:
    try:
        now = datetime.now(timezone.utc)
        dow = now.weekday()
        if not allow_weekend and dow in (5, 6):
            logger.info("Market likely closed (weekend).")
            return False
        if mt5 is None:
            return True
        tick = mt5.symbol_info_tick(symbol)
        if tick is None:
            logger.info("No tick available for %s.", symbol)
            return False
        tick_ts = datetime.fromtimestamp(tick.time, tz=timezone.utc)
        age_sec = (now - tick_ts).total_seconds()
        bar_min = minutes_per_bar(tf)
        thresh = max(bar_min * 120, 1800)
        if age_sec > thresh:
            logger.info("Market likely closed (last tick %.1f min ago).", age_sec / 60.0)
            return False
        return True
    except Exception as e:
        logger.warning("Market check failed: %s", e)
        return True

def main():
    load_dotenv()
    p = argparse.ArgumentParser(description="Live DRL trading bot (MT5)")
    p.add_argument("--symbol", default=os.getenv("TRAINING_SYMBOL","EURUSD"))
    p.add_argument("--timeframe", default=os.getenv("TIMEFRAME","M15"))
    p.add_argument("--model", default=None)
    p.add_argument("--features", default="notebooks/models/selected_features.json")
    p.add_argument("--volume", type=float, default=float(os.getenv("VOLUME","0.01")))
    p.add_argument("--live", action="store_true")
    p.add_argument("--dry-run", action="store_true")
    p.add_argument("--order-comment", default=os.getenv("ORDER_COMMENT","DRL-Live"))
    p.add_argument("--log-file", default="logs/live_bot.log")
    p.add_argument("--skip-market-check", action="store_true", help="Disable market closed guard")
    args = p.parse_args()

    symbol = args.symbol; tf = args.timeframe
    dry_run = not args.live or args.dry_run
    vol = max(args.volume, 0.0); comment = args.order_comment
    allow_weekend = os.getenv("WEEKEND_TRADING","false").lower() in ("1","true","yes")

    logger = setup_logging(Path(args.log_file))
    logger.info("Connecting to MT5...")
    ok = broker.open_session(
        login=int(os.getenv("MT5_LOGIN","0")) or None,
        password=os.getenv("MT5_PASSWORD"),
        server=os.getenv("MT5_SERVER"),
        path=os.getenv("MT5_PATH"),
    )
    logger.info("MT5 connected: %s", ok)

    model_path = args.model or f"notebooks/models/ppo_{symbol}_{tf}.zip"
    mp = Path(model_path)
    if not mp.exists():
        alt = Path("notebooks") / "notebooks/models" / Path(f"ppo_{symbol}_{tf}.zip")
        if alt.exists():
            mp = alt
    if not mp.exists():
        logger.error("Model not found: %s", model_path)
        logger.error("Also checked notebooks/models/ppo_%s_%s.zip", symbol, tf)
        sys.exit(2)

    try:
        import json
        with open(args.features, "r", encoding="utf-8") as f:
            feature_cols = json.load(f)
    except Exception as e:
        logger.error("Failed to load features from %s: %s", args.features, e); sys.exit(2)

    logger.info("Loading model: %s", mp)
    model = PPO.load(mp.as_posix())

    logger.info("Starting live loop | symbol=%s timeframe=%s dry_run=%s volume=%.3f", symbol, tf, dry_run, vol)
    try:
        while not STOP:
            try:
                open_ok = True if args.skip_market_check else is_market_open(symbol, tf, logger, allow_weekend)
                if not open_ok:
                    secs = min(900, seconds_to_next_bar(tf))
                    logger.info("Sleeping %.0f seconds until next check (market closed).", secs)
                    time.sleep(secs)
                    continue

                df = broker.fetch_last_n(symbol, tf, n=500)
                if df is None or len(df) < 50:
                    logger.warning("Insufficient bars fetched; sleeping 5s"); time.sleep(5); continue

                df_feat = features.add_indicators(df.copy())
                last_row = df_feat.iloc[-1]
                print_bar_context(logger, last_row)

                obs = last_row[list(feature_cols)].astype(np.float32).values
                if np.isnan(obs).any() or np.isinf(obs).any():
                    logger.warning("Obs has NaN/Inf; skipping this tick"); time.sleep(5); continue

                action, _ = model.predict(obs, deterministic=True)
                a = scalar_action(action); sig = action_to_signals(a)
                print_header(logger, symbol, a, sig)

                positions_snapshot(logger, symbol)
                _ = close_opposite_positions_if_any(logger, symbol, sig, dry_run)

                if sig["hold"]:
                    logger.info("Appropriate position already exists or HOLD signal. No new order.")
                else:
                    _ = place_signal_order(logger, symbol, sig, vol, dry_run, comment)

                secs = seconds_to_next_bar(tf)
                logger.info("Waiting for new signals...")
                logger.info("Calculated sleep time: %.3f seconds", secs)
                time.sleep(secs)

            except Exception as e:
                logger.exception("Loop error: %s", e); time.sleep(5)
    finally:
        try: broker.close_session()
        except Exception: pass
        logger.info("Shutdown complete.")

if __name__ == "__main__":
    main()