From 7242dcb9e117d822479e8ab6fe5fc7509da07040 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 1 Sep 2026 18:55:58 -0400 Subject: [PATCH] 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`)) --- .../20260829T175542Z_e09e306a_prompt_io.md | 46 +++++ ...20260829T175542Z_e09e306a_prompt_io.raw.md | 43 +++++ piker/brokers/binance/feed.py | 133 +++++++------ tests/test_binance.py | 182 +++++++++++++----- 4 files changed, 297 insertions(+), 107 deletions(-) create mode 100644 ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.md b/ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.md new file mode 100644 index 00000000..2d769445 --- /dev/null +++ b/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. diff --git a/ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.raw.md b/ai/prompt-io/opencode/20260829T175542Z_e09e306a_prompt_io.raw.md new file mode 100644 index 00000000..7798c7b0 --- /dev/null +++ b/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. diff --git a/piker/brokers/binance/feed.py b/piker/brokers/binance/feed.py index 9f6af268..ee0b9e92 100644 --- a/piker/brokers/binance/feed.py +++ b/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 - # ^XXX https://developers.binance.com/docs/derivatives/change-log#2025-12-29 + nq: float|None = None # Normal quantity without RPI-order trades + # ^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}/' - + '/'.join( - f'{symbol.lower()}@{sub_name}' - for symbol in symbols - ) + f'{wss_url}/{symbol.lower()}@{sub_name}' 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'} - 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 len(symbols) != 1: + raise ValueError( + 'Binance quote readiness requires exactly one symbol; ' + f'got {symbols!r}' + ) - if not missing: - assert first_trade - return first_trade + symbol: str = symbols[0] + msg_iter = aiter(msg_stream) + + 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'{symbols!r}; no live {missing_str} data received' + f'Binance quote startup timed out after ' + f'{startup_timeout}s; ' + f'no live data received for {symbol}' ) 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' - ) + except StopAsyncIteration as err: + raise DataUnavailable( + f'Binance quote streams closed during startup; ' + 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) - 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 + mkt: MktPair + pair: Pair + mkt, pair = await get_mkt_info(symbols[0]) + mkt_mode: str = ( + 'usdtm_futes' + if isinstance(pair, FutesPair) + else 'spot' + ) - # build out init msgs according to latest spec - init_msgs.append( - FeedInit(mkt_info=mkt) - ) + # build out init msgs according to latest spec + init_msgs: list[FeedInit] = [ + FeedInit(mkt_info=mkt), + ] - assert mkt_mode ws_symbols: list[str] = [mkt.bs_mktid] async with open_quote_stream( mkt_mode=mkt_mode, diff --git a/tests/test_binance.py b/tests/test_binance.py index 31370ee8..67a390bd 100644 --- a/tests/test_binance.py +++ b/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, - ) - - 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' + startup_timeout=0.01, ) 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)