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
|