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..cd01e49f1 --- /dev/null +++ b/lumibot/brokers/tastytrade.py @@ -0,0 +1,1233 @@ +""" +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``. + +Functional surface: + +- Authentication via OAuth (``provider_secret`` + ``refresh_token``) +- Account selection by account number +- 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 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``) +- Native websocket streaming (``AlertStreamer`` + ``DXLinkStreamer``) +- Market data on ``TastytradeData`` (chains, quotes, historical bars) +""" + +import asyncio +import datetime +import os +import re +import threading +import traceback +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 +from lumibot.trading_builtins import PollingStream + +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 + 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 + + +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) + 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(): + 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" + POLL_EVENT = PollingStream.POLL_EVENT + + 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, + polling_interval: float = 5.0, + ): + 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 = 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: + is_test = self._parse_truthy(os.environ.get("TASTYTRADE_SANDBOX")) + + 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.polling_interval = polling_interval + 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 _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 + # ------------------------------------------------------------------ + 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_from_input(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 + + @classmethod + def _tt_position_to_asset(cls, tp) -> Optional[Asset]: + """Convert a Tastytrade position to a Lumibot Asset. + + 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) + 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}. (Futures / future options / " + f"crypto support is a follow-up.)", + "yellow", + )) + return None + + # ------------------------------------------------------------------ + # 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": + """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 = { + "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. + "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_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.") + 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 + # 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, + 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]: + # 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 + + 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 + # 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=signed_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] + + # 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("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.") + # 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 + + 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(): + 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): + """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)``. + + 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( + "Order identifier is not set; cannot modify. Did you submit it?" + ) + 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: + 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") + 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=signed_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 + # ------------------------------------------------------------------ + _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) + # 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, + strategy_object=None) -> Optional[Order]: + """Convert a Tastytrade ``PlacedOrder`` into a Lumibot ``Order``. + + 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: + """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 [] + + # ------------------------------------------------------------------ + # 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): + return PollingStream(self.polling_interval) + + def _register_stream_events(self): + 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): + 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() + # 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": + 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, + ) + 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}" + ) + 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. + + # 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 + # ------------------------------------------------------------------ + 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..9388ce61e --- /dev/null +++ b/lumibot/data_sources/tastytrade_data.py @@ -0,0 +1,237 @@ +import datetime +from decimal import Decimal +from typing import Callable, 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__) + +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 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" + SOURCE = "Tastytrade" + + def __init__( + self, + session=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 + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + def _await(self, coro): + if self._runner is not None: + return self._runner(coro) + import asyncio + return asyncio.run(coro) + + 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]: + 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( + self, + asset: Asset, + 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_historical_prices is not yet implemented. " + "Use a separate historical bar source for now.", + "yellow", + )) + return None 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/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 new file mode 100644 index 000000000..87c4091ec --- /dev/null +++ b/tests/test_tastytrade_broker_smoke_apitest.py @@ -0,0 +1,808 @@ +""" +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 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) +# --------------------------------------------------------------------------- + +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() + + +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.""" + import concurrent.futures + import threading + import time + + from lumibot.brokers.tastytrade import _AsyncBridge + + bridge = _AsyncBridge() + try: + # 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: + cancelled.set() + raise + + with pytest.raises(concurrent.futures.TimeoutError): + bridge.run(_slow(), timeout=0.05) + + # Give cancellation a moment to propagate to the loop thread. + for _ in range(20): + if cancelled.is_set(): + break + time.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): + """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() + + +# --------------------------------------------------------------------------- +# 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 + + # 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 + 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 + # 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" + # 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() + + +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 + 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() + + +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 + # 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 = [leg.action for leg 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 + + +# --------------------------------------------------------------------------- +# 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" + # 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: + 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_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 + + leg = MagicMock() + leg.fills = [] + placed = MagicMock() + placed.legs = [leg] + 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) +# --------------------------------------------------------------------------- + +@pytest.mark.apitest +@pytest.mark.tastytrade +@pytest.mark.skipif( + not all(os.environ.get(k) for k in ( + "TASTYTRADE_CLIENT_SECRET", + "TASTYTRADE_REFRESH_TOKEN", + "TASTYTRADE_ACCOUNT_NUMBER", + )), + reason="Tastytrade credentials not configured.", +) +def test_live_sandbox_balances_and_positions(): + """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, is_test=True) + 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()