From 8f6ba1781de85f5d3007f606a2a0183a694c342e Mon Sep 17 00:00:00 2001 From: miguelangelo78 Date: Sun, 26 Jul 2026 14:16:54 +0000 Subject: [PATCH 1/4] Add TickerAll hosted MT5 API broker and data source Adds an optional broker and data source for the hosted TickerAll MetaTrader 5 API (https://tickerall.com), so a Lumibot strategy can trade any MT5 account (Forex, metals, indices, CFDs, crypto) on any operating system with no local MetaTrader 5 terminal installed. Addresses the MT5 integration request in #977. The integration is fully additive: two new modules plus one line each in the broker and data-source registries and the docs toctree. No existing code paths are changed. - lumibot/brokers/tickerall.py: TickerAll(Broker) with account balances, open positions, market/limit/stop orders (with optional stop-loss/take-profit), cancel and modify, and a PollingStream fill loop (modeled on the ccxt and bitunix brokers). Unsupported order types (stop_limit, trailing_stop) are rejected with a clear message rather than being silently mishandled. - lumibot/data_sources/tickerall_data.py: TickerAllData(DataSource) with historical bars, last price and quotes. Missing bars return None rather than fabricated data. - docsrc/brokers.tickerall.rst: usage documentation, wired into the brokers toctree. - tests/test_broker_tickerall.py: unit tests with the hosted client mocked (skipped when the optional tickerall package is not installed). The tickerall package is an optional dependency, imported lazily, so Lumibot stays importable without it. --- docsrc/brokers.rst | 1 + docsrc/brokers.tickerall.rst | 141 ++++++++ lumibot/brokers/__init__.py | 1 + lumibot/brokers/tickerall.py | 466 +++++++++++++++++++++++++ lumibot/data_sources/__init__.py | 1 + lumibot/data_sources/tickerall_data.py | 287 +++++++++++++++ tests/test_broker_tickerall.py | 249 +++++++++++++ 7 files changed, 1146 insertions(+) create mode 100644 docsrc/brokers.tickerall.rst create mode 100644 lumibot/brokers/tickerall.py create mode 100644 lumibot/data_sources/tickerall_data.py create mode 100644 tests/test_broker_tickerall.py 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. Create an account at `tickerall.com `_ and connect one + or more MetaTrader 5 broker accounts in the dashboard. +2. Generate an API key. +3. 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..e0f11ab22 --- /dev/null +++ b/lumibot/brokers/tickerall.py @@ -0,0 +1,466 @@ +""" +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). +- 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 _parse_broker_position(self, position, strategy): + from lumibot.entities import Position + + symbol = position.symbol + asset = self._asset_from_symbol(symbol) + volume = float(position.volume) + # SHORT positions carry a negative quantity in Lumibot. + quantity = -volume if str(position.side).upper() == "SELL" else volume + entry = float(position.entry_price) if position.entry_price is not None else None + + pos = Position(strategy, asset, quantity, avg_fill_price=entry) + # Fields Lumibot attaches dynamically (not constructor args). + if position.current_price is not None: + pos.current_price = float(position.current_price) + if position.profit is not None: + pos.pnl = float(position.profit) + # Keep the broker ticket for closing/reconciliation. + pos.broker_ticket = int(position.ticket) + 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 + result = [] + for p in detail.positions: + parsed = self._parse_broker_position(p, strategy_name) + if parsed is not None: + result.append(parsed) + return result + + 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 + for p in detail.positions: + if p.symbol == target: + return self._parse_broker_position(p, strategy_name) + return 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) + + 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 + if order.is_filled() or order.is_canceled(): + 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")) + + # ── 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: filled if a position + # for the asset now exists, otherwise treat as canceled externally. + if order.asset.symbol in open_positions: + 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) 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..180f2388b --- /dev/null +++ b/lumibot/data_sources/tickerall_data.py @@ -0,0 +1,287 @@ +""" +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 +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 {} + + api_key = self._cfg("API_KEY") or self._cfg("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") + 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") + 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 + logger.warning(f"Could not fetch symbol list from TickerAll: {e}") + self._symbols = [] + 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/tests/test_broker_tickerall.py b/tests/test_broker_tickerall.py new file mode 100644 index 000000000..3c7e7ac8d --- /dev/null +++ b/tests/test_broker_tickerall.py @@ -0,0 +1,249 @@ +"""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 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) + + # ── 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) + + +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")), {}) + + +if __name__ == "__main__": + unittest.main() From 0b22787738c6d3df82b51c450135849c69a4e768 Mon Sep 17 00:00:00 2001 From: miguelangelo78 Date: Sun, 26 Jul 2026 17:46:57 +0000 Subject: [PATCH 2/4] Harden TickerAll broker after full live-strategy verification Found and fixed by running a complete Strategy-through-Trader flow live on both a netting and a hedging demo account: - cancel_order: do not skip when the local order status is "cancelling". The strategy sets that status right before calling the broker, and is_canceled() treats "cancelling" as canceled, so the broker cancel was being silently skipped and the order stayed open at the broker. - close_position / sell_all: use the hosted API's native position-close (by ticket) instead of the base broker's offsetting sell order. On a netting account the offsetting fill could leave a transient phantom position in the tracker; closing by ticket flattens cleanly. Adds cancel_open_orders. - positions: aggregate the broker positions for a symbol into one net Lumibot position (BUY +, SELL -), so hedging accounts (which can hold several positions per symbol) map onto Lumibot's one-position-per-asset model, and remember every underlying ticket for closing. - reconcile the position tracker from the broker snapshot right after a fill. Tests: adds coverage for the cancelling-status guard, the terminal-status skip, the polling fill/cancel reconciliation, and hedging aggregation (24 total, all passing). --- lumibot/brokers/tickerall.py | 133 ++++++++++++++++++++++++++------- tests/test_broker_tickerall.py | 66 ++++++++++++++++ 2 files changed, 174 insertions(+), 25 deletions(-) diff --git a/lumibot/brokers/tickerall.py b/lumibot/brokers/tickerall.py index e0f11ab22..2df25dafd 100644 --- a/lumibot/brokers/tickerall.py +++ b/lumibot/brokers/tickerall.py @@ -17,6 +17,10 @@ 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 @@ -142,44 +146,57 @@ def get_historical_account_value(self) -> dict: return {"hourly": None, "daily": None} # ── positions ──────────────────────────────────────────────────────────── - def _parse_broker_position(self, position, strategy): + 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 - symbol = position.symbol - asset = self._asset_from_symbol(symbol) - volume = float(position.volume) - # SHORT positions carry a negative quantity in Lumibot. - quantity = -volume if str(position.side).upper() == "SELL" else volume - entry = float(position.entry_price) if position.entry_price is not None else None - - pos = Position(strategy, asset, quantity, avg_fill_price=entry) + 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 position.current_price is not None: - pos.current_price = float(position.current_price) - if position.profit is not None: - pos.pnl = float(position.profit) - # Keep the broker ticket for closing/reconciliation. - pos.broker_ticket = int(position.ticket) + 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 - result = [] + by_symbol: dict = {} for p in detail.positions: - parsed = self._parse_broker_position(p, strategy_name) - if parsed is not None: - result.append(parsed) - return result + 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 - for p in detail.positions: - if p.symbol == target: - return self._parse_broker_position(p, strategy_name) - return None + 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: @@ -333,6 +350,12 @@ 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.""" @@ -347,7 +370,11 @@ def _dispatch(self, event, **payload): def cancel_order(self, order) -> None: if order is None or not order.identifier: return - if order.is_filled() or order.is_canceled(): + # 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)) @@ -373,6 +400,62 @@ def _modify_order(self, order, limit_price: float | None = None, stop_price: flo 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 diff --git a/tests/test_broker_tickerall.py b/tests/test_broker_tickerall.py index 3c7e7ac8d..b599249c3 100644 --- a/tests/test_broker_tickerall.py +++ b/tests/test_broker_tickerall.py @@ -119,6 +119,20 @@ def test_long_position_is_positive_quantity(self): 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() @@ -193,6 +207,58 @@ def test_cancel_order(self): 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 + 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_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): From 6e045827e79ecaf05e6fa8b379a7df3e0ee0f8f0 Mon Sep 17 00:00:00 2001 From: miguelangelo78 Date: Sun, 26 Jul 2026 17:56:49 +0000 Subject: [PATCH 3/4] Declare tickerall as an optional dependency (lumibot[tickerall]) --- docsrc/brokers.tickerall.rst | 12 +++++++++--- setup.py | 5 +++++ 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/docsrc/brokers.tickerall.rst b/docsrc/brokers.tickerall.rst index 02acdfa84..24f809d99 100644 --- a/docsrc/brokers.tickerall.rst +++ b/docsrc/brokers.tickerall.rst @@ -10,10 +10,16 @@ Python package, which is Windows-only and requires a running terminal. How to Use TickerAll -------------------- -1. Create an account at `tickerall.com `_ and connect one +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. -2. Generate an API key. -3. Set the environment variables below (or pass a ``config`` dict to the broker). +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 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", From 2bbf5f4dc0ae50b9809dfb5b315f5e404078f3fd Mon Sep 17 00:00:00 2001 From: miguelangelo78 Date: Sun, 26 Jul 2026 18:18:05 +0000 Subject: [PATCH 4/4] Address PR review feedback - Read the documented TICKERALL_API_KEY / TICKERALL_ACCOUNT_ID / TICKERALL_BASE_URL environment variables: credentials now fall back from the config dict to os.environ, so the documented env-var path actually authenticates and selects an account. - Do not permanently cache a transient symbol-list fetch failure. On error, leave the cache unresolved so the next call retries, instead of disabling symbol resolution for the whole process lifetime. - Resolve a departed pending order as filled vs canceled from its terminal state in order history (deal_count > 0 means it produced a fill), not from whether a position exists. A fill that nets a netting position to exactly zero removes the position from the snapshot, which the position-existence check would have mis-reported as a cancel. Falls back to the position check only when history is unavailable. Adds regression tests for env-var credentials, the symbol-fetch retry, and the net-zero pending fill/cancel resolution (28 tests, all passing). --- lumibot/brokers/tickerall.py | 29 +++++++++++++++-- lumibot/data_sources/tickerall_data.py | 22 ++++++++++--- tests/test_broker_tickerall.py | 43 ++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 7 deletions(-) diff --git a/lumibot/brokers/tickerall.py b/lumibot/brokers/tickerall.py index 2df25dafd..c2014d1ec 100644 --- a/lumibot/brokers/tickerall.py +++ b/lumibot/brokers/tickerall.py @@ -538,12 +538,35 @@ def do_polling(self): # We cancelled it; the cancel event was already dispatched. self._cancelled_tickets.discard(str(order.identifier)) continue - # Left the pending list without our cancel: filled if a position - # for the asset now exists, otherwise treat as canceled externally. - if order.asset.symbol in open_positions: + # 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/tickerall_data.py b/lumibot/data_sources/tickerall_data.py index 180f2388b..5a43a4f0f 100644 --- a/lumibot/data_sources/tickerall_data.py +++ b/lumibot/data_sources/tickerall_data.py @@ -15,6 +15,7 @@ from __future__ import annotations import datetime as dt +import os from decimal import Decimal from threading import RLock @@ -60,7 +61,13 @@ def __init__(self, config, **kwargs): self.name = "tickerall" self._config = config or {} - api_key = self._cfg("API_KEY") or self._cfg("TICKERALL_API_KEY") + # 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'] " @@ -76,10 +83,14 @@ def __init__(self, config, **kwargs): "Install it with: pip install tickerall" ) from e - base_url = self._cfg("BASE_URL") + 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") + 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 @@ -128,8 +139,11 @@ def _ensure_symbols(self) -> list: 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}") - self._symbols = [] + return [] return self._symbols def resolve_symbol(self, asset: Asset | str) -> str: diff --git a/tests/test_broker_tickerall.py b/tests/test_broker_tickerall.py index b599249c3..19f6829ee 100644 --- a/tests/test_broker_tickerall.py +++ b/tests/test_broker_tickerall.py @@ -4,6 +4,7 @@ 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 @@ -231,6 +232,7 @@ def test_cancel_order_skips_terminal(self): 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) @@ -238,6 +240,33 @@ def _tracked_pending(self, broker, identifier="555"): 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. @@ -310,6 +339,20 @@ def test_get_last_price_candle_fallback(self): 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()