diff --git a/ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.md b/ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.md new file mode 100644 index 00000000..1055a6bf --- /dev/null +++ b/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. diff --git a/ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.raw.md b/ai/prompt-io/opencode/20260826T222600Z_5f631e19_prompt_io.raw.md new file mode 100644 index 00000000..e11ca0ed --- /dev/null +++ b/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. diff --git a/piker/brokers/binance/feed.py b/piker/brokers/binance/feed.py index 5a0a4175..4611f0ef 100644 --- a/piker/brokers/binance/feed.py +++ b/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 `.` 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 -async def subscribe( - ws: NoBsWs, +def get_quote_routes( + mkt_mode: str, 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]), + ) + ) -): - # setup subs + 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 + ) - subid: int = next(iter_subids) - # trade data (aka L1) - # https://binance-docs.github.io/apidocs/spot/en/#symbol-order-book-ticker - l1_sub = make_sub(symbols, 'bookTicker', subid) - await ws.send_msg(l1_sub) +async def relay_messages( + ws: NoBsWs, + send_chan: trio.MemorySendChannel, +) -> None: + ''' + Relay one Binance websocket into a merged quote channel. - # aggregate (each order clear by taker **not** by maker) - # trades data: - # https://binance-docs.github.io/apidocs/spot/en/#aggregate-trade-streams - agg_trades_sub = make_sub(symbols, 'aggTrade', subid) - await ws.send_msg(agg_trades_sub) + ''' + async with ( + send_chan, + aclosing(stream_messages(ws)) as msg_gen, + ): + async for msg in msg_gen: + await send_chan.send(msg) - # might get ack from ws server, or maybe some - # other msg still in transit.. - res = await ws.recv_msg() - subid: str|None = res.get('id') - if subid: - assert res['id'] == subid - yield +@acm +async def open_quote_stream( + mkt_mode: str, + symbols: list[str], +) -> AsyncGenerator[trio.MemoryReceiveChannel, None]: + ''' + Open and merge the websocket routes required by a market. - subs = [] - for sym in symbols: - subs.append("{sym}@aggTrade") - subs.append("{sym}@bookTicker") + ''' + quote_routes: tuple[str, ...] = get_quote_routes( + mkt_mode=mkt_mode, + symbols=symbols, + ) - # unsub from all pairs on teardown - if ws.connected(): - await ws.send_msg({ - "method": "UNSUBSCRIBE", - "params": subs, - "id": subid, - }) + send, recv = trio.open_memory_channel(616) + async with AsyncExitStack() as stack: + websockets: list[NoBsWs] = [] + try: + for wss_url in quote_routes: + ws: NoBsWs = await stack.enter_async_context( + open_autorecon_ws( + url=wss_url, + connect_timeout=16, + ) + ) + websockets.append(ws) - # XXX: do we need to ack the unsub? - # await ws.recv_msg() + 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 + + async with trio.open_nursery() as tn: + for ws in websockets: + tn.start_soon( + relay_messages, + ws, + send.clone(), + ) + + await send.aclose() + async with recv: + try: + yield recv + finally: + tn.cancel_scope.cancel() + + +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 - - # TODO: for sanity, but remove eventually Xp - if 'future' in mkt.type_key: - assert 'fstream' in wss_url - - async with ( - open_autorecon_ws( - url=wss_url, - 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) + assert mkt_mode + ws_symbols: list[str] = [mkt.bs_mktid] + async with open_quote_stream( + mkt_mode=mkt_mode, + symbols=ws_symbols, + ) as msg_stream: + quote: dict = await wait_for_live_quotes( + msg_stream=msg_stream, + symbols=ws_symbols, + ) 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: diff --git a/piker/brokers/binance/venues.py b/piker/brokers/binance/venues.py index fb6e7f4f..d3627b8a 100644 --- a/piker/brokers/binance/venues.py +++ b/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] diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index dee510f8..60c12639 100644 --- a/piker/data/_web_bs.py +++ b/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, ) ) - await nobsws._connected.wait() + 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: diff --git a/tests/test_binance.py b/tests/test_binance.py new file mode 100644 index 00000000..31370ee8 --- /dev/null +++ b/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)