Skip to content

Commit ec7b859

Browse files
martinkersnermartinkersner
authored
feat: async WebSocket client for all data types (on the generated WS surface) (#168)
* feat: async WebSocket ticker client (pilot) on generated WS surface 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. * fix: filter empty-result subscribe acks in WS client 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. * feat: generalize WS client to all data types; exclude orderbook 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.
1 parent 98fd6c2 commit ec7b859

7 files changed

Lines changed: 855 additions & 3 deletions

File tree

‎datamaxi/_ws_endpoints.py‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
"""
2+
Auto-generated WebSocket channel registry from the datamaxi-backend WS surface.
3+
DO NOT EDIT — regenerate with: make ws-python
4+
5+
Source: pkg/apiws/app.go route table + protobuf/*.proto + Go view structs.
6+
The subscribe `param` formats come from overrides/ws_channels.json (the one
7+
piece not derivable from those sources — see parse_ws.py).
8+
"""
9+
10+
WS_BASE_PATH = "/ws/v1"
11+
WS_AUTH_HEADER = "X-DTMX-APIKEY"
12+
13+
WS_CHANNELS = {
14+
"/announcement/listing": {
15+
"plan": "pro_plus",
16+
"market": None,
17+
"message": "ListingMessage",
18+
"param": None,
19+
"subscribe": True,
20+
"unsubscribe": True,
21+
},
22+
"/announcement/listing/internal": {
23+
"plan": "pro_plus",
24+
"market": None,
25+
"message": "InternalListingMessage",
26+
"param": None,
27+
"subscribe": True,
28+
"unsubscribe": True,
29+
},
30+
"/forex": {
31+
"plan": "basic",
32+
"market": None,
33+
"message": "ForexMessage",
34+
"param": "SYMBOL",
35+
"subscribe": True,
36+
"unsubscribe": True,
37+
"param_note": "fx pair e.g. USD-KRW, no exchange",
38+
},
39+
"/funding-rate": {
40+
"plan": "basic",
41+
"market": None,
42+
"message": "FundingRateMessage",
43+
"param": "SYMBOL@exchange",
44+
"subscribe": True,
45+
"unsubscribe": True,
46+
"param_note": "UPPER symbol + lower exchange; verified live (handler.go doc comment is wrong — Bisonai/datamaxi-backend#7926)",
47+
},
48+
"/liquidation": {
49+
"plan": "basic",
50+
"market": None,
51+
"message": "LiquidationMessage",
52+
"param": "SYMBOL@exchange",
53+
"subscribe": True,
54+
"unsubscribe": False,
55+
},
56+
"/liquidation/feed": {
57+
"plan": "basic",
58+
"market": None,
59+
"message": "LiquidationMessage",
60+
"param": None,
61+
"subscribe": False,
62+
"unsubscribe": False,
63+
},
64+
"/open-interest": {
65+
"plan": "basic",
66+
"market": None,
67+
"message": "OpenInterestMessage",
68+
"param": "SYMBOL@exchange",
69+
"subscribe": True,
70+
"unsubscribe": False,
71+
},
72+
"/premium": {
73+
"plan": "basic",
74+
"market": None,
75+
"message": "PremiumMessage",
76+
"param": "src:tgt:tokenId:srcQuote:tgtQuote:srcMkt:tgtMkt",
77+
"subscribe": True,
78+
"unsubscribe": True,
79+
},
80+
"/ticker/futures": {
81+
"plan": "basic",
82+
"market": "futures",
83+
"message": "TickerMessage",
84+
"param": "SYMBOL@exchange[@currency@conversionBase]",
85+
"subscribe": True,
86+
"unsubscribe": True,
87+
},
88+
"/ticker/spot": {
89+
"plan": "basic",
90+
"market": "spot",
91+
"message": "TickerMessage",
92+
"param": "SYMBOL@exchange[@currency@conversionBase]",
93+
"subscribe": True,
94+
"unsubscribe": True,
95+
},
96+
}

‎datamaxi/_ws_models.py‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
"""
2+
Auto-generated WebSocket message models from the datamaxi-backend WS surface.
3+
DO NOT EDIT — regenerate with: make ws-python
4+
5+
Source: pkg/apiws/app.go route table + protobuf/*.proto + Go view structs.
6+
The subscribe `param` formats come from overrides/ws_channels.json (the one
7+
piece not derivable from those sources — see parse_ws.py).
8+
"""
9+
10+
from __future__ import annotations
11+
12+
from typing import Any, Dict, List, TypedDict # noqa: F401
13+
14+
15+
# source: gostruct:pkg/apiforex/app.go::Forex
16+
ForexMessage = TypedDict(
17+
"ForexMessage",
18+
{
19+
"s": str,
20+
"d": int,
21+
"r": float,
22+
},
23+
total=False,
24+
)
25+
26+
27+
# source: gostruct:pkg/apifundingrate/types.go::Snapshot
28+
FundingRateMessage = TypedDict(
29+
"FundingRateMessage",
30+
{
31+
"f": float,
32+
"i": int,
33+
"e": str,
34+
"id": str,
35+
"s": str,
36+
"b": str,
37+
"q": str,
38+
"d": int,
39+
"p": int,
40+
},
41+
total=False,
42+
)
43+
44+
45+
# source: gostruct:pkg/apiannouncement/app.go::InternalListing
46+
InternalListingMessage = TypedDict(
47+
"InternalListingMessage",
48+
{
49+
"s": str,
50+
"e": str,
51+
"b": str,
52+
"t": str,
53+
"u": str,
54+
"d": int,
55+
},
56+
total=False,
57+
)
58+
59+
60+
# source: proto:protobuf/liquidation.proto::liquidation
61+
LiquidationMessage = TypedDict(
62+
"LiquidationMessage",
63+
{
64+
"id": str,
65+
"e": str,
66+
"d": str,
67+
"s": str,
68+
"b": str,
69+
"q": str,
70+
"sd": str,
71+
"p": float,
72+
"pusd": float,
73+
"pfiat": float,
74+
"v": float,
75+
"vusd": float,
76+
"vfiat": float,
77+
"pt": Dict[str, Any],
78+
"pa": str,
79+
"src": Any,
80+
},
81+
total=False,
82+
)
83+
84+
85+
# source: gostruct:pkg/apiannouncement/app.go::Listing
86+
ListingMessage = TypedDict(
87+
"ListingMessage",
88+
{
89+
"e": str,
90+
"b": str,
91+
"q": str,
92+
"u": str,
93+
"d": int,
94+
},
95+
total=False,
96+
)
97+
98+
99+
# source: proto:protobuf/open-interest.proto::open_interest
100+
OpenInterestMessage = TypedDict(
101+
"OpenInterestMessage",
102+
{
103+
"id": str,
104+
"e": str,
105+
"d": str,
106+
"s": str,
107+
"b": str,
108+
"q": str,
109+
"oi": float,
110+
"oiusd": float,
111+
"oifiat": float,
112+
"pt": Dict[str, Any],
113+
"pa": str,
114+
},
115+
total=False,
116+
)
117+
118+
119+
# source: gostruct:pkg/apipremium/handlers/dataapipremiumws/handler.go::PremiumOut
120+
PremiumMessage = TypedDict(
121+
"PremiumMessage",
122+
{
123+
"key": str,
124+
"source_exchange": str,
125+
"target_exchange": str,
126+
"token_id": str,
127+
"source_base": str,
128+
"source_quote": str,
129+
"target_quote": str,
130+
"source_market": str,
131+
"target_market": str,
132+
"premium": float,
133+
"source_price": float,
134+
"target_price": float,
135+
"timestamp": int,
136+
},
137+
total=False,
138+
)
139+
140+
141+
# source: gostruct:pkg/apiticker/app.go::View
142+
TickerMessage = TypedDict(
143+
"TickerMessage",
144+
{
145+
"p": float,
146+
"v": float,
147+
"p24h": float,
148+
"pc": float,
149+
"hb": float,
150+
"la": float,
151+
"ud": float,
152+
"ld": float,
153+
"e": str,
154+
"s": str,
155+
"b": str,
156+
"q": str,
157+
"d": int,
158+
"m": str,
159+
},
160+
total=False,
161+
)

0 commit comments

Comments
 (0)