Fix Binance USD-M websocket routing

Binance retired its legacy futures websocket and split quote
traffic across `/public` and `/market`. Route `bookTicker` and
`aggTrade` through dedicated URLs, merge both sockets, and accept
the new `st` stream-type field.

Bound connection and first-quote startup so failed or silent
streams raise `DataUnavailable` instead of hanging `datad` and
chart consumers.

Also,
- derive feed mode from resolved `FutesPair` metadata
- keep direct URL subscriptions stable across reconnects
- cover USD-M routes, payloads, and startup timeouts

Prompt-IO: ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.md

@@ -0,0 +1,46 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: unavailable
+timestamp: 2026-08-26T22:26:00Z
+git_ref: 5f631e19
+scope: code
+substantive: true
+raw_file: 20260826T222600Z_5f631e19_prompt_io.raw.md
+---
+
+## Prompt
+
+Diagnose and fix the hanging Binance USD-M perpetual feed for
+`NVDAUSDT.USDTM.PERP.BINANCE` by comparing piker with Binance's current
+API and changelog. Ensure unsupported or silent websocket feeds fail
+loudly into `datad` and chart consumers instead of blocking startup.
+
+## Response summary
+
+Migrates Binance USD-M market data from the retired legacy websocket to
+the split public and market routes, handles Binance's new aggregate-trade
+schema, and introduces bounded connection and first-data startup. Direct
+stream URLs avoid reconnect-time subscription-ACK races, while targeted
+tests preserve routing and fail-loud behavior.
+
+## Files changed
+
+- `piker/brokers/binance/feed.py` - route, merge, normalize, and bound
+  Binance quote streams.
+- `piker/brokers/binance/venues.py` - define current USD-M websocket
+  endpoints.
+- `piker/data/_web_bs.py` - support bounded websocket context startup.
+- `tests/test_binance.py` - cover routes, schemas, and startup failures.
+- `ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.raw.md` -
+  retain the generated-output record.
+- `ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.md` - record
+  structured provenance for the code changes.
+
+## Human edits
+
+The human explicitly required fail-loud behavior rather than merely
+updating the endpoint, and supplied a direct CLI reproduction proving
+both `datad.binance` and the chart caller blocked. Those requirements
+drove the connection and first-data deadlines and their regressions. The
+changes remain uncommitted for human review.

ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.raw.md

@@ -0,0 +1,57 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-26T22:26:00Z
+git_ref: 5f631e19
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt
+
+The user reported that selecting
+`NVDAUSDT.USDTM.PERP.BINANCE` connected Binance's USD-M
+websocket but never loaded the feed or downstream chart. They asked
+OpenCode to compare Binance's API changelog with piker's FQME and feed
+code, diagnose the failure, and solve it.
+
+The user clarified that future provider failures must raise loudly
+instead of silently blocking feed and chart startup. They also confirmed
+the same hang from
+`piker -l info --pdb chart nvdausdt.usdtm.perp.binance`.
+
+## Generated code
+
+> `git diff HEAD~1..HEAD -- piker/brokers/binance/feed.py`
+
+Routes USD-M book-ticker and aggregate-trade data through Binance's new
+`/public` and `/market` websocket paths, merges both sockets into the
+existing quote stream, accepts the new `st` aggregate-trade field, and
+requires both stream classes before declaring startup complete. It also
+removes the misleading missing-provider warning, derives feed mode from
+resolved pair metadata, and raises `DataUnavailable` for connection or
+live-data startup timeouts.
+
+> `git diff HEAD~1..HEAD -- piker/brokers/binance/venues.py`
+
+Defines Binance's current public and market USD-M websocket endpoints
+and makes the market endpoint the regular futures websocket API root.
+
+> `git diff HEAD~1..HEAD -- piker/data/_web_bs.py`
+
+Adds an optional connection deadline to `open_autorecon_ws()` so callers
+can fail context entry when repeated websocket handshakes never become
+connected.
+
+> `git diff HEAD~1..HEAD -- tests/test_binance.py`
+
+Adds regressions for exact Binance stream routing, silent live-data
+startup, silent websocket connection startup, and the current USD-M
+aggregate-trade payload schema.
+
+## Verification output
+
+`pytest -q tests/test_binance.py` passes all four tests. Compile checks,
+focused Ruff fatal-error checks, and `git diff --check` pass. Live probes
+through piker's merged stream received both required stream classes and
+returned normalized trade quotes for `NVDAUSDT` USD-M and `BTCUSDT`
+spot.

piker/brokers/binance/feed.py

@@ -22,18 +22,14 @@ from __future__ import annotations
 from contextlib import (
     asynccontextmanager as acm,
     aclosing,
+    AsyncExitStack,
 )
 from datetime import datetime
-from functools import (
-    partial,
-)
-import itertools
 from pprint import pformat
 from typing import (
     Any,
     AsyncGenerator,
     Callable,
-    Generator,
 )
 import time
 
@@ -77,6 +73,8 @@ from .venues import (
     Pair,
     FutesPair,
     get_api_eps,
+    _futes_market_ws,
+    _futes_public_ws,
 )
 
 log = get_logger(name=__name__)
@@ -110,12 +108,13 @@ class AggTrade(Struct, frozen=True):
     M: bool|None = None  # Ignore
     nq: float|None = None # Normal quantity without the trades involving RPI orders
     # ^XXX https://developers.binance.com/docs/derivatives/change-log#2025-12-29
+    st: int|None = None  # 1 for USD-M, 2 for COIN-M
 
 
 async def stream_messages(
     ws: NoBsWs,
 
-) -> AsyncGenerator[NoBsWs, dict]:
+) -> AsyncGenerator[tuple[str, dict], None]:
 
     # TODO: match syntax here!
     msg: dict[str, Any]
@@ -204,31 +203,6 @@ async def stream_messages(
                 yield 'trade', piker_quote
 
 
-def make_sub(
-    pairs: list[str],
-    sub_name: str,
-    uid: int,
-) -> dict[str, str]:
-    '''
-    Create a request subscription packet `dict`.
-
-    - spot:
-      https://binance-docs.github.io/apidocs/spot/en/#live-subscribing-unsubscribing-to-streams
-
-    - futes:
-      https://binance-docs.github.io/apidocs/futures/en/#websocket-market-streams
-
-    '''
-    return {
-        'method': 'SUBSCRIBE',
-        'params': [
-            f'{pair.lower()}@{sub_name}'
-            for pair in pairs
-        ],
-        'id': uid
-    }
-
-
 # TODO, why aren't frame resp `log.info()`s showing in upstream
 # code?!
 @acm
@@ -306,10 +280,6 @@ async def get_mkt_info(
 
     # uppercase since kraken bs_mktid is always upper
     if 'binance' not in fqme.lower():
-        log.warning(
-            f'Missing `.<provider>` part in fqme ??\n'
-            f'fqme: {fqme!r}\n'
-        )
         fqme += '.binance'
 
     mkt_mode: str = ''
@@ -439,54 +409,137 @@ async def get_mkt_info(
         return mkt, pair
 
 
-@acm
+def get_quote_routes(
-async def subscribe(
+    mkt_mode: str,
-    ws: NoBsWs,
     symbols: list[str],
+) -> tuple[str, ...]:
+    '''
+    Render websocket URLs for each required Binance stream class.
 
-    # defined once at import time to keep a global state B)
+    '''
-    iter_subids: Generator[int, None, None] = itertools.count(),
+    sub_names: tuple[str, str] = ('bookTicker', 'aggTrade')
+    if mkt_mode == 'usdtm_futes':
+        return tuple(
+            f'{wss_url}/'
+            + '/'.join(
+                f'{symbol.lower()}@{sub_name}'
+                for symbol in symbols
+            )
+            for wss_url, sub_name in (
+                (_futes_public_ws, sub_names[0]),
+                (_futes_market_ws, sub_names[1]),
+            )
+        )
 
+    wss_url: str = get_api_eps(mkt_mode)[1]
+    return tuple(
+        f'{wss_url}/{symbol.lower()}@{sub_name}'
+        for symbol in symbols
+        for sub_name in sub_names
+    )
+
+
+async def relay_messages(
+    ws: NoBsWs,
+    send_chan: trio.MemorySendChannel,
+) -> None:
+    '''
+    Relay one Binance websocket into a merged quote channel.
+
+    '''
+    async with (
+        send_chan,
+        aclosing(stream_messages(ws)) as msg_gen,
     ):
-    # setup subs
+        async for msg in msg_gen:
+            await send_chan.send(msg)
 
-    subid: int = next(iter_subids)
 
-    # trade data (aka L1)
+@acm
-    # https://binance-docs.github.io/apidocs/spot/en/#symbol-order-book-ticker
+async def open_quote_stream(
-    l1_sub = make_sub(symbols, 'bookTicker', subid)
+    mkt_mode: str,
-    await ws.send_msg(l1_sub)
+    symbols: list[str],
+) -> AsyncGenerator[trio.MemoryReceiveChannel, None]:
+    '''
+    Open and merge the websocket routes required by a market.
 
-    # aggregate (each order clear by taker **not** by maker)
+    '''
-    # trades data:
+    quote_routes: tuple[str, ...] = get_quote_routes(
-    # https://binance-docs.github.io/apidocs/spot/en/#aggregate-trade-streams
+        mkt_mode=mkt_mode,
-    agg_trades_sub = make_sub(symbols, 'aggTrade', subid)
+        symbols=symbols,
-    await ws.send_msg(agg_trades_sub)
+    )
 
-    # might get ack from ws server, or maybe some
+    send, recv = trio.open_memory_channel(616)
-    # other msg still in transit..
+    async with AsyncExitStack() as stack:
-    res = await ws.recv_msg()
+        websockets: list[NoBsWs] = []
-    subid: str|None = res.get('id')
+        try:
-    if subid:
+            for wss_url in quote_routes:
-        assert res['id'] == subid
+                ws: NoBsWs = await stack.enter_async_context(
+                    open_autorecon_ws(
+                        url=wss_url,
+                        connect_timeout=16,
+                    )
+                )
+                websockets.append(ws)
 
-    yield
+        except TimeoutError as err:
+            routes_str: str = '\n'.join(quote_routes)
+            raise DataUnavailable(
+                f'Binance websocket connection timed out for '
+                f'{symbols!r}:\n'
+                f'{routes_str}'
+            ) from err
 
-    subs = []
+        async with trio.open_nursery() as tn:
-    for sym in symbols:
+            for ws in websockets:
-        subs.append("{sym}@aggTrade")
+                tn.start_soon(
-        subs.append("{sym}@bookTicker")
+                    relay_messages,
+                    ws,
+                    send.clone(),
+                )
 
-    # unsub from all pairs on teardown
+            await send.aclose()
-    if ws.connected():
+            async with recv:
-        await ws.send_msg({
+                try:
-            "method": "UNSUBSCRIBE",
+                    yield recv
-            "params": subs,
+                finally:
-            "id": subid,
+                    tn.cancel_scope.cancel()
-        })
 
-        # XXX: do we need to ack the unsub?
+
-        # await ws.recv_msg()
+async def wait_for_live_quotes(
+    msg_stream: AsyncGenerator | trio.MemoryReceiveChannel,
+    symbols: list[str],
+    timeout: float = 16,
+) -> dict:
+    '''
+    Require both L1 and trade data before declaring a feed live.
+
+    '''
+    missing: set[str] = {'l1', 'trade'}
+    first_trade: dict|None = None
+    try:
+        with trio.fail_after(timeout):
+            async for typ, quote in msg_stream:
+                missing.discard(typ)
+                if typ == 'trade' and first_trade is None:
+                    first_trade = quote
+
+                if not missing:
+                    assert first_trade
+                    return first_trade
+
+    except trio.TooSlowError as err:
+        missing_str: str = ', '.join(sorted(missing))
+        raise DataUnavailable(
+            f'Binance feed startup timed out after {timeout}s for '
+            f'{symbols!r}; no live {missing_str} data received'
+        ) from err
+
+    missing_str: str = ', '.join(sorted(missing))
+    raise DataUnavailable(
+        f'Binance quote streams closed during startup for '
+        f'{symbols!r}; no live {missing_str} data received'
+    )
 
 
 async def stream_quotes(
@@ -503,43 +556,40 @@ async def stream_quotes(
     async with (
         tractor.trionics.maybe_raise_from_masking_exc(),
         send_chan as send_chan,
-        open_cached_client('binance') as client,
     ):
         init_msgs: list[FeedInit] = []
+        mkt_mode: str|None = None
         for sym in symbols:
             mkt: MktPair
             pair: Pair
             mkt, pair = await get_mkt_info(sym)
+            pair_mode: str = (
+                'usdtm_futes'
+                if isinstance(pair, FutesPair)
+                else 'spot'
+            )
+            if mkt_mode not in {None, pair_mode}:
+                raise DataUnavailable(
+                    'Binance cannot mix spot and futures symbols in '
+                    f'one feed: {symbols!r}'
+                )
+            mkt_mode = pair_mode
 
             # build out init msgs according to latest spec
             init_msgs.append(
                 FeedInit(mkt_info=mkt)
             )
 
-        wss_url: str = get_api_eps(client.mkt_mode)[1]  # 2nd elem is wss url
+        assert mkt_mode
-
+        ws_symbols: list[str] = [mkt.bs_mktid]
-        # TODO: for sanity, but remove eventually Xp
+        async with open_quote_stream(
-        if 'future' in mkt.type_key:
+            mkt_mode=mkt_mode,
-            assert 'fstream' in wss_url
+            symbols=ws_symbols,
-
+        ) as msg_stream:
-        async with (
+            quote: dict = await wait_for_live_quotes(
-            open_autorecon_ws(
+                msg_stream=msg_stream,
-                url=wss_url,
+                symbols=ws_symbols,
-                fixture=partial(
+            )
-                    subscribe,
-                    symbols=[mkt.bs_mktid],
-                ),
-            ) as ws,
-
-            # avoid stream-gen closure from breaking trio..
-            aclosing(stream_messages(ws)) as msg_gen,
-        ):
-            # log.info('WAITING ON FIRST LIVE QUOTE..')
-            typ, quote = await anext(msg_gen)
-
-            # pull a first quote and deliver
-            while typ != 'trade':
-                typ, quote = await anext(msg_gen)
 
             task_status.started((init_msgs, quote))
 
@@ -556,7 +606,7 @@ async def stream_quotes(
             topic: str = mkt.bs_fqme
 
             # start streaming
-            async for typ, quote in msg_gen:
+            async for typ, quote in msg_stream:
                 # period = time.time() - last
                 # hz = 1/period if period else float('inf')
                 # if hz > 60:

piker/brokers/binance/venues.py

@@ -42,7 +42,10 @@ _spot_ws: str = 'wss://stream.binance.com/ws'
 # or this one? ..
 # 'wss://ws-api.binance.com:443/ws-api/v3',
 
-# https://binance-docs.github.io/apidocs/futures/en/#websocket-market-streams
+# https://developers.binance.com/docs/derivatives/usds-margined-futures/
+# websocket-market-streams/Important-WebSocket-Change-Notice
+_futes_public_ws: str = f'wss://fstream.{_domain}/public/ws'
+_futes_market_ws: str = f'wss://fstream.{_domain}/market/ws'
 _futes_ws: str = f'wss://fstream.{_domain}/ws'
 _auth_futes_ws: str = 'wss://fstream-auth.{_domain}/ws'
 
@@ -79,7 +82,7 @@ def get_api_eps(venue: MarketType) -> tuple[str, str]:
         ),
         'usdtm_futes': (
             _futes_url,
-            _futes_ws,
+            _futes_market_ws,
         ),
     }[venue]
 

piker/data/_web_bs.py

@@ -342,6 +342,7 @@ async def open_autorecon_ws(
     url: str,
 
     fixture: AsyncContextManager|None = None,
+    connect_timeout: float|None = None,
 
     # time in sec between msgs received before
     # we presume connection might need a reset.
@@ -391,7 +392,19 @@ async def open_autorecon_ws(
                     reset_after=reset_after,
                 )
             )
+            try:
+                if connect_timeout is None:
                     await nobsws._connected.wait()
+                else:
+                    with trio.fail_after(connect_timeout):
+                        await nobsws._connected.wait()
+
+            except trio.TooSlowError as err:
+                raise TimeoutError(
+                    f'Websocket connection timed out after '
+                    f'{connect_timeout}s: {url}'
+                ) from err
+
             assert nobsws._cs
             assert nobsws.connected()
             try:

tests/test_binance.py

@@ -0,0 +1,169 @@
+'''
+Binance websocket routing and feed-startup regressions.
+
+'''
+import pytest
+import trio
+
+from piker.brokers._util import DataUnavailable
+from piker.brokers.binance.feed import (
+    get_quote_routes,
+    stream_messages,
+    wait_for_live_quotes,
+)
+from piker.brokers.binance.venues import (
+    _futes_market_ws,
+    _futes_public_ws,
+    _spot_ws,
+)
+from piker.data import _web_bs
+
+
+def test_usdtm_websocket_routes() -> None:
+    '''
+    Binance retired its legacy USD-M ``/ws`` route in April 2026.
+
+    Market and high-frequency public traffic now use separate base
+    paths. Verify piker retains both routes so a
+    future endpoint edit cannot silently send a valid subscription to
+    a socket which never publishes that traffic class.
+
+    '''
+    routes: tuple[str, ...] = get_quote_routes(
+        mkt_mode='usdtm_futes',
+        symbols=['NVDAUSDT'],
+    )
+    assert routes == (
+        f'{_futes_public_ws}/nvdausdt@bookTicker',
+        f'{_futes_market_ws}/nvdausdt@aggTrade',
+    )
+    assert get_quote_routes(
+        mkt_mode='spot',
+        symbols=['BTCUSDT'],
+    ) == (
+        f'{_spot_ws}/btcusdt@bookTicker',
+        f'{_spot_ws}/btcusdt@aggTrade',
+    )
+
+
+def test_silent_binance_feed_startup_fails() -> None:
+    '''
+    A retired Binance websocket accepted piker's connection but sent
+    no aggregate trades, leaving ``stream_quotes()`` blocked before
+    ``task_status.started()`` and hanging every chart consumer.
+
+    Keep an empty receive channel open to reproduce a connected but
+    silent socket. The controlled deadline proves startup raises
+    ``DataUnavailable`` with the absent stream classes instead of
+    waiting indefinitely for a first quote.
+
+    '''
+    async def main() -> None:
+        send, recv = trio.open_memory_channel(1)
+        async with send, recv:
+            with pytest.raises(
+                DataUnavailable,
+                match='no live l1, trade data received',
+            ):
+                await wait_for_live_quotes(
+                    msg_stream=recv,
+                    symbols=['NVDAUSDT'],
+                    timeout=0.01,
+                )
+
+    trio.run(main)
+
+
+def test_websocket_connection_startup_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A websocket handshake which never completed previously blocked
+    before Binance's live-data timeout started. Generic feed startup
+    then waited 616 seconds, effectively hanging the chart like a
+    connected socket which never published quotes.
+
+    Replace the reconnect task with one which reports task startup
+    but never sets ``NoBsWs._connected``. The bounded context entry
+    proves ``open_autorecon_ws()`` raises a URL-bearing
+    ``TimeoutError`` and cancels its nursery instead of waiting
+    indefinitely.
+
+    '''
+    async def never_connect(
+        *args,
+        task_status=trio.TASK_STATUS_IGNORED,
+        **kwargs,
+    ) -> None:
+        task_status.started()
+        await trio.sleep_forever()
+
+    monkeypatch.setattr(
+        _web_bs,
+        '_reconnect_forever',
+        never_connect,
+    )
+
+    async def main() -> None:
+        with pytest.raises(
+            TimeoutError,
+            match='wss://silent.invalid/ws',
+        ):
+            async with _web_bs.open_autorecon_ws(
+                url='wss://silent.invalid/ws',
+                connect_timeout=0.01,
+            ):
+                pytest.fail(
+                    'silent websocket unexpectedly connected'
+                )
+
+    trio.run(main)
+
+
+def test_usdtm_aggregate_trade_stream_type() -> None:
+    '''
+    Binance added ``st`` to USD-M aggregate trades during its June
+    2026 UM/CM websocket integration. The first valid NVDAUSDT trade
+    then raised ``TypeError`` while constructing piker's strict
+    ``AggTrade`` struct, terminating the feed relay during startup.
+
+    Feed one representative USD-M frame through ``stream_messages()``
+    and assert its trade quote is emitted. This proves the new stream
+    discriminator is accepted without weakening parsing of the fields
+    used by downstream consumers.
+
+    '''
+    class MockWs:
+        def __init__(self) -> None:
+            self._msgs = iter([{
+                'e': 'aggTrade',
+                'E': 1787782452314,
+                'a': 8430770,
+                's': 'NVDAUSDT',
+                'p': '217.91000',
+                'q': '0.92',
+                'nq': '0.92',
+                'f': 20677382,
+                'l': 20677382,
+                'T': 1787782452162,
+                'm': True,
+                'st': 1,
+            }])
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self) -> dict:
+            try:
+                return next(self._msgs)
+            except StopIteration:
+                raise StopAsyncIteration from None
+
+    async def main() -> None:
+        msg_gen = stream_messages(MockWs())
+        typ, quote = await anext(msg_gen)
+        assert typ == 'trade'
+        assert quote['symbol'] == 'NVDAUSDT'
+        assert quote['last'] == 217.91
+
+    trio.run(main)