diff --git a/mt5bridge_ccxt/mt5bridge.py b/mt5bridge_ccxt/mt5bridge.py index 28a35c3..4c35ccf 100644 --- a/mt5bridge_ccxt/mt5bridge.py +++ b/mt5bridge_ccxt/mt5bridge.py @@ -21,6 +21,7 @@ Architecture: MetaTrader 5 client """ +import asyncio import json import time from datetime import datetime, timezone @@ -87,28 +88,26 @@ MT5_ORDER_TYPE = { } # MT5 retcode -> CCXT-friendly status string -# Source: MQL5 docs, ENUM_TRADE_REQUEST_RETCODE MT5_RETCODE_STATUS = { - 10009: "open", # TRADE_RETCODE_DONE - request executed - 10010: "filled", # TRADE_RETCODE_PLACED - pending order placed - 10011: "canceled", # TRADE_RETCODE_DONE_PARTIAL - 10012: "canceled", # TRADE_RETCODE_ERROR - 10013: "rejected", # TRADE_RETCODE_TIMEOUT - 10014: "rejected", # TRADE_RETCODE_INVALID - 10015: "rejected", # TRADE_RETCODE_INVALID_VOLUME - 10016: "rejected", # TRADE_RETCODE_INVALID_PRICE - 10017: "rejected", # TRADE_RETCODE_BUSY - 10018: "rejected", # TRADE_RETCODE_NO_CHANGES - 10019: "rejected", # TRADE_RETCODE_LOCKED - 10020: "rejected", # TRADE_RETCODE_FROZEN - 10021: "rejected", # TRADE_RETCODE_INVALID_FILL - 10022: "rejected", # TRADE_RETCODE_CONNECTION - 10023: "rejected", # TRADE_RETCODE_ONLY_REAL - 10024: "rejected", # TRADE_RETCODE_LIMIT_ORDERS - 10025: "rejected", # TRADE_RETCODE_LIMIT_VOLUME + 10009: "open", + 10010: "filled", + 10011: "canceled", + 10012: "canceled", + 10013: "rejected", + 10014: "rejected", + 10015: "rejected", + 10016: "rejected", + 10017: "rejected", + 10018: "rejected", + 10019: "rejected", + 10020: "rejected", + 10021: "rejected", + 10022: "rejected", + 10023: "rejected", + 10024: "rejected", + 10025: "rejected", } -# MT5 retcode -> True if it indicates insufficient funds MT5_RETCODE_INSUFFICIENT_FUNDS = {10019, 10020, 10021, 10022, 10025} @@ -124,9 +123,7 @@ class mt5bridge(ccxt.Exchange): secret: not used (placeholder for CCXT compatibility) host: base URL of MT5Bridge (default: http://localhost:8080) timeout: request timeout in ms (default: 30000) - symbols: optional dict mapping CCXT symbols to MT5 symbols, e.g. - {"XAU/USD": "XAUUSDc", "EUR/USD": "EURUSDc"} - If not provided, the wrapper strips "/" from CCXT symbols. + symbols: optional dict mapping CCXT symbols to MT5 symbols default_volume_step: default lot rounding (default 0.01) Example: @@ -143,7 +140,7 @@ class mt5bridge(ccxt.Exchange): id = "mt5bridge" name = "MT5 Bridge" countries = ["*"] - version = "0.1.0" + version = "0.2.0" rateLimit = 100 certified = False pro = False @@ -185,6 +182,16 @@ class mt5bridge(ccxt.Exchange): "setMarginMode": False, "transfer": False, "withdraw": False, + # ── Watching (polling-based, see watch_* methods) ── + "watchBalance": False, + "watchMyTrades": False, + "watchOHLCV": True, # polling-based + "watchOrderBook": False, + "watchOrders": False, + "watchPositions": False, + "watchTicker": True, # polling-based + "watchTickers": False, + "watchTrades": False, } timeframes = { @@ -197,6 +204,18 @@ class mt5bridge(ccxt.Exchange): "1d": "TIMEFRAME_D1", } + # Polling intervals per timeframe (used by watch_ohlcv as the minimum + # interval to check for a new bar) + WATCH_OHLCV_POLL_MS = { + "1m": 5000, + "5m": 10000, + "15m": 15000, + "30m": 30000, + "1h": 60000, + "4h": 120000, + "1d": 300000, + } + urls = { "api": "http://localhost:8080", } @@ -239,16 +258,14 @@ class mt5bridge(ccxt.Exchange): def __init__(self, config={}): super().__init__(config) - # Configuration self.host = (config.get("host") or self.urls["api"]).rstrip("/") self.apiKey = config.get("apiKey") or self.apiKey or "" self.timeout = config.get("timeout") or 30000 self.default_volume_step = config.get("default_volume_step", 0.01) self.symbol_map = config.get("symbols", {}) or {} - # Reverse map for displaying MT5 symbols as CCXT symbols self._mt5_to_ccxt = {v: k for k, v in self.symbol_map.items()} - # HTTP session with retry logic + # HTTP session with retry self.session = requests.Session() retry = Retry( total=3, @@ -266,47 +283,25 @@ class mt5bridge(ccxt.Exchange): "User-Agent": f"mt5bridge-ccxt/{self.version}", }) - # MQL5 helper self.mql5 = Mql5Bridge(self) # ──────── Internal helpers ──────── def _resolve_mt5_symbol(self, symbol): - """Convert CCXT symbol (XAU/USD) to MT5 symbol (XAUUSDc).""" if not symbol: return symbol if symbol in self.symbol_map: return self.symbol_map[symbol] - # Fallback: strip slash return symbol.replace("/", "") def _resolve_ccxt_symbol(self, mt5_symbol): - """Convert MT5 symbol back to CCXT format for display.""" if not mt5_symbol: return mt5_symbol if mt5_symbol in self._mt5_to_ccxt: return self._mt5_to_ccxt[mt5_symbol] - # Best-effort fallback return mt5_symbol def _request(self, method, path, params=None, json_body=None): - """Make HTTP request to MT5Bridge with proper error handling. - - Args: - method: HTTP method (GET, POST, DELETE) - path: API path (without host) - params: query parameters - json_body: JSON request body for POST - - Returns: - dict: response JSON - - Raises: - Mt5BridgeAuthError: HTTP 401 - Mt5BridgeInvalidRequestError: HTTP 400 - Mt5BridgeNotConnectedError: HTTP 503 - Mt5BridgeError: other failures - """ url = f"{self.host}/{path.lstrip('/')}" timeout_s = self.timeout / 1000.0 @@ -329,7 +324,6 @@ class mt5bridge(ccxt.Exchange): except requests.exceptions.RequestException as e: raise Mt5BridgeError(f"Request failed: {e}") from e - # Map HTTP status codes to exceptions if response.status_code == 401: raise Mt5BridgeAuthError("Unauthorized: check your API key") if response.status_code == 503: @@ -356,7 +350,6 @@ class mt5bridge(ccxt.Exchange): ) from e def _extract_detail(self, response, fallback): - """Extract 'detail' field from a JSON error response.""" try: body = response.json() return body.get("detail") or fallback @@ -364,13 +357,6 @@ class mt5bridge(ccxt.Exchange): return fallback or response.text[:200] def _parse_datetime(self, dt_str): - """Parse ISO datetime to millisecond timestamp. - - Handles formats: - "yyyy-MM-ddTHH:mm:ss" - "yyyy-MM-ddTHH:mm:ss.fff000" - "yyyy-MM-dd" - """ if not dt_str: return None try: @@ -387,11 +373,9 @@ class mt5bridge(ccxt.Exchange): return None def _get_order_type_code(self, order_type, side): - """Map CCXT (order_type, side) to MT5 order type code.""" key = (order_type, side) if key in MT5_ORDER_TYPE: return MT5_ORDER_TYPE[key] - # Fallback: try the more general mapping valid_types = ("market", "limit", "stop", "stop_limit") if order_type not in valid_types: raise InvalidOrder( @@ -400,15 +384,12 @@ class mt5bridge(ccxt.Exchange): ) if side not in ("buy", "sell"): raise InvalidOrder(f"Unsupported side '{side}'. Valid: buy, sell") - # Should not reach here given valid_types raise InvalidOrder(f"Unsupported combination: {order_type} + {side}") def _get_status_from_retcode(self, retcode): - """Map MT5 retcode to CCXT status string.""" return MT5_RETCODE_STATUS.get(int(retcode), "rejected") def _format_volume(self, amount): - """Round volume to nearest step.""" if amount is None: return 0.0 try: @@ -416,14 +397,19 @@ class mt5bridge(ccxt.Exchange): except (TypeError, ValueError): return 0.0 + @staticmethod + def safe_float(d, key, default=None): + try: + v = d.get(key) + if v is None: + return default + return float(v) + except (TypeError, ValueError): + return default + # ──────── Public API: status & markets ──────── def fetch_status(self, params={}): - """Check MT5Bridge and MT5 connection status. - - Returns: - dict: {"status": "ok"|"maintenance", "updated": ms, "info": ...} - """ try: r = self._request("GET", "/health") except Mt5BridgeNotConnectedError: @@ -443,27 +429,14 @@ class mt5bridge(ccxt.Exchange): } def fetch_markets(self, params={}): - """Fetch available markets. - - Note: MT5Bridge doesn't expose a "list all symbols" endpoint. - Markets are built from the `symbols` config (CCXT -> MT5 mapping). - To add a market, populate the `symbols` config dict. - - Returns: - list of market dicts (CCXT format) - """ if not self.symbol_map: - # Try to discover from positions / tickers return [] - markets = [] for ccxt_symbol, mt5_symbol in self.symbol_map.items(): markets.append(self._build_market(ccxt_symbol, mt5_symbol)) return markets def _build_market(self, ccxt_symbol, mt5_symbol): - """Build a CCXT market dict from symbol info.""" - # Try to get symbol info (optional - don't fail if unavailable) info_data = {} try: r = self._request("GET", f"/symbols/{mt5_symbol}") @@ -484,7 +457,7 @@ class mt5bridge(ccxt.Exchange): "baseId": base, "quoteId": quote, "active": True, - "type": "swap", # MT5 OTC/CFD + "type": "swap", "spot": False, "margin": True, "swap": True, @@ -513,18 +486,9 @@ class mt5bridge(ccxt.Exchange): "info": info_data, } - # ──────── Public API: market data ──────── + # ──────── Public API: market data (fetch_*) ──────── def fetch_ticker(self, symbol, params={}): - """Fetch current ticker for a symbol. - - Args: - symbol: CCXT symbol (e.g. "XAU/USD") - params: extra params - - Returns: - dict: CCXT ticker - """ mt5_symbol = self._resolve_mt5_symbol(symbol) r = self._request("GET", f"/symbols/{mt5_symbol}/tick") data = r["data"][0] @@ -558,20 +522,6 @@ class mt5bridge(ccxt.Exchange): } def fetch_ohlcv(self, symbol, timeframe="1h", since=None, limit=None, params={}): - """Fetch OHLCV candlestick data. - - Args: - symbol: CCXT symbol (e.g. "XAU/USD") - timeframe: "1m" | "5m" | "15m" | "30m" | "1h" | "4h" | "1d" - since: start timestamp in milliseconds (optional) - limit: max number of bars (default 100, max 10000) - params: extra params. Can include: - - "date_from": explicit start date "yyyy-MM-dd" - - "date_to": explicit end date "yyyy-MM-dd" - - Returns: - list of [timestamp, open, high, low, close, volume] - """ if timeframe not in self.timeframes: raise BadRequest( f"Unsupported timeframe '{timeframe}'. " @@ -581,7 +531,6 @@ class mt5bridge(ccxt.Exchange): mt5_tf = self.timeframes[timeframe] limit = min(limit or 100, 10000) - # Decide endpoint if "date_from" in params and "date_to" in params: query = { "symbol": mt5_symbol, @@ -623,24 +572,9 @@ class mt5bridge(ccxt.Exchange): ]) return result - @staticmethod - def safe_float(d, key, default=None): - try: - v = d.get(key) - if v is None: - return default - return float(v) - except (TypeError, ValueError): - return default - # ──────── Public API: account & positions ──────── def fetch_balance(self, params={}): - """Fetch account balance. - - Returns: - dict: {info, timestamp, free, used, total, debt} - """ r = self._request("GET", "/account") acc = r["data"][0] currency = acc.get("currency") or "USD" @@ -657,15 +591,6 @@ class mt5bridge(ccxt.Exchange): } def fetch_positions(self, symbols=None, params={}): - """Fetch open positions. - - Args: - symbols: optional list of CCXT symbols to filter - params: extra params - - Returns: - list of CCXT position dicts - """ r = self._request("GET", "/positions") positions = r.get("data", []) @@ -718,30 +643,14 @@ class mt5bridge(ccxt.Exchange): return result def fetch_position(self, symbol, params={}): - """Fetch a single position by symbol. - - Note: MT5 can have multiple positions per symbol; this returns the - most recently opened one (or None if no position exists). - """ positions = self.fetch_positions([symbol], params) if not positions: return None - # Return the one with highest ticket (most recent) return max(positions, key=lambda p: int(p["info"].get("ticket", 0))) # ──────── Public API: orders & trades ──────── def fetch_open_orders(self, symbol=None, since=None, limit=None, params={}): - """Fetch open (pending) orders. - - Args: - symbol: optional CCXT symbol to filter - since: not used by MT5Bridge - limit: not used - - Returns: - list of order dicts - """ mt5_symbol = self._resolve_mt5_symbol(symbol) if symbol else None query = {"symbol": mt5_symbol} if mt5_symbol else {} r = self._request("GET", "/orders", params=query) @@ -749,34 +658,14 @@ class mt5bridge(ccxt.Exchange): return [self._parse_order(o, status="open") for o in orders] def fetch_closed_orders(self, symbol=None, since=None, limit=None, params={}): - """Fetch closed orders via history. - - MT5Bridge exposes history as deals, not orders. We approximate - closed orders by reading the deal history and pairing entries - with exits. - """ - # For simplicity, return my_trades; for proper order tracking - # one would need to join deals with the order they belong to. return self.fetch_my_trades(symbol, since, limit, params) def fetch_my_trades(self, symbol=None, since=None, limit=None, params={}): - """Fetch historical trades. - - Args: - symbol: optional CCXT symbol - since: start timestamp in ms - limit: not used (returns all in range) - params: can include "date_from", "date_to" - - Returns: - list of CCXT trade dicts - """ if "date_from" in params and "date_to" in params: date_from = params["date_from"] date_to = params["date_to"] else: if since is None: - # Default: last 7 days since_ts = int((time.time() - 7 * 86400) * 1000) else: since_ts = int(since) @@ -826,7 +715,6 @@ class mt5bridge(ccxt.Exchange): return trades def _parse_order(self, o, status="open"): - """Convert MT5Bridge order dict to CCXT order dict.""" ts = self._parse_datetime(o.get("time")) ccxt_sym = self._resolve_ccxt_symbol(o.get("symbol", "")) mt5_type = int(o.get("type", 0)) @@ -868,30 +756,6 @@ class mt5bridge(ccxt.Exchange): # ──────── Public API: trading ──────── def create_order(self, symbol, type, side, amount, price=None, params={}): - """Create a new order. - - Args: - symbol: CCXT symbol (e.g. "XAU/USD") - type: "market" | "limit" | "stop" | "stop_limit" - side: "buy" | "sell" - amount: volume in lots (e.g. 0.01) - price: required for limit/stop orders - params: extra params: - - sl: stop loss price (default 0) - - tp: take profit price (default 0) - - magic: magic number (default 0) - - comment: order comment, max 31 chars - - deviation: slippage in points (default 10) - - check_first: if True, validate via /order/check first - - Returns: - dict: CCXT order - - Raises: - InvalidOrder: order was rejected by broker - InsufficientFunds: retcode indicates not enough margin - ArgumentsRequired: missing required argument - """ if type not in ("market", "limit", "stop", "stop_limit"): raise InvalidOrder( f"Unsupported order type '{type}'. " @@ -921,7 +785,6 @@ class mt5bridge(ccxt.Exchange): "deviation": int(params.get("deviation", 10)), } - # Optional pre-check if params.get("check_first", False): check = self._request("POST", "/order/check", json_body=body) check_result = check.get("data", {}) @@ -936,7 +799,6 @@ class mt5bridge(ccxt.Exchange): f"comment={check_result.get('comment', '')}" ) - # Send order r = self._request("POST", "/order/send", json_body={"request": body}) result = r.get("data", {}) retcode = int(result.get("retcode", 0)) @@ -953,7 +815,6 @@ class mt5bridge(ccxt.Exchange): return self._make_order_from_result(result, symbol, type, side, body) def _make_order_from_result(self, result, symbol, type, side, body): - """Build a CCXT order dict from a successful order result.""" ts = self.milliseconds() order_id = str(result.get("order", "")) return { @@ -980,22 +841,6 @@ class mt5bridge(ccxt.Exchange): } def cancel_order(self, id, symbol=None, params={}): - """Cancel a pending order (async via MQL5 helper EA). - - Note: MT5Bridge doesn't expose a synchronous cancel endpoint. This - method sends a GlobalVariable command that the MQL5 helper EA - (Mt5BridgeHelper.mq5) reads and acts on by calling OrderDelete(). - - For reliable cancellation, deploy the helper EA on a separate chart. - - Args: - id: order ticket (string or int) - symbol: optional symbol (unused) - params: extra params - - Returns: - dict: {"info": ..., "id": str, "status": "canceling"} - """ ticket = int(id) result = self.mql5.send_cancel_command(ticket) return { @@ -1003,3 +848,114 @@ class mt5bridge(ccxt.Exchange): "id": str(id), "status": "canceling", } + + # ───────────────────────────────────────────────────────────────── + # Watching (polling-based implementations of CCXT watch_* methods) + # ───────────────────────────────────────────────────────────────── + # + # MT5Bridge doesn't expose WebSocket or SSE. The watch_* methods + # below poll the synchronous fetch_* endpoints until a new value + # is detected, then return it. This is functionally equivalent to + # a real watch for low-frequency use cases. + # + # For high-frequency real-time streaming, upgrade MT5Bridge to + # support WebSocket (see README of this package). + + async def watch_ticker(self, symbol, params={}): + """Watch the ticker for a symbol by polling. + + Polls the /symbols/{symbol}/tick endpoint until a new tick is + detected (timestamp changes, or bid/ask changes), then returns it. + + Args: + symbol: CCXT symbol + params: extra params: + - poll_interval_ms: polling interval in ms (default 500) + - max_wait_ms: max time to wait before returning current + value even if no change (default 30000) + + Returns: + dict: CCXT ticker + + Example: + >>> while True: + ... ticker = await exchange.watch_ticker("XAU/USD") + ... print(ticker["bid"], ticker["ask"]) + """ + poll_interval_ms = params.get("poll_interval_ms", 500) + max_wait_ms = params.get("max_wait_ms", 30000) + last_signature = None + loop = asyncio.get_event_loop() + deadline = loop.time() + (max_wait_ms / 1000.0) + + while True: + ticker = await loop.run_in_executor(None, self.fetch_ticker, symbol) + # Use (timestamp, bid, ask) as the change signature + signature = (ticker.get("timestamp"), ticker.get("bid"), ticker.get("ask")) + + if last_signature is None or signature != last_signature: + last_signature = signature + return ticker + + # No change yet + now = loop.time() + if now >= deadline: + return ticker # return current value, even if unchanged + + await asyncio.sleep(poll_interval_ms / 1000.0) + + async def watch_ohlcv(self, symbol, timeframe="1m", since=None, limit=None, params={}): + """Watch OHLCV for a symbol by polling. + + Polls fetch_ohlcv until a new bar is detected (latest bar's + timestamp changes), then returns the new bars. + + Args: + symbol: CCXT symbol + timeframe: CCXT timeframe + since: optional start timestamp in ms + limit: number of bars to return (default 100) + params: extra params: + - poll_interval_ms: minimum polling interval in ms + (default is auto-set per timeframe) + - max_wait_ms: max time to wait before returning (default 600000) + + Returns: + list of [timestamp, open, high, low, close, volume] + """ + poll_interval_ms = params.get("poll_interval_ms", + self.WATCH_OHLCV_POLL_MS.get(timeframe, 30000)) + max_wait_ms = params.get("max_wait_ms", 600000) + last_bar_time = None + loop = asyncio.get_event_loop() + deadline = loop.time() + (max_wait_ms / 1000.0) + fetch_limit = min(limit or 2, 5) # only need last few bars to detect new one + + def _fetch(): + return self.fetch_ohlcv(symbol, timeframe, since=since, limit=fetch_limit, params=params) + + while True: + ohlcv = await loop.run_in_executor(None, _fetch) + if ohlcv: + latest_time = ohlcv[-1][0] + if last_bar_time is None or latest_time != last_bar_time: + last_bar_time = latest_time + # Return full requested limit, not just the last few + if limit and limit > fetch_limit: + return await loop.run_in_executor( + None, + lambda: self.fetch_ohlcv( + symbol, timeframe, since=since, limit=limit, params=params + ), + ) + return ohlcv + + now = loop.time() + if now >= deadline: + # Return what we have, even if no new bar + if ohlcv: + return ohlcv + # No data at all; let the user see the empty result + return ohlcv + + await asyncio.sleep(poll_interval_ms / 1000.0)