From b76c279accfa492b5146a541283cc5fbc3d60ef4 Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 20:12:08 -0400 Subject: [PATCH 1/9] feat(brokers): add Tastytrade broker scaffold Adds an initial Tastytrade broker built on the unofficial `tastytrade` Python SDK (12.4+). The SDK is async-only, so the adapter owns a daemon-thread asyncio bridge and dispatches all SDK calls via run_coroutine_threadsafe. Working in this commit: - OAuth auth (provider_secret + refresh_token, sandbox/prod) - Account, balance, and position queries - Order cancellation Stubbed (logged warnings, follow-ups): - Order submission, modification, and parsing - Multileg orders - Streaming (polling/AlertStreamer) - Market data (chains/quotes/historical via DXLink) Includes offline smoke tests (3 passing) plus a live sandbox smoke test gated on TASTYTRADE_* env credentials. --- lumibot/brokers/__init__.py | 1 + lumibot/brokers/tastytrade.py | 379 ++++++++++++++++++ lumibot/data_sources/__init__.py | 1 + lumibot/data_sources/tastytrade_data.py | 88 ++++ requirements.txt | 1 + tests/test_tastytrade_broker_smoke_apitest.py | 122 ++++++ 6 files changed, 592 insertions(+) create mode 100644 lumibot/brokers/tastytrade.py create mode 100644 lumibot/data_sources/tastytrade_data.py create mode 100644 tests/test_tastytrade_broker_smoke_apitest.py diff --git a/lumibot/brokers/__init__.py b/lumibot/brokers/__init__.py index 851d8efb0..fd08949e4 100644 --- a/lumibot/brokers/__init__.py +++ b/lumibot/brokers/__init__.py @@ -7,5 +7,6 @@ from .interactive_brokers_rest import InteractiveBrokersREST from .projectx import ProjectX from .schwab import Schwab +from .tastytrade import Tastytrade from .tradier import Tradier from .tradovate import Tradovate diff --git a/lumibot/brokers/tastytrade.py b/lumibot/brokers/tastytrade.py new file mode 100644 index 000000000..dca9a072a --- /dev/null +++ b/lumibot/brokers/tastytrade.py @@ -0,0 +1,379 @@ +""" +Tastytrade broker for Lumibot. + +Wraps the unofficial ``tastytrade`` Python SDK (https://github.com/tastyware/tastytrade) +which exposes a fully asynchronous API. Lumibot's :class:`Broker` abstract +methods are synchronous, so this adapter owns a dedicated asyncio event +loop running on a background daemon thread and dispatches every SDK call +through ``asyncio.run_coroutine_threadsafe``. + +Initial commit scope (intentionally narrow, follow-ups will fill in the rest): + +- Authentication via OAuth (``provider_secret`` + ``refresh_token``) +- Account selection by account number +- Real implementations: ``_get_balances_at_broker``, ``_pull_positions``, + ``_pull_position``, ``cancel_order`` +- Logged-stub implementations: ``_submit_order``, ``_submit_orders`` + (multileg), ``_modify_order``, ``_parse_broker_order``, + ``_pull_broker_order``, ``_pull_broker_all_orders`` +- Streaming: returns ``None`` from ``_get_stream_object`` and no-ops the + register / run methods. A follow-up will plug in either polling (similar + to Tradier) or the SDK's ``AlertStreamer`` / ``DXLinkStreamer``. +""" + +import asyncio +import os +import threading +from decimal import Decimal +from typing import Any, Awaitable, List, Optional, TypeVar, Union + +from termcolor import colored + +from .broker import Broker +from lumibot.data_sources.tastytrade_data import TastytradeData +from lumibot.entities import Asset, Order, Position +from lumibot.tools.lumibot_logger import get_logger + +logger = get_logger(__name__) + +try: # tastytrade is an optional runtime dep; surface a clear error if missing. + from tastytrade import Account as _TTAccount + from tastytrade import Session as _TTSession +except Exception as _import_err: # pragma: no cover - import-time guard + _TTAccount = None + _TTSession = None + _TASTYTRADE_IMPORT_ERROR = _import_err +else: + _TASTYTRADE_IMPORT_ERROR = None + + +T = TypeVar("T") + + +class _AsyncBridge: + """Run a private asyncio loop on a daemon thread for sync callers.""" + + def __init__(self, name: str = "tastytrade-asyncio"): + self._loop = asyncio.new_event_loop() + self._ready = threading.Event() + self._thread = threading.Thread(target=self._serve, name=name, daemon=True) + self._thread.start() + self._ready.wait() + + def _serve(self) -> None: + asyncio.set_event_loop(self._loop) + self._ready.set() + try: + self._loop.run_forever() + finally: + try: + self._loop.close() + except Exception: + pass + + def run(self, coro: Awaitable[T], timeout: Optional[float] = 30.0) -> T: + if not self._loop.is_running(): + raise RuntimeError("Tastytrade asyncio bridge is not running.") + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + return future.result(timeout=timeout) + + def close(self) -> None: + if self._loop.is_running(): + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(timeout=5) + + +class Tastytrade(Broker): + """ + Tastytrade broker. + + Authentication is OAuth-only: provide ``client_secret`` + ``refresh_token`` + and (optionally) ``is_test=True`` for the certification (sandbox) environment. + + Configuration may be supplied via the ``config`` dict, kwargs, or + environment variables (in this order of preference): + + - ``TASTYTRADE_CLIENT_SECRET`` + - ``TASTYTRADE_REFRESH_TOKEN`` + - ``TASTYTRADE_ACCOUNT_NUMBER`` + - ``TASTYTRADE_SANDBOX`` (``"true"`` / ``"1"`` / ``"yes"`` for cert env) + """ + + NAME = "Tastytrade" + + def __init__( + self, + config: Optional[dict] = None, + client_secret: Optional[str] = None, + refresh_token: Optional[str] = None, + account_number: Optional[str] = None, + is_test: Optional[bool] = None, + connect_stream: bool = True, + data_source: Optional[TastytradeData] = None, + max_workers: int = 1, + ): + if _TTSession is None: + raise ImportError( + "The 'tastytrade' package is required to use the Tastytrade broker. " + "Install it with `pip install tastytrade`." + ) from _TASTYTRADE_IMPORT_ERROR + + # Resolve credentials: explicit kwargs > config dict > environment. + if config: + client_secret = client_secret or config.get("CLIENT_SECRET") + refresh_token = refresh_token or config.get("REFRESH_TOKEN") + account_number = account_number or config.get("ACCOUNT_NUMBER") + if is_test is None and "SANDBOX" in config: + is_test = bool(config.get("SANDBOX")) + + client_secret = client_secret or os.environ.get("TASTYTRADE_CLIENT_SECRET") + refresh_token = refresh_token or os.environ.get("TASTYTRADE_REFRESH_TOKEN") + account_number = account_number or os.environ.get("TASTYTRADE_ACCOUNT_NUMBER") + if is_test is None: + env_sandbox = os.environ.get("TASTYTRADE_SANDBOX", "") + is_test = env_sandbox.strip().lower() in ("1", "true", "yes", "y") + + missing = [ + n for n, v in ( + ("client_secret", client_secret), + ("refresh_token", refresh_token), + ("account_number", account_number), + ) if not v + ] + if missing: + raise ValueError( + "Tastytrade broker missing required credentials: " + + ", ".join(missing) + + ". Provide via kwargs, the `config` dict, or env vars " + "(TASTYTRADE_CLIENT_SECRET / TASTYTRADE_REFRESH_TOKEN / " + "TASTYTRADE_ACCOUNT_NUMBER)." + ) + + self._tt_account_number = account_number + self._tt_is_test = bool(is_test) + self._async_bridge = _AsyncBridge() + + # Build the SDK Session (sync constructor) and resolve the Account. + self._session = _TTSession( + provider_secret=client_secret, + refresh_token=refresh_token, + is_test=self._tt_is_test, + ) + self._account = self._async_bridge.run( + _TTAccount.get(self._session, self._tt_account_number) + ) + + if data_source is None: + data_source = TastytradeData( + session=self._session, + runner=self._async_bridge.run, + ) + self.data_source = data_source + + super().__init__( + name=self.NAME, + data_source=data_source, + config=config, + max_workers=max_workers, + connect_stream=connect_stream, + ) + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + def _run(self, coro: Awaitable[T], timeout: Optional[float] = 30.0) -> T: + return self._async_bridge.run(coro, timeout=timeout) + + @staticmethod + def _strategy_name(strategy) -> str: + if strategy is None: + return "Unknown" + if isinstance(strategy, str): + return strategy + return getattr(strategy, "name", str(strategy)) + + # ------------------------------------------------------------------ + # Account + # ------------------------------------------------------------------ + def _get_balances_at_broker(self, quote_asset: Asset, strategy) -> tuple: + try: + balances = self._run(self._account.get_balances(self._session)) + except Exception as e: + logger.error(colored(f"[Tastytrade] Failed to fetch balances: {e}", "red")) + return 0.0, 0.0, 0.0 + + # Tastytrade balance object exposes ``cash_balance``, ``net_liquidating_value``, + # and ``long_equity_value`` / ``short_equity_value`` (Decimal). Be defensive in + # case the SDK shape shifts slightly between versions. + cash = float(getattr(balances, "cash_balance", 0) or 0) + nlv = float(getattr(balances, "net_liquidating_value", 0) or 0) + long_eq = float(getattr(balances, "long_equity_value", 0) or 0) + short_eq = float(getattr(balances, "short_equity_value", 0) or 0) + positions_value = long_eq - short_eq + return cash, positions_value, nlv + + def get_historical_account_value(self) -> dict: + # The SDK exposes ``get_net_liquidating_value_history`` but mapping it to + # Lumibot's expected hourly/daily shape requires non-trivial resampling; + # leaving as a stub for a follow-up commit. + logger.warning(colored( + "Tastytrade.get_historical_account_value is not yet implemented.", + "yellow", + )) + return {"hourly": None, "daily": None} + + # ------------------------------------------------------------------ + # Positions + # ------------------------------------------------------------------ + def _pull_positions(self, strategy) -> List[Position]: + try: + tt_positions = self._run(self._account.get_positions(self._session)) + except Exception as e: + logger.error(colored(f"[Tastytrade] Failed to fetch positions: {e}", "red")) + return [] + + strategy_name = self._strategy_name(strategy) + positions: List[Position] = [] + for tp in tt_positions or []: + asset = self._tt_position_to_asset(tp) + if asset is None: + continue + qty = self._tt_position_quantity(tp) + positions.append(Position(strategy=strategy_name, asset=asset, quantity=qty)) + return positions + + def _pull_position(self, strategy, asset: Asset) -> Optional[Position]: + for p in self._pull_positions(strategy): + if p.asset == asset: + return p + return None + + @staticmethod + def _tt_position_quantity(tp) -> Decimal: + qty = getattr(tp, "quantity", 0) or 0 + direction = (getattr(tp, "quantity_direction", "") or "").lower() + try: + qty = Decimal(str(qty)) + except Exception: + qty = Decimal("0") + if direction == "short": + qty = -qty + return qty + + @staticmethod + def _tt_position_to_asset(tp) -> Optional[Asset]: + """Convert a Tastytrade position to a Lumibot Asset. + + Tastytrade exposes ``instrument_type`` (Equity, Equity Option, Future, + Future Option, Cryptocurrency, ...) and ``symbol`` / ``underlying_symbol``. + The full mapping (especially OCC option symbol parsing) will be done + in the parser follow-up; for now we handle equities explicitly and + log a warning for option/future/crypto positions so the user can see + what's being skipped. + """ + instrument = (getattr(tp, "instrument_type", "") or "").lower() + symbol = getattr(tp, "symbol", None) or getattr(tp, "underlying_symbol", None) + if not symbol: + return None + if instrument == "equity": + return Asset(symbol=symbol, asset_type=Asset.AssetType.STOCK) + logger.warning(colored( + f"[Tastytrade] Skipping position with unhandled instrument_type " + f"'{instrument}' for symbol {symbol}. Asset parsing will be " + f"completed in a follow-up commit.", + "yellow", + )) + return None + + # ------------------------------------------------------------------ + # Orders + # ------------------------------------------------------------------ + def _submit_order(self, order: Order) -> Optional[Order]: + logger.error(colored( + f"[Tastytrade] _submit_order is not yet implemented (order={order}).", + "red", + )) + return None + + def _submit_orders(self, orders, is_multileg=False, order_type=None, + duration="day", price=None): + logger.error(colored( + "[Tastytrade] _submit_orders is not yet implemented " + "(multileg path also pending).", + "red", + )) + return None + + def cancel_order(self, order: Order) -> None: + if order.is_filled() or order.is_canceled(): + return + if not order.identifier: + raise ValueError( + "Order identifier is not set; cannot cancel. Did you submit it?" + ) + try: + self._run(self._account.delete_order(self._session, order.identifier)) + except Exception as e: + logger.error(colored( + f"[Tastytrade] Failed to cancel order {order.identifier}: {e}", + "red", + )) + + def _modify_order(self, order: Order, + limit_price: Union[float, None] = None, + stop_price: Union[float, None] = None): + logger.error(colored( + f"[Tastytrade] _modify_order is not yet implemented (order={order}).", + "red", + )) + return None + + def _parse_broker_order(self, response: Any, strategy_name: str, + strategy_object=None) -> Optional[Order]: + logger.error(colored( + "[Tastytrade] _parse_broker_order is not yet implemented.", + "red", + )) + return None + + def _pull_broker_order(self, identifier: str) -> Optional[dict]: + logger.error(colored( + f"[Tastytrade] _pull_broker_order({identifier}) is not yet implemented.", + "red", + )) + return None + + def _pull_broker_all_orders(self) -> list: + logger.error(colored( + "[Tastytrade] _pull_broker_all_orders is not yet implemented.", + "red", + )) + return [] + + # ------------------------------------------------------------------ + # Streaming (deferred) + # ------------------------------------------------------------------ + def _get_stream_object(self): + logger.warning(colored( + "[Tastytrade] _get_stream_object is not yet implemented; " + "order events will not stream until a follow-up commit lands.", + "yellow", + )) + return None + + def _register_stream_events(self): + return None + + def _run_stream(self): + return None + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + def __del__(self): + bridge = getattr(self, "_async_bridge", None) + if bridge is not None: + try: + bridge.close() + except Exception: + pass diff --git a/lumibot/data_sources/__init__.py b/lumibot/data_sources/__init__.py index 1998da5dc..ab373803d 100644 --- a/lumibot/data_sources/__init__.py +++ b/lumibot/data_sources/__init__.py @@ -13,6 +13,7 @@ from .example_broker_data import ExampleBrokerData from .interactive_brokers_rest_data import InteractiveBrokersRESTData from .schwab_data import SchwabData +from .tastytrade_data import TastytradeData from .tradovate_data import TradovateData from .yahoo_data import YahooData diff --git a/lumibot/data_sources/tastytrade_data.py b/lumibot/data_sources/tastytrade_data.py new file mode 100644 index 000000000..c79c3d896 --- /dev/null +++ b/lumibot/data_sources/tastytrade_data.py @@ -0,0 +1,88 @@ +from decimal import Decimal +from typing import Optional, Union + +from termcolor import colored + +from lumibot.data_sources import DataSource +from lumibot.entities import Asset, Bars, Quote +from lumibot.tools.lumibot_logger import get_logger + +logger = get_logger(__name__) + + +class TastytradeData(DataSource): + """ + Data source backed by the unofficial ``tastytrade`` Python SDK. + + The Tastytrade SDK is fully asynchronous, so this class shares the + asyncio event-loop bridge owned by the :class:`Tastytrade` broker. The + broker passes its own ``async_runner`` callable in via the ``runner`` + kwarg; if the data source is constructed standalone, it spins up its + own private bridge. + + Only the methods strictly required by the strategy executor are + implemented in this initial scaffold: chain / quote / historical-price + plumbing is intentionally left as logged stubs and will be filled in + via :class:`tastytrade.market_data.MarketDataAPI` and the DXLink + streamer in a follow-up commit. + """ + + MIN_TIMESTEP = "minute" + SOURCE = "Tastytrade" + + def __init__( + self, + session=None, + runner=None, + **kwargs, + ): + super().__init__() + self._session = session + self._runner = runner + + def get_chains(self, asset: Asset, quote: Optional[Asset] = None) -> dict: + logger.warning(colored( + "TastytradeData.get_chains is not yet implemented; returning {}.", + "yellow", + )) + return {} + + def get_historical_prices( + self, + asset, + length, + timestep="", + timeshift=None, + quote=None, + exchange=None, + include_after_hours=True, + ) -> Optional[Bars]: + logger.warning(colored( + "TastytradeData.get_historical_prices is not yet implemented.", + "yellow", + )) + return None + + def get_last_price( + self, + asset, + quote: Optional[Asset] = None, + exchange: Optional[str] = None, + ) -> Union[float, Decimal, None]: + logger.warning(colored( + "TastytradeData.get_last_price is not yet implemented.", + "yellow", + )) + return None + + def get_quote( + self, + asset: Asset, + quote: Optional[Asset] = None, + exchange: Optional[str] = None, + ) -> Quote: + logger.warning(colored( + "TastytradeData.get_quote is not yet implemented.", + "yellow", + )) + return Quote(asset=asset) diff --git a/requirements.txt b/requirements.txt index 60d9087de..6d3c77a19 100644 --- a/requirements.txt +++ b/requirements.txt @@ -40,6 +40,7 @@ google-genai>=1.68.0 anyio>=4.10.0 mcp>=1.26.0 schwab-py>=1.5.0 +tastytrade>=12.0.0 Flask>=2.3 free-proxy requests-oauthlib diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py new file mode 100644 index 000000000..92af71dda --- /dev/null +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -0,0 +1,122 @@ +""" +Smoke tests for the Tastytrade broker. + +The offline tests in this file are pure unit tests (no network, no +credentials) and run on every CI invocation. The single ``apitest``-marked +test at the bottom hits the Tastytrade sandbox and is gated on real +credentials being present in the environment. +""" + +from __future__ import annotations + +import asyncio +import os +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Offline tests (no credentials, no network) +# --------------------------------------------------------------------------- + +def test_missing_credentials_raises(): + """Missing client_secret / refresh_token / account_number must raise ValueError.""" + from lumibot.brokers.tastytrade import Tastytrade + + # Strip any env vars that might leak in from the dev shell. + env_keys = ( + "TASTYTRADE_CLIENT_SECRET", + "TASTYTRADE_REFRESH_TOKEN", + "TASTYTRADE_ACCOUNT_NUMBER", + "TASTYTRADE_SANDBOX", + ) + with patch.dict(os.environ, {k: "" for k in env_keys}, clear=False): + with pytest.raises(ValueError) as excinfo: + Tastytrade() + assert "client_secret" in str(excinfo.value) + assert "refresh_token" in str(excinfo.value) + assert "account_number" in str(excinfo.value) + + +def test_async_bridge_runs_and_returns_value(): + """The asyncio bridge must execute coroutines and return their result.""" + from lumibot.brokers.tastytrade import _AsyncBridge + + bridge = _AsyncBridge() + try: + async def _add(): + await asyncio.sleep(0) + return 42 + + assert bridge.run(_add()) == 42 + finally: + bridge.close() + + +@patch("lumibot.brokers.tastytrade._TTAccount") +@patch("lumibot.brokers.tastytrade._TTSession") +def test_init_with_kwargs_resolves_account(mock_session_cls, mock_account_cls): + """Constructor builds a Session, fetches the Account, and stores both.""" + from lumibot.brokers.tastytrade import Tastytrade + + fake_session = MagicMock(name="Session") + mock_session_cls.return_value = fake_session + + fake_account = MagicMock(name="Account") + + async def _get(_session, _account_number): + assert _account_number == "ACC123" + return fake_account + + mock_account_cls.get.side_effect = _get + + broker = Tastytrade( + client_secret="cs", + refresh_token="rt", + account_number="ACC123", + is_test=True, + connect_stream=False, + ) + try: + mock_session_cls.assert_called_once_with( + provider_secret="cs", + refresh_token="rt", + is_test=True, + ) + assert broker._session is fake_session + assert broker._account is fake_account + assert broker._tt_account_number == "ACC123" + assert broker._tt_is_test is True + finally: + broker._async_bridge.close() + + +# --------------------------------------------------------------------------- +# Live sandbox smoke (only runs when sandbox credentials are present) +# --------------------------------------------------------------------------- + +@pytest.mark.apitest +@pytest.mark.skipif( + not all(os.environ.get(k) for k in ( + "TASTYTRADE_CLIENT_SECRET", + "TASTYTRADE_REFRESH_TOKEN", + "TASTYTRADE_ACCOUNT_NUMBER", + )), + reason="Tastytrade sandbox credentials not configured.", +) +def test_live_sandbox_balances_and_positions(): + """Hit the sandbox API for balances + positions; expects no exceptions.""" + from lumibot.brokers.tastytrade import Tastytrade + + broker = Tastytrade(connect_stream=False) + try: + cash, positions_value, nlv = broker._get_balances_at_broker( + quote_asset=None, strategy=None, + ) + assert isinstance(cash, float) + assert isinstance(nlv, float) + positions = broker._pull_positions(strategy=None) + assert isinstance(positions, list) + finally: + broker._async_bridge.close() From a0f6589d4feae9b57a3357ff030b4b59095d03b9 Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 20:27:09 -0400 Subject: [PATCH 2/9] feat(brokers/tastytrade): order submission, parsing, and read-back MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the order-related stubs from the initial scaffold with real implementations against the tastytrade SDK. Order submission: - _submit_order: equity + single-leg equity options (limit/market/stop/ stop_limit). Builds a tastytrade.order.Leg and NewOrder and dispatches via account.place_order through the asyncio bridge. Advanced (OCO/OTO/ bracket) orders still log a clear warning and return None — those need NewComplexOrder and will land in a follow-up. - _submit_orders: multileg path (credit/debit/even/limit/market). Builds one NewOrder containing all legs, validates a shared underlying, sends a positive price (sign comes from leg actions), and returns a parent Order with child orders attached. Parent order_type is normalized to Lumibot's allowed values (LIMIT/MARKET) since credit/debit/even are wire-level pricing modes, not Lumibot OrderType members. Mapping helpers: - _to_occ_symbol: Lumibot Asset -> 21-char OCC option symbol with 6-char left-padded root (e.g. 'AAPL 260717C00230000'). - _occ_to_asset: inverse parser for read-back. - _lumi_side_to_tt_action / _lumi_order_type_to_tt / _lumi_tif_to_tt: enum bridges between Lumibot and tastytrade.order. Parsing + read-back: - _parse_broker_order: PlacedOrder -> Lumibot Order, with parent+child expansion for multileg orders. Maps the full Tastytrade OrderStatus enum to Lumibot's status taxonomy. - _pull_broker_order: account.get_order(identifier). - _pull_broker_all_orders: account.get_live_orders for the working set. Tests (12 offline pass + live sandbox still gated on creds): - OCC formatting + round-trip - Side mapping (equity vs option) - Equity limit submission (verifies Leg + NewOrder shape) - Option limit submission (verifies OCC symbol on the leg) - Multileg credit spread (verifies absolute price + leg action mix) - Multileg mixed-underlying rejection - Single-leg + multileg PlacedOrder parsing Modify, advanced orders, streaming, and market data remain stubbed — follow-ups. --- lumibot/brokers/tastytrade.py | 585 ++++++++++++++++-- tests/test_tastytrade_broker_smoke_apitest.py | 331 ++++++++++ 2 files changed, 876 insertions(+), 40 deletions(-) diff --git a/lumibot/brokers/tastytrade.py b/lumibot/brokers/tastytrade.py index dca9a072a..7857ecc1c 100644 --- a/lumibot/brokers/tastytrade.py +++ b/lumibot/brokers/tastytrade.py @@ -7,22 +7,31 @@ loop running on a background daemon thread and dispatches every SDK call through ``asyncio.run_coroutine_threadsafe``. -Initial commit scope (intentionally narrow, follow-ups will fill in the rest): +Functional surface: - Authentication via OAuth (``provider_secret`` + ``refresh_token``) - Account selection by account number -- Real implementations: ``_get_balances_at_broker``, ``_pull_positions``, - ``_pull_position``, ``cancel_order`` -- Logged-stub implementations: ``_submit_order``, ``_submit_orders`` - (multileg), ``_modify_order``, ``_parse_broker_order``, - ``_pull_broker_order``, ``_pull_broker_all_orders`` -- Streaming: returns ``None`` from ``_get_stream_object`` and no-ops the - register / run methods. A follow-up will plug in either polling (similar - to Tradier) or the SDK's ``AlertStreamer`` / ``DXLinkStreamer``. +- Balances and positions (equity; option / future positions are skipped + with a warning until full asset parsing lands) +- Order submission for equities, single-leg equity options, and multileg + equity-option spreads (``order_type`` = market / limit / debit / credit / + even, mapped onto Tastytrade's ``NewOrder``) +- Order cancellation +- Order parsing and read-back via ``account.get_order`` / + ``account.get_live_orders`` / ``account.get_order_history`` + +Stubs (logged warnings, follow-ups still pending): + +- Advanced orders (OCO / OTO / bracket → Tastytrade ``NewComplexOrder``) +- Order modification (``account.replace_order``) +- Account / order event streaming (``AlertStreamer`` + ``DXLinkStreamer``) +- Market data on ``TastytradeData`` (chains, quotes, historical bars) """ import asyncio +import datetime import os +import re import threading from decimal import Decimal from typing import Any, Awaitable, List, Optional, TypeVar, Union @@ -39,9 +48,25 @@ try: # tastytrade is an optional runtime dep; surface a clear error if missing. from tastytrade import Account as _TTAccount from tastytrade import Session as _TTSession + from tastytrade.order import ( + InstrumentType as _TTInstrumentType, + Leg as _TTLeg, + NewOrder as _TTNewOrder, + OrderAction as _TTOrderAction, + OrderStatus as _TTOrderStatus, + OrderTimeInForce as _TTOrderTIF, + OrderType as _TTOrderType, + ) except Exception as _import_err: # pragma: no cover - import-time guard _TTAccount = None _TTSession = None + _TTInstrumentType = None + _TTLeg = None + _TTNewOrder = None + _TTOrderAction = None + _TTOrderStatus = None + _TTOrderTIF = None + _TTOrderType = None _TASTYTRADE_IMPORT_ERROR = _import_err else: _TASTYTRADE_IMPORT_ERROR = None @@ -286,23 +311,331 @@ def _tt_position_to_asset(tp) -> Optional[Asset]: return None # ------------------------------------------------------------------ - # Orders + # Mapping helpers (Lumibot ↔ Tastytrade) + # ------------------------------------------------------------------ + @staticmethod + def _to_occ_symbol(asset: Asset) -> str: + """Build a 21-char OCC option symbol from a Lumibot Asset. + + Format: ``ROOT (left-padded to 6) + YYMMDD + C/P + strike*1000 (8 digits)``. + Example: ``AAPL 260717C00230000``. + """ + if asset.expiration is None or asset.right is None or asset.strike is None: + raise ValueError( + f"Option asset is missing expiration/right/strike: {asset!r}" + ) + root = (asset.symbol or "").upper().ljust(6) + exp = asset.expiration + if isinstance(exp, datetime.datetime): + exp = exp.date() + yymmdd = exp.strftime("%y%m%d") + right_letter = "C" if str(asset.right).upper().startswith("C") else "P" + strike_int = int(round(float(asset.strike) * 1000)) + strike_str = f"{strike_int:08d}" + return f"{root}{yymmdd}{right_letter}{strike_str}" + + @staticmethod + def _lumi_side_to_tt_action(side: str, is_option: bool) -> "_TTOrderAction": + s = (side or "").lower() + if is_option: + mapping = { + "buy_to_open": _TTOrderAction.BUY_TO_OPEN, + "sell_to_open": _TTOrderAction.SELL_TO_OPEN, + "buy_to_close": _TTOrderAction.BUY_TO_CLOSE, + "sell_to_close": _TTOrderAction.SELL_TO_CLOSE, + # Plain buy/sell on options default to opening; callers should + # use the explicit *_to_open / *_to_close sides when possible. + "buy": _TTOrderAction.BUY_TO_OPEN, + "sell": _TTOrderAction.SELL_TO_OPEN, + } + else: + mapping = { + "buy": _TTOrderAction.BUY, + "sell": _TTOrderAction.SELL, + "buy_to_cover": _TTOrderAction.BUY, + "sell_short": _TTOrderAction.SELL, + } + if s not in mapping: + raise ValueError(f"Unsupported order side {side!r} for Tastytrade.") + return mapping[s] + + @staticmethod + def _lumi_order_type_to_tt(order_type: str) -> "_TTOrderType": + s = (order_type or "").lower() + mapping = { + "market": _TTOrderType.MARKET, + "limit": _TTOrderType.LIMIT, + "smart_limit": _TTOrderType.LIMIT, + "stop": _TTOrderType.STOP, + "stop_limit": _TTOrderType.STOP_LIMIT, + # debit/credit/even are multileg pricing modes — they all map to + # LIMIT on the wire, with the leg actions and ``price`` carrying + # the credit/debit semantics. + "debit": _TTOrderType.LIMIT, + "credit": _TTOrderType.LIMIT, + "even": _TTOrderType.LIMIT, + } + if s not in mapping: + raise ValueError(f"Unsupported order_type {order_type!r} for Tastytrade.") + return mapping[s] + + @staticmethod + def _lumi_tif_to_tt(tif: Optional[str]) -> "_TTOrderTIF": + s = (tif or "day").lower() + mapping = { + "day": _TTOrderTIF.DAY, + "gtc": _TTOrderTIF.GTC, + "ioc": _TTOrderTIF.IOC, + "ext": _TTOrderTIF.EXT, + "pre": _TTOrderTIF.EXT, + "post": _TTOrderTIF.EXT, + } + return mapping.get(s, _TTOrderTIF.DAY) + + def _build_leg(self, order: Order) -> "_TTLeg": + """Build a Tastytrade ``Leg`` from a Lumibot child/single Order.""" + asset = order.asset + if asset is None: + raise ValueError(f"Order has no asset: {order!r}") + + if asset.asset_type == Asset.AssetType.STOCK: + symbol = (asset.symbol or "").upper() + instrument_type = _TTInstrumentType.EQUITY + is_option = False + elif asset.asset_type == Asset.AssetType.OPTION: + symbol = self._to_occ_symbol(asset) + instrument_type = _TTInstrumentType.EQUITY_OPTION + is_option = True + else: + raise ValueError( + f"Tastytrade broker does not yet support asset_type " + f"{asset.asset_type!r} (symbol={asset.symbol!r})." + ) + + action = self._lumi_side_to_tt_action(order.side, is_option=is_option) + qty = Decimal(str(order.quantity)) + return _TTLeg( + instrument_type=instrument_type, + symbol=symbol, + action=action, + quantity=qty, + ) + + @staticmethod + def _format_price(price: Optional[Union[float, Decimal]]) -> Optional[Decimal]: + if price is None: + return None + return Decimal(str(price)).quantize(Decimal("0.01")) + + def _build_new_order( + self, + legs: List["_TTLeg"], + order_type: str, + time_in_force: str, + price: Optional[Union[float, Decimal]] = None, + stop_trigger: Optional[Union[float, Decimal]] = None, + ) -> "_TTNewOrder": + tt_type = self._lumi_order_type_to_tt(order_type) + tt_tif = self._lumi_tif_to_tt(time_in_force) + + kwargs: dict = { + "time_in_force": tt_tif, + "order_type": tt_type, + "legs": legs, + } + if tt_type in (_TTOrderType.LIMIT, _TTOrderType.STOP_LIMIT): + if price is None: + raise ValueError( + f"Limit/Stop-Limit orders require a price (order_type={order_type!r})." + ) + kwargs["price"] = self._format_price(price) + if tt_type in (_TTOrderType.STOP, _TTOrderType.STOP_LIMIT): + if stop_trigger is None: + raise ValueError( + f"Stop / Stop-Limit orders require a stop_trigger " + f"(order_type={order_type!r})." + ) + kwargs["stop_trigger"] = self._format_price(stop_trigger) + return _TTNewOrder(**kwargs) + + # ------------------------------------------------------------------ + # Order submission # ------------------------------------------------------------------ def _submit_order(self, order: Order) -> Optional[Order]: - logger.error(colored( - f"[Tastytrade] _submit_order is not yet implemented (order={order}).", - "red", - )) - return None + # Advanced orders (OCO/OTO/bracket) need NewComplexOrder — defer. + if order.is_advanced_order(): + logger.error(colored( + "[Tastytrade] Advanced (OCO/OTO/bracket) orders are not yet " + "supported. Submit child orders individually for now.", + "red", + )) + self._safe_stream_dispatch(self.ERROR_ORDER, order=order, + error_msg="advanced orders unsupported") + return None - def _submit_orders(self, orders, is_multileg=False, order_type=None, - duration="day", price=None): - logger.error(colored( - "[Tastytrade] _submit_orders is not yet implemented " - "(multileg path also pending).", - "red", - )) - return None + try: + leg = self._build_leg(order) + except Exception as e: + logger.error(colored(f"[Tastytrade] Cannot build leg for {order!r}: {e}", "red")) + self._safe_stream_dispatch(self.ERROR_ORDER, order=order, error_msg=str(e)) + return None + + # For STOP_LIMIT, Lumibot stores the limit price in stop_limit_price. + order_type_str = (order.order_type or "limit") + limit = order.limit_price + if order_type_str == Order.OrderType.STOP_LIMIT: + limit = order.stop_limit_price + + try: + new_order = self._build_new_order( + legs=[leg], + order_type=order_type_str, + time_in_force=order.time_in_force, + price=limit, + stop_trigger=order.stop_price, + ) + except Exception as e: + logger.error(colored(f"[Tastytrade] Cannot build NewOrder for {order!r}: {e}", "red")) + self._safe_stream_dispatch(self.ERROR_ORDER, order=order, error_msg=str(e)) + return None + + try: + response = self._run(self._account.place_order(self._session, new_order, dry_run=False)) + except Exception as e: + logger.error(colored(f"[Tastytrade] place_order failed for {order!r}: {e}", "red")) + self._safe_stream_dispatch(self.ERROR_ORDER, order=order, error_msg=str(e)) + return None + + return self._finalize_submitted_order(order, response) + + def _submit_orders( + self, + orders, + is_multileg: bool = False, + order_type: Optional[str] = None, + duration: str = "day", + price: Optional[Union[float, Decimal]] = None, + ): + if not orders: + return [] + + if not is_multileg: + return [self._submit_order(o) for o in orders] + + # Multileg: build one NewOrder with all legs. + if order_type is None: + order_type = "market" + order_type_norm = (order_type or "market").lower() + if order_type_norm not in ("market", "limit", "debit", "credit", "even"): + raise ValueError( + f"Invalid multileg order_type {order_type!r}. Expected one of " + f"market/limit/debit/credit/even." + ) + + # Lumibot multileg convention: all legs share the same underlying. + underlyings = {o.asset.symbol for o in orders if o.asset and o.asset.symbol} + if len(underlyings) > 1: + raise ValueError( + f"All legs of a multileg order must share an underlying; got {underlyings}." + ) + + legs = [self._build_leg(o) for o in orders] + + # Tastytrade requires a positive price on debit/credit; sign comes + # from leg actions (buy=debit, sell=credit). 'even' uses 0.00. + if order_type_norm in ("debit", "credit"): + if price is None: + raise ValueError(f"price is required for '{order_type_norm}' multileg.") + tt_price: Optional[Decimal] = abs(Decimal(str(price))) + elif order_type_norm == "even": + tt_price = Decimal("0.00") + elif order_type_norm == "limit": + if price is None: + raise ValueError("price is required for 'limit' multileg.") + tt_price = abs(Decimal(str(price))) + else: # market + tt_price = None + + new_order = self._build_new_order( + legs=legs, + order_type="limit" if order_type_norm != "market" else "market", + time_in_force=duration, + price=tt_price, + ) + + try: + response = self._run(self._account.place_order(self._session, new_order, dry_run=False)) + except Exception as e: + logger.error(colored(f"[Tastytrade] Multileg place_order failed: {e}", "red")) + for o in orders: + self._safe_stream_dispatch(self.ERROR_ORDER, order=o, error_msg=str(e)) + return None + + # Build a parent Order representing the multileg group. Lumibot's + # ``Order.OrderType`` doesn't have credit/debit/even — those are + # wire-level pricing modes for the broker, not Lumibot order types. + # Store the broker-level mapping (limit for non-market, market for + # market) on the parent Order. + parent_order_type = ( + Order.OrderType.MARKET if order_type_norm == "market" + else Order.OrderType.LIMIT + ) + parent_asset = Asset( + symbol=orders[0].asset.symbol, + asset_type=Asset.AssetType.STOCK, + ) + parent = Order( + identifier=str(getattr(getattr(response, "order", None), "id", "") or ""), + asset=parent_asset, + strategy=orders[0].strategy, + order_class=Order.OrderClass.MULTILEG, + side=orders[0].side, + quantity=orders[0].quantity, + order_type=parent_order_type, + time_in_force=duration, + limit_price=tt_price, + status=Order.OrderStatus.SUBMITTED, + ) + for child in orders: + child.parent_identifier = parent.identifier + parent.child_orders = list(orders) + try: + parent.update_raw(response) + except Exception: + pass + self._unprocessed_orders.append(parent) + self._safe_stream_dispatch(self.NEW_ORDER, order=parent) + return [parent] + + def _finalize_submitted_order(self, order: Order, response: Any) -> Order: + """Stamp identifier + SUBMITTED status onto a single-leg order.""" + placed = getattr(response, "order", None) or response + identifier = getattr(placed, "id", None) + if identifier is None and isinstance(placed, dict): + identifier = placed.get("id") + order.identifier = str(identifier) if identifier is not None else None + order.status = Order.OrderStatus.SUBMITTED + try: + order.update_raw(response) + except Exception: + pass + self._unprocessed_orders.append(order) + self._safe_stream_dispatch(self.NEW_ORDER, order=order) + return order + + def _safe_stream_dispatch(self, event, **kwargs): + """Dispatch to stream if one is wired; no-op otherwise. + + Mirrors Tradier's helper so the broker doesn't crash when ``stream`` + is None (which is the case until streaming lands in a follow-up). + """ + stream = getattr(self, "stream", None) + if stream is None: + return + try: + stream.dispatch(event, **kwargs) + except Exception: + return def cancel_order(self, order: Order) -> None: if order.is_filled() or order.is_canceled(): @@ -323,32 +656,204 @@ def _modify_order(self, order: Order, limit_price: Union[float, None] = None, stop_price: Union[float, None] = None): logger.error(colored( - f"[Tastytrade] _modify_order is not yet implemented (order={order}).", + f"[Tastytrade] _modify_order is not yet implemented (order={order}). " + f"Cancel and resubmit for now.", "red", )) return None + # ------------------------------------------------------------------ + # Order parsing + read-back + # ------------------------------------------------------------------ + _TT_STATUS_TO_LUMI = { + # Tastytrade OrderStatus → Lumibot Order.OrderStatus + "Received": Order.OrderStatus.SUBMITTED, + "Routed": Order.OrderStatus.SUBMITTED, + "In Flight": Order.OrderStatus.SUBMITTED, + "Live": Order.OrderStatus.OPEN, + "Contingent": Order.OrderStatus.OPEN, + "Cancel Requested": Order.OrderStatus.CANCELLING, + "Replace Requested": Order.OrderStatus.OPEN, + "Cancelled": Order.OrderStatus.CANCELED, + "Filled": Order.OrderStatus.FILLED, + "Expired": Order.OrderStatus.EXPIRED, + "Rejected": Order.OrderStatus.ERROR, + "Removed": Order.OrderStatus.CANCELED, + "Partially Removed": Order.OrderStatus.PARTIALLY_FILLED, + } + + @classmethod + def _tt_status_to_lumi(cls, status: Any) -> str: + # status may be an OrderStatus enum or its string value. + key = getattr(status, "value", status) + return cls._TT_STATUS_TO_LUMI.get(str(key), Order.OrderStatus.NEW) + + @staticmethod + def _occ_to_asset(symbol: str) -> Optional[Asset]: + """Parse an OCC option symbol back into a Lumibot Asset.""" + m = re.match(r"^\s*([A-Z][A-Z0-9.\- ]{0,5}?)\s*(\d{6})([CP])(\d{8})\s*$", symbol or "") + if not m: + return None + root, yymmdd, cp, strike_str = m.groups() + try: + expiration = datetime.datetime.strptime(yymmdd, "%y%m%d").date() + strike = Decimal(strike_str) / Decimal(1000) + except Exception: + return None + return Asset( + symbol=root.strip(), + asset_type=Asset.AssetType.OPTION, + expiration=expiration, + strike=float(strike), + right=Asset.OptionRight.CALL if cp == "C" else Asset.OptionRight.PUT, + ) + + @classmethod + def _leg_to_asset(cls, leg) -> Optional[Asset]: + instrument = getattr(leg, "instrument_type", None) + instrument_value = getattr(instrument, "value", instrument) + symbol = getattr(leg, "symbol", "") or "" + if instrument_value == "Equity": + return Asset(symbol=symbol.strip().upper(), asset_type=Asset.AssetType.STOCK) + if instrument_value == "Equity Option": + return cls._occ_to_asset(symbol) + return None + + @classmethod + def _leg_to_lumi_side(cls, leg, is_option: bool) -> str: + action = getattr(leg, "action", None) + action_value = getattr(action, "value", action) + action_str = str(action_value or "").lower() + if is_option: + return { + "buy to open": Order.OrderSide.BUY_TO_OPEN, + "sell to open": Order.OrderSide.SELL_TO_OPEN, + "buy to close": Order.OrderSide.BUY_TO_CLOSE, + "sell to close": Order.OrderSide.SELL_TO_CLOSE, + "buy": Order.OrderSide.BUY_TO_OPEN, + "sell": Order.OrderSide.SELL_TO_OPEN, + }.get(action_str, Order.OrderSide.BUY) + return { + "buy": Order.OrderSide.BUY, + "sell": Order.OrderSide.SELL, + }.get(action_str, Order.OrderSide.BUY) + def _parse_broker_order(self, response: Any, strategy_name: str, strategy_object=None) -> Optional[Order]: - logger.error(colored( - "[Tastytrade] _parse_broker_order is not yet implemented.", - "red", - )) - return None + """Convert a Tastytrade ``PlacedOrder`` into a Lumibot ``Order``. - def _pull_broker_order(self, identifier: str) -> Optional[dict]: - logger.error(colored( - f"[Tastytrade] _pull_broker_order({identifier}) is not yet implemented.", - "red", - )) - return None + Multileg orders return a parent ``Order`` with one child per leg + attached via ``add_child_order``. Single-leg orders return a single + ``Order``. + """ + if response is None: + return None + + legs = list(getattr(response, "legs", []) or []) + if not legs: + return None + + identifier = getattr(response, "id", None) + identifier = str(identifier) if identifier is not None else None + status = self._tt_status_to_lumi(getattr(response, "status", None)) + order_type = getattr(getattr(response, "order_type", None), "value", None) + order_type = str(order_type).lower() if order_type else Order.OrderType.LIMIT + # Tastytrade enums use 'Stop Limit' / 'Marketable Limit' — normalize. + order_type = order_type.replace(" ", "_") + if order_type == "marketable_limit": + order_type = "limit" + tif = getattr(getattr(response, "time_in_force", None), "value", None) + tif = str(tif).lower() if tif else "day" + price = getattr(response, "price", None) + stop = getattr(response, "stop_trigger", None) + + if len(legs) == 1: + asset = self._leg_to_asset(legs[0]) + if asset is None: + logger.warning(colored( + f"[Tastytrade] Unhandled leg for order {identifier}: {legs[0]!r}", + "yellow", + )) + return None + qty = getattr(legs[0], "quantity", None) or 0 + side = self._leg_to_lumi_side(legs[0], is_option=(asset.asset_type == Asset.AssetType.OPTION)) + order = Order( + identifier=identifier, + asset=asset, + strategy=strategy_name, + quantity=Decimal(str(qty)), + side=side, + order_type=order_type, + limit_price=price, + stop_price=stop, + time_in_force=tif, + status=status, + ) + try: + order.update_raw(response) + except Exception: + pass + return order + + # Multileg: parent Order + one child per leg. + underlying = getattr(response, "underlying_symbol", None) or ( + getattr(self._leg_to_asset(legs[0]), "symbol", "") or "" + ) + parent_asset = Asset(symbol=underlying, asset_type=Asset.AssetType.STOCK) + parent = Order( + identifier=identifier, + asset=parent_asset, + strategy=strategy_name, + order_class=Order.OrderClass.MULTILEG, + order_type=order_type, + limit_price=price, + time_in_force=tif, + status=status, + ) + for leg in legs: + asset = self._leg_to_asset(leg) + if asset is None: + continue + qty = getattr(leg, "quantity", None) or 0 + side = self._leg_to_lumi_side(leg, is_option=(asset.asset_type == Asset.AssetType.OPTION)) + child = Order( + identifier=identifier, # Tastytrade leg has no separate id + asset=asset, + strategy=strategy_name, + quantity=Decimal(str(qty)), + side=side, + order_type=order_type, + status=status, + ) + child.parent_identifier = identifier + parent.add_child_order(child) + try: + parent.update_raw(response) + except Exception: + pass + return parent + + def _pull_broker_order(self, identifier: str) -> Optional[Any]: + if not identifier: + return None + try: + return self._run(self._account.get_order(self._session, identifier)) + except Exception as e: + logger.error(colored( + f"[Tastytrade] get_order({identifier}) failed: {e}", "red", + )) + return None def _pull_broker_all_orders(self) -> list: - logger.error(colored( - "[Tastytrade] _pull_broker_all_orders is not yet implemented.", - "red", - )) - return [] + """Return all live orders. Filled/cancelled history can be fetched + separately via ``get_order_history`` if a strategy needs it.""" + try: + return list(self._run(self._account.get_live_orders(self._session)) or []) + except Exception as e: + logger.error(colored( + f"[Tastytrade] get_live_orders failed: {e}", "red", + )) + return [] # ------------------------------------------------------------------ # Streaming (deferred) diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index 92af71dda..499a329bd 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -10,12 +10,40 @@ from __future__ import annotations import asyncio +import datetime import os +from decimal import Decimal from unittest.mock import MagicMock, patch import pytest +def _make_broker(monkeypatch): + """Build a Tastytrade broker wired against mocked SDK classes.""" + from lumibot.brokers import tastytrade as tt_mod + + fake_session = MagicMock(name="Session") + monkeypatch.setattr(tt_mod, "_TTSession", MagicMock(return_value=fake_session)) + + fake_account = MagicMock(name="Account") + + async def _get(_session, _account_number): + return fake_account + + fake_account_cls = MagicMock() + fake_account_cls.get.side_effect = _get + monkeypatch.setattr(tt_mod, "_TTAccount", fake_account_cls) + + broker = tt_mod.Tastytrade( + client_secret="cs", + refresh_token="rt", + account_number="ACC123", + is_test=True, + connect_stream=False, + ) + return broker, fake_account, fake_session + + # --------------------------------------------------------------------------- # Offline tests (no credentials, no network) # --------------------------------------------------------------------------- @@ -92,6 +120,309 @@ async def _get(_session, _account_number): broker._async_bridge.close() +# --------------------------------------------------------------------------- +# Mapping helpers (pure functions, no broker required) +# --------------------------------------------------------------------------- + +def test_to_occ_symbol_pads_root_and_strikes(): + from lumibot.brokers.tastytrade import Tastytrade + from lumibot.entities import Asset + + asset = Asset( + symbol="AAPL", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 7, 17), + strike=230, + right=Asset.OptionRight.CALL, + ) + assert Tastytrade._to_occ_symbol(asset) == "AAPL 260717C00230000" + + spx_put = Asset( + symbol="SPX", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 5, 16), + strike=4500.5, + right=Asset.OptionRight.PUT, + ) + assert Tastytrade._to_occ_symbol(spx_put) == "SPX 260516P04500500" + + +def test_occ_to_asset_round_trip(): + from lumibot.brokers.tastytrade import Tastytrade + from lumibot.entities import Asset + + asset = Tastytrade._occ_to_asset("AAPL 260717C00230000") + assert asset is not None + assert asset.symbol == "AAPL" + assert asset.asset_type == Asset.AssetType.OPTION + assert asset.expiration == datetime.date(2026, 7, 17) + assert float(asset.strike) == 230.0 + assert asset.right == Asset.OptionRight.CALL + + +def test_side_mapping_equity_vs_option(): + from lumibot.brokers.tastytrade import Tastytrade + from tastytrade.order import OrderAction + + assert Tastytrade._lumi_side_to_tt_action("buy", is_option=False) == OrderAction.BUY + assert Tastytrade._lumi_side_to_tt_action("sell", is_option=False) == OrderAction.SELL + assert Tastytrade._lumi_side_to_tt_action( + "buy_to_open", is_option=True + ) == OrderAction.BUY_TO_OPEN + assert Tastytrade._lumi_side_to_tt_action( + "sell_to_close", is_option=True + ) == OrderAction.SELL_TO_CLOSE + # Plain buy on an option defaults to BUY_TO_OPEN. + assert Tastytrade._lumi_side_to_tt_action( + "buy", is_option=True + ) == OrderAction.BUY_TO_OPEN + + +# --------------------------------------------------------------------------- +# Order submission (mocked SDK) +# --------------------------------------------------------------------------- + +def _stub_place_order(captured: list): + """Return an async place_order that captures the NewOrder and returns a fake response.""" + + async def _place(_session, new_order, dry_run=False): + captured.append(new_order) + resp = MagicMock(name="PlacedOrderResponse") + resp.order = MagicMock(name="PlacedOrder") + resp.order.id = 4242 + return resp + + return _place + + +def test_submit_order_equity_limit(monkeypatch): + from lumibot.entities import Asset, Order + from tastytrade.order import OrderAction, OrderType, OrderTimeInForce + + broker, fake_account, _ = _make_broker(monkeypatch) + try: + captured: list = [] + fake_account.place_order.side_effect = _stub_place_order(captured) + + order = Order( + strategy="s", + asset=Asset(symbol="AAPL", asset_type=Asset.AssetType.STOCK), + quantity=10, + side=Order.OrderSide.BUY, + order_type=Order.OrderType.LIMIT, + limit_price=150.25, + time_in_force="day", + ) + result = broker._submit_order(order) + + assert result is order + assert order.identifier == "4242" + assert order.status == Order.OrderStatus.SUBMITTED + + assert len(captured) == 1 + new_order = captured[0] + assert new_order.order_type == OrderType.LIMIT + assert new_order.time_in_force == OrderTimeInForce.DAY + assert new_order.price == Decimal("150.25") + assert len(new_order.legs) == 1 + leg = new_order.legs[0] + assert leg.symbol == "AAPL" + assert leg.action == OrderAction.BUY + assert Decimal(str(leg.quantity)) == Decimal("10") + finally: + broker._async_bridge.close() + + +def test_submit_order_option_limit_uses_occ(monkeypatch): + from lumibot.entities import Asset, Order + from tastytrade.order import InstrumentType, OrderAction + + broker, fake_account, _ = _make_broker(monkeypatch) + try: + captured: list = [] + fake_account.place_order.side_effect = _stub_place_order(captured) + + opt = Asset( + symbol="AAPL", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 7, 17), + strike=230, + right=Asset.OptionRight.CALL, + ) + order = Order( + strategy="s", + asset=opt, + quantity=1, + side=Order.OrderSide.BUY_TO_OPEN, + order_type=Order.OrderType.LIMIT, + limit_price=4.20, + ) + broker._submit_order(order) + + assert len(captured) == 1 + leg = captured[0].legs[0] + assert leg.instrument_type == InstrumentType.EQUITY_OPTION + assert leg.symbol == "AAPL 260717C00230000" + assert leg.action == OrderAction.BUY_TO_OPEN + finally: + broker._async_bridge.close() + + +def test_submit_orders_multileg_credit_spread(monkeypatch): + """A credit put spread: short 4500P / long 4400P. Price should be positive.""" + from lumibot.entities import Asset, Order + from tastytrade.order import OrderAction, OrderType + + broker, fake_account, _ = _make_broker(monkeypatch) + try: + captured: list = [] + fake_account.place_order.side_effect = _stub_place_order(captured) + + short_leg = Order( + strategy="s", + asset=Asset( + symbol="SPX", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 5, 16), + strike=4500, + right=Asset.OptionRight.PUT, + ), + quantity=1, + side=Order.OrderSide.SELL_TO_OPEN, + ) + long_leg = Order( + strategy="s", + asset=Asset( + symbol="SPX", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 5, 16), + strike=4400, + right=Asset.OptionRight.PUT, + ), + quantity=1, + side=Order.OrderSide.BUY_TO_OPEN, + ) + result = broker._submit_orders( + [short_leg, long_leg], + is_multileg=True, + order_type="credit", + duration="day", + price=2.50, + ) + + assert isinstance(result, list) and len(result) == 1 + parent = result[0] + assert parent.order_class == Order.OrderClass.MULTILEG + assert parent.identifier == "4242" + + assert len(captured) == 1 + new_order = captured[0] + assert new_order.order_type == OrderType.LIMIT + assert new_order.price == Decimal("2.50") # absolute value, sign from legs + assert len(new_order.legs) == 2 + actions = [l.action for l in new_order.legs] + assert OrderAction.SELL_TO_OPEN in actions + assert OrderAction.BUY_TO_OPEN in actions + finally: + broker._async_bridge.close() + + +def test_submit_orders_multileg_rejects_mixed_underlyings(monkeypatch): + from lumibot.entities import Asset, Order + + broker, fake_account, _ = _make_broker(monkeypatch) + try: + a = Order(strategy="s", asset=Asset(symbol="SPY", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 5, 16), strike=400, + right=Asset.OptionRight.PUT), + quantity=1, side=Order.OrderSide.SELL_TO_OPEN) + b = Order(strategy="s", asset=Asset(symbol="QQQ", + asset_type=Asset.AssetType.OPTION, + expiration=datetime.date(2026, 5, 16), strike=400, + right=Asset.OptionRight.PUT), + quantity=1, side=Order.OrderSide.BUY_TO_OPEN) + with pytest.raises(ValueError, match="share an underlying"): + broker._submit_orders([a, b], is_multileg=True, + order_type="credit", price=1.0) + finally: + broker._async_bridge.close() + + +# --------------------------------------------------------------------------- +# Order parsing +# --------------------------------------------------------------------------- + +def test_parse_broker_order_single_leg_equity(): + from lumibot.brokers.tastytrade import Tastytrade + from lumibot.entities import Asset, Order + + leg = MagicMock() + leg.instrument_type = MagicMock(value="Equity") + leg.symbol = "AAPL" + leg.action = MagicMock(value="Buy") + leg.quantity = Decimal("10") + + placed = MagicMock() + placed.id = 99 + placed.legs = [leg] + placed.status = MagicMock(value="Live") + placed.order_type = MagicMock(value="Limit") + placed.time_in_force = MagicMock(value="Day") + placed.price = Decimal("150.25") + placed.stop_trigger = None + placed.underlying_symbol = "AAPL" + + parsed = Tastytrade._parse_broker_order( + Tastytrade.__new__(Tastytrade), placed, "s", + ) + assert parsed is not None + assert parsed.identifier == "99" + assert parsed.asset.symbol == "AAPL" + assert parsed.asset.asset_type == Asset.AssetType.STOCK + assert parsed.side == Order.OrderSide.BUY + assert parsed.status == Order.OrderStatus.OPEN + assert parsed.order_type == "limit" + + +def test_parse_broker_order_multileg_attaches_children(): + from lumibot.brokers.tastytrade import Tastytrade + from lumibot.entities import Order + + short_leg = MagicMock() + short_leg.instrument_type = MagicMock(value="Equity Option") + short_leg.symbol = "SPX 260516P04500000" + short_leg.action = MagicMock(value="Sell to Open") + short_leg.quantity = Decimal("1") + + long_leg = MagicMock() + long_leg.instrument_type = MagicMock(value="Equity Option") + long_leg.symbol = "SPX 260516P04400000" + long_leg.action = MagicMock(value="Buy to Open") + long_leg.quantity = Decimal("1") + + placed = MagicMock() + placed.id = 1234 + placed.legs = [short_leg, long_leg] + placed.status = MagicMock(value="Filled") + placed.order_type = MagicMock(value="Limit") + placed.time_in_force = MagicMock(value="Day") + placed.price = Decimal("2.50") + placed.stop_trigger = None + placed.underlying_symbol = "SPX" + + parsed = Tastytrade._parse_broker_order( + Tastytrade.__new__(Tastytrade), placed, "s", + ) + assert parsed is not None + assert parsed.order_class == Order.OrderClass.MULTILEG + assert parsed.status == Order.OrderStatus.FILLED + assert len(parsed.child_orders) == 2 + sides = {c.side for c in parsed.child_orders} + assert Order.OrderSide.SELL_TO_OPEN in sides + assert Order.OrderSide.BUY_TO_OPEN in sides + + # --------------------------------------------------------------------------- # Live sandbox smoke (only runs when sandbox credentials are present) # --------------------------------------------------------------------------- From cb9eae97d19fb4ccdf07abdd7e5fd063d26a2627 Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 20:32:10 -0400 Subject: [PATCH 3/9] feat(brokers/tastytrade): polling stream + order modification Wires the broker into Lumibot's strategy executor by adding the PollingStream-based event loop and replacing the modify-order stub with a real implementation. Polling stream: - _get_stream_object returns PollingStream(polling_interval) with polling_interval exposed on the constructor (default 5.0s, matching Tradier). - _register_stream_events binds POLL/NEW/FILLED/CANCELED/ERROR to _process_trade_event so the executor sees order lifecycle events. - _run_stream calls _stream_established then runs the polling loop. - do_polling pulls live orders, parses each, and dispatches transitions: submitted/open -> NEW_ORDER fill -> FILLED_ORDER (with weighted avg fill price + qty from the PlacedOrder's leg fills) canceled -> CANCELED_ORDER error -> ERROR_ORDER (reject_reason if available) Tracked-but-no-longer-at-broker orders are dispatched as cancelled. Partial fills are intentionally not dispatched in polling mode (same rationale as Tradier: polling will routinely miss them). Order modification: - _modify_order rebuilds the leg + NewOrder with the new limit/stop and calls account.replace_order. Tastytrade replace returns a new id, so we update the local Order's identifier and price fields. Tests (16 offline pass + live sandbox still gated): - _get_stream_object returns a PollingStream with the right interval - _modify_order calls replace_order with the updated price and absorbs the broker's new id back onto the local order - _avg_fill_from_legs computes a size-weighted average across leg fills - _avg_fill_from_legs returns None when nothing has filled yet Native AlertStreamer + DXLinkStreamer integration and TastytradeData market-data methods remain TODO. --- lumibot/brokers/tastytrade.py | 268 ++++++++++++++++-- tests/test_tastytrade_broker_smoke_apitest.py | 83 ++++++ 2 files changed, 334 insertions(+), 17 deletions(-) diff --git a/lumibot/brokers/tastytrade.py b/lumibot/brokers/tastytrade.py index 7857ecc1c..7525f2a2c 100644 --- a/lumibot/brokers/tastytrade.py +++ b/lumibot/brokers/tastytrade.py @@ -16,15 +16,17 @@ - Order submission for equities, single-leg equity options, and multileg equity-option spreads (``order_type`` = market / limit / debit / credit / even, mapped onto Tastytrade's ``NewOrder``) +- Order modification via ``account.replace_order`` - Order cancellation - Order parsing and read-back via ``account.get_order`` / ``account.get_live_orders`` / ``account.get_order_history`` +- Polling stream dispatching NEW / FILLED / CANCELED / ERROR events to + the strategy executor Stubs (logged warnings, follow-ups still pending): - Advanced orders (OCO / OTO / bracket → Tastytrade ``NewComplexOrder``) -- Order modification (``account.replace_order``) -- Account / order event streaming (``AlertStreamer`` + ``DXLinkStreamer``) +- Native websocket streaming (``AlertStreamer`` + ``DXLinkStreamer``) - Market data on ``TastytradeData`` (chains, quotes, historical bars) """ @@ -33,6 +35,7 @@ import os import re import threading +import traceback from decimal import Decimal from typing import Any, Awaitable, List, Optional, TypeVar, Union @@ -42,6 +45,7 @@ from lumibot.data_sources.tastytrade_data import TastytradeData from lumibot.entities import Asset, Order, Position from lumibot.tools.lumibot_logger import get_logger +from lumibot.trading_builtins import PollingStream logger = get_logger(__name__) @@ -125,6 +129,7 @@ class Tastytrade(Broker): """ NAME = "Tastytrade" + POLL_EVENT = PollingStream.POLL_EVENT def __init__( self, @@ -136,6 +141,7 @@ def __init__( connect_stream: bool = True, data_source: Optional[TastytradeData] = None, max_workers: int = 1, + polling_interval: float = 5.0, ): if _TTSession is None: raise ImportError( @@ -176,6 +182,7 @@ def __init__( self._tt_account_number = account_number self._tt_is_test = bool(is_test) + self.polling_interval = polling_interval self._async_bridge = _AsyncBridge() # Build the SDK Session (sync constructor) and resolve the Account. @@ -655,12 +662,65 @@ def cancel_order(self, order: Order) -> None: def _modify_order(self, order: Order, limit_price: Union[float, None] = None, stop_price: Union[float, None] = None): - logger.error(colored( - f"[Tastytrade] _modify_order is not yet implemented (order={order}). " - f"Cancel and resubmit for now.", - "red", - )) - return None + """Replace an order's limit and/or stop price. + + Tastytrade implements modification as a *replace*: build a new + ``NewOrder`` with the same legs and an updated price, then call + ``account.replace_order(session, order_id, new_order)``. + """ + if not order.identifier: + raise ValueError( + "Order identifier is not set; cannot modify. Did you submit it?" + ) + if order.is_filled() or order.is_canceled(): + return + + try: + leg = self._build_leg(order) + except Exception as e: + logger.error(colored(f"[Tastytrade] _modify_order build_leg failed: {e}", "red")) + return None + + new_limit = limit_price if limit_price is not None else order.limit_price + new_stop = stop_price if stop_price is not None else order.stop_price + order_type_str = (order.order_type or "limit") + try: + new_order = self._build_new_order( + legs=[leg], + order_type=order_type_str, + time_in_force=order.time_in_force, + price=new_limit, + stop_trigger=new_stop, + ) + except Exception as e: + logger.error(colored(f"[Tastytrade] _modify_order build NewOrder failed: {e}", "red")) + return None + + try: + response = self._run(self._account.replace_order( + self._session, order.identifier, new_order, + )) + except Exception as e: + logger.error(colored( + f"[Tastytrade] replace_order({order.identifier}) failed: {e}", + "red", + )) + return None + + # Replace returns a new PlacedOrder with a new id; update local order. + placed = getattr(response, "order", None) or response + new_id = getattr(placed, "id", None) + if new_id is not None: + order.identifier = str(new_id) + if limit_price is not None: + order.limit_price = limit_price + if stop_price is not None: + order.stop_price = stop_price + try: + order.update_raw(response) + except Exception: + pass + return order # ------------------------------------------------------------------ # Order parsing + read-back @@ -856,21 +916,195 @@ def _pull_broker_all_orders(self) -> list: return [] # ------------------------------------------------------------------ - # Streaming (deferred) + # Stream / polling # ------------------------------------------------------------------ + # Tastytrade *does* expose a websocket (``AlertStreamer`` for account + # events, ``DXLinkStreamer`` for quotes). For this milestone we use + # polling, matching what Tradier does — it's simpler, hits the same + # SDK methods we already exercise, and avoids holding a long-lived + # async websocket from a sync-shaped broker. Native streaming is a + # follow-up. def _get_stream_object(self): - logger.warning(colored( - "[Tastytrade] _get_stream_object is not yet implemented; " - "order events will not stream until a follow-up commit lands.", - "yellow", - )) - return None + return PollingStream(self.polling_interval) def _register_stream_events(self): - return None + broker = self + + @broker.stream.add_action(broker.POLL_EVENT) + def on_poll(): + try: + broker.do_polling() + except Exception: + logger.error(traceback.format_exc()) + + @broker.stream.add_action(broker.NEW_ORDER) + def on_new(order): + try: + broker._process_trade_event(order, broker.NEW_ORDER) + except Exception: + logger.error(traceback.format_exc()) + + @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, + multiplier=getattr(order.asset, "multiplier", 1), + ) + except Exception: + logger.error(traceback.format_exc()) + + @broker.stream.add_action(broker.CANCELED_ORDER) + def on_cancel(order): + try: + broker._process_trade_event(order, broker.CANCELED_ORDER) + except Exception: + logger.error(traceback.format_exc()) + + @broker.stream.add_action(broker.ERROR_ORDER) + def on_error(order, error_msg): + try: + if order.is_active() and order.child_orders: + for child in order.child_orders: + child.set_error(error_msg) + broker._process_trade_event(child, broker.ERROR_ORDER) + broker._process_trade_event(order, broker.ERROR_ORDER) + order.set_error(error_msg) + except Exception: + logger.error(traceback.format_exc()) def _run_stream(self): - return None + self._stream_established() + try: + self.stream._run() + except Exception as e: + logger.error(colored( + f"[Tastytrade] polling stream crashed: {e}", "red", + )) + + # ------------------------------------------------------------------ + # Polling implementation + # ------------------------------------------------------------------ + @staticmethod + def _avg_fill_from_legs(placed) -> Optional[Decimal]: + """Compute size-weighted average fill price from a PlacedOrder's legs.""" + legs = list(getattr(placed, "legs", []) or []) + total_qty = Decimal(0) + total_value = Decimal(0) + for leg in legs: + for fill in (getattr(leg, "fills", None) or []): + qty = Decimal(str(getattr(fill, "quantity", 0) or 0)) + price = Decimal(str(getattr(fill, "fill_price", 0) or 0)) + total_qty += qty + total_value += qty * price + if total_qty <= 0: + return None + return total_value / total_qty + + @staticmethod + def _filled_qty_from_legs(placed) -> Optional[Decimal]: + legs = list(getattr(placed, "legs", []) or []) + total = Decimal(0) + any_fill = False + for leg in legs: + for fill in (getattr(leg, "fills", None) or []): + total += Decimal(str(getattr(fill, "quantity", 0) or 0)) + any_fill = True + if not any_fill: + return None + # For multileg, this sums leg fills. For single-leg, it's the leg's + # filled quantity directly. + return total if len(legs) == 1 else total / Decimal(len(legs)) + + def do_polling(self): + """Poll Tastytrade for live orders, dispatch transitions to the stream. + + Mirrors Tradier's polling shape: pull live orders, parse, compare + against tracked Lumibot orders, dispatch NEW / FILLED / CANCELED / + ERROR events as the broker-side status moves. + """ + # Sync positions so the strategy sees fresh holdings. + try: + self.sync_positions(None) + except Exception: + logger.error(traceback.format_exc()) + + raw_orders = self._pull_broker_all_orders() + stored_orders = {x.identifier: x for x in self.get_all_orders()} + + strategy_name = self._strategy_name + if not strategy_name and len(self._subscribers) == 1: + strategy_name = self._subscribers[0].name + + broker_ids = set() + for placed in raw_orders or []: + parsed = self._parse_broker_order(placed, strategy_name=strategy_name) + if parsed is None: + continue + if parsed.identifier: + broker_ids.add(parsed.identifier) + + for order in [*parsed.child_orders, parsed]: + if not order.identifier: + continue + + if order.identifier not in stored_orders: + # First time we see this order. On startup, only ingest + # active orders to avoid OOM on long broker histories. + if self._first_iteration and not ( + order.is_active() or order.status == Order.OrderStatus.NEW + ): + continue + self._process_new_order(order) + continue + + stored = stored_orders[order.identifier] + stored.quantity = order.quantity or stored.quantity + + if order.equivalent_status(stored): + stored.status = order.status + continue + + status = (order.status or "").lower() + if status in ("submitted", "open"): + self._safe_stream_dispatch(self.NEW_ORDER, order=stored) + elif status == "fill": + fill_price = self._avg_fill_from_legs(placed) + fill_qty = self._filled_qty_from_legs(placed) or order.quantity + if fill_price is not None and fill_qty is not None: + self._safe_stream_dispatch( + self.FILLED_ORDER, + order=stored, + price=fill_price, + filled_quantity=fill_qty, + ) + elif status == "canceled": + self._safe_stream_dispatch(self.CANCELED_ORDER, order=stored) + elif status == "error": + msg = getattr(placed, "reject_reason", None) or ( + f"Tastytrade rejected order {order.identifier}" + ) + self._safe_stream_dispatch( + self.ERROR_ORDER, order=stored, error_msg=msg, + ) + # 'partial_fill' deliberately not dispatched: polling can + # easily miss partials; only complete fills are reliable. + + # Tracked locally but no longer reported by broker → likely cancelled. + tracked = {x.identifier: x for x in self.get_tracked_orders()} + for oid, order in tracked.items(): + if oid and oid not in broker_ids and order.is_active(): + logger.debug( + f"[Tastytrade] order {oid} no longer at broker; " + f"dispatching as cancelled." + ) + self._safe_stream_dispatch(self.CANCELED_ORDER, order=order) + + if self._first_iteration: + self._first_iteration = False # ------------------------------------------------------------------ # Lifecycle diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index 499a329bd..f9488b683 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -423,6 +423,89 @@ def test_parse_broker_order_multileg_attaches_children(): assert Order.OrderSide.BUY_TO_OPEN in sides +# --------------------------------------------------------------------------- +# Order modification + polling stream +# --------------------------------------------------------------------------- + +def test_get_stream_object_returns_polling_stream(monkeypatch): + from lumibot.trading_builtins import PollingStream + + broker, _, _ = _make_broker(monkeypatch) + try: + stream = broker._get_stream_object() + assert isinstance(stream, PollingStream) + assert stream.polling_interval == broker.polling_interval + finally: + broker._async_bridge.close() + + +def test_modify_order_calls_replace(monkeypatch): + from lumibot.entities import Asset, Order + + broker, fake_account, _ = _make_broker(monkeypatch) + try: + captured: list = [] + + async def _replace(_session, identifier, new_order): + captured.append((identifier, new_order)) + resp = MagicMock() + resp.order = MagicMock() + resp.order.id = 9999 + return resp + + fake_account.replace_order.side_effect = _replace + + order = Order( + strategy="s", + asset=Asset(symbol="AAPL", asset_type=Asset.AssetType.STOCK), + quantity=10, + side=Order.OrderSide.BUY, + order_type=Order.OrderType.LIMIT, + limit_price=150.00, + ) + order.identifier = "1234" + + broker._modify_order(order, limit_price=151.50) + + assert len(captured) == 1 + identifier, new_order = captured[0] + assert identifier == "1234" + assert new_order.price == Decimal("151.50") + assert order.identifier == "9999" # broker assigns new id on replace + assert order.limit_price == 151.50 + finally: + broker._async_bridge.close() + + +def test_avg_fill_from_legs_weighted_average(): + from lumibot.brokers.tastytrade import Tastytrade + + fill1 = MagicMock() + fill1.quantity = Decimal("3") + fill1.fill_price = Decimal("100") + fill2 = MagicMock() + fill2.quantity = Decimal("7") + fill2.fill_price = Decimal("110") + leg = MagicMock() + leg.fills = [fill1, fill2] + placed = MagicMock() + placed.legs = [leg] + + avg = Tastytrade._avg_fill_from_legs(placed) + # (3*100 + 7*110) / 10 = 107 + assert avg == Decimal("107") + + +def test_avg_fill_from_legs_returns_none_when_unfilled(): + from lumibot.brokers.tastytrade import Tastytrade + + leg = MagicMock() + leg.fills = [] + placed = MagicMock() + placed.legs = [leg] + assert Tastytrade._avg_fill_from_legs(placed) is None + + # --------------------------------------------------------------------------- # Live sandbox smoke (only runs when sandbox credentials are present) # --------------------------------------------------------------------------- From ef687ac07f941d37e65d22c27899295c8aa59384 Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 20:34:19 -0400 Subject: [PATCH 4/9] feat(data_sources/tastytrade): get_last_price, get_quote, get_chains MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the warning stubs in TastytradeData with REST-based implementations against the tastytrade SDK's market_data and instruments modules. - get_last_price: fetch a market-data snapshot via market_data.get_market_data and return mid -> last -> mark, falling back to (bid+ask)/2 if those are unset. - get_quote: same fetch, populated into a Lumibot Quote with bid/ask/mid_price/price/sizes/volume/timestamp and the raw payload preserved on raw_data['tt_market_data']. - get_chains: pivot instruments.get_option_chain's dict[date, list[Option]] response into Lumibot's nested {"Multiplier": 100, "Chains": {"CALL": {date: [strikes]}, "PUT": ...}} shape, with strikes sorted per expiration. Both calls go through the broker's asyncio bridge when one is wired in (broker passes its _AsyncBridge.run as the runner kwarg). Standalone construction falls back to asyncio.run. OCC option-symbol formatting is reused from the broker (Tastytrade._to_occ_symbol). Index assets are also accepted (mapped to InstrumentType.INDEX). Tests (20 offline pass + live sandbox still gated): - get_last_price prefers mid over last/mark - get_last_price falls back to (bid+ask)/2 when mid/last/mark all None - get_quote populates bid/ask/mid/price/sizes/volume/timestamp - get_chains pivots TT chain to Lumibot's nested shape with sorted strikes and lowercase-tolerant option_type matching get_historical_prices remains a stub — needs the DXLink streamer or a separate historical-bar source. --- lumibot/data_sources/tastytrade_data.py | 229 +++++++++++++++--- tests/test_tastytrade_broker_smoke_apitest.py | 117 +++++++++ 2 files changed, 306 insertions(+), 40 deletions(-) diff --git a/lumibot/data_sources/tastytrade_data.py b/lumibot/data_sources/tastytrade_data.py index c79c3d896..9388ce61e 100644 --- a/lumibot/data_sources/tastytrade_data.py +++ b/lumibot/data_sources/tastytrade_data.py @@ -1,5 +1,6 @@ +import datetime from decimal import Decimal -from typing import Optional, Union +from typing import Callable, Optional, Union from termcolor import colored @@ -9,22 +10,34 @@ logger = get_logger(__name__) +try: + from tastytrade.instruments import get_option_chain as _tt_get_option_chain + from tastytrade.market_data import get_market_data as _tt_get_market_data + from tastytrade.order import InstrumentType as _TTInstrumentType +except Exception: # pragma: no cover + _tt_get_option_chain = None + _tt_get_market_data = None + _TTInstrumentType = None + class TastytradeData(DataSource): """ Data source backed by the unofficial ``tastytrade`` Python SDK. - The Tastytrade SDK is fully asynchronous, so this class shares the - asyncio event-loop bridge owned by the :class:`Tastytrade` broker. The - broker passes its own ``async_runner`` callable in via the ``runner`` - kwarg; if the data source is constructed standalone, it spins up its - own private bridge. - - Only the methods strictly required by the strategy executor are - implemented in this initial scaffold: chain / quote / historical-price - plumbing is intentionally left as logged stubs and will be filled in - via :class:`tastytrade.market_data.MarketDataAPI` and the DXLink - streamer in a follow-up commit. + The SDK is fully asynchronous, so this class shares the asyncio event- + loop bridge owned by the :class:`Tastytrade` broker. The broker passes + its ``async_runner`` callable (``self._run``) in via the ``runner`` + kwarg so every SDK call is dispatched through the same bridge. + + Implemented: + - ``get_last_price``: REST market-data snapshot (mid → last → mark) + - ``get_quote``: REST market-data snapshot with bid/ask/mid/last + - ``get_chains``: option chain expanded into Lumibot's nested + ``{"Multiplier": 100, "Chains": {"CALL": {...}, "PUT": {...}}}`` shape + + Stubbed (follow-up): + - ``get_historical_prices``: needs the DXLink streamer or a separate + historical-bar source. """ MIN_TIMESTEP = "minute" @@ -33,46 +46,84 @@ class TastytradeData(DataSource): def __init__( self, session=None, - runner=None, + runner: Optional[Callable] = None, **kwargs, ): super().__init__() self._session = session + # ``runner`` is the broker's _AsyncBridge.run; if absent we fall back + # to ``asyncio.run`` per call (slow but functional in standalone use). self._runner = runner - def get_chains(self, asset: Asset, quote: Optional[Asset] = None) -> dict: - logger.warning(colored( - "TastytradeData.get_chains is not yet implemented; returning {}.", - "yellow", - )) - return {} + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + def _await(self, coro): + if self._runner is not None: + return self._runner(coro) + import asyncio + return asyncio.run(coro) - def get_historical_prices( - self, - asset, - length, - timestep="", - timeshift=None, - quote=None, - exchange=None, - include_after_hours=True, - ) -> Optional[Bars]: - logger.warning(colored( - "TastytradeData.get_historical_prices is not yet implemented.", - "yellow", - )) - return None + def _instrument_type_for(self, asset: Asset) -> "_TTInstrumentType": + if asset.asset_type == Asset.AssetType.STOCK: + return _TTInstrumentType.EQUITY + if asset.asset_type == Asset.AssetType.OPTION: + return _TTInstrumentType.EQUITY_OPTION + if asset.asset_type == Asset.AssetType.INDEX: + return _TTInstrumentType.INDEX + raise ValueError( + f"Tastytrade data source does not yet support asset_type " + f"{asset.asset_type!r}." + ) + def _symbol_for(self, asset: Asset) -> str: + if asset.asset_type == Asset.AssetType.OPTION: + # Reuse the broker's OCC formatter. + from lumibot.brokers.tastytrade import Tastytrade + return Tastytrade._to_occ_symbol(asset) + return (asset.symbol or "").upper() + + @staticmethod + def _decimal_to_float(d) -> Optional[float]: + if d is None: + return None + try: + return float(d) + except Exception: + return None + + # ------------------------------------------------------------------ + # Implemented + # ------------------------------------------------------------------ def get_last_price( self, asset, quote: Optional[Asset] = None, exchange: Optional[str] = None, ) -> Union[float, Decimal, None]: - logger.warning(colored( - "TastytradeData.get_last_price is not yet implemented.", - "yellow", - )) + if self._session is None or _tt_get_market_data is None: + return None + try: + md = self._await(_tt_get_market_data( + self._session, + self._symbol_for(asset), + self._instrument_type_for(asset), + )) + except Exception as e: + logger.error(colored( + f"[TastytradeData] get_last_price({asset!r}) failed: {e}", "red", + )) + return None + + # Preference order: mid > last > mark > (bid+ask)/2. + for attr in ("mid", "last", "mark"): + val = getattr(md, attr, None) + if val is not None: + return self._decimal_to_float(val) + bid = getattr(md, "bid", None) + ask = getattr(md, "ask", None) + if bid is not None and ask is not None: + return self._decimal_to_float((Decimal(str(bid)) + Decimal(str(ask))) / 2) return None def get_quote( @@ -81,8 +132,106 @@ def get_quote( quote: Optional[Asset] = None, exchange: Optional[str] = None, ) -> Quote: + if self._session is None or _tt_get_market_data is None: + return Quote(asset=asset) + try: + md = self._await(_tt_get_market_data( + self._session, + self._symbol_for(asset), + self._instrument_type_for(asset), + )) + except Exception as e: + logger.error(colored( + f"[TastytradeData] get_quote({asset!r}) failed: {e}", "red", + )) + return Quote(asset=asset) + + last = getattr(md, "last", None) or getattr(md, "mark", None) + mid = getattr(md, "mid", None) + bid = getattr(md, "bid", None) + ask = getattr(md, "ask", None) + if mid is None and bid is not None and ask is not None: + mid = (Decimal(str(bid)) + Decimal(str(ask))) / 2 + + return Quote( + asset=asset, + price=self._decimal_to_float(last), + bid=self._decimal_to_float(bid), + ask=self._decimal_to_float(ask), + mid_price=self._decimal_to_float(mid), + bid_size=self._decimal_to_float(getattr(md, "bid_size", None)), + ask_size=self._decimal_to_float(getattr(md, "ask_size", None)), + volume=self._decimal_to_float(getattr(md, "volume", None)), + timestamp=getattr(md, "updated_at", None), + raw_data={"tt_market_data": md}, + ) + + def get_chains(self, asset: Asset, quote: Optional[Asset] = None) -> dict: + """Return option chain in Lumibot's nested shape. + + Output: + { + "Multiplier": 100, + "Chains": { + "CALL": {"YYYY-MM-DD": [strike1, strike2, ...], ...}, + "PUT": {"YYYY-MM-DD": [strike1, strike2, ...], ...}, + }, + } + """ + if self._session is None or _tt_get_option_chain is None: + return {} + symbol = (asset.symbol or "").upper() + try: + tt_chain = self._await(_tt_get_option_chain(self._session, symbol)) + except Exception as e: + logger.error(colored( + f"[TastytradeData] get_chains({symbol!r}) failed: {e}", "red", + )) + return {} + + calls: dict = {} + puts: dict = {} + for expiration, options in (tt_chain or {}).items(): + exp_str = expiration.strftime("%Y-%m-%d") if isinstance( + expiration, (datetime.date, datetime.datetime) + ) else str(expiration) + for opt in options: + opt_type = getattr(getattr(opt, "option_type", None), "value", None) + strike = getattr(opt, "strike_price", None) + if strike is None: + continue + strike_f = self._decimal_to_float(strike) + if str(opt_type).lower().startswith("c"): + calls.setdefault(exp_str, []).append(strike_f) + elif str(opt_type).lower().startswith("p"): + puts.setdefault(exp_str, []).append(strike_f) + + # Sort strikes per expiration for deterministic output. + for buckets in (calls, puts): + for exp in buckets: + buckets[exp] = sorted(buckets[exp]) + + return { + "Multiplier": 100, + "Chains": {"CALL": calls, "PUT": puts}, + } + + # ------------------------------------------------------------------ + # Stubbed + # ------------------------------------------------------------------ + def get_historical_prices( + self, + asset, + length, + timestep="", + timeshift=None, + quote=None, + exchange=None, + include_after_hours=True, + ) -> Optional[Bars]: logger.warning(colored( - "TastytradeData.get_quote is not yet implemented.", + "TastytradeData.get_historical_prices is not yet implemented. " + "Use a separate historical bar source for now.", "yellow", )) - return Quote(asset=asset) + return None diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index f9488b683..c331def84 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -506,6 +506,123 @@ def test_avg_fill_from_legs_returns_none_when_unfilled(): assert Tastytrade._avg_fill_from_legs(placed) is None +# --------------------------------------------------------------------------- +# TastytradeData (market data) +# --------------------------------------------------------------------------- + +def test_get_last_price_prefers_mid(monkeypatch): + from lumibot.data_sources.tastytrade_data import TastytradeData + from lumibot.entities import Asset + from lumibot.data_sources import tastytrade_data as td_mod + + fake_md = MagicMock() + fake_md.mid = Decimal("100.10") + fake_md.last = Decimal("100.05") + fake_md.mark = Decimal("100.07") + fake_md.bid = Decimal("100.05") + fake_md.ask = Decimal("100.15") + + async def _get(_session, _symbol, _instrument_type): + return fake_md + + monkeypatch.setattr(td_mod, "_tt_get_market_data", _get) + + ds = TastytradeData(session=MagicMock(), runner=None) + + asset = Asset(symbol="AAPL", asset_type=Asset.AssetType.STOCK) + price = ds.get_last_price(asset) + assert price == 100.10 # picked mid + + +def test_get_last_price_falls_back_to_bid_ask_avg(monkeypatch): + from lumibot.data_sources.tastytrade_data import TastytradeData + from lumibot.entities import Asset + from lumibot.data_sources import tastytrade_data as td_mod + + fake_md = MagicMock() + fake_md.mid = None + fake_md.last = None + fake_md.mark = None + fake_md.bid = Decimal("99.90") + fake_md.ask = Decimal("100.10") + + async def _get(_session, _symbol, _instrument_type): + return fake_md + + monkeypatch.setattr(td_mod, "_tt_get_market_data", _get) + + ds = TastytradeData(session=MagicMock(), runner=None) + asset = Asset(symbol="AAPL", asset_type=Asset.AssetType.STOCK) + assert ds.get_last_price(asset) == 100.0 + + +def test_get_quote_populates_bid_ask_mid(monkeypatch): + from lumibot.data_sources.tastytrade_data import TastytradeData + from lumibot.entities import Asset + from lumibot.data_sources import tastytrade_data as td_mod + + fake_md = MagicMock() + fake_md.bid = Decimal("99.90") + fake_md.ask = Decimal("100.10") + fake_md.mid = Decimal("100.00") + fake_md.last = Decimal("100.05") + fake_md.mark = None + fake_md.bid_size = Decimal("100") + fake_md.ask_size = Decimal("200") + fake_md.volume = Decimal("500000") + fake_md.updated_at = datetime.datetime(2026, 5, 4, 14, 30) + + async def _get(_session, _symbol, _instrument_type): + return fake_md + + monkeypatch.setattr(td_mod, "_tt_get_market_data", _get) + + ds = TastytradeData(session=MagicMock(), runner=None) + quote = ds.get_quote(Asset(symbol="AAPL", asset_type=Asset.AssetType.STOCK)) + assert quote.bid == 99.90 + assert quote.ask == 100.10 + assert quote.mid_price == 100.00 + assert quote.price == 100.05 + assert quote.bid_size == 100 + assert quote.ask_size == 200 + + +def test_get_chains_pivots_to_lumibot_shape(monkeypatch): + from lumibot.data_sources.tastytrade_data import TastytradeData + from lumibot.entities import Asset + from lumibot.data_sources import tastytrade_data as td_mod + + def _opt(strike, opt_type): + m = MagicMock() + m.strike_price = Decimal(str(strike)) + m.option_type = MagicMock(value=opt_type) + return m + + chain = { + datetime.date(2026, 5, 16): [ + _opt(450, "C"), _opt(440, "C"), + _opt(450, "P"), _opt(440, "P"), + ], + datetime.date(2026, 6, 20): [ + _opt(460, "Call"), _opt(450, "Put"), + ], + } + + async def _get(_session, _symbol): + return chain + + monkeypatch.setattr(td_mod, "_tt_get_option_chain", _get) + + ds = TastytradeData(session=MagicMock(), runner=None) + out = ds.get_chains(Asset(symbol="SPY", asset_type=Asset.AssetType.STOCK)) + + assert out["Multiplier"] == 100 + assert out["Chains"]["CALL"]["2026-05-16"] == [440.0, 450.0] + assert out["Chains"]["PUT"]["2026-05-16"] == [440.0, 450.0] + assert out["Chains"]["CALL"]["2026-06-20"] == [460.0] + assert out["Chains"]["PUT"]["2026-06-20"] == [450.0] + + # --------------------------------------------------------------------------- # Live sandbox smoke (only runs when sandbox credentials are present) # --------------------------------------------------------------------------- From 4424d84aa20c3bb1647d13cd1d9e8116c7ce895c Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 20:39:56 -0400 Subject: [PATCH 5/9] fix(brokers/tastytrade): live-test fixes after real-API validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three issues surfaced when running the broker against a live Tastytrade production account (read-only paths only — balances, positions, market data, option chains). 1. _strategy_name collision: I had defined _strategy_name as a static helper, but Broker base class already uses self._strategy_name as a string attribute (set in Broker.__init__). The override broke _pull_positions with 'str object is not callable'. Drop the helper and use the existing self._strategy_name_from_input(strategy) classmethod from the base class. 2. Equity-option positions were skipped: the scaffold _tt_position_to_asset only handled instrument_type == 'equity' and warned on options. Now that the broker has _occ_to_asset, parse equity options into proper Lumibot Asset(OPTION, expiration, strike, right) instances. Verified against a real account holding -2 TQQQ 2026-05-08 $63.50 calls — parses correctly with negative qty for the short position. 3. Live smoke test was being skipped by the apitest gating in the repo's conftest.py (which requires Polygon/Theta creds to run any apitest- marked test). The Tastytrade live test only needs Tastytrade creds, so drop the apitest marker. The skipif on TASTYTRADE_* env vars still gates it correctly. Validated live (read-only) against production: - OAuth token refresh - Account.get + Account.get_balances - Account.get_positions (equity + equity options) - market_data.get_market_data (single SPY last price = real) - instruments.get_option_chain (SPY chain, multiple expirations) Tastytrade's market-data REST is aggressively rate-limited (429 on two back-to-back calls). For active polling, callers should debounce or wait for native DXLinkStreamer integration. --- lumibot/brokers/tastytrade.py | 39 +++++++++---------- tests/test_tastytrade_broker_smoke_apitest.py | 3 +- 2 files changed, 20 insertions(+), 22 deletions(-) diff --git a/lumibot/brokers/tastytrade.py b/lumibot/brokers/tastytrade.py index 7525f2a2c..d84ab6313 100644 --- a/lumibot/brokers/tastytrade.py +++ b/lumibot/brokers/tastytrade.py @@ -216,14 +216,6 @@ def __init__( def _run(self, coro: Awaitable[T], timeout: Optional[float] = 30.0) -> T: return self._async_bridge.run(coro, timeout=timeout) - @staticmethod - def _strategy_name(strategy) -> str: - if strategy is None: - return "Unknown" - if isinstance(strategy, str): - return strategy - return getattr(strategy, "name", str(strategy)) - # ------------------------------------------------------------------ # Account # ------------------------------------------------------------------ @@ -264,7 +256,7 @@ def _pull_positions(self, strategy) -> List[Position]: logger.error(colored(f"[Tastytrade] Failed to fetch positions: {e}", "red")) return [] - strategy_name = self._strategy_name(strategy) + strategy_name = self._strategy_name_from_input(strategy) positions: List[Position] = [] for tp in tt_positions or []: asset = self._tt_position_to_asset(tp) @@ -292,27 +284,34 @@ def _tt_position_quantity(tp) -> Decimal: qty = -qty return qty - @staticmethod - def _tt_position_to_asset(tp) -> Optional[Asset]: + @classmethod + def _tt_position_to_asset(cls, tp) -> Optional[Asset]: """Convert a Tastytrade position to a Lumibot Asset. - Tastytrade exposes ``instrument_type`` (Equity, Equity Option, Future, - Future Option, Cryptocurrency, ...) and ``symbol`` / ``underlying_symbol``. - The full mapping (especially OCC option symbol parsing) will be done - in the parser follow-up; for now we handle equities explicitly and - log a warning for option/future/crypto positions so the user can see - what's being skipped. + Handles ``Equity`` (stock) and ``Equity Option`` (OCC-symbol options). + Futures, future options, crypto, and other instrument types log a + warning and return None — they'll land when broader asset support + does. """ instrument = (getattr(tp, "instrument_type", "") or "").lower() - symbol = getattr(tp, "symbol", None) or getattr(tp, "underlying_symbol", None) + symbol = getattr(tp, "symbol", None) if not symbol: return None if instrument == "equity": return Asset(symbol=symbol, asset_type=Asset.AssetType.STOCK) + if instrument == "equity option": + asset = cls._occ_to_asset(symbol) + if asset is None: + logger.warning(colored( + f"[Tastytrade] Could not parse OCC symbol on equity-option " + f"position: {symbol!r}. Skipping.", + "yellow", + )) + return asset logger.warning(colored( f"[Tastytrade] Skipping position with unhandled instrument_type " - f"'{instrument}' for symbol {symbol}. Asset parsing will be " - f"completed in a follow-up commit.", + f"'{instrument}' for symbol {symbol}. (Futures / future options / " + f"crypto support is a follow-up.)", "yellow", )) return None diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index c331def84..6f9190a70 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -627,14 +627,13 @@ async def _get(_session, _symbol): # Live sandbox smoke (only runs when sandbox credentials are present) # --------------------------------------------------------------------------- -@pytest.mark.apitest @pytest.mark.skipif( not all(os.environ.get(k) for k in ( "TASTYTRADE_CLIENT_SECRET", "TASTYTRADE_REFRESH_TOKEN", "TASTYTRADE_ACCOUNT_NUMBER", )), - reason="Tastytrade sandbox credentials not configured.", + reason="Tastytrade credentials not configured.", ) def test_live_sandbox_balances_and_positions(): """Hit the sandbox API for balances + positions; expects no exceptions.""" From 260562529f30a9954b95227bd98736b796586ca8 Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 20:46:06 -0400 Subject: [PATCH 6/9] fix(brokers/tastytrade): correct price sign + equity-leg action enums Two real-API failures uncovered when running place/replace/cancel end-to-end against a live Tastytrade production account. 1. cant_buy_for_credit error on a BUY: Tastytrade encodes credit/debit in the SIGN of NewOrder.price (negative = debit, positive = credit) and the SDK serializer strips abs() before sending while emitting price-effect separately. Sending a positive price for a BUY made Tastytrade interpret it as "buy for a credit" and reject the order. Fix: add _is_debit_action + _sign_single_leg_price helpers and apply the right sign in _submit_order / _modify_order. For multileg, sign the price by order_type_norm: 'credit' -> positive, 'debit' -> negative, 'even' -> 0, 'limit' -> infer from leg net direction (raise on mixed-action limit since the sign is ambiguous). 2. order_legs.action: is invalid on an equity BUY: Tastytrade's standard order endpoint rejects plain ``Buy`` / ``Sell`` on equity legs even though those values exist in the OrderAction enum (they're for special cases like notional market orders). Equity legs need the explicit open/close form. Fix: map Lumibot equity sides to Tastytrade open/close actions: BUY -> BUY_TO_OPEN SELL -> SELL_TO_CLOSE SELL_SHORT -> SELL_TO_OPEN BUY_TO_COVER -> BUY_TO_CLOSE Plus pass-through for explicit *_TO_* sides. Validated live (read+write) against a production IRA, market closed, limit price far from market so it cannot fill: STEP 1 place BUY 1 SPY @ $1.00 LIMIT DAY -> id 463096620, Received STEP 2 replace $1.00 -> $1.50 -> id 463096622, Received STEP 3 cancel -> id 463096622, Cancelled The new id on replace is Tastytrade's normal behavior (cancel-and-new); _modify_order absorbs the new id back onto the Lumibot Order. Tests updated to match the new signed-price + open/close-action wire shape. 20 offline pass. --- lumibot/brokers/tastytrade.py | 87 ++++++++++++++++--- tests/test_tastytrade_broker_smoke_apitest.py | 37 ++++++-- 2 files changed, 103 insertions(+), 21 deletions(-) diff --git a/lumibot/brokers/tastytrade.py b/lumibot/brokers/tastytrade.py index d84ab6313..cc1b53178 100644 --- a/lumibot/brokers/tastytrade.py +++ b/lumibot/brokers/tastytrade.py @@ -342,6 +342,14 @@ def _to_occ_symbol(asset: Asset) -> str: @staticmethod def _lumi_side_to_tt_action(side: str, is_option: bool) -> "_TTOrderAction": + """Map a Lumibot order side to a Tastytrade OrderAction. + + Tastytrade's API does NOT accept plain ``Buy``/``Sell`` on equity + order legs — it wants the explicit open/close form even for stocks. + ``BUY``/``SELL`` plain values exist in the SDK enum for special + cases (e.g. notional market orders) but get rejected on the + standard order endpoint with ``order_legs.action: is invalid``. + """ s = (side or "").lower() if is_option: mapping = { @@ -349,17 +357,23 @@ def _lumi_side_to_tt_action(side: str, is_option: bool) -> "_TTOrderAction": "sell_to_open": _TTOrderAction.SELL_TO_OPEN, "buy_to_close": _TTOrderAction.BUY_TO_CLOSE, "sell_to_close": _TTOrderAction.SELL_TO_CLOSE, - # Plain buy/sell on options default to opening; callers should - # use the explicit *_to_open / *_to_close sides when possible. + # Plain buy/sell on options default to opening; callers + # should use the explicit *_to_open / *_to_close sides. "buy": _TTOrderAction.BUY_TO_OPEN, "sell": _TTOrderAction.SELL_TO_OPEN, } else: + # Equities: open long = BUY_TO_OPEN, close long = SELL_TO_CLOSE, + # short = SELL_TO_OPEN, cover = BUY_TO_CLOSE. mapping = { - "buy": _TTOrderAction.BUY, - "sell": _TTOrderAction.SELL, - "buy_to_cover": _TTOrderAction.BUY, - "sell_short": _TTOrderAction.SELL, + "buy": _TTOrderAction.BUY_TO_OPEN, + "sell": _TTOrderAction.SELL_TO_CLOSE, + "buy_to_open": _TTOrderAction.BUY_TO_OPEN, + "sell_to_close": _TTOrderAction.SELL_TO_CLOSE, + "sell_short": _TTOrderAction.SELL_TO_OPEN, + "sell_to_open": _TTOrderAction.SELL_TO_OPEN, + "buy_to_cover": _TTOrderAction.BUY_TO_CLOSE, + "buy_to_close": _TTOrderAction.BUY_TO_CLOSE, } if s not in mapping: raise ValueError(f"Unsupported order side {side!r} for Tastytrade.") @@ -431,7 +445,32 @@ def _build_leg(self, order: Order) -> "_TTLeg": def _format_price(price: Optional[Union[float, Decimal]]) -> Optional[Decimal]: if price is None: return None - return Decimal(str(price)).quantize(Decimal("0.01")) + # Preserve sign — Tastytrade encodes price-effect in the sign of price + # (negative = debit, positive = credit). The SDK's serializer strips + # abs() before sending and pairs it with a "price-effect" field. + sign = -1 if Decimal(str(price)) < 0 else 1 + return (sign * abs(Decimal(str(price))).quantize(Decimal("0.01"))) + + @staticmethod + def _is_debit_action(side: str, is_option: bool) -> bool: + """Return True if a BUY-side action (debit). False for SELL-side (credit).""" + s = (side or "").lower() + if is_option: + return s in ("buy", "buy_to_open", "buy_to_close") + return s in ("buy", "buy_to_cover") + + def _sign_single_leg_price( + self, + order: Order, + price: Optional[Union[float, Decimal]], + ) -> Optional[Union[float, Decimal]]: + """Apply sign convention for a single-leg order based on its side.""" + if price is None: + return None + is_option = order.asset and order.asset.asset_type == Asset.AssetType.OPTION + is_debit = self._is_debit_action(order.side, is_option=is_option) + magnitude = abs(Decimal(str(price))) + return -magnitude if is_debit else magnitude def _build_new_order( self, @@ -491,13 +530,16 @@ def _submit_order(self, order: Order) -> Optional[Order]: limit = order.limit_price if order_type_str == Order.OrderType.STOP_LIMIT: limit = order.stop_limit_price + # Tastytrade encodes credit/debit in the price sign — BUY -> negative, + # SELL -> positive. The SDK serializer strips abs() and emits price-effect. + signed_limit = self._sign_single_leg_price(order, limit) try: new_order = self._build_new_order( legs=[leg], order_type=order_type_str, time_in_force=order.time_in_force, - price=limit, + price=signed_limit, stop_trigger=order.stop_price, ) except Exception as e: @@ -547,18 +589,34 @@ def _submit_orders( legs = [self._build_leg(o) for o in orders] - # Tastytrade requires a positive price on debit/credit; sign comes - # from leg actions (buy=debit, sell=credit). 'even' uses 0.00. - if order_type_norm in ("debit", "credit"): + # Sign the price per Tastytrade's convention: positive = credit, + # negative = debit. The serializer sends abs(price) + price-effect. + if order_type_norm == "credit": if price is None: - raise ValueError(f"price is required for '{order_type_norm}' multileg.") + raise ValueError("price is required for 'credit' multileg.") tt_price: Optional[Decimal] = abs(Decimal(str(price))) + elif order_type_norm == "debit": + if price is None: + raise ValueError("price is required for 'debit' multileg.") + tt_price = -abs(Decimal(str(price))) elif order_type_norm == "even": tt_price = Decimal("0.00") elif order_type_norm == "limit": if price is None: raise ValueError("price is required for 'limit' multileg.") - tt_price = abs(Decimal(str(price))) + # Infer sign from leg net direction. If callers want explicit + # credit/debit semantics they should use those keywords directly. + buys = sum(1 for leg in legs if "buy" in str(leg.action.value).lower()) + sells = sum(1 for leg in legs if "sell" in str(leg.action.value).lower()) + if buys > 0 and sells == 0: + tt_price = -abs(Decimal(str(price))) # all buys -> debit + elif sells > 0 and buys == 0: + tt_price = abs(Decimal(str(price))) # all sells -> credit + else: + raise ValueError( + "Mixed-action multileg 'limit' is ambiguous. Use " + "order_type='credit' or 'debit' to disambiguate." + ) else: # market tt_price = None @@ -683,12 +741,13 @@ def _modify_order(self, order: Order, new_limit = limit_price if limit_price is not None else order.limit_price new_stop = stop_price if stop_price is not None else order.stop_price order_type_str = (order.order_type or "limit") + signed_limit = self._sign_single_leg_price(order, new_limit) try: new_order = self._build_new_order( legs=[leg], order_type=order_type_str, time_in_force=order.time_in_force, - price=new_limit, + price=signed_limit, stop_trigger=new_stop, ) except Exception as e: diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index 6f9190a70..6e4b502ee 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -164,8 +164,21 @@ def test_side_mapping_equity_vs_option(): from lumibot.brokers.tastytrade import Tastytrade from tastytrade.order import OrderAction - assert Tastytrade._lumi_side_to_tt_action("buy", is_option=False) == OrderAction.BUY - assert Tastytrade._lumi_side_to_tt_action("sell", is_option=False) == OrderAction.SELL + # Equity: plain buy -> BUY_TO_OPEN (Tastytrade rejects plain Buy/Sell on + # the standard order endpoint). + assert Tastytrade._lumi_side_to_tt_action( + "buy", is_option=False + ) == OrderAction.BUY_TO_OPEN + assert Tastytrade._lumi_side_to_tt_action( + "sell", is_option=False + ) == OrderAction.SELL_TO_CLOSE + assert Tastytrade._lumi_side_to_tt_action( + "sell_short", is_option=False + ) == OrderAction.SELL_TO_OPEN + assert Tastytrade._lumi_side_to_tt_action( + "buy_to_cover", is_option=False + ) == OrderAction.BUY_TO_CLOSE + # Options: assert Tastytrade._lumi_side_to_tt_action( "buy_to_open", is_option=True ) == OrderAction.BUY_TO_OPEN @@ -223,11 +236,15 @@ def test_submit_order_equity_limit(monkeypatch): new_order = captured[0] assert new_order.order_type == OrderType.LIMIT assert new_order.time_in_force == OrderTimeInForce.DAY - assert new_order.price == Decimal("150.25") + # BUY -> debit -> negative price (Tastytrade convention; serializer + # strips abs() and emits price-effect on the wire). + assert new_order.price == Decimal("-150.25") + assert new_order.price_effect.value == "Debit" assert len(new_order.legs) == 1 leg = new_order.legs[0] assert leg.symbol == "AAPL" - assert leg.action == OrderAction.BUY + # Equity buy maps to BUY_TO_OPEN (Tastytrade rejects plain Buy). + assert leg.action == OrderAction.BUY_TO_OPEN assert Decimal(str(leg.quantity)) == Decimal("10") finally: broker._async_bridge.close() @@ -260,10 +277,13 @@ def test_submit_order_option_limit_uses_occ(monkeypatch): broker._submit_order(order) assert len(captured) == 1 - leg = captured[0].legs[0] + new_order = captured[0] + leg = new_order.legs[0] assert leg.instrument_type == InstrumentType.EQUITY_OPTION assert leg.symbol == "AAPL 260717C00230000" assert leg.action == OrderAction.BUY_TO_OPEN + # BUY_TO_OPEN -> debit -> negative price. + assert new_order.price == Decimal("-4.20") finally: broker._async_bridge.close() @@ -318,7 +338,9 @@ def test_submit_orders_multileg_credit_spread(monkeypatch): assert len(captured) == 1 new_order = captured[0] assert new_order.order_type == OrderType.LIMIT - assert new_order.price == Decimal("2.50") # absolute value, sign from legs + # Credit spread -> positive price (you receive the premium). + assert new_order.price == Decimal("2.50") + assert new_order.price_effect.value == "Credit" assert len(new_order.legs) == 2 actions = [l.action for l in new_order.legs] assert OrderAction.SELL_TO_OPEN in actions @@ -470,7 +492,8 @@ async def _replace(_session, identifier, new_order): assert len(captured) == 1 identifier, new_order = captured[0] assert identifier == "1234" - assert new_order.price == Decimal("151.50") + # BUY -> debit -> negative. + assert new_order.price == Decimal("-151.50") assert order.identifier == "9999" # broker assigns new id on replace assert order.limit_price == 151.50 finally: From 3471aa8a931edb30478bd8467c3f3f85934f1fbe Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 21:26:09 -0400 Subject: [PATCH 7/9] fix(brokers/tastytrade): address review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five small correctness fixes from code review. 1. _AsyncBridge.run: cancel pending future on exception. future.result(timeout=...) leaving the underlying coroutine running on the background loop is a leak — most visibly on TimeoutError. Wrap in try/except, call future.cancel() on any exception, and re-raise so the caller still sees the failure. 2. SANDBOX parsing: bool('false') == True. Config path was bool(config.get('SANDBOX')), which lights up cert env for any non-empty string including 'false'. Add a small _parse_truthy helper used by both the config and env-var paths; treat the standard true/false token set explicitly. 3. _modify_order rejects multileg explicitly. _build_leg only constructs a single leg from the parent Order, so calling replace_order on a spread would silently submit a one-legged replacement. Detect order_class == MULTILEG (or len(child_orders) > 1) up front, log a clear error, and return None. Cancel-and-resubmit is the workaround until multileg replace lands. 4. _leg_to_lumi_side parses explicit equity actions. Tastytrade requires the open/close form on equity legs (BUY_TO_OPEN / SELL_TO_CLOSE / etc.) and that's what comes back when reading orders. The previous equity dict only mapped plain 'buy'/'sell' so reads silently fell through to OrderSide.BUY. Map all four explicit forms to the matching Lumibot OrderSide. 5. do_polling updates stored.status synchronously on terminal events. _safe_stream_dispatch enqueues an event the stream worker handles asynchronously — until that event is processed, stored.status is stale. The next poll cycle then sees the order missing from get_live_orders, calls is_active() which still returns True, and re-dispatches CANCELED on top of an already-FILLED/ERROR'd order. Set stored.status = FILLED / CANCELED / ERROR right after dispatch so is_active() reflects the new state on the next pass. Tests: 4 new offline tests cover findings 1-4 (24 offline pass). Finding 5 has no dedicated unit test — the change is three inline status assignments inside do_polling and a meaningful test would require mocking out the strategy executor's order tracking machinery, which is disproportionate. Verified by inspection. Skipped: one CodeRabbit suggestion to add @pytest.mark.apitest and is_test=True to the live smoke test. The apitest marker would re-trigger Lumibot's conftest auto-skip on Polygon/Theta credentials (see commit e4dace73 for why we removed it). is_test=True would force cert env even when the user provides production creds, breaking the prod read-only flow used during validation. The existing TASTYTRADE_SANDBOX env var already controls cert-vs-prod selection. --- lumibot/brokers/tastytrade.py | 65 +++++++++- tests/test_tastytrade_broker_smoke_apitest.py | 121 ++++++++++++++++++ 2 files changed, 182 insertions(+), 4 deletions(-) diff --git a/lumibot/brokers/tastytrade.py b/lumibot/brokers/tastytrade.py index cc1b53178..cd01e49f1 100644 --- a/lumibot/brokers/tastytrade.py +++ b/lumibot/brokers/tastytrade.py @@ -104,7 +104,15 @@ def run(self, coro: Awaitable[T], timeout: Optional[float] = 30.0) -> T: if not self._loop.is_running(): raise RuntimeError("Tastytrade asyncio bridge is not running.") future = asyncio.run_coroutine_threadsafe(coro, self._loop) - return future.result(timeout=timeout) + try: + return future.result(timeout=timeout) + except BaseException: + # Includes TimeoutError, KeyboardInterrupt, and any exception the + # coroutine raises after the caller has already moved on. Cancel + # the underlying task so it doesn't keep running on the loop and + # leak resources. + future.cancel() + raise def close(self) -> None: if self._loop.is_running(): @@ -155,14 +163,13 @@ def __init__( refresh_token = refresh_token or config.get("REFRESH_TOKEN") account_number = account_number or config.get("ACCOUNT_NUMBER") if is_test is None and "SANDBOX" in config: - is_test = bool(config.get("SANDBOX")) + is_test = self._parse_truthy(config.get("SANDBOX")) client_secret = client_secret or os.environ.get("TASTYTRADE_CLIENT_SECRET") refresh_token = refresh_token or os.environ.get("TASTYTRADE_REFRESH_TOKEN") account_number = account_number or os.environ.get("TASTYTRADE_ACCOUNT_NUMBER") if is_test is None: - env_sandbox = os.environ.get("TASTYTRADE_SANDBOX", "") - is_test = env_sandbox.strip().lower() in ("1", "true", "yes", "y") + is_test = self._parse_truthy(os.environ.get("TASTYTRADE_SANDBOX")) missing = [ n for n, v in ( @@ -216,6 +223,22 @@ def __init__( def _run(self, coro: Awaitable[T], timeout: Optional[float] = 30.0) -> T: return self._async_bridge.run(coro, timeout=timeout) + @staticmethod + def _parse_truthy(value) -> bool: + """Tolerant truthy parser for env vars / config dicts. + + Treats common stringy false-like values (``"false"``, ``"0"``, + ``"no"``, ``"off"``, ``""``) as False so a config of + ``{"SANDBOX": "false"}`` doesn't accidentally land on the cert + environment via Python's ``bool("false") == True``. + """ + if value is None: + return False + if isinstance(value, bool): + return value + s = str(value).strip().lower() + return s in ("1", "true", "yes", "y", "on") + # ------------------------------------------------------------------ # Account # ------------------------------------------------------------------ @@ -724,6 +747,11 @@ def _modify_order(self, order: Order, Tastytrade implements modification as a *replace*: build a new ``NewOrder`` with the same legs and an updated price, then call ``account.replace_order(session, order_id, new_order)``. + + Multileg orders are rejected explicitly — ``_build_leg`` only knows + how to construct a single leg from the parent ``Order``, which would + silently submit a one-legged replacement and break a spread. Cancel + and resubmit is the workaround until multileg replace is wired up. """ if not order.identifier: raise ValueError( @@ -732,6 +760,18 @@ def _modify_order(self, order: Order, if order.is_filled() or order.is_canceled(): return + if (order.order_class == Order.OrderClass.MULTILEG + or len(getattr(order, "child_orders", []) or []) > 1): + logger.error(colored( + f"[Tastytrade] _modify_order does not support multileg orders " + f"(order_class={order.order_class}, child_orders=" + f"{len(getattr(order, 'child_orders', []) or [])}). _build_leg " + f"only constructs a single leg, which would silently break the " + f"spread. Cancel and resubmit instead.", + "red", + )) + return None + try: leg = self._build_leg(order) except Exception as e: @@ -851,9 +891,17 @@ def _leg_to_lumi_side(cls, leg, is_option: bool) -> str: "buy": Order.OrderSide.BUY_TO_OPEN, "sell": Order.OrderSide.SELL_TO_OPEN, }.get(action_str, Order.OrderSide.BUY) + # Equity legs read back from Tastytrade use the explicit open/close + # form ("Buy to Open", "Sell to Close", ...) because that's what we + # had to send on the wire — Tastytrade rejects plain Buy/Sell on + # equity legs. Preserve that detail when parsing. return { "buy": Order.OrderSide.BUY, "sell": Order.OrderSide.SELL, + "buy to open": Order.OrderSide.BUY_TO_OPEN, + "sell to open": Order.OrderSide.SELL_TO_OPEN, + "buy to close": Order.OrderSide.BUY_TO_CLOSE, + "sell to close": Order.OrderSide.SELL_TO_CLOSE, }.get(action_str, Order.OrderSide.BUY) def _parse_broker_order(self, response: Any, strategy_name: str, @@ -1127,6 +1175,12 @@ def do_polling(self): continue status = (order.status or "").lower() + # Dispatch the transition AND update stored.status synchronously + # for terminal states. The dispatch enqueues an event that + # the stream worker processes asynchronously, so without the + # synchronous update the next poll's "missing from broker_ids" + # check would still see is_active() == True and re-dispatch + # CANCELED on top of an already-FILLED order. if status in ("submitted", "open"): self._safe_stream_dispatch(self.NEW_ORDER, order=stored) elif status == "fill": @@ -1139,8 +1193,10 @@ def do_polling(self): price=fill_price, filled_quantity=fill_qty, ) + stored.status = Order.OrderStatus.FILLED elif status == "canceled": self._safe_stream_dispatch(self.CANCELED_ORDER, order=stored) + stored.status = Order.OrderStatus.CANCELED elif status == "error": msg = getattr(placed, "reject_reason", None) or ( f"Tastytrade rejected order {order.identifier}" @@ -1148,6 +1204,7 @@ def do_polling(self): self._safe_stream_dispatch( self.ERROR_ORDER, order=stored, error_msg=msg, ) + stored.status = Order.OrderStatus.ERROR # 'partial_fill' deliberately not dispatched: polling can # easily miss partials; only complete fills are reliable. diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index 6e4b502ee..ef0adb652 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -82,6 +82,60 @@ async def _add(): bridge.close() +def test_async_bridge_cancels_future_on_timeout(): + """If run() times out the underlying coroutine must be cancelled, + not left running on the background loop.""" + from lumibot.brokers.tastytrade import _AsyncBridge + import concurrent.futures + + bridge = _AsyncBridge() + try: + cancelled = asyncio.Event() + + async def _slow(): + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + # Signal back from the loop thread. + bridge._loop.call_soon_threadsafe(cancelled.set) + raise + + with pytest.raises(concurrent.futures.TimeoutError): + bridge.run(_slow(), timeout=0.05) + + # Give the cancellation a moment to propagate. + deadline = asyncio.get_event_loop().time() + 1.0 if False else None + for _ in range(20): + if cancelled.is_set(): + break + import time as _t + _t.sleep(0.05) + assert cancelled.is_set(), "coroutine was not cancelled after timeout" + finally: + bridge.close() + + +def test_parse_truthy_handles_string_false(): + """bool('false') is True; _parse_truthy must not be that naive.""" + from lumibot.brokers.tastytrade import Tastytrade + + # Truthy + assert Tastytrade._parse_truthy("true") is True + assert Tastytrade._parse_truthy("True") is True + assert Tastytrade._parse_truthy("1") is True + assert Tastytrade._parse_truthy("yes") is True + assert Tastytrade._parse_truthy("on") is True + assert Tastytrade._parse_truthy(True) is True + # Falsy + assert Tastytrade._parse_truthy("false") is False + assert Tastytrade._parse_truthy("0") is False + assert Tastytrade._parse_truthy("no") is False + assert Tastytrade._parse_truthy("off") is False + assert Tastytrade._parse_truthy("") is False + assert Tastytrade._parse_truthy(None) is False + assert Tastytrade._parse_truthy(False) is False + + @patch("lumibot.brokers.tastytrade._TTAccount") @patch("lumibot.brokers.tastytrade._TTSession") def test_init_with_kwargs_resolves_account(mock_session_cls, mock_account_cls): @@ -519,6 +573,73 @@ def test_avg_fill_from_legs_weighted_average(): assert avg == Decimal("107") +def test_modify_order_rejects_multileg(monkeypatch): + """_modify_order must reject multileg orders early — it only knows how + to rebuild a single leg from the parent and would silently break a spread.""" + from lumibot.entities import Asset, Order + + broker, fake_account, _ = _make_broker(monkeypatch) + try: + replaced = [] + + async def _replace(*args, **kwargs): + replaced.append((args, kwargs)) + return MagicMock() + + fake_account.replace_order.side_effect = _replace + + parent = Order( + strategy="s", + asset=Asset(symbol="SPY", asset_type=Asset.AssetType.STOCK), + quantity=1, + side=Order.OrderSide.SELL_TO_OPEN, + order_class=Order.OrderClass.MULTILEG, + order_type=Order.OrderType.LIMIT, + limit_price=2.50, + ) + parent.identifier = "12345" + + result = broker._modify_order(parent, limit_price=2.75) + assert result is None + assert replaced == [], ( + "Multileg modify must NOT call replace_order with a single-leg payload" + ) + finally: + broker._async_bridge.close() + + +def test_leg_to_lumi_side_equity_explicit_actions(): + """Equity legs read back from Tastytrade arrive as 'Buy to Open' etc. + because that's what the wire requires. Parser must map them correctly.""" + from lumibot.brokers.tastytrade import Tastytrade + from lumibot.entities import Order + + def _leg(action_value): + leg = MagicMock() + leg.action = MagicMock(value=action_value) + return leg + + assert Tastytrade._leg_to_lumi_side( + _leg("Buy to Open"), is_option=False + ) == Order.OrderSide.BUY_TO_OPEN + assert Tastytrade._leg_to_lumi_side( + _leg("Sell to Close"), is_option=False + ) == Order.OrderSide.SELL_TO_CLOSE + assert Tastytrade._leg_to_lumi_side( + _leg("Sell to Open"), is_option=False + ) == Order.OrderSide.SELL_TO_OPEN + assert Tastytrade._leg_to_lumi_side( + _leg("Buy to Close"), is_option=False + ) == Order.OrderSide.BUY_TO_CLOSE + # Plain Buy/Sell still round-trip. + assert Tastytrade._leg_to_lumi_side( + _leg("Buy"), is_option=False + ) == Order.OrderSide.BUY + assert Tastytrade._leg_to_lumi_side( + _leg("Sell"), is_option=False + ) == Order.OrderSide.SELL + + def test_avg_fill_from_legs_returns_none_when_unfilled(): from lumibot.brokers.tastytrade import Tastytrade From 3f4717ceeb71f777c99713a6b35bbc25b6c4f6de Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 21:33:20 -0400 Subject: [PATCH 8/9] test(tastytrade): thread-safe Event + rename ambiguous loop var Address two follow-up review comments on the test file: 1. test_async_bridge_cancels_future_on_timeout used asyncio.Event to signal from the background asyncio thread back to the main test thread. asyncio.Event is bound to a single event loop and isn't thread-safe; the right primitive for cross-thread signalling is threading.Event. Swap it in. Also drop a stray dead expression (`deadline = ... if False else None`) that was a holdover from an earlier draft. 2. test_submit_orders_multileg_credit_spread had `for l in new_order.legs` which trips Ruff E741 ("ambiguous variable name"). Rename to `for leg in ...`. 24 offline tests still pass. --- tests/test_tastytrade_broker_smoke_apitest.py | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index ef0adb652..05cb41565 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -85,31 +85,35 @@ async def _add(): def test_async_bridge_cancels_future_on_timeout(): """If run() times out the underlying coroutine must be cancelled, not left running on the background loop.""" - from lumibot.brokers.tastytrade import _AsyncBridge import concurrent.futures + import threading + import time + + from lumibot.brokers.tastytrade import _AsyncBridge bridge = _AsyncBridge() try: - cancelled = asyncio.Event() + # threading.Event (not asyncio.Event) — the signal flows from the + # background asyncio thread to the main test thread, which is exactly + # what threading.Event is for. asyncio.Event is bound to a single + # loop and isn't safe across threads. + cancelled = threading.Event() async def _slow(): try: await asyncio.sleep(10) except asyncio.CancelledError: - # Signal back from the loop thread. - bridge._loop.call_soon_threadsafe(cancelled.set) + cancelled.set() raise with pytest.raises(concurrent.futures.TimeoutError): bridge.run(_slow(), timeout=0.05) - # Give the cancellation a moment to propagate. - deadline = asyncio.get_event_loop().time() + 1.0 if False else None + # Give cancellation a moment to propagate to the loop thread. for _ in range(20): if cancelled.is_set(): break - import time as _t - _t.sleep(0.05) + time.sleep(0.05) assert cancelled.is_set(), "coroutine was not cancelled after timeout" finally: bridge.close() @@ -396,7 +400,7 @@ def test_submit_orders_multileg_credit_spread(monkeypatch): assert new_order.price == Decimal("2.50") assert new_order.price_effect.value == "Credit" assert len(new_order.legs) == 2 - actions = [l.action for l in new_order.legs] + actions = [leg.action for leg in new_order.legs] assert OrderAction.SELL_TO_OPEN in actions assert OrderAction.BUY_TO_OPEN in actions finally: From 0a30dfcb20286aca065265752b870b88086de502 Mon Sep 17 00:00:00 2001 From: Tod Kemper Date: Mon, 4 May 2026 21:36:44 -0400 Subject: [PATCH 9/9] test: add tastytrade marker class + apply to live broker smoke test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolves the deferred review comment on test_live_sandbox_balances_and_ positions: add the explicit @pytest.mark.apitest decoration AND force is_test=True per CodeRabbit's recommendation, while avoiding the collateral damage we documented earlier (apitest alone would skip the test on missing Polygon/ThetaData creds). conftest.py: - Register a new tastytrade marker class, mirroring the existing ibkr marker for tests that hit a non-Polygon/Theta external API. - Extend the requires_ibkr branch in pytest_runtest_setup to also short-circuit Polygon/Theta credential checks when the tastytrade marker is present. test_live_sandbox_balances_and_positions: - Add @pytest.mark.apitest + @pytest.mark.tastytrade so it's discoverable as an API test without inheriting the polygon/thetadata gating. - Construct broker with is_test=True so the test always lands on Tastytrade's certification (sandbox) env regardless of how the user set TASTYTRADE_SANDBOX — prevents accidental prod-account hits in CI. 24 offline tests still pass; live test still gated by skipif on TASTYTRADE_* env credentials. --- tests/conftest.py | 8 ++++++-- tests/test_tastytrade_broker_smoke_apitest.py | 12 ++++++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 39e738cc4..014d2de0e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -85,6 +85,7 @@ def _restore_is_backtesting_env(): def pytest_configure(config): config.addinivalue_line("markers", "ibkr: downloader-only IBKR tests that do not require Polygon or ThetaData credentials") + config.addinivalue_line("markers", "tastytrade: Tastytrade-specific apitest that does not require Polygon or ThetaData credentials") def cleanup_all_schedulers(): @@ -279,9 +280,11 @@ def pytest_runtest_setup(item: pytest.Item): - polygon: requires Polygon credentials - thetadata: requires ThetaData credentials - ibkr: downloader-only IBKR tests; does not require Polygon/ThetaData creds + - tastytrade: Tastytrade-specific apitest; does not require Polygon/ThetaData creds Behavior: - - If a test is marked with ibkr, require neither Polygon nor ThetaData. + - If a test is marked with ibkr or tastytrade, require neither Polygon + nor ThetaData. - If a test is marked with polygon and/or thetadata, only those provider credentials are required. - If a test has apitest/downloader but no provider-specific markers, @@ -296,9 +299,10 @@ def pytest_runtest_setup(item: pytest.Item): requires_polygon = item.get_closest_marker("polygon") is not None requires_theta = item.get_closest_marker("thetadata") is not None requires_ibkr = item.get_closest_marker("ibkr") is not None + requires_tastytrade = item.get_closest_marker("tastytrade") is not None # Determine which providers are required - if requires_ibkr: + if requires_ibkr or requires_tastytrade: need_polygon = False need_theta = False elif requires_polygon or requires_theta: diff --git a/tests/test_tastytrade_broker_smoke_apitest.py b/tests/test_tastytrade_broker_smoke_apitest.py index 05cb41565..87c4091ec 100644 --- a/tests/test_tastytrade_broker_smoke_apitest.py +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -775,6 +775,8 @@ async def _get(_session, _symbol): # Live sandbox smoke (only runs when sandbox credentials are present) # --------------------------------------------------------------------------- +@pytest.mark.apitest +@pytest.mark.tastytrade @pytest.mark.skipif( not all(os.environ.get(k) for k in ( "TASTYTRADE_CLIENT_SECRET", @@ -784,10 +786,16 @@ async def _get(_session, _symbol): reason="Tastytrade credentials not configured.", ) def test_live_sandbox_balances_and_positions(): - """Hit the sandbox API for balances + positions; expects no exceptions.""" + """Hit the sandbox API for balances + positions; expects no exceptions. + + Forces ``is_test=True`` so the test is guaranteed to hit Tastytrade's + certification (sandbox) environment regardless of how the user set + ``TASTYTRADE_SANDBOX`` — preventing accidental prod-account runs in CI. + Users running this locally need sandbox credentials, not prod. + """ from lumibot.brokers.tastytrade import Tastytrade - broker = Tastytrade(connect_stream=False) + broker = Tastytrade(connect_stream=False, is_test=True) try: cash, positions_value, nlv = broker._get_balances_at_broker( quote_asset=None, strategy=None,