mt5bridge-ccxt/mt5bridge_ccxt/mql5_bridge.py

"""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")