99 lines
4.1 KiB
Python
99 lines
4.1 KiB
Python
import time
|
|
import pandas as pd
|
|
import MetaTrader5 as mt5
|
|
from utils.logger import get_logger
|
|
|
|
log = get_logger()
|
|
|
|
TF = {"M1": mt5.TIMEFRAME_M1, "M5": mt5.TIMEFRAME_M5, "M15": mt5.TIMEFRAME_M15,
|
|
"M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1, "H4": mt5.TIMEFRAME_H4, "D1": mt5.TIMEFRAME_D1}
|
|
|
|
class MT5:
|
|
def __init__(self, cfg: dict, magic: int, deviation: int):
|
|
self.cfg, self.magic, self.deviation = cfg, magic, deviation
|
|
|
|
# ---------- connexion ----------
|
|
def connect(self, retries: int = 5) -> bool:
|
|
for i in range(retries):
|
|
kw = {"login": int(self.cfg["login"]), "password": self.cfg["password"], "server": self.cfg["server"]}
|
|
if self.cfg.get("path"):
|
|
kw["path"] = self.cfg["path"]
|
|
if mt5.initialize(**kw):
|
|
a = mt5.account_info()
|
|
log.info(f"Connecté: compte {a.login} | {a.server} | solde {a.balance} {a.currency}")
|
|
return True
|
|
log.warning(f"Connexion échouée ({mt5.last_error()}) tentative {i + 1}/{retries}")
|
|
time.sleep(3)
|
|
return False
|
|
|
|
def ensure(self) -> bool:
|
|
ti = mt5.terminal_info()
|
|
if ti and ti.connected:
|
|
return True
|
|
log.warning("Connexion perdue, reconnexion...")
|
|
mt5.shutdown()
|
|
return self.connect()
|
|
|
|
def shutdown(self):
|
|
mt5.shutdown()
|
|
|
|
# ---------- données ----------
|
|
def rates(self, symbol: str, tf: str, n: int = 500):
|
|
mt5.symbol_select(symbol, True)
|
|
r = mt5.copy_rates_from_pos(symbol, TF[tf], 0, n)
|
|
if r is None or len(r) == 0:
|
|
return None
|
|
df = pd.DataFrame(r)
|
|
df["time"] = pd.to_datetime(df["time"], unit="s")
|
|
return df
|
|
|
|
def info(self, symbol):
|
|
return mt5.symbol_info(symbol)
|
|
|
|
def tick(self, symbol):
|
|
return mt5.symbol_info_tick(symbol)
|
|
|
|
def account(self):
|
|
return mt5.account_info()
|
|
|
|
def positions(self, symbol=None):
|
|
p = mt5.positions_get(symbol=symbol) if symbol else mt5.positions_get()
|
|
return [x for x in (p or []) if x.magic == self.magic]
|
|
|
|
# ---------- ordres ----------
|
|
def _filling(self, info):
|
|
fm = info.filling_mode
|
|
if fm & 1: return mt5.ORDER_FILLING_FOK
|
|
if fm & 2: return mt5.ORDER_FILLING_IOC
|
|
return mt5.ORDER_FILLING_RETURN
|
|
|
|
def open_trade(self, symbol, side, lots, sl, tp, comment=""):
|
|
info, t = self.info(symbol), self.tick(symbol)
|
|
price = t.ask if side == "buy" else t.bid
|
|
req = {"action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lots,
|
|
"type": mt5.ORDER_TYPE_BUY if side == "buy" else mt5.ORDER_TYPE_SELL,
|
|
"price": price, "sl": round(sl, info.digits), "tp": round(tp, info.digits),
|
|
"deviation": self.deviation, "magic": self.magic, "comment": comment[:30],
|
|
"type_time": mt5.ORDER_TIME_GTC, "type_filling": self._filling(info)}
|
|
res = mt5.order_send(req)
|
|
ok = res is not None and res.retcode == mt5.TRADE_RETCODE_DONE
|
|
log.info(f"ORDER {side} {symbol} {lots} sl={sl} tp={tp} -> {'OK' if ok else res}")
|
|
return ok
|
|
|
|
def close_position(self, pos):
|
|
info, t = self.info(pos.symbol), self.tick(pos.symbol)
|
|
buy = pos.type == mt5.POSITION_TYPE_BUY
|
|
req = {"action": mt5.TRADE_ACTION_DEAL, "symbol": pos.symbol, "volume": pos.volume,
|
|
"type": mt5.ORDER_TYPE_SELL if buy else mt5.ORDER_TYPE_BUY,
|
|
"position": pos.ticket, "price": t.bid if buy else t.ask,
|
|
"deviation": self.deviation, "magic": self.magic, "comment": "close",
|
|
"type_time": mt5.ORDER_TIME_GTC, "type_filling": self._filling(info)}
|
|
res = mt5.order_send(req)
|
|
return res is not None and res.retcode == mt5.TRADE_RETCODE_DONE
|
|
|
|
def modify_sl(self, pos, new_sl):
|
|
info = self.info(pos.symbol)
|
|
req = {"action": mt5.TRADE_ACTION_SLTP, "position": pos.ticket, "symbol": pos.symbol,
|
|
"sl": round(new_sl, info.digits), "tp": pos.tp}
|
|
res = mt5.order_send(req)
|
|
return res is not None and res.retcode == mt5.TRADE_RETCODE_DONE
|