streamer errors

Author: Graeme22

tastytrade/__init__.py

@@ -7,6 +7,7 @@
 VERSION = "9.12"
 
 __version__ = VERSION
+version_str = f"tastyware/tastytrade:v{VERSION}"
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)

tastytrade/order.py

@@ -5,7 +5,7 @@
 
 from pydantic import computed_field, field_serializer, model_validator
 
-from tastytrade import VERSION
+from tastytrade import version_str
 from tastytrade.utils import (
     PriceEffect,
     TastytradeJsonDataclass,
@@ -241,7 +241,7 @@ class NewOrder(TastytradeJsonDataclass):
 
     time_in_force: OrderTimeInForce
     order_type: OrderType
-    source: str = f"tastyware/tastytrade:v{VERSION}"
+    source: str = version_str
     legs: list[Leg]
     gtc_date: Optional[date] = None
     #: For a stop/stop limit order. If the latter, use price for the limit price
@@ -277,7 +277,7 @@ class NewComplexOrder(TastytradeJsonDataclass):
     """
 
     orders: list[NewOrder]
-    source: str = f"tastyware/tastytrade:v{VERSION}"
+    source: str = version_str
     trigger_order: Optional[NewOrder] = None
     type: ComplexOrderType = ComplexOrderType.OCO
 

tastytrade/session.py

@@ -341,6 +341,10 @@ def __init__(
             headers=self.sync_client.headers.copy(),
             proxy=proxy,
         )
+        #: expiration for session token
+        self.session_expiration = datetime.fromisoformat(
+            data["session-expiration"].replace("Z", "+00:00")
+        )
 
         # Pull streamer tokens and urls
         url = (
@@ -349,10 +353,15 @@ def __init__(
             else "/quote-streamer-tokens"
         )
         data = self._get(url)
+
         #: Auth token for dxfeed websocket
         self.streamer_token = data["token"]
         #: URL for dxfeed websocket
         self.dxlink_url = data["dxlink-url"]
+        #: expiration for streamer token
+        self.streamer_expiration = datetime.fromisoformat(
+            data["expires-at"].replace("Z", "+00:00")
+        )
 
     def __enter__(self):
         return self

tastytrade/streamer.py

@@ -21,7 +21,7 @@
 from websockets.asyncio.client import ClientConnection, connect
 from websockets.exceptions import ConnectionClosed
 
-from tastytrade import logger
+from tastytrade import logger, version_str
 from tastytrade.account import Account, AccountBalance, CurrentPosition, TradingStatus
 from tastytrade.dxfeed import (
     Event,
@@ -48,7 +48,7 @@
 CERT_STREAMER_URL = "wss://streamer.cert.tastyworks.com"
 STREAMER_URL = "wss://streamer.tastyworks.com"
 
-DXLINK_VERSION = "0.1-js/1.0.0-beta.4"
+DXLINK_VERSION = "0.1-DXF-JS/0.3.0"
 
 
 class QuoteAlert(TastytradeJsonDataclass):
@@ -361,6 +361,7 @@ async def _subscribe(
             "auth-token": self.token,
             "action": subscription.value,
             "request-id": self.request_id,
+            "source": version_str,
         }
         if value:
             message["value"] = value
@@ -512,17 +513,23 @@ async def _connect(self) -> None:
                         logger.debug("Feed configured: %s", message)
                     elif message["type"] == "KEEPALIVE":
                         pass
+                    elif message["type"] == "ERROR":
+                        logger.error(f"Fatal streamer error: {message['message']}")
+                        asyncio.create_task(self.close())
+                        return
                     else:
-                        logger.error(f"Streamer error: {message}")
+                        logger.error(f"Unknown message: {message}")
             except ConnectionClosed as e:
                 logger.error(f"Websocket connection closed with {e}")
                 if e.rcvd and e.rcvd.code == 1009:
                     logger.error(
                         "Subscription message too long! Try reducing the number of symbols."
                     )
+                    asyncio.create_task(self.close())
                     return
             except asyncio.CancelledError:
                 logger.debug("Websocket interrupted, cancelling main loop.")
+                asyncio.create_task(self.close())
                 return
             logger.debug("Websocket connection closed, retrying...")
             reconnecting = True