296 lines
12 KiB
Python
296 lines
12 KiB
Python
|
#!/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()
|