16 KiB
16 KiB
In [ ]:
#!/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()