add streamer reconnection callbacks (#186)

* add streamer reconnection callbacks

* add reconnect args to create()

* change create to __await__

Author: Graeme22

docs/account-streamer.rst

@@ -1,6 +1,9 @@
 Account Streamer
 ================
 
+Basic usage
+-----------
+
 The account streamer is used to track account-level updates, such as order fills, watchlist updates and quote alerts.
 Typically, you'll want a separate task running for the account streamer, which can then notify your application about important events.
 
@@ -35,3 +38,24 @@ Probably the most important information the account streamer handles is order fi
 
         async for order in streamer.listen(PlacedOrder):
             print(order)
+
+Retry callback
+--------------
+
+The account streamer has a special "callback" function which can be used to execute arbitrary code whenever the websocket reconnects. This is useful for re-subscribing to whatever alerts you wanted to subscribe to initially (in fact, you can probably use the same function/code you use when initializing the connection).
+The callback function should look something like this:
+
+.. code-block:: python
+
+    async def callback(streamer: AlertStreamer, arg1, arg2):
+        await streamer.subscribe_quote_alerts()
+
+The requirements are that the first parameter be the `AlertStreamer` instance, and the function should be asynchronous. Other than that, you have the flexibility to decide what arguments you want to use.
+This callback can then be used when creating the streamer:
+
+.. code-block:: python
+
+    async with AlertStreamer(session, reconnect_fn=callback, reconnect_args=(arg1, arg2)) as streamer:
+        # ...
+
+The reconnection uses `websockets`' exponential backoff algorithm, which can be configured through environment variables `here <https://websockets.readthedocs.io/en/14.1/reference/variables.html>`_.

docs/data-streamer.rst

@@ -10,7 +10,7 @@ You can create a streamer using an active production session:
 .. code-block:: python
 
    from tastytrade import DXLinkStreamer
-   streamer = await DXLinkStreamer.create(session)
+   streamer = await DXLinkStreamer(session)
 
 Or, you can create a streamer using an asynchronous context manager:
 
@@ -110,7 +110,7 @@ For example, we can use the streamer to create an option chain that will continu
            # the `streamer_symbol` property is the symbol used by the streamer
            streamer_symbols = [o.streamer_symbol for o in options]
 
-           streamer = await DXLinkStreamer.create(session)
+           streamer = await DXLinkStreamer(session)
            # subscribe to quotes and greeks for all options on that date
            await streamer.subscribe(Quote, [symbol] + streamer_symbols)
            await streamer.subscribe(Greeks, streamer_symbols)
@@ -146,3 +146,24 @@ Now, we can access the quotes and greeks at any time, and they'll be up-to-date
    print(live_prices.quotes[symbol], live_prices.greeks[symbol])
 
 >>> Quote(eventSymbol='.SPY230721C387', eventTime=0, sequence=0, timeNanoPart=0, bidTime=1689365699000, bidExchangeCode='X', bidPrice=62.01, bidSize=50.0, askTime=1689365699000, askExchangeCode='X', askPrice=62.83, askSize=50.0) Greeks(eventSymbol='.SPY230721C387', eventTime=0, eventFlags=0, index=7255910303911641088, time=1689398266363, sequence=0, price=62.6049270064687, volatility=0.536152815048564, delta=0.971506591907638, gamma=0.001814464566110275, theta=-0.1440768557397271, rho=0.0831882577866199, vega=0.0436861878838861)
+
+Retry callback
+--------------
+
+The data streamer has a special "callback" function which can be used to execute arbitrary code whenever the websocket reconnects. This is useful for re-subscribing to whatever events you wanted to subscribe to initially (in fact, you can probably use the same function/code you use when initializing the connection).
+The callback function should look something like this:
+
+.. code-block:: python
+
+    async def callback(streamer: DXLinkStreamer, arg1, arg2):
+        await streamer.subscribe(Quote, ['SPY'])
+
+The requirements are that the first parameter be the `DXLinkStreamer` instance, and the function should be asynchronous. Other than that, you have the flexibility to decide what arguments you want to use.
+This callback can then be used when creating the streamer:
+
+.. code-block:: python
+
+    async with DXLinkStreamer(session, reconnect_fn=callback, reconnect_args=(arg1, arg2)) as streamer:
+        # ...
+
+The reconnection uses `websockets`' exponential backoff algorithm, which can be configured through environment variables `here <https://websockets.readthedocs.io/en/14.1/reference/variables.html>`_.

pyproject.toml

@@ -16,7 +16,7 @@ dependencies = [
     "httpx>=0.27.2",
     "pandas-market-calendars>=4.4.1",
     "pydantic>=2.9.2",
-    "websockets>=14.1",
+    "websockets>=14.1,<15",
 ]
 
 [project.urls]
@@ -29,7 +29,7 @@ dev-dependencies = [
     "pytest-aio>=1.5.0",
     "pytest-cov>=5.0.0",
     "ruff>=0.6.9",
-    "pyright>=1.1.384",
+    "pyright>=1.1.390",
 ]
 
 [tool.setuptools.package-data]