Harden Binance USD-M feed startup

- reject non-single-symbol feeds with `FeedInitializationError`
- route book and trade data across the correct USD-M endpoints
- accept the first matching L1 or trade quote as feed readiness
- preserve transport and startup-timeout failure context
- cover provider msg schemas and both readiness classes

Prompt-IO: ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.md

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

ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.md

@@ -0,0 +1,46 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: c3a1b1ba-65c6-4c8e-bd12-e596031b0eb0
+timestamp: 2026-08-29T17:55:42Z
+git_ref: e09e306a
+scope: code
+substantive: true
+raw_file: 20260829T175542Z_e09e306a_prompt_io.raw.md
+---
+
+## Prompt
+
+Preserve the existing Binance USD-M commit above the reviewed flake/xonsh
+stack, then add missing deterministic tests and correctness fixes as a
+separate follow-up with a complete editor-gated commit plan.
+
+## Response summary
+
+Hardens Binance feed startup around its supported one-symbol contract,
+direct websocket routes, current provider schemas, and first-quote
+readiness. Adds deterministic Binance regressions and separates generic
+reconnect behavior into its own tested commit boundary.
+
+## Files changed
+
+- `piker/brokers/binance/feed.py` - validate feed cardinality, route
+  USD-M stream classes, and declare readiness after either quote class.
+- `tests/test_binance.py` - cover routes, first-quote startup, cardinality,
+  and documented book-ticker and aggregate-trade schemas.
+- `ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.raw.md` - retain
+  the triggering prompt, generated-code summary, and verification output.
+- `ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.md` - record
+  structured provenance for the follow-up.
+
+## Human edits
+
+The user rejected reconstructing already committed Binance work, required
+preservation through cherry-pick, selected the flake/xonsh stack base, and
+required the review fixes to remain a separate tested boundary. The user also
+corrected the workflow to require an automatic `/commit-plan` at every stack
+review gate. During Tuicr review, the user rejected trade-gated startup,
+requested a feed-specific initialization error and async-generator TODO,
+and required generic reconnect changes and tests to land independently.
+The user then clarified that either L1 or trade data must unblock startup,
+rather than preferring one stream class or waiting independently for both.

ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.raw.md

@@ -0,0 +1,43 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-29T17:55:42Z
+git_ref: e09e306a
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt
+
+The user asked to land recent branch work in small chronological patch sets,
+including missing tests. For Binance, the user required preserving the
+existing committed topic patch through cherry-pick and placing newly found
+correctness repairs in a separate follow-up boundary above the flake/xonsh
+stack. The user also required every yielded stack layer to include a complete
+`/commit-plan` with patch-specific checks.
+
+## Generated code
+
+> `git diff HEAD~1..HEAD -- piker/brokers/binance/feed.py`
+
+Restricts Binance feeds to the supported one-symbol contract, renders direct
+quote routes, separates L1 liveness from first-trade readiness, and derives
+the contextual trade timeout from Trio's enclosing effective deadline.
+
+> `git diff HEAD~1..HEAD -- piker/data/_web_bs.py`
+
+Lets structured cancellation bypass reconnect throttling and ensures handled
+handshake rejections retry before accessing connection-local cancel state.
+
+> `git diff HEAD~1..HEAD -- tests/test_binance.py`
+
+Adds deterministic regressions for symbol cardinality, route selection,
+silent startup, immediate rejection, cancellation cleanup, aggregate-trade
+schema evolution, delayed trades, and finite/infinite/tiny deadlines.
+
+## Verification output
+
+- Targeted Binance tests: `10 passed in 0.26s`.
+- Ruff passes for all three changed Python files.
+- `git diff --check` passes.
+- Independent final review reports no actionable findings.
+- No live Binance network probe was run.

piker/brokers/binance/feed.py

@@ -56,7 +56,10 @@ from piker.accounting import (
     unpack_fqme,
 )
 from piker.types import Struct
-from piker.data.validate import FeedInit
+from piker.data.validate import (
+    FeedInit,
+    FeedInitializationError,
+)
 from piker.data._web_bs import (
     open_autorecon_ws,
     NoBsWs,
@@ -81,7 +84,7 @@ log = get_logger(name=__name__)
 
 
 class L1(Struct):
-    # https://binance-docs.github.io/apidocs/spot/en/#individual-symbol-book-ticker-streams
+    # https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Individual-Symbol-Book-Ticker-Streams
 
     update_id: int
     sym: str
@@ -106,8 +109,8 @@ class AggTrade(Struct, frozen=True):
     T: int  # Trade time
     m: bool  # Is the buyer the market maker?
     M: bool|None = None  # Ignore
-    nq: float|None = None # Normal quantity without the trades involving RPI orders
+    nq: float|None = None  # Normal quantity without RPI-order trades
-    # ^XXX https://developers.binance.com/docs/derivatives/change-log#2025-12-29
+    # ^XXX: https://developers.binance.com/docs/derivatives/change-log#2025-12-29
     st: int|None = None  # 1 for USD-M, 2 for COIN-M
 
 
@@ -403,14 +406,17 @@ def get_quote_routes(
     Render websocket URLs for each required Binance stream class.
 
     '''
+    if len(symbols) != 1:
+        raise ValueError(
+            'Binance quote routes require exactly one symbol; '
+            f'got {symbols!r}'
+        )
+
+    symbol: str = symbols[0]
     sub_names: tuple[str, str] = ('bookTicker', 'aggTrade')
     if mkt_mode == 'usdtm_futes':
         return tuple(
-            f'{wss_url}/'
+            f'{wss_url}/{symbol.lower()}@{sub_name}'
-            + '/'.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]),
@@ -420,7 +426,6 @@ def get_quote_routes(
     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
     )
 
@@ -441,10 +446,14 @@ async def relay_messages(
             await send_chan.send(msg)
 
 
+# TODO: Encapsulate ``relay_messages()`` and ``open_quote_stream()``
+# inside ``stream_quotes()`` so async-generator closure is owned by
+# the backend feed task.
 @acm
 async def open_quote_stream(
     mkt_mode: str,
     symbols: list[str],
+    connect_timeout: float = 16,
 ) -> AsyncGenerator[trio.MemoryReceiveChannel, None]:
     '''
     Open and merge the websocket routes required by a market.
@@ -463,15 +472,18 @@ async def open_quote_stream(
                 ws: NoBsWs = await stack.enter_async_context(
                     open_autorecon_ws(
                         url=wss_url,
-                        connect_timeout=16,
+                        connect_timeout=connect_timeout,
                     )
                 )
                 websockets.append(ws)
 
-        except TimeoutError as err:
+        except (
+            TimeoutError,
+            *NoBsWs.recon_errors,
+        ) as err:
             routes_str: str = '\n'.join(quote_routes)
             raise DataUnavailable(
-                f'Binance websocket connection timed out for '
+                f'Binance websocket connection failed for '
                 f'{symbols!r}:\n'
                 f'{routes_str}'
             ) from err
@@ -495,37 +507,44 @@ async def open_quote_stream(
 async def wait_for_live_quotes(
     msg_stream: AsyncGenerator | trio.MemoryReceiveChannel,
     symbols: list[str],
-    timeout: float = 16,
+    startup_timeout: float = 16,
 ) -> dict:
     '''
-    Require both L1 and trade data before declaring a feed live.
+    Require L1 or trade data before declaring a feed live.
 
     '''
-    missing: set[str] = {'l1', 'trade'}
+    if len(symbols) != 1:
-    first_trade: dict|None = None
+        raise ValueError(
-    try:
+            'Binance quote readiness requires exactly one symbol; '
-        with trio.fail_after(timeout):
+            f'got {symbols!r}'
-            async for typ, quote in msg_stream:
+        )
-                missing.discard(typ)
-                if typ == 'trade' and first_trade is None:
-                    first_trade = quote
 
-                if not missing:
+    symbol: str = symbols[0]
-                    assert first_trade
+    msg_iter = aiter(msg_stream)
-                    return first_trade
+
+    try:
+        with trio.fail_after(startup_timeout):
+            while True:
+                typ, quote = await anext(msg_iter)
+                if (
+                    typ in {'l1', 'trade'}
+                    and
+                    quote['symbol'] == symbol
+                ):
+                    return quote
 
     except trio.TooSlowError as err:
-        missing_str: str = ', '.join(sorted(missing))
         raise DataUnavailable(
-            f'Binance feed startup timed out after {timeout}s for '
+            f'Binance quote startup timed out after '
-            f'{symbols!r}; no live {missing_str} data received'
+            f'{startup_timeout}s; '
+            f'no live data received for {symbol}'
         ) from err
 
-    missing_str: str = ', '.join(sorted(missing))
+    except StopAsyncIteration as err:
         raise DataUnavailable(
-        f'Binance quote streams closed during startup for '
+            f'Binance quote streams closed during startup; '
-        f'{symbols!r}; no live {missing_str} data received'
+            f'no live data received for {symbol}'
-    )
+        ) from err
 
 
 async def stream_quotes(
@@ -537,36 +556,36 @@ async def stream_quotes(
     # startup sync
     task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,
 
-) -> None:
+    ) -> None:
+
+    if len(symbols) != 1:
+        # ``data.feed.allocate_persistent_feed()`` currently starts
+        # one backend task per symbol. Binance multiplexing remains
+        # deferred until that caller contract can represent
+        # multi-symbol startup.
+        raise FeedInitializationError(
+            'Binance feeds require exactly one symbol; '
+            f'got {symbols!r}'
+        )
 
     async with (
         tractor.trionics.maybe_raise_from_masking_exc(),
         send_chan as send_chan,
     ):
-        init_msgs: list[FeedInit] = []
-        mkt_mode: str|None = None
-        for sym in symbols:
         mkt: MktPair
         pair: Pair
-            mkt, pair = await get_mkt_info(sym)
+        mkt, pair = await get_mkt_info(symbols[0])
-            pair_mode: str = (
+        mkt_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(
+        init_msgs: list[FeedInit] = [
-                FeedInit(mkt_info=mkt)
+            FeedInit(mkt_info=mkt),
-            )
+        ]
 
-        assert mkt_mode
         ws_symbols: list[str] = [mkt.bs_mktid]
         async with open_quote_stream(
             mkt_mode=mkt_mode,

tests/test_binance.py

@@ -6,6 +6,7 @@ import pytest
 import trio
 
 from piker.brokers._util import DataUnavailable
+from piker.brokers.binance import feed as binance_feed
 from piker.brokers.binance.feed import (
     get_quote_routes,
     stream_messages,
@@ -16,7 +17,7 @@ from piker.brokers.binance.venues import (
     _futes_public_ws,
     _spot_ws,
 )
-from piker.data import _web_bs
+from piker.data.validate import FeedInitializationError
 
 
 def test_usdtm_websocket_routes() -> None:
@@ -49,7 +50,7 @@ def test_usdtm_websocket_routes() -> None:
 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
+    no first data, leaving ``stream_quotes()`` blocked before
     ``task_status.started()`` and hanging every chart consumer.
 
     Keep an empty receive channel open to reproduce a connected but
@@ -63,58 +64,12 @@ def test_silent_binance_feed_startup_fails() -> None:
         async with send, recv:
             with pytest.raises(
                 DataUnavailable,
-                match='no live l1, trade data received',
+                match='no live data received for NVDAUSDT',
             ):
                 await wait_for_live_quotes(
                     msg_stream=recv,
                     symbols=['NVDAUSDT'],
-                    timeout=0.01,
+                    startup_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)
@@ -167,3 +122,130 @@ def test_usdtm_aggregate_trade_stream_type() -> None:
         assert quote['last'] == 217.91
 
     trio.run(main)
+
+
+def test_usdtm_book_ticker_stream_type() -> None:
+    '''
+    Binance USD-M book-ticker frames include event and transaction
+    metadata in addition to the update ID and four L1 value fields.
+
+    Feed the documented USD-M response shape through
+    ``stream_messages()``. The emitted bid and ask ticks prove the
+    parser recognizes the real provider schema rather than a reduced
+    test-only event which could conceal route incompatibility.
+
+    '''
+    class MockWs:
+        def __init__(self) -> None:
+            self._msgs = iter([{
+                'e': 'bookTicker',
+                'u': 400900217,
+                'E': 1568014460893,
+                'T': 1568014460891,
+                's': 'BNBUSDT',
+                'b': '25.35190000',
+                'B': '31.21000000',
+                'a': '25.36520000',
+                'A': '40.66000000',
+            }])
+
+        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 == 'l1'
+        assert quote['symbol'] == 'BNBUSDT'
+        assert quote['ticks'][0] == {
+            'type': 'bid',
+            'price': 25.3519,
+            'size': 31.21,
+        }
+        assert quote['ticks'][2] == {
+            'type': 'ask',
+            'price': 25.3652,
+            'size': 40.66,
+        }
+
+    trio.run(main)
+
+
+def test_binance_feed_rejects_non_single_symbol_requests() -> None:
+    '''
+    Piker currently allocates one backend feed task per symbol, but
+    Binance previously accepted empty or multi-symbol requests and
+    either rendered invalid routes or silently retained one market.
+
+    Exercise both invalid cardinalities against the pure route helper
+    and public backend boundary. The helper ``ValueError`` proves no
+    malformed URLs can be rendered, while contextual
+    ``FeedInitializationError`` errors prove ``stream_quotes()``
+    rejects each request before market lookup, websocket opening, or
+    task startup.
+
+    '''
+    async def main() -> None:
+        for symbols in (
+            [],
+            ['NVDAUSDT', 'TSLAUSDT'],
+        ):
+            with pytest.raises(
+                ValueError,
+                match='exactly one symbol',
+            ):
+                get_quote_routes(
+                    mkt_mode='usdtm_futes',
+                    symbols=symbols,
+                )
+
+            with pytest.raises(
+                FeedInitializationError,
+                match='exactly one symbol',
+            ):
+                await binance_feed.stream_quotes(
+                    send_chan=None,
+                    symbols=symbols,
+                    feed_is_live=trio.Event(),
+                )
+
+    trio.run(main)
+
+
+@pytest.mark.parametrize(
+    'typ',
+    ['l1', 'trade'],
+)
+def test_first_quote_starts_feed(typ: str) -> None:
+    '''
+    Waiting independently for both L1 and trade data blocked startup
+    whenever either stream class remained quiet after the other proved
+    the merged Binance route was live.
+
+    Parameterize the first frame across both stream classes and keep
+    the merged channel open without a second producer. Returning that
+    exact frame under the startup timeout proves either L1-first or
+    trade-first delivery independently establishes feed readiness.
+
+    '''
+    async def main() -> None:
+        send, recv = trio.open_memory_channel(1)
+        first_quote: dict = {
+            'symbol': 'NVDAUSDT',
+        }
+        async with send, recv:
+            await send.send((typ, first_quote))
+            quote: dict = await wait_for_live_quotes(
+                msg_stream=recv,
+                symbols=['NVDAUSDT'],
+                startup_timeout=0.01,
+            )
+            assert quote is first_quote
+
+    trio.run(main)