feat: async WebSocket client for all data types (on the generated WS surface)#168
Merged
Conversation
Validates the codegen WS pipeline end-to-end for Python. Consumes the
generated datamaxi/_ws_endpoints.py (WS_CHANNELS/WS_BASE_PATH/WS_AUTH_HEADER)
and datamaxi/_ws_models.py (TickerMessage) — both emitted by datamaxi-codegen
from the backend WS surface (route table + protobuf + Go view structs).
datamaxi/aio/ws.py: AsyncDatamaxiWS + AsyncWSConnection + TickerChannel on the
websockets lib (optional [ws] extra). One connection per channel path; SUBSCRIBE
/UNSUBSCRIBE/PING protocol; app-level PING keepalive (~30s vs the ~90s proxy
idle); reconnect-with-resubscribe; ack filtering; channel-stream routing
(caller filters by msg['s']/['e'] — the wire has no channel tag).
async with AsyncDatamaxiWS(api_key) as ws:
async for msg in ws.ticker.subscribe('BTC-USDT@binance', market='spot'):
...
Tests (tests/test_ws.py) run a real in-process websockets server: auth header,
path from the generated registry, subscribe protocol, streamed data, ack
filtering, multi-symbol multiplexing, keepalive PING. importorskip-guarded;
websockets added to requirements-test. Generated _ws_*.py excluded from
black/flake8 like _endpoints.py. import datamaxi does not load websockets.
Suite: 205 passed.
Live testing surfaced that when the accepted param list is empty (e.g. the
announcement channel), the server omits `result` and sends just {"id": N}.
The ack filter only matched {"result":..., "id":...}, so {"id": N} leaked
as a data message. Detect an ack as any dict whose keys are a subset of
{"result", "id"} (data payloads always carry other fields — and may
themselves include an "id" token, so presence of "id" alone isn't enough).
Verified live against the announcement channel.
Expand the async WS client from ticker-only to every DataMaxi+ WS data type, driven by the generated WS_CHANNELS registry. Accessors: ticker (market-keyed), forex, premium, funding_rate, open_interest, liquidation (subscribe), liquidation_feed (firehose), announcement / announcement_internal (Pro+). Generic channel types (Subscription / MarketSubscription / Feed) over the same AsyncWSConnection transport. Regenerate _ws_endpoints.py + _ws_models.py from codegen with orderbook excluded (unsupported product, backend removal tracked in Bisonai/datamaxi-backend#7927) -> 10 channels / 8 models. Tests: orderbook-excluded, every channel maps to a generated model + a client accessor, param_format from registry, plus forex-subscribe and feed-stream over the mock server. Live-verified through the client against prod (ticker, open_interest, liquidation_feed, premium). Suite: 211 passed.
martinkersner
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Async WebSocket client covering every DataMaxi+ WS data type, driven by the codegen-generated WS surface and validated against production.
datamaxi/_ws_endpoints.py(WS_CHANNELS/WS_BASE_PATH/WS_AUTH_HEADER) anddatamaxi/_ws_models.py(per-channelTypedDicts), both emitted by datamaxi-codegen #35. Excluded from black/flake8 like_endpoints.py.datamaxi/aio/ws.py—AsyncDatamaxiWSon thewebsocketslib (new optional[ws]extra), with generic channel types (Subscription/MarketSubscription/Feed) over oneAsyncWSConnectiontransport.SUBSCRIBE/UNSUBSCRIBE/PINGprotocol, app-level keepalive (~30s vs the ~90s proxy idle), reconnect-with-resubscribe, ack filtering.Accessors (all from the generated registry):
ticker(market-keyed),forex,premium,funding_rate,open_interest,liquidation,liquidation_feed,announcement/announcement_internal(Pro+). Orderbook is intentionally excluded (unsupported product — backend removal tracked in datamaxi-backend#7927).Routing: channel-stream —
subscribe()yields every message on the channel; callers filter bymsg["s"]/["e"](the wire tags messages by payload fields, not a channel id).Validated against production
Live-connected to
wss://api.datamaxiplus.comand confirmed:total=Falsesuperset, since protojson omits zero fields).str(e.g.open_interest.d) vs Go-json int64 →int(e.g.ticker.d).SYMBOL@exchange(the backend doc comment was wrong → #7926).{"id": N}(server omitsresult) and were leaking as data; now filtered (dict with keys ⊆{"result","id"}), re-verified live.Tests
tests/test_ws.pyruns a real in-processwebsocketsserver: auth header, path from the generated registry, subscribe protocol, streamed data, ack filtering (incl. empty-result), multi-symbol multiplexing, keepalive PING, forex-subscribe, feed-stream, orderbook-excluded, and every channel maps to a generated model + client accessor.importorskip-guarded;websocketsinrequirements-test.pytest -m "not integration": 211 passed, 11 skipped.import datamaxidoes not load websockets (extra stays optional).Scope / next
Depends on codegen #35 (the generated
_ws_*.py). Remaining un-live-tested: plain/liquidation(sporadic), announcement data (rare), and reconnect. Rust stays parked.Known failures
None.