"""MQL5 GlobalVariable bridge for reading indicator signals from MT5. This module provides convenient access to GlobalVariables on the MT5 side, enabling Python strategies to read signals written by MQL5 indicators (such as the Alpha Trend indicator) without re-implementing the indicator logic in Python. Example: >>> exchange = mt5bridge_ccxt.mt5bridge({...}) >>> signal = exchange.mql5.alpha_trend_signal("XAUUSD", "H1") >>> if signal["trend"] == 1: ... exchange.create_order("XAU/USD", "market", "buy", 0.01) """ class Mql5Bridge: """Helper for reading and writing MQL5 GlobalVariables via MT5Bridge.""" # Standard command keys used to signal the MQL5 helper EA CMD_CANCEL_PREFIX = "MT5BRIDGE_CMD_CANCEL_" CMD_MODIFY_PREFIX = "MT5BRIDGE_CMD_MODIFY_" CMD_CLOSE_PREFIX = "MT5BRIDGE_CMD_CLOSE_" LAST_CMD_KEY = "MT5BRIDGE_LAST_CMD" def __init__(self, exchange): self._exchange = exchange # ──────── Low-level GlobalVariable API ──────── def list(self): """List all global variables on MT5. Returns: list of dict: [{"name": "AT_Trend_XAUUSD", "value": 1.0}, ...] """ return self._exchange.private_get_gvar()["data"] def get(self, name, default=None): """Get a single global variable value. Args: name: GlobalVariable name default: value to return if variable does not exist Returns: float or default """ try: return self._exchange.private_get_gvar_name({"name": name})["value"] except Exception: return default def set(self, name, value): """Set a global variable value. Args: name: GlobalVariable name value: numeric value (int or float) Returns: dict with name and value """ return self._exchange.private_post_gvar_name( {"name": name, "value": float(value)} ) def delete(self, name): """Delete a global variable. Args: name: GlobalVariable name Returns: dict with "deleted" key """ return self._exchange.private_delete_gvar_name({"name": name}) # ──────── High-level indicator helpers ──────── def alpha_trend_signal(self, symbol, timeframe, length=14, atr_mult=1.0, use_volume=0, show_signals=1): """Read Alpha Trend indicator signal from MT5 GlobalVariables. This expects the Alpha Trend indicator (Alpha Trend.mq5) to be loaded on a chart of `symbol` with matching parameters. Args: symbol: MT5 symbol (e.g. "XAUUSDc") timeframe: MT5 timeframe name (e.g. "M1", "M5", "H1", "D1") length: indicator length (default 14) atr_mult: ATR multiplier (default 1.0) use_volume: 0=No (RSI), 1=Yes (MFI) - matches `inp_change_calc` show_signals: 0=No, 1=Yes - matches `inp_signals` Returns: dict with keys: alpha, offset, trend, buy, sell (any value may be None if the indicator is not loaded) """ period_str = f"PERIOD_{timeframe}" prefix = ( f"AT_{symbol}_{period_str}" f"_L{length}_A{atr_mult:.1f}_C{use_volume}_S{show_signals}" ) return { "alpha": self.get(f"{prefix}_Alpha"), "offset": self.get(f"{prefix}_Offset"), "trend": self.get(f"{prefix}_Trend"), "buy": self.get(f"{prefix}_Buy"), "sell": self.get(f"{prefix}_Sell"), } def read_signal(self, key, default=None): """Read a custom signal by key (convention: MY_SIGNAL_{SYMBOL}_{TF}).""" return self.get(key, default) # ──────── Command helpers (used by MQL5 helper EA) ──────── def send_cancel_command(self, ticket): """Send an async cancel command for a pending order. The MQL5 helper EA (Mt5BridgeHelper.mq5) reads this GlobalVariable and calls OrderDelete() on the broker side. Args: ticket: order/position ticket to cancel/close Returns: dict with "command_key" and "ticket" """ key = f"{self.CMD_CANCEL_PREFIX}{ticket}" self.set(key, _now_seconds()) self.set(self.LAST_CMD_KEY, float(ticket)) return {"command_key": key, "ticket": ticket} def send_modify_command(self, ticket, sl=None, tp=None, price=None): """Send an async modify command (SL/TP/price) for an order or position. Args: ticket: order/position ticket sl: new stop loss (None to leave unchanged) tp: new take profit (None to leave unchanged) price: new open price (for pending orders, None to leave unchanged) Returns: dict with "command_key" and "ticket" """ import json key = f"{self.CMD_MODIFY_PREFIX}{ticket}" payload = {"ts": _now_seconds()} if sl is not None: payload["sl"] = float(sl) if tp is not None: payload["tp"] = float(tp) if price is not None: payload["price"] = float(price) # Encode the payload in the GV value (use NaN to mean "no value") encoded = _pack_payload(payload) self.set(key, encoded) self.set(self.LAST_CMD_KEY, float(ticket)) return {"command_key": key, "ticket": ticket, "payload": payload} def send_close_command(self, ticket, volume=None): """Send an async close command for an open position. Args: ticket: position ticket volume: volume to close (None = close full position) Returns: dict with "command_key" and "ticket" """ key = f"{self.CMD_CLOSE_PREFIX}{ticket}" payload = {"ts": _now_seconds()} if volume is not None: payload["volume"] = float(volume) encoded = _pack_payload(payload) self.set(key, encoded) self.set(self.LAST_CMD_KEY, float(ticket)) return {"command_key": key, "ticket": ticket, "volume": volume} def _now_seconds(): """Current Unix timestamp in seconds.""" import time return time.time() def _pack_payload(payload): """Encode a payload dict into a float for GlobalVariable storage. Uses bit-packing of a short JSON string into the mantissa. For simplicity, we use a stable hash and store the JSON separately via a side-channel GlobalVariable when the payload is non-empty. """ import json import struct json_str = json.dumps(payload, separators=(",", ":")) # Use 8-byte float64 to store up to 53 bits ≈ 7 ASCII chars # For longer payloads, fall back to NaN if len(json_str) <= 7: try: return float(json_str) except ValueError: pass # Fall back: store timestamp only, ignore detailed payload # (helper EA will use sensible defaults for missing fields) return float("nan")