diff --git a/docsrc/brokers.rst b/docsrc/brokers.rst index 5f60fede7..bd696ea53 100644 --- a/docsrc/brokers.rst +++ b/docsrc/brokers.rst @@ -33,5 +33,6 @@ Broker setup is easier on `BotSpot `_ is a hosted MetaTrader 5 API. This broker +lets a Lumibot strategy trade any MetaTrader 5 account (Forex, metals, indices, +CFDs, crypto) through that hosted API, so it runs on **any operating system with +no local MetaTrader 5 terminal installed** - unlike the official MetaTrader5 +Python package, which is Windows-only and requires a running terminal. + +How to Use TickerAll +-------------------- + +1. Install Lumibot with the optional TickerAll dependency: + + .. code-block:: shell + + pip install lumibot[tickerall] + +2. Create an account at `tickerall.com `_ and connect one + or more MetaTrader 5 broker accounts in the dashboard. +3. Generate an API key. +4. Set the environment variables below (or pass a ``config`` dict to the broker). + +The hosted API supports market data (historical bars, last price, quotes), +account balances and open positions, and order management (market, limit and +stop orders, with optional stop-loss and take-profit). + +**Environment Variables** + +Set the following in your ``.env`` file or system environment: + +.. code-block:: shell + + TICKERALL_API_KEY=your_tickerall_api_key + # Optional: only needed when the API key has more than one connected account + TICKERALL_ACCOUNT_ID=your_connected_account_id + +Instruments are addressed by their MetaTrader 5 symbol (for example +``EURUSDm``, ``XAUUSDm`` or ``BTCUSD``). Construct assets with the ``forex`` +asset type so open positions reconcile against the orders you submit: + +.. code-block:: python + + from lumibot.entities import Asset + + asset = Asset("EURUSDm", asset_type="forex") + +Example Usage +------------- + +**Creating the broker** + +.. code-block:: python + + from lumibot.brokers import TickerAll + from lumibot.traders import Trader + + config = { + "API_KEY": "your_tickerall_api_key", + # "ACCOUNT_ID": "your_connected_account_id", # only if the key has several accounts + } + broker = TickerAll(config) + + trader = Trader() + strategy = MyStrategy(broker=broker) + trader.add_strategy(strategy) + trader.run_all() + +**Placing a market order with stop-loss and take-profit** + +.. code-block:: python + + from lumibot.entities import Asset, Order + + asset = Asset("EURUSDm", asset_type="forex") + order = self.create_order( + asset=asset, + quantity=0.10, # volume in lots + side=Order.OrderSide.BUY, + order_type=Order.OrderType.MARKET, + # stop-loss / take-profit are attached as a bracket order + secondary_stop_price=1.0800, # stop-loss price + secondary_limit_price=1.1000, # take-profit price + ) + submitted_order = self.submit_order(order) + if submitted_order: + self.log_message(f"Placed order: ID={submitted_order.identifier}, Status={submitted_order.status}") + +**Placing a limit order** + +.. code-block:: python + + asset = Asset("XAUUSDm", asset_type="forex") + order = self.create_order( + asset=asset, + quantity=0.10, + side=Order.OrderSide.BUY, + order_type=Order.OrderType.LIMIT, + limit_price=2350.00, + ) + self.submit_order(order) + +**Reading bars, last price and account value** + +.. code-block:: python + + asset = Asset("EURUSDm", asset_type="forex") + bars = self.get_historical_prices(asset, 100, "day") + last = self.get_last_price(asset) + cash = self.get_cash() + self.log_message(f"Cash {cash}, last {last}, {len(bars.df)} bars") + +**Closing a position** + +.. code-block:: python + + asset = Asset("EURUSDm", asset_type="forex") + self.close_position(asset) + +**Cancelling open (pending) orders** + +.. code-block:: python + + for order in self.get_orders(): + if order.is_active(): + self.cancel_order(order) + +.. note:: + The hosted API offers ``market``, ``limit`` and ``stop`` orders. ``stop_limit`` + and ``trailing_stop`` order types are not available and are rejected with a + clear message rather than being silently mishandled. Order volume is + expressed in lots. + +Documentation +--------------- + +.. automodule:: lumibot.brokers.tickerall + :noindex: + :members: + :undoc-members: + :show-inheritance: + +.. automodule:: lumibot.data_sources.tickerall_data + :noindex: + :members: + :undoc-members: + :show-inheritance: diff --git a/lumibot/brokers/__init__.py b/lumibot/brokers/__init__.py index 947ab89ed..8943442f3 100644 --- a/lumibot/brokers/__init__.py +++ b/lumibot/brokers/__init__.py @@ -14,6 +14,7 @@ "ProjectX": "projectx", "Polymarket": "polymarket", "Schwab": "schwab", + "TickerAll": "tickerall", "Tradier": "tradier", "Tradovate": "tradovate", } diff --git a/lumibot/brokers/tickerall.py b/lumibot/brokers/tickerall.py new file mode 100644 index 000000000..c2014d1ec --- /dev/null +++ b/lumibot/brokers/tickerall.py @@ -0,0 +1,572 @@ +""" +TickerAll hosted MT5 API broker for Lumibot. + +Connects Lumibot to the hosted TickerAll MetaTrader 5 API +(https://tickerall.com), giving strategies access to the many global brokers +that run on MetaTrader 5 (Forex, metals, indices, CFDs, crypto). Because trading +goes through a hosted API, this broker runs on any OS with **no local +MetaTrader 5 terminal** installed - unlike the official MetaTrader5 Python +package, which is Windows-only and needs a running terminal. + +Design notes +------------ +- Modeled on the polling brokers in this codebase (``ccxt.py`` for the + hosted-API shape, ``bitunix.py`` for the ``PollingStream`` fill loop). +- Order fills: market orders fill synchronously, so the fill event is dispatched + from ``_submit_order``. Pending (limit/stop) orders are reported as NEW when + placed; their eventual state is reconciled by the polling loop, and open + positions are always kept in sync from the broker snapshot (the source of + truth for what is actually open). +- Closing positions: ``close_position`` / ``sell_all`` use the hosted API's + NATIVE position-close (by ticket) rather than the base broker's offsetting + sell order. On a netting account this flattens cleanly and avoids leaving a + transient phantom position in the tracker. +- Asset type: MT5 instruments are addressed by a single opaque symbol string + (e.g. ``EURUSDm``, ``XAUUSDm``, ``BTCUSD``). This integration treats every MT5 + instrument under a single canonical asset type (``forex``) so that positions + parsed from the broker match orders submitted by a strategy (Lumibot keys + positions on symbol AND asset type). Construct assets as + ``Asset("EURUSDm", asset_type="forex")``. +- Supported order types: ``market``, ``limit``, ``stop``. ``stop_limit`` and + ``trailing_stop`` are not offered by the hosted API and are rejected with a + clear message rather than silently mishandled. + +License: MIT +""" + +from __future__ import annotations + +import os + +from lumibot.tools.lumibot_logger import get_logger + +from .broker import Broker + +logger = get_logger(__name__) + + +def _colored(*args, **kwargs): + try: + from termcolor import colored + return colored(*args, **kwargs) + except Exception: + return args[0] + + +class TickerAll(Broker): + """Broker that trades MT5 accounts through the hosted TickerAll API.""" + + NAME = "TickerAll" + POLL_EVENT = "poll" + DEFAULT_POLL_INTERVAL = 5 # seconds between polling cycles + + # Lumibot order types the hosted MT5 API can place. + _SUPPORTED_ORDER_TYPES = {"market", "limit", "stop"} + + def __init__( + self, + config=None, + data_source=None, + connect_stream: bool = True, + poll_interval: float | None = None, + max_workers: int = 1, + **kwargs, + ): + from lumibot.data_sources import TickerAllData + + if data_source is None: + data_source = TickerAllData(config) + if not isinstance(data_source, TickerAllData): + raise ValueError(f"TickerAll broker's data source must be a TickerAllData, got {type(data_source)}") + + # Share the hosted-API client with the data source (single connection). + self.api = data_source.api + self.poll_interval = poll_interval or self.DEFAULT_POLL_INTERVAL + # Tickets we cancelled ourselves, so the poll loop routes them to CANCELED + # (rather than guessing they filled). + self._cancelled_tickets: set[str] = set() + + # MT5 markets are ~24/5 (closed weekends). Default to always-tradeable + # (the broker enforces real session hours and rejects out-of-session + # orders); a user can pass config["MARKET"] to use a market calendar. + cfg_market = config.get("MARKET") if isinstance(config, dict) else None + desired_market = cfg_market or os.environ.get("MARKET") or "24/7" + + super().__init__( + name=self.NAME, + connect_stream=connect_stream, + data_source=data_source, + config=config, + max_workers=max_workers, + **kwargs, + ) + # The base __init__ resets self.market (defaulting to NASDAQ for an + # unknown data source); apply the MT5 market setting after it. + self.market = desired_market + + # ── shared helpers ─────────────────────────────────────────────────────── + @property + def account_id(self) -> str: + return self.data_source.account_id + + def _resolve_symbol(self, asset) -> str: + return self.data_source.resolve_symbol(asset) + + def _asset_from_symbol(self, symbol: str): + from lumibot.data_sources.tickerall_data import CANONICAL_ASSET_TYPE + from lumibot.entities import Asset + + return Asset(symbol=symbol, asset_type=CANONICAL_ASSET_TYPE) + + # ── balances ───────────────────────────────────────────────────────────── + def _get_balances_at_broker(self, quote_asset, strategy) -> tuple: + """Return (cash, positions_value, portfolio_value) as floats. + + cash = account balance (booked) + portfolio_value = account equity (balance + floating P&L) = net liquidation + positions_value = portfolio_value - cash (value tied up in open positions) + """ + detail = self.api.accounts.get(self.account_id) + acc = detail.account + if acc is None: + logger.warning( + f"TickerAll account snapshot unavailable ({detail.hint or 'offline'}); " + "reporting zero balances." + ) + return 0.0, 0.0, 0.0 + + # NOTE: a balance of 0.0 is a valid account state - never treat 0 as "missing". + cash = float(acc.balance) + equity = float(acc.equity) if acc.equity is not None else cash + positions_value = equity - cash + return cash, positions_value, equity + + def get_historical_account_value(self) -> dict: + # The hosted API does not expose a historical equity curve. + return {"hourly": None, "daily": None} + + # ── positions ──────────────────────────────────────────────────────────── + def _net_position(self, plist, strategy): + """Aggregate the broker positions for one symbol into a single net + Lumibot position (BUY +, SELL -). + + A netting account has at most one position per symbol; a HEDGING account + can hold several (long and short) at once. Lumibot keys positions on the + asset (one per symbol), so we present the net exposure and remember every + underlying ticket for closing. + """ + from lumibot.entities import Position + + net = notional = total_vol = pnl = 0.0 + current = None + tickets = [] + for p in plist: + vol = float(p.volume) + net += -vol if str(p.side).upper() == "SELL" else vol + total_vol += vol + if p.entry_price is not None: + notional += vol * float(p.entry_price) + if p.profit is not None: + pnl += float(p.profit) + if current is None and p.current_price is not None: + current = float(p.current_price) + tickets.append(int(p.ticket)) + + asset = self._asset_from_symbol(plist[0].symbol) + avg = (notional / total_vol) if total_vol else None + pos = Position(strategy, asset, round(net, 8), avg_fill_price=avg) + # Fields Lumibot attaches dynamically (not constructor args). + if current is not None: + pos.current_price = current + pos.pnl = pnl + pos.broker_tickets = tickets # every underlying ticket (hedging may have >1) + pos.broker_ticket = tickets[0] if tickets else None + return pos + + def _pull_positions(self, strategy) -> list: + detail = self.api.accounts.get(self.account_id) + strategy_name = self._strategy_name_from_input(strategy) if strategy is not None else self._strategy_name + by_symbol: dict = {} + for p in detail.positions: + by_symbol.setdefault(p.symbol, []).append(p) + return [self._net_position(plist, strategy_name) for plist in by_symbol.values()] + + def _pull_position(self, strategy, asset): + target = self._resolve_symbol(asset) + detail = self.api.accounts.get(self.account_id) + strategy_name = self._strategy_name_from_input(strategy) if strategy is not None else self._strategy_name + plist = [p for p in detail.positions if p.symbol == target] + return self._net_position(plist, strategy_name) if plist else None + + # ── orders (pull side) ─────────────────────────────────────────────────── + def _pull_broker_all_orders(self) -> list: + """Open (pending) orders at the broker. Filled market orders are not + listed here - they become positions immediately.""" + try: + return list(self.api.orders.list_pending(self.account_id)) + except Exception as e: + logger.warning(f"Could not pull pending orders from TickerAll: {e}") + return [] + + def _pull_broker_order(self, identifier: str): + for o in self._pull_broker_all_orders(): + if str(o.ticket) == str(identifier): + return o + return None + + def _parse_broker_order(self, response, strategy_name, strategy_object=None): + from lumibot.entities import Order + + symbol = getattr(response, "symbol", None) + asset = self._asset_from_symbol(symbol) + side = "buy" if str(getattr(response, "side", "BUY")).upper() == "BUY" else "sell" + volume = float(getattr(response, "volume", 0) or 0) + + # Map the MT5 pending-order type to a Lumibot order type. + raw_type = str(getattr(response, "order_type", None) or getattr(response, "type", "")).lower() + if "limit" in raw_type: + order_type = Order.OrderType.LIMIT + elif "stop" in raw_type: + order_type = Order.OrderType.STOP + else: + order_type = Order.OrderType.LIMIT + + limit_price = getattr(response, "limit_price", None) or getattr(response, "price", None) + stop_price = getattr(response, "price", None) if order_type == Order.OrderType.STOP else None + sl = getattr(response, "stop_loss", None) or None + tp = getattr(response, "take_profit", None) or None + + order = Order( + strategy_name, + asset, + volume, + side, + order_type=order_type, + limit_price=float(limit_price) if limit_price else None, + stop_price=float(stop_price) if stop_price else None, + secondary_stop_price=float(sl) if sl else None, + secondary_limit_price=float(tp) if tp else None, + ) + order.set_identifier(str(getattr(response, "ticket", ""))) + order.status = "open" + order.update_raw(response) + return order + + # ── order submission ───────────────────────────────────────────────────── + @staticmethod + def _extract_sl_tp(order): + """Pull stop-loss / take-profit prices off a (possibly bracket) order.""" + sl = getattr(order, "secondary_stop_price", None) or getattr(order, "stop_loss_price", None) + tp = getattr(order, "secondary_limit_price", None) or getattr(order, "take_profit_price", None) + # Fall back to scanning child orders (bracket / OTO shape). + for child in getattr(order, "child_orders", None) or []: + ctype = str(getattr(child, "order_type", "")).lower() + if tp is None and "limit" in ctype and getattr(child, "limit_price", None): + tp = child.limit_price + if sl is None and "stop" in ctype and getattr(child, "stop_price", None): + sl = child.stop_price + return (float(sl) if sl else None, float(tp) if tp else None) + + def _submit_order(self, order): + + # --- validate quantity --- + if getattr(order, "quantity", None) is None or float(order.quantity) <= 0: + logger.warning(f"Order {order} rejected: quantity must be > 0.") + order.set_error("Order quantity must be greater than 0.") + return order + + # --- map order type --- + order_type = str(order.order_type).lower() + if order_type not in self._SUPPORTED_ORDER_TYPES: + msg = ( + f"Order type '{order_type}' is not supported by the TickerAll hosted MT5 API. " + f"Supported types: market, limit, stop." + ) + logger.error(_colored(msg, "red")) + order.set_error(msg) + return order + + side = "BUY" if str(order.side).lower().startswith("buy") else "SELL" + symbol = self._resolve_symbol(order.asset) + volume = float(order.quantity) + sl, tp = self._extract_sl_tp(order) + + # Price for pending orders (limit -> limit_price, stop -> stop_price). + price = None + if order_type == "limit": + price = float(order.limit_price) if order.limit_price is not None else None + elif order_type == "stop": + price = float(order.stop_price) if order.stop_price is not None else None + + try: + result = self.api.orders.place( + self.account_id, + type=order_type, + symbol=symbol, + side=side, + volume=volume, + price=price, + stop_loss=sl, + take_profit=tp, + comment=(order.tag or "lumibot"), + ) + except Exception as e: + msg = f"{order} did not go through. Error: {e}" + logger.error(_colored(msg, "red")) + order.set_error(e) + return order + + order.set_identifier(str(result.ticket)) + order.update_raw(result) # marks the order transmitted so the base tracks it + + if order_type == "market": + # Market orders fill immediately; report the fill so on_filled_order fires. + fill_price = self._fill_price(result, symbol) + filled_qty = float(result.volume) if getattr(result, "volume", None) else volume + self._report_market_fill(order, fill_price, filled_qty) + else: + # Pending order now resting at the broker. + order.status = "new" + self._dispatch(self.NEW_ORDER, order=order) + + return order + + def _fill_price(self, result, symbol: str) -> float: + """Best available fill price for a just-filled market order.""" + if getattr(result, "price", None): + return float(result.price) + # Fall back to the resulting position's entry price, then last price. + try: + detail = self.api.accounts.get(self.account_id) + for p in detail.positions: + if int(p.ticket) == int(result.ticket) and p.entry_price is not None: + return float(p.entry_price) + except Exception: + pass + last = self.data_source.get_last_price(self._asset_from_symbol(symbol)) + return float(last) if last else 0.0 + + def _report_market_fill(self, order, price, quantity): + """Dispatch NEW then FILLED for a synchronously-filled market order.""" + self._dispatch(self.NEW_ORDER, order=order) + self._dispatch(self.FILLED_ORDER, order=order, price=price, filled_quantity=quantity) + # Force an immediate position reconciliation from the broker snapshot so a + # close-via-offset order does not leave a transient phantom position in the + # tracker (on a netting account the offsetting fill nets to flat). + stream = getattr(self, "stream", None) + if stream is not None: + stream.dispatch(self.POLL_EVENT) + + def _dispatch(self, event, **payload): + """Queue an event on the polling stream, or process it inline if no stream.""" + stream = getattr(self, "stream", None) + if stream is not None: + stream.dispatch(event, **payload) + else: + order = payload.pop("order", None) + self._process_trade_event(order, event, **payload) + + # ── cancel / modify ────────────────────────────────────────────────────── + def cancel_order(self, order) -> None: + if order is None or not order.identifier: + return + # Skip only if the order is already terminal. Do NOT skip on a local + # "cancelling" status - the caller sets that right before calling this, + # and is_canceled() treats "cancelling" as canceled. (Broker adapters + # must not treat local CANCELLING as terminal before sending the cancel.) + if order.is_filled() or str(order.status).lower() in ("canceled", "cancelled", "expired"): + return + try: + self.api.orders.cancel_pending(self.account_id, int(order.identifier)) + self._cancelled_tickets.add(str(order.identifier)) + self._dispatch(self.CANCELED_ORDER, order=order) + except Exception as e: + logger.error(_colored(f"Could not cancel order {order}: {e}", "red")) + + def _modify_order(self, order, limit_price: float | None = None, stop_price: float | None = None): + if order is None or not order.identifier: + return + new_price = limit_price if limit_price is not None else stop_price + try: + self.api.orders.modify_pending( + self.account_id, + int(order.identifier), + price=float(new_price) if new_price is not None else None, + ) + if limit_price is not None: + order.limit_price = limit_price + if stop_price is not None: + order.stop_price = stop_price + except Exception as e: + logger.error(_colored(f"Could not modify order {order}: {e}", "red")) + + # ── closing positions ──────────────────────────────────────────────────── + def close_position(self, strategy_name: str, asset, fraction: float = 1.00): + """Close a position using the hosted API's NATIVE position-close. + + MT5 positions are closed directly by ticket, not by an offsetting order. + Using the native close (rather than the base broker's offsetting sell) + flattens cleanly on a netting account and avoids the transient phantom + position an offsetting-order fill would leave in the tracker. + """ + symbol = self._resolve_symbol(asset) + closed = 0 + for p in self.api.accounts.get(self.account_id).positions: + if p.symbol != symbol: + continue + vol = round(float(p.volume) * float(fraction), 8) if fraction and float(fraction) < 1.0 else None + try: + self.api.positions.close(self.account_id, int(p.ticket), volume=vol) + closed += 1 + except Exception as e: + logger.error(_colored(f"Could not close position {p.ticket}: {e}", "red")) + if closed: + self._force_position_sync() + return None + + def sell_all(self, strategy_name, cancel_open_orders=True, strategy=None, is_multileg=False): + """Flatten all positions via the native position-close (see close_position).""" + logger.warning(_colored(f"Closing all positions for {strategy_name}", "yellow")) + if cancel_open_orders: + self.cancel_open_orders(strategy_name) + closed = 0 + for p in self.api.accounts.get(self.account_id).positions: + try: + self.api.positions.close(self.account_id, int(p.ticket)) + closed += 1 + except Exception as e: + logger.error(_colored(f"Could not close position {p.ticket}: {e}", "red")) + if closed: + self._force_position_sync() + + def cancel_open_orders(self, strategy_name=None): + """Cancel every pending order at the broker.""" + for o in self._pull_broker_all_orders(): + try: + self.api.orders.cancel_pending(self.account_id, int(o.ticket)) + self._cancelled_tickets.add(str(o.ticket)) + except Exception as e: + logger.error(_colored(f"Could not cancel pending order {o.ticket}: {e}", "red")) + + def _force_position_sync(self): + """Reconcile the tracker from the broker snapshot right away.""" + stream = getattr(self, "stream", None) + if stream is not None: + stream.dispatch(self.POLL_EVENT) + else: + self.sync_positions(None) + + # ── polling stream ─────────────────────────────────────────────────────── + def _get_stream_object(self): + from lumibot.trading_builtins import PollingStream + + return PollingStream(self.poll_interval) + + def _register_stream_events(self): + broker = self + + @broker.stream.add_action(broker.POLL_EVENT) + def on_poll(): + try: + broker.do_polling() + except Exception as e: + logger.error(f"TickerAll polling error: {e}") + + @broker.stream.add_action(broker.NEW_ORDER) + def on_new(order): + try: + broker._process_trade_event(order, broker.NEW_ORDER) + except Exception as e: + logger.error(f"TickerAll new-order event error: {e}") + + @broker.stream.add_action(broker.FILLED_ORDER) + def on_fill(order, price, filled_quantity): + try: + broker._process_trade_event( + order, broker.FILLED_ORDER, price=price, filled_quantity=filled_quantity + ) + except Exception as e: + logger.error(f"TickerAll fill event error: {e}") + + @broker.stream.add_action(broker.CANCELED_ORDER) + def on_cancel(order): + try: + broker._process_trade_event(order, broker.CANCELED_ORDER) + except Exception as e: + logger.error(f"TickerAll cancel event error: {e}") + + def _run_stream(self): + self._stream_established() + try: + self.stream._run() + except Exception as e: + logger.error(f"Error running TickerAll polling stream: {e}") + + def do_polling(self): + """Reconcile positions and pending orders once per poll cycle. + + Positions are the source of truth for what is open, so they are always + synced from the broker snapshot. Pending (limit/stop) orders are + reconciled against the tracked order set: newly-seen pending orders fire + NEW; a tracked order that has left the pending list is reported FILLED + when a matching position exists, otherwise CANCELED. + """ + # 1) Positions: authoritative sync from the broker snapshot. + self.sync_positions(None) + + # 2) Pending orders. + raw_orders = self._pull_broker_all_orders() + broker_ids = {str(o.ticket) for o in raw_orders} + stored = {o.identifier: o for o in self.get_all_orders() if o.identifier} + + for o in raw_orders: + parsed = self._parse_broker_order(o, strategy_name=self._strategy_name) + if parsed is None: + continue + if parsed.identifier not in stored: + # A pending order we have not tracked yet. + self._process_trade_event(parsed, self.NEW_ORDER) + + # Tracked active orders that are no longer pending at the broker. + open_positions = {p.asset.symbol for p in self._filled_positions.get_list()} + for order in self.get_tracked_orders(): + if not order.identifier or not order.is_active(): + continue + if order.identifier in broker_ids: + continue # still pending + if str(order.identifier) in self._cancelled_tickets: + # We cancelled it; the cancel event was already dispatched. + self._cancelled_tickets.discard(str(order.identifier)) + continue + # Left the pending list without our cancel. A fill that nets a + # netting position to exactly zero removes the position from the + # snapshot, so "a position exists" is not a reliable filled/canceled + # signal. Resolve it from the order's terminal state in history + # (deal_count > 0 means it produced a fill), and only fall back to + # the position-existence check when history is unavailable. + terminal = self._pending_terminal_state(order.identifier) + filled = (terminal == "filled") if terminal is not None else (order.asset.symbol in open_positions) + if filled: + price = order.limit_price or order.stop_price or order.avg_fill_price or 0.0 + self._process_trade_event( + order, self.FILLED_ORDER, price=float(price), filled_quantity=float(order.quantity) + ) + else: + self._process_trade_event(order, self.CANCELED_ORDER) + + def _pending_terminal_state(self, identifier): + """Resolve whether a pending order that left the book was filled or + canceled, from recent order history. ``deal_count > 0`` (or a filled + state) means it produced a fill. Returns 'filled', 'canceled', or None + when it cannot be determined. + """ + try: + for h in self.api.history.orders(self.account_id, limit=50): + if str(h.order_ticket) == str(identifier): + dealt = int(getattr(h, "deal_count", 0) or 0) > 0 + if dealt or "fill" in str(getattr(h, "state", "")).lower(): + return "filled" + return "canceled" + except Exception as e: + logger.debug(f"Could not resolve terminal state for order {identifier}: {e}") + return None diff --git a/lumibot/data_sources/__init__.py b/lumibot/data_sources/__init__.py index 659da1e34..f7d96ea4d 100644 --- a/lumibot/data_sources/__init__.py +++ b/lumibot/data_sources/__init__.py @@ -23,6 +23,7 @@ "PolygonDataBacktesting": "polygon_backtesting", "ProjectXData": "projectx_data", "SchwabData": "schwab_data", + "TickerAllData": "tickerall_data", "TradierData": "tradier_data", "TradovateData": "tradovate_data", "UnavailabeTimestep": "exceptions", diff --git a/lumibot/data_sources/tickerall_data.py b/lumibot/data_sources/tickerall_data.py new file mode 100644 index 000000000..5a43a4f0f --- /dev/null +++ b/lumibot/data_sources/tickerall_data.py @@ -0,0 +1,301 @@ +""" +TickerAll hosted MT5 API data source for Lumibot. + +Connects to the hosted TickerAll MetaTrader 5 API (https://tickerall.com) to +serve historical bars, last prices, and quotes for MT5 instruments (Forex, +metals, indices, CFDs, crypto). Because the data comes from a hosted API, this +data source runs on any OS with no local MetaTrader 5 terminal installed. + +It is paired with ``lumibot.brokers.TickerAll`` (which shares this data source's +client), but can also be used stand-alone for research/backtesting-data pulls. + +License: MIT +""" + +from __future__ import annotations + +import datetime as dt +import os +from decimal import Decimal +from threading import RLock + +import pandas as pd + +from lumibot.entities import Asset, Bars, Quote +from lumibot.tools.lumibot_logger import get_logger + +from .data_source import DataSource + +logger = get_logger(__name__) + +# MT5 timeframes the hosted API serves natively. +_MT5_TIMEFRAMES = {"M1", "M5", "M15", "M30", "H1", "H4", "D1", "W1", "MN1"} +# Minutes-per-bar, used to translate a ``timeshift`` timedelta into a bar count. +_TF_MINUTES = { + "M1": 1, "M5": 5, "M15": 15, "M30": 30, + "H1": 60, "H4": 240, "D1": 1440, "W1": 10080, "MN1": 43200, +} +# The canonical asset type this integration uses for every MT5 instrument, so +# that positions parsed from the broker match orders submitted by a strategy +# (Lumibot keys positions on symbol AND asset_type). See the broker docstring. +CANONICAL_ASSET_TYPE = "forex" + + +class TickerAllData(DataSource): + """Serves market data from the hosted TickerAll MT5 API.""" + + SOURCE = "TICKERALL" + MIN_TIMESTEP = "minute" + IS_BACKTESTING_DATA_SOURCE = False + # Lumibot canonical timestep -> the hosted API's timeframe string. + TIMESTEP_MAPPING = [ + {"timestep": "minute", "representations": ["M1"]}, + {"timestep": "hour", "representations": ["H1"]}, + {"timestep": "day", "representations": ["D1"]}, + {"timestep": "week", "representations": ["W1"]}, + {"timestep": "month", "representations": ["MN1"]}, + ] + + def __init__(self, config, **kwargs): + super().__init__(**kwargs) + self.name = "tickerall" + self._config = config or {} + + # Credentials come from the config dict, and fall back to the documented + # TICKERALL_* environment variables so either path works. + api_key = ( + self._cfg("API_KEY") + or self._cfg("TICKERALL_API_KEY") + or os.environ.get("TICKERALL_API_KEY") + ) + if not api_key: + raise ValueError( + "TickerAll data source requires an API key. Set config['API_KEY'] " + "or the TICKERALL_API_KEY environment variable." + ) + + # Import lazily so lumibot stays importable without the optional dependency. + try: + from tickerall import Tickerall + except ImportError as e: + raise ImportError( + "The 'tickerall' package is required for the TickerAll data source. " + "Install it with: pip install tickerall" + ) from e + + base_url = self._cfg("BASE_URL") or os.environ.get("TICKERALL_BASE_URL") + self.api = Tickerall(api_key=api_key, base_url=base_url) if base_url else Tickerall(api_key=api_key) + + self._configured_account_id = ( + self._cfg("ACCOUNT_ID") + or self._cfg("TICKERALL_ACCOUNT_ID") + or os.environ.get("TICKERALL_ACCOUNT_ID") + ) + self._account_id: str | None = None + self._symbols: list | None = None + self._stream = None + self._subscribed: set[str] = set() + # Reentrant: account_id and _ensure_symbols both guard state under this + # lock, and _ensure_symbols resolves account_id while holding it. + self._lock = RLock() + + # ── config helper ──────────────────────────────────────────────────────── + def _cfg(self, key): + if isinstance(self._config, dict): + return self._config.get(key) + return getattr(self._config, key, None) + + # ── account / symbol resolution ────────────────────────────────────────── + @property + def account_id(self) -> str: + if self._account_id is not None: + return self._account_id + with self._lock: + if self._account_id is not None: + return self._account_id + if self._configured_account_id: + self._account_id = self._configured_account_id + return self._account_id + accounts = self.api.accounts.list() + if len(accounts) == 1: + self._account_id = accounts[0].id + elif not accounts: + raise ValueError( + "No accounts are connected to this TickerAll API key. Connect one " + "in the dashboard, or set config['ACCOUNT_ID']." + ) + else: + ids = ", ".join(f"{a.id} ({a.server} #{a.account_number})" for a in accounts) + raise ValueError( + "This TickerAll API key has multiple accounts; set config['ACCOUNT_ID'] " + f"to one of: {ids}" + ) + return self._account_id + + def _ensure_symbols(self) -> list: + if self._symbols is None: + with self._lock: + if self._symbols is None: + try: + self._symbols = list(self.api.accounts.symbols(self.account_id)) + except Exception as e: # pragma: no cover - network dependent + # Do NOT cache the failure - leave _symbols None so a + # transient hiccup retries on the next call instead of + # permanently disabling symbol resolution. + logger.warning(f"Could not fetch symbol list from TickerAll: {e}") + return [] + return self._symbols + + def resolve_symbol(self, asset: Asset | str) -> str: + """Map a lumibot Asset/str to the broker's real (case-sensitive) symbol. + + Lumibot upper-cases asset symbols at construction, but MT5 broker symbols + are case-sensitive (e.g. ``EURUSDm``). Resolve case-insensitively against + the account's live symbol list; fall back to the requested symbol. + """ + requested = asset.symbol if isinstance(asset, Asset) else str(asset) + symbols = self._ensure_symbols() + if requested in symbols: + return requested + lowered = requested.lower() + for s in symbols: + if s.lower() == lowered: + return s + return requested + + def _to_timeframe(self, timestep: str) -> str: + """Translate a lumibot timestep (or a raw MT5 timeframe) to an API timeframe.""" + if not timestep: + timestep = self.MIN_TIMESTEP + ts = str(timestep).strip() + # Accept a raw MT5 timeframe (M5, M15, H4, ...) passed through directly. + if ts.upper() in _MT5_TIMEFRAMES: + return ts.upper() + # Otherwise map the lumibot canonical timestep via TIMESTEP_MAPPING. + try: + return self._parse_source_timestep(ts, reverse=True) + except Exception: + # Common shorthand fallbacks. + fallback = {"minute": "M1", "hour": "H1", "day": "D1", "week": "W1", "month": "MN1"} + return fallback.get(ts.lower(), "M1") + + # ── market data ────────────────────────────────────────────────────────── + def get_historical_prices( + self, asset, length, timestep="", timeshift=None, quote=None, + exchange=None, include_after_hours=True, **kwargs + ) -> Bars | None: + """Return the most recent ``length`` bars as a Bars object, or None if unavailable.""" + if isinstance(asset, str): + asset = Asset(symbol=asset, asset_type=CANONICAL_ASSET_TYPE) + if exchange is not None: + logger.warning(f"TickerAllData ignores the 'exchange' parameter (got {exchange}).") + + tf = self._to_timeframe(timestep or self.get_timestep()) + symbol = self.resolve_symbol(asset) + + # If a timeshift is requested, fetch extra bars and trim the most-recent ones. + shift_bars = 0 + if timeshift is not None: + if not isinstance(timeshift, dt.timedelta): + timeshift = dt.timedelta(days=timeshift) + shift_bars = int(timeshift.total_seconds() // 60 // _TF_MINUTES.get(tf, 1)) + + want = int(length) + shift_bars + candles = self.api.candles.get(self.account_id, symbol=symbol, count=want, timeframe=tf) + if not candles: + # RULE #1: no fabricated data - surface the absence honestly. + logger.warning(f"No {tf} bars returned by TickerAll for {symbol}.") + return None + + candles = sorted(candles, key=lambda c: c.timestamp) + if shift_bars: + candles = candles[:-shift_bars] if shift_bars < len(candles) else [] + candles = candles[-int(length):] + if not candles: + return None + + df = pd.DataFrame( + [ + { + "datetime": c.timestamp, + "open": c.open, + "high": c.high, + "low": c.low, + "close": c.close, + "volume": c.tick_volume, + } + for c in candles + ] + ) + df["datetime"] = pd.to_datetime(df["datetime"], unit="s", utc=True) + df = df.set_index("datetime") + return Bars(df, self.SOURCE, asset, quote=quote, raw=df) + + def get_last_price(self, asset, quote=None, exchange=None, **kwargs) -> float | Decimal | None: + """Return the latest price (mid of live bid/ask; falls back to last close).""" + symbol = self.resolve_symbol(asset) + tick = self._latest_tick(symbol) + if tick is not None: + bid, ask = tick + if bid and ask: + return (float(bid) + float(ask)) / 2.0 + return float(ask or bid) + # Fallback: most recent 1-minute close. + candles = self.api.candles.get(self.account_id, symbol=symbol, count=1, timeframe="M1") + if candles: + return float(candles[-1].close) + logger.warning(f"No last price available for {symbol} via TickerAll.") + return None + + def get_quote(self, asset: Asset, quote: Asset = None, exchange: str = None) -> Quote: + """Return a Quote with live bid/ask when available.""" + symbol = self.resolve_symbol(asset) + tick = self._latest_tick(symbol) + if tick is not None: + bid, ask = tick + mid = (float(bid) + float(ask)) / 2.0 if (bid and ask) else float(ask or bid) + return Quote(asset=asset, price=mid, bid=float(bid) if bid else None, + ask=float(ask) if ask else None) + candles = self.api.candles.get(self.account_id, symbol=symbol, count=1, timeframe="M1") + if candles: + c = candles[-1] + return Quote(asset=asset, price=float(c.close), + bid=float(c.bid) if c.bid else None, ask=None) + return Quote(asset=asset) + + def get_chains(self, asset: Asset, quote: Asset = None, exchange: str = None) -> dict: + # MT5 instruments have no option chains. + return {} + + # ── live tick helpers ──────────────────────────────────────────────────── + def _latest_tick(self, symbol: str): + """Return (bid, ask) from the live stream, or None if unavailable.""" + try: + stream = self._ensure_stream() + if symbol not in self._subscribed: + stream.subscribe_ticks(self.account_id, [symbol]) + self._subscribed.add(symbol) + ev = stream.wait_for_tick(symbol, account_id=self.account_id, timeout=6.0) + return (ev.bid, ev.ask) + except Exception as e: + logger.debug(f"No live tick for {symbol} ({e}); falling back to candle close.") + return None + + def _ensure_stream(self): + if self._stream is None or not self._stream.is_connected(): + self._stream = self.api.stream.connect(timeout=15.0) + self._subscribed = set() + return self._stream + + def close(self): + """Release the stream and HTTP client.""" + try: + if self._stream is not None: + self._stream.close() + self._stream = None + except Exception: + pass + try: + self.api.close() + except Exception: + pass diff --git a/setup.py b/setup.py index b2320a33f..c3b240359 100644 --- a/setup.py +++ b/setup.py @@ -120,7 +120,12 @@ def _maybe_copy_theta_terminal(self): "thetadata": [ "thetadata", ], + # Optional dependency to enable the TickerAll hosted MT5 API broker + "tickerall": [ + "tickerall>=0.1.16", + ], }, + keywords=[ "algorithmic-trading", "backtesting", diff --git a/tests/test_broker_tickerall.py b/tests/test_broker_tickerall.py new file mode 100644 index 000000000..19f6829ee --- /dev/null +++ b/tests/test_broker_tickerall.py @@ -0,0 +1,358 @@ +"""Unit tests for the TickerAll hosted MT5 API broker + data source. + +The hosted-API client is mocked, so these run in CI with no network or +credentials. The whole module is skipped if the optional ``tickerall`` package +is not installed. +""" +import os +import unittest +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +pytest.importorskip("tickerall") + +from lumibot.brokers import TickerAll # noqa: E402 +from lumibot.data_sources import TickerAllData # noqa: E402 +from lumibot.entities import Asset, Order # noqa: E402 + + +def _account_detail(balance=1000.0, equity=1050.0, positions=None, is_demo=True, has_account=True): + account = None + if has_account: + account = SimpleNamespace( + name="Demo", account_type="demo", leverage=100, balance=balance, + broker_name="TestBroker", currency="USD", equity=equity, margin=10.0, + free_margin=equity - 10.0, margin_level=200.0, + ) + return SimpleNamespace( + id="acc1", broker="mt5", server="Test-MT5", account_number="12345678", + is_demo=is_demo, status="CONNECTED", account=account, positions=positions or [], hint=None, + ) + + +def _position(ticket=1, symbol="EURUSDm", side="BUY", volume=0.1, entry=1.10, current=1.11, profit=1.0): + return SimpleNamespace( + ticket=ticket, symbol=symbol, side=side, volume=volume, stop_loss=0.0, take_profit=0.0, + magic=0, comment="", swap=0.0, commission=0.0, open_time=None, + entry_price=entry, current_price=current, profit=profit, last_update=None, + ) + + +class TestTickerAllBroker(unittest.TestCase): + def setUp(self): + self.patcher = patch("tickerall.Tickerall") + self.MockTA = self.patcher.start() + self.client = MagicMock() + self.MockTA.return_value = self.client + self.client.accounts.symbols.return_value = ["EURUSDm", "BTCUSDm", "XAUUSDm"] + self.cfg = {"API_KEY": "test_key", "ACCOUNT_ID": "acc1"} + + def tearDown(self): + self.patcher.stop() + + def _broker(self, connect_stream=False): + ds = TickerAllData(self.cfg) + broker = TickerAll(self.cfg, data_source=ds, connect_stream=connect_stream) + broker.stream = MagicMock() # capture dispatched events without a real thread + return broker + + # ── construction ───────────────────────────────────────────────────────── + def test_initialization(self): + broker = self._broker() + self.assertEqual(broker.NAME, "TickerAll") + self.assertEqual(broker.api, self.client) + self.assertEqual(broker.account_id, "acc1") + # MT5 default market is always-tradeable unless a calendar is configured. + self.assertEqual(broker.market, "24/7") + + def test_requires_tickerall_data_source(self): + with self.assertRaises(ValueError): + TickerAll(self.cfg, data_source=object(), connect_stream=False) + + # ── symbol resolution (case-sensitive MT5 symbols) ──────────────────────── + def test_symbol_resolution_case_insensitive(self): + broker = self._broker() + # Lumibot upper-cases the symbol; it must resolve back to the real casing. + self.assertEqual(broker._resolve_symbol(Asset("EURUSDM", asset_type="forex")), "EURUSDm") + self.assertEqual(broker._resolve_symbol("btcusdm"), "BTCUSDm") + # Unknown symbol falls back to the requested value. + self.assertEqual(broker._resolve_symbol("UNKNOWNm"), "UNKNOWNm") + + # ── balances ────────────────────────────────────────────────────────────── + def test_balances(self): + broker = self._broker() + self.client.accounts.get.return_value = _account_detail(balance=1000.0, equity=1050.0) + cash, positions_value, portfolio = broker._get_balances_at_broker(Asset("USD", asset_type="forex"), "s") + self.assertEqual(cash, 1000.0) + self.assertAlmostEqual(positions_value, 50.0) + self.assertEqual(portfolio, 1050.0) + + def test_balance_of_zero_is_valid(self): + # A balance of 0.0 is a real, valid account state (never treat 0 as missing). + broker = self._broker() + self.client.accounts.get.return_value = _account_detail(balance=0.0, equity=0.0) + cash, positions_value, portfolio = broker._get_balances_at_broker(Asset("USD", asset_type="forex"), "s") + self.assertEqual((cash, positions_value, portfolio), (0.0, 0.0, 0.0)) + + def test_balances_no_account_snapshot(self): + broker = self._broker() + self.client.accounts.get.return_value = _account_detail(has_account=False) + self.assertEqual(broker._get_balances_at_broker(Asset("USD", asset_type="forex"), "s"), (0.0, 0.0, 0.0)) + + # ── positions ────────────────────────────────────────────────────────────── + def test_short_position_is_negative_quantity(self): + broker = self._broker() + self.client.accounts.get.return_value = _account_detail( + positions=[_position(side="SELL", volume=0.2, symbol="EURUSDm")] + ) + positions = broker._pull_positions("s") + self.assertEqual(len(positions), 1) + self.assertEqual(float(positions[0].quantity), -0.2) # SELL -> negative + self.assertEqual(positions[0].broker_ticket, 1) + + def test_long_position_is_positive_quantity(self): + broker = self._broker() + self.client.accounts.get.return_value = _account_detail( + positions=[_position(side="BUY", volume=0.15)] + ) + positions = broker._pull_positions("s") + self.assertEqual(float(positions[0].quantity), 0.15) + + def test_hedging_positions_aggregate_to_net(self): + # A hedging account can hold several positions per symbol; they must + # aggregate into one net Lumibot position (BUY +, SELL -), tracking + # every underlying ticket for closing. + broker = self._broker() + self.client.accounts.get.return_value = _account_detail(positions=[ + _position(ticket=1, symbol="EURUSDm", side="BUY", volume=0.2), + _position(ticket=2, symbol="EURUSDm", side="SELL", volume=0.05), + ]) + positions = broker._pull_positions("s") + self.assertEqual(len(positions), 1) # one net position, not two colliding + self.assertAlmostEqual(float(positions[0].quantity), 0.15) # 0.20 - 0.05 + self.assertEqual(sorted(positions[0].broker_tickets), [1, 2]) + + # ── order type mapping ────────────────────────────────────────────────────── + def test_submit_market_order_dispatches_fill(self): + broker = self._broker() + self.client.orders.place.return_value = SimpleNamespace( + ticket=999, symbol="EURUSDm", side="BUY", type="market", volume=0.1, + status="DONE", timestamp="", price=1.105, stop_loss=None, take_profit=None, comment="", + ) + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", order_type=Order.OrderType.MARKET) + broker._submit_order(order) + + self.client.orders.place.assert_called_once() + _, kwargs = self.client.orders.place.call_args + self.assertEqual(kwargs["type"], "market") + self.assertEqual(kwargs["side"], "BUY") + self.assertEqual(kwargs["volume"], 0.1) + self.assertEqual(order.identifier, "999") + # A market fill dispatches NEW then FILLED on the stream. + events = [c.args[0] for c in broker.stream.dispatch.call_args_list] + self.assertIn(broker.NEW_ORDER, events) + self.assertIn(broker.FILLED_ORDER, events) + + def test_submit_limit_order_is_pending(self): + broker = self._broker() + self.client.orders.place.return_value = SimpleNamespace( + ticket=1001, symbol="EURUSDm", side="BUY", type="limit", volume=0.1, + status="PLACED", timestamp="", price=1.08, stop_loss=None, take_profit=None, comment="", + ) + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", + order_type=Order.OrderType.LIMIT, limit_price=1.08) + broker._submit_order(order) + _, kwargs = self.client.orders.place.call_args + self.assertEqual(kwargs["type"], "limit") + self.assertEqual(kwargs["price"], 1.08) + self.assertEqual(order.status, "new") # resting pending order + + def test_submit_rejects_unsupported_order_type(self): + broker = self._broker() + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", + order_type=Order.OrderType.STOP_LIMIT, limit_price=1.1, stop_price=1.09) + result = broker._submit_order(order) + self.assertEqual(result.status, "error") + self.client.orders.place.assert_not_called() + + def test_submit_rejects_zero_quantity(self): + broker = self._broker() + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.0, "buy", order_type=Order.OrderType.MARKET) + result = broker._submit_order(order) + self.assertEqual(result.status, "error") + self.client.orders.place.assert_not_called() + + def test_market_order_forwards_sl_tp(self): + broker = self._broker() + self.client.orders.place.return_value = SimpleNamespace( + ticket=1, symbol="EURUSDm", side="BUY", type="market", volume=0.1, + status="DONE", timestamp="", price=1.1, stop_loss=1.08, take_profit=1.12, comment="", + ) + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", + order_type=Order.OrderType.MARKET, + secondary_stop_price=1.08, secondary_limit_price=1.12) + broker._submit_order(order) + _, kwargs = self.client.orders.place.call_args + self.assertEqual(kwargs["stop_loss"], 1.08) + self.assertEqual(kwargs["take_profit"], 1.12) + + # ── cancel ────────────────────────────────────────────────────────────────── + def test_cancel_order(self): + broker = self._broker() + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", order_type=Order.OrderType.LIMIT, + limit_price=1.05) + order.set_identifier("777") + order.status = "new" + broker.cancel_order(order) + self.client.orders.cancel_pending.assert_called_once_with("acc1", 777) + + def test_cancel_order_proceeds_on_cancelling_status(self): + # Strategy.cancel_order sets status to "cancelling" right before calling + # the broker; is_canceled() treats "cancelling" as canceled, so the guard + # must NOT skip on it (otherwise the broker cancel never fires). + broker = self._broker() + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", order_type=Order.OrderType.LIMIT, + limit_price=1.05) + order.set_identifier("778") + order.status = "cancelling" + broker.cancel_order(order) + self.client.orders.cancel_pending.assert_called_once_with("acc1", 778) + + def test_cancel_order_skips_terminal(self): + broker = self._broker() + order = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", order_type=Order.OrderType.LIMIT, + limit_price=1.05) + order.set_identifier("779") + order.status = "canceled" + broker.cancel_order(order) + self.client.orders.cancel_pending.assert_not_called() + + def _tracked_pending(self, broker, identifier="555"): + """Track a pending limit order as active (as after a real submit).""" + broker.stream = None # do_polling processes events inline + self.client.history.orders.return_value = [] # default: no history match + o = Order("s", Asset("EURUSDm", asset_type="forex"), 0.1, "buy", + order_type=Order.OrderType.LIMIT, limit_price=1.0) + o.set_identifier(identifier) + o.update_raw({}) + broker._process_trade_event(o, broker.NEW_ORDER) + return o + + def test_do_polling_history_reports_fill_when_position_netted_to_zero(self): + # A pending order that fills and nets a netting position to exactly zero + # leaves NO position in the snapshot; history (deal_count > 0) must still + # report it as FILLED, not CANCELED. + broker = self._broker() + o = self._tracked_pending(broker, identifier="600") + self.client.orders.list_pending.return_value = [] + self.client.accounts.get.return_value = _account_detail(positions=[]) # net zero -> no position + self.client.history.orders.return_value = [ + SimpleNamespace(order_ticket="600", symbol="EURUSDm", side="BUY", volume=0.1, + price=1.0, time="", position_id="0", state="filled", deal_count=1), + ] + broker.do_polling() + self.assertTrue(o.is_filled()) + + def test_do_polling_history_deal_count_zero_is_cancel(self): + broker = self._broker() + o = self._tracked_pending(broker, identifier="601") + self.client.orders.list_pending.return_value = [] + self.client.accounts.get.return_value = _account_detail(positions=[]) + self.client.history.orders.return_value = [ + SimpleNamespace(order_ticket="601", symbol="EURUSDm", side="BUY", volume=0.1, + price=1.0, time="", position_id="0", state="canceled", deal_count=0), + ] + broker.do_polling() + self.assertTrue(o.is_canceled()) + + def test_do_polling_fills_pending_when_position_appears(self): + # A tracked pending order that left the broker list AND has a matching + # position -> reconciled as FILLED. + broker = self._broker() + o = self._tracked_pending(broker) + self.client.orders.list_pending.return_value = [] + self.client.accounts.get.return_value = _account_detail( + positions=[_position(symbol="EURUSDm", side="BUY", volume=0.1)] + ) + broker.do_polling() + self.assertTrue(o.is_filled()) + + def test_do_polling_cancels_pending_when_no_position(self): + # A tracked pending order that vanished with no matching position -> CANCELED. + broker = self._broker() + o = self._tracked_pending(broker, identifier="556") + self.client.orders.list_pending.return_value = [] + self.client.accounts.get.return_value = _account_detail(positions=[]) + broker.do_polling() + self.assertTrue(o.is_canceled()) + + +class TestTickerAllData(unittest.TestCase): + def setUp(self): + self.patcher = patch("tickerall.Tickerall") + self.MockTA = self.patcher.start() + self.client = MagicMock() + self.MockTA.return_value = self.client + self.client.accounts.symbols.return_value = ["EURUSDm", "BTCUSDm"] + self.ds = TickerAllData({"API_KEY": "k", "ACCOUNT_ID": "acc1"}) + + def tearDown(self): + self.patcher.stop() + + def test_timeframe_mapping(self): + self.assertEqual(self.ds._to_timeframe("day"), "D1") + self.assertEqual(self.ds._to_timeframe("minute"), "M1") + self.assertEqual(self.ds._to_timeframe("hour"), "H1") + # Raw MT5 timeframes pass through. + self.assertEqual(self.ds._to_timeframe("M5"), "M5") + self.assertEqual(self.ds._to_timeframe("H4"), "H4") + + def test_get_historical_prices_builds_bars(self): + self.client.candles.get.return_value = [ + SimpleNamespace(timestamp=1_700_000_000 + i * 86400, open=1.1, high=1.2, low=1.0, + close=1.15, bid=1.149, tick_volume=100, spread=2) + for i in range(5) + ] + bars = self.ds.get_historical_prices(Asset("EURUSDm", asset_type="forex"), 5, "day") + self.assertIsNotNone(bars) + self.assertEqual(len(bars.df), 5) + for col in ("open", "high", "low", "close", "volume"): + self.assertIn(col, bars.df.columns) + self.assertIsNotNone(bars.df.index.tz) # tz-aware index + + def test_get_historical_prices_empty_returns_none(self): + # RULE #1: no fabricated data - absence is surfaced honestly. + self.client.candles.get.return_value = [] + self.assertIsNone(self.ds.get_historical_prices(Asset("EURUSDm", asset_type="forex"), 5, "day")) + + def test_get_last_price_candle_fallback(self): + # No live tick available -> fall back to the latest candle close. + self.client.stream.connect.side_effect = Exception("no stream") + self.client.candles.get.return_value = [ + SimpleNamespace(timestamp=1_700_000_000, open=1.1, high=1.2, low=1.0, + close=1.2345, bid=1.234, tick_volume=1, spread=2) + ] + self.assertEqual(self.ds.get_last_price(Asset("EURUSDm", asset_type="forex")), 1.2345) + + def test_get_chains_empty(self): + self.assertEqual(self.ds.get_chains(Asset("EURUSDm", asset_type="forex")), {}) + + def test_env_var_credentials(self): + # The documented TICKERALL_* environment variables must authenticate and + # select an account with no config dict passed. + with patch.dict(os.environ, {"TICKERALL_API_KEY": "envkey", "TICKERALL_ACCOUNT_ID": "envacct"}, clear=False): + ds = TickerAllData({}) + self.MockTA.assert_called_with(api_key="envkey") + self.assertEqual(ds._configured_account_id, "envacct") + + def test_symbol_fetch_failure_not_cached(self): + # A transient symbol-fetch failure must not be cached permanently. + self.client.accounts.symbols.side_effect = [Exception("hiccup"), ["EURUSDm", "BTCUSDm"]] + self.assertEqual(self.ds._ensure_symbols(), []) # first call fails, returns [] + self.assertEqual(self.ds._ensure_symbols(), ["EURUSDm", "BTCUSDm"]) # retries, succeeds + + +if __name__ == "__main__": + unittest.main()