asterdex_sdk/futures/models/
websocket.rs

1// US-011: WebSocket message types — complete struct definitions and manual Deserialize impl.
2// The outer combined-stream wrapper { "stream": "...", "data": {...} } is already unwrapped
3// by the Wave 5 WebSocketClient before these types are constructed.
4
5use serde::Deserialize;
6
7// ---------------------------------------------------------------------------
8// Event structs
9// ---------------------------------------------------------------------------
10
11/// Aggregate trade event — dispatched when `"e": "aggTrade"`.
12#[derive(Debug, Deserialize)]
13pub struct AggTradeEvent {
14    #[serde(rename = "e")]
15    pub event_type: String,
16    #[serde(rename = "E")]
17    pub event_time: u64,
18    #[serde(rename = "s")]
19    pub symbol: String,
20    #[serde(rename = "a")]
21    pub agg_trade_id: u64,
22    #[serde(rename = "p")]
23    pub price: String,
24    #[serde(rename = "q")]
25    pub quantity: String,
26    #[serde(rename = "f")]
27    pub first_trade_id: u64,
28    #[serde(rename = "l")]
29    pub last_trade_id: u64,
30    #[serde(rename = "T")]
31    pub trade_time: u64,
32    #[serde(rename = "m")]
33    pub is_buyer_maker: bool,
34}
35
36/// Mark price event — dispatched when `"e": "markPriceUpdate"`.
37#[derive(Debug, Deserialize)]
38pub struct MarkPriceEvent {
39    #[serde(rename = "e")]
40    pub event_type: String,
41    #[serde(rename = "E")]
42    pub event_time: u64,
43    #[serde(rename = "s")]
44    pub symbol: String,
45    #[serde(rename = "p")]
46    pub mark_price: String,
47    #[serde(rename = "i")]
48    pub index_price: String,
49    #[serde(rename = "P")]
50    pub estimated_settle_price: String,
51    #[serde(rename = "r")]
52    pub last_funding_rate: String,
53    #[serde(rename = "R", default)]
54    pub interest_rate: String,
55    #[serde(rename = "T")]
56    pub next_funding_time: u64,
57}
58
59/// Kline/candlestick event — dispatched when `"e": "kline"`.
60#[derive(Debug, Deserialize)]
61pub struct KlineEvent {
62    #[serde(rename = "e")]
63    pub event_type: String,
64    #[serde(rename = "E")]
65    pub event_time: u64,
66    #[serde(rename = "s")]
67    pub symbol: String,
68    #[serde(rename = "k")]
69    pub kline: KlineData,
70}
71
72/// Inner kline data nested under the `k` field of a `KlineEvent`.
73#[derive(Debug, Deserialize)]
74pub struct KlineData {
75    #[serde(rename = "t")]
76    pub start_time: u64,
77    #[serde(rename = "T")]
78    pub close_time: u64,
79    #[serde(rename = "s")]
80    pub symbol: String,
81    /// Interval string e.g. `"1m"`, `"1h"`.
82    #[serde(rename = "i")]
83    pub interval: String,
84    #[serde(rename = "f")]
85    pub first_trade_id: i64,
86    #[serde(rename = "L")]
87    pub last_trade_id: i64,
88    #[serde(rename = "o")]
89    pub open: String,
90    #[serde(rename = "c")]
91    pub close: String,
92    #[serde(rename = "h")]
93    pub high: String,
94    #[serde(rename = "l")]
95    pub low: String,
96    #[serde(rename = "v")]
97    pub volume: String,
98    #[serde(rename = "n")]
99    pub trade_count: u64,
100    #[serde(rename = "x")]
101    pub is_closed: bool,
102    #[serde(rename = "q")]
103    pub quote_volume: String,
104    #[serde(rename = "V")]
105    pub taker_buy_volume: String,
106    #[serde(rename = "Q")]
107    pub taker_buy_quote_volume: String,
108}
109
110/// 24-hour mini ticker event — dispatched when `"e": "24hrMiniTicker"`.
111#[derive(Debug, Deserialize)]
112pub struct MiniTickerEvent {
113    #[serde(rename = "e")]
114    pub event_type: String,
115    #[serde(rename = "E")]
116    pub event_time: u64,
117    #[serde(rename = "s")]
118    pub symbol: String,
119    #[serde(rename = "c")]
120    pub close_price: String,
121    #[serde(rename = "o")]
122    pub open_price: String,
123    #[serde(rename = "h")]
124    pub high_price: String,
125    #[serde(rename = "l")]
126    pub low_price: String,
127    #[serde(rename = "v")]
128    pub volume: String,
129    #[serde(rename = "q")]
130    pub quote_volume: String,
131}
132
133/// Full 24-hour ticker event — dispatched when `"e": "24hrTicker"`.
134#[derive(Debug, Deserialize)]
135pub struct TickerEvent {
136    #[serde(rename = "e")]
137    pub event_type: String,
138    #[serde(rename = "E")]
139    pub event_time: u64,
140    #[serde(rename = "s")]
141    pub symbol: String,
142    #[serde(rename = "p")]
143    pub price_change: String,
144    #[serde(rename = "P")]
145    pub price_change_percent: String,
146    #[serde(rename = "w")]
147    pub weighted_avg_price: String,
148    #[serde(rename = "c")]
149    pub last_price: String,
150    #[serde(rename = "Q")]
151    pub last_qty: String,
152    #[serde(rename = "o")]
153    pub open_price: String,
154    #[serde(rename = "h")]
155    pub high_price: String,
156    #[serde(rename = "l")]
157    pub low_price: String,
158    #[serde(rename = "v")]
159    pub volume: String,
160    #[serde(rename = "q")]
161    pub quote_volume: String,
162    #[serde(rename = "O")]
163    pub open_time: u64,
164    #[serde(rename = "C")]
165    pub close_time: u64,
166    #[serde(rename = "F")]
167    pub first_trade_id: i64,
168    #[serde(rename = "L")]
169    pub last_trade_id: i64,
170    #[serde(rename = "n")]
171    pub trade_count: u64,
172}
173
174/// Best-bid/ask book ticker event — no `"e"` field; detected by `"u"` key presence without `"e"`.
175#[derive(Debug, Deserialize)]
176pub struct BookTickerEvent {
177    #[serde(rename = "u")]
178    pub update_id: u64,
179    #[serde(rename = "s")]
180    pub symbol: String,
181    #[serde(rename = "b")]
182    pub best_bid_price: String,
183    #[serde(rename = "B")]
184    pub best_bid_qty: String,
185    #[serde(rename = "a")]
186    pub best_ask_price: String,
187    #[serde(rename = "A")]
188    pub best_ask_qty: String,
189}
190
191/// Liquidation force order event — dispatched when `"e": "forceOrder"`.
192#[derive(Debug, Deserialize)]
193pub struct ForceOrderEvent {
194    #[serde(rename = "e")]
195    pub event_type: String,
196    #[serde(rename = "E")]
197    pub event_time: u64,
198    #[serde(rename = "o")]
199    pub order: ForceOrderData,
200}
201
202/// Inner order data for a force-order event.
203#[derive(Debug, Deserialize)]
204pub struct ForceOrderData {
205    #[serde(rename = "s")]
206    pub symbol: String,
207    #[serde(rename = "S")]
208    pub side: String,
209    #[serde(rename = "o")]
210    pub order_type: String,
211    #[serde(rename = "f")]
212    pub time_in_force: String,
213    #[serde(rename = "q")]
214    pub quantity: String,
215    #[serde(rename = "p")]
216    pub price: String,
217    /// May be `"0"` for orders not yet filled — kept as String (BR-010).
218    #[serde(rename = "ap")]
219    pub average_price: String,
220    #[serde(rename = "X")]
221    pub order_status: String,
222    #[serde(rename = "l")]
223    pub last_filled_qty: String,
224    #[serde(rename = "z")]
225    pub accumulated_qty: String,
226    #[serde(rename = "T")]
227    pub trade_time: u64,
228}
229
230/// Partial depth snapshot event — no `"e"` field; detected by `"lastUpdateId"` key.
231/// Dispatched for `<symbol>@depth5`, `@depth10`, `@depth20` streams.
232#[derive(Debug, Deserialize)]
233pub struct PartialDepthEvent {
234    #[serde(rename = "lastUpdateId")]
235    pub last_update_id: u64,
236    pub bids: Vec<[String; 2]>,
237    pub asks: Vec<[String; 2]>,
238}
239
240/// Differential depth stream event — dispatched when `"e": "depthUpdate"`.
241#[derive(Debug, Deserialize)]
242pub struct DiffDepthEvent {
243    #[serde(rename = "e")]
244    pub event_type: String,
245    #[serde(rename = "E")]
246    pub event_time: u64,
247    #[serde(rename = "s")]
248    pub symbol: String,
249    #[serde(rename = "U")]
250    pub first_update_id: u64,
251    #[serde(rename = "u")]
252    pub final_update_id: u64,
253    #[serde(rename = "pu")]
254    pub prev_final_update_id: u64,
255    #[serde(rename = "b")]
256    pub bids: Vec<[String; 2]>,
257    #[serde(rename = "a")]
258    pub asks: Vec<[String; 2]>,
259}
260
261// ---------------------------------------------------------------------------
262// WebSocketMessage enum — manual Deserialize impl
263// ---------------------------------------------------------------------------
264
265/// Typed WebSocket message.
266///
267/// Deserialization dispatches on the `e` (event_type) field for object messages.
268/// Array JSON (all-symbol streams such as `!markPrice@arr`) is detected by the
269/// top-level value being a JSON array and dispatched on the first element's `e`.
270///
271/// Special cases without an `"e"` field:
272/// - `PartialDepth` — detected by the presence of `"lastUpdateId"`.
273/// - `BookTicker` — detected by `"u"` key with no `"e"` field.
274///
275/// Any unrecognised or malformed message is mapped to `Unknown(serde_json::Value)`
276/// and a `tracing::warn!` is emitted — the SDK never panics on unknown events.
277#[derive(Debug)]
278pub enum WebSocketMessage {
279    /// Single aggregate trade event (`"e": "aggTrade"`).
280    AggTrade(AggTradeEvent),
281    /// Single mark price update (`"e": "markPriceUpdate"`).
282    MarkPrice(MarkPriceEvent),
283    /// All-symbol mark price array (`!markPrice@arr`).
284    AllMarkPrices(Vec<MarkPriceEvent>),
285    /// Kline/candlestick event (`"e": "kline"`).
286    Kline(KlineEvent),
287    /// 24-hour mini ticker for a single symbol.
288    MiniTicker(MiniTickerEvent),
289    /// All-symbol 24-hour mini tickers (`!miniTicker@arr`).
290    AllMiniTickers(Vec<MiniTickerEvent>),
291    /// Full 24-hour ticker for a single symbol.
292    Ticker(TickerEvent),
293    /// All-symbol full 24-hour tickers.
294    AllTickers(Vec<TickerEvent>),
295    /// Best bid/ask book ticker (detected by `"u"` key without `"e"`).
296    BookTicker(BookTickerEvent),
297    /// Liquidation force order event (`"e": "forceOrder"`).
298    ForceOrder(ForceOrderEvent),
299    /// All liquidation force orders.
300    AllForceOrders(Vec<ForceOrderEvent>),
301    /// Partial depth snapshot (detected by `"lastUpdateId"` key).
302    PartialDepth(PartialDepthEvent),
303    /// Differential depth update (`"e": "depthUpdate"`).
304    DiffDepth(DiffDepthEvent),
305    /// Unknown or unrecognised message — captured as raw JSON for forward compatibility.
306    Unknown(serde_json::Value),
307}
308
309// US-011: manual Deserialize for WebSocketMessage — dispatches on "e" field.
310impl<'de> serde::Deserialize<'de> for WebSocketMessage {
311    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
312        let v = serde_json::Value::deserialize(deserializer)?;
313
314        // ---------------------------------------------------------------
315        // Array = all-symbol stream (e.g. !markPrice@arr, !miniTicker@arr)
316        // ---------------------------------------------------------------
317        if let Some(arr) = v.as_array() {
318            if arr.is_empty() {
319                tracing::warn!("Received empty WebSocket array message");
320                return Ok(WebSocketMessage::Unknown(v));
321            }
322            let first = &arr[0];
323            match first.get("e").and_then(|e| e.as_str()).unwrap_or("") {
324                "markPriceUpdate" => {
325                    match serde_json::from_value::<Vec<MarkPriceEvent>>(v.clone()) {
326                        Ok(events) => return Ok(WebSocketMessage::AllMarkPrices(events)),
327                        Err(err) => {
328                            tracing::warn!(error = ?err, "Failed to parse AllMarkPrices array");
329                            return Ok(WebSocketMessage::Unknown(v));
330                        }
331                    }
332                }
333                "24hrMiniTicker" => {
334                    match serde_json::from_value::<Vec<MiniTickerEvent>>(v.clone()) {
335                        Ok(events) => return Ok(WebSocketMessage::AllMiniTickers(events)),
336                        Err(err) => {
337                            tracing::warn!(error = ?err, "Failed to parse AllMiniTickers array");
338                            return Ok(WebSocketMessage::Unknown(v));
339                        }
340                    }
341                }
342                "24hrTicker" => {
343                    match serde_json::from_value::<Vec<TickerEvent>>(v.clone()) {
344                        Ok(events) => return Ok(WebSocketMessage::AllTickers(events)),
345                        Err(err) => {
346                            tracing::warn!(error = ?err, "Failed to parse AllTickers array");
347                            return Ok(WebSocketMessage::Unknown(v));
348                        }
349                    }
350                }
351                "forceOrder" => {
352                    match serde_json::from_value::<Vec<ForceOrderEvent>>(v.clone()) {
353                        Ok(events) => return Ok(WebSocketMessage::AllForceOrders(events)),
354                        Err(err) => {
355                            tracing::warn!(error = ?err, "Failed to parse AllForceOrders array");
356                            return Ok(WebSocketMessage::Unknown(v));
357                        }
358                    }
359                }
360                other => {
361                    tracing::warn!(event_type = other, "Unknown WebSocket array event type");
362                    return Ok(WebSocketMessage::Unknown(v));
363                }
364            }
365        }
366
367        // ---------------------------------------------------------------
368        // Object messages — check for special no-"e" cases first
369        // ---------------------------------------------------------------
370
371        // PartialDepth: has "lastUpdateId" but no "e" field.
372        if v.get("e").is_none() && v.get("lastUpdateId").is_some() {
373            match serde_json::from_value::<PartialDepthEvent>(v.clone()) {
374                Ok(event) => return Ok(WebSocketMessage::PartialDepth(event)),
375                Err(err) => {
376                    tracing::warn!(error = ?err, "Failed to parse PartialDepth event");
377                    return Ok(WebSocketMessage::Unknown(v));
378                }
379            }
380        }
381
382        // BookTicker: has "u" (update_id) but no "e" field.
383        if v.get("e").is_none() && v.get("u").is_some() {
384            match serde_json::from_value::<BookTickerEvent>(v.clone()) {
385                Ok(event) => return Ok(WebSocketMessage::BookTicker(event)),
386                Err(err) => {
387                    tracing::warn!(error = ?err, "Failed to parse BookTicker event");
388                    return Ok(WebSocketMessage::Unknown(v));
389                }
390            }
391        }
392
393        // ---------------------------------------------------------------
394        // Dispatch by "e" field value
395        // ---------------------------------------------------------------
396        let event_type = v.get("e").and_then(|e| e.as_str()).unwrap_or("");
397
398        /// Helper: deserialize or fall back to Unknown with a warning.
399        macro_rules! parse_event {
400            ($variant:ident, $type:ty, $label:literal) => {
401                match serde_json::from_value::<$type>(v.clone()) {
402                    Ok(event) => Ok(WebSocketMessage::$variant(event)),
403                    Err(err) => {
404                        tracing::warn!(
405                            error = ?err,
406                            event_type = $label,
407                            "Failed to parse known WebSocket event; falling back to Unknown"
408                        );
409                        Ok(WebSocketMessage::Unknown(v))
410                    }
411                }
412            };
413        }
414
415        match event_type {
416            "aggTrade" => parse_event!(AggTrade, AggTradeEvent, "aggTrade"),
417            "markPriceUpdate" => parse_event!(MarkPrice, MarkPriceEvent, "markPriceUpdate"),
418            "kline" => parse_event!(Kline, KlineEvent, "kline"),
419            "24hrMiniTicker" => parse_event!(MiniTicker, MiniTickerEvent, "24hrMiniTicker"),
420            "24hrTicker" => parse_event!(Ticker, TickerEvent, "24hrTicker"),
421            "forceOrder" => parse_event!(ForceOrder, ForceOrderEvent, "forceOrder"),
422            "depthUpdate" => parse_event!(DiffDepth, DiffDepthEvent, "depthUpdate"),
423            other => {
424                tracing::warn!(event_type = other, "Unknown WebSocket message event type");
425                Ok(WebSocketMessage::Unknown(v))
426            }
427        }
428    }
429}
430
431// ---------------------------------------------------------------------------
432// Unit tests
433// ---------------------------------------------------------------------------
434
435#[cfg(test)]
436mod tests {
437    use super::*;
438
439    #[test]
440    fn agg_trade_deserializes_correctly() {
441        let json = r#"{
442            "e": "aggTrade",
443            "E": 1625184000000,
444            "s": "BTCUSDT",
445            "a": 12345,
446            "p": "30000.00",
447            "q": "0.001",
448            "f": 100,
449            "l": 105,
450            "T": 1625184000000,
451            "m": false
452        }"#;
453        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
454        assert!(matches!(msg, WebSocketMessage::AggTrade(ref e) if e.symbol == "BTCUSDT"));
455        if let WebSocketMessage::AggTrade(event) = msg {
456            assert_eq!(event.price, "30000.00");
457            assert_eq!(event.quantity, "0.001");
458            assert!(!event.is_buyer_maker);
459        }
460    }
461
462    #[test]
463    fn unknown_event_type_maps_to_unknown_variant() {
464        let json = r#"{"e":"futureData","x":"someValue"}"#;
465        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
466        assert!(matches!(msg, WebSocketMessage::Unknown(_)));
467    }
468
469    #[test]
470    fn kline_event_deserializes_correctly() {
471        let json = r#"{
472            "e": "kline",
473            "E": 1625184000000,
474            "s": "BTCUSDT",
475            "k": {
476                "t": 1625184000000, "T": 1625184059999, "s": "BTCUSDT",
477                "i": "1m", "f": 1, "L": 10,
478                "o": "30000", "c": "30100", "h": "30200", "l": "29900",
479                "v": "100", "n": 10, "x": false,
480                "q": "3010000", "V": "50", "Q": "1505000"
481            }
482        }"#;
483        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
484        assert!(matches!(msg, WebSocketMessage::Kline(ref e) if e.kline.interval == "1m"));
485        if let WebSocketMessage::Kline(event) = msg {
486            assert!(!event.kline.is_closed);
487            assert_eq!(event.kline.open, "30000");
488            assert_eq!(event.kline.close, "30100");
489        }
490    }
491
492    #[test]
493    fn partial_depth_deserializes_correctly() {
494        let json = r#"{"lastUpdateId":123456,"bids":[["30000","1.5"],["29999","2.0"]],"asks":[["30001","0.5"]]}"#;
495        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
496        assert!(matches!(msg, WebSocketMessage::PartialDepth(_)));
497        if let WebSocketMessage::PartialDepth(event) = msg {
498            assert_eq!(event.bids[0][0], "30000");
499            assert_eq!(event.bids[0][1], "1.5");
500        }
501    }
502
503    #[test]
504    fn all_mark_prices_array_deserializes_correctly() {
505        let json = r#"[{"e":"markPriceUpdate","E":1625184000000,"s":"BTCUSDT","p":"30000","i":"30001","P":"29999","r":"0.0001","T":1625198400000}]"#;
506        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
507        assert!(matches!(msg, WebSocketMessage::AllMarkPrices(_)));
508        if let WebSocketMessage::AllMarkPrices(events) = msg {
509            assert_eq!(events.len(), 1);
510            assert_eq!(events[0].symbol, "BTCUSDT");
511        }
512    }
513
514    #[test]
515    fn mark_price_single_deserializes_correctly() {
516        let json = r#"{
517            "e": "markPriceUpdate",
518            "E": 1625184000000,
519            "s": "BTCUSDT",
520            "p": "30000.00",
521            "i": "30001.00",
522            "P": "29999.00",
523            "r": "0.0001",
524            "R": "0.000050",
525            "T": 1625198400000
526        }"#;
527        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
528        assert!(matches!(msg, WebSocketMessage::MarkPrice(_)));
529        if let WebSocketMessage::MarkPrice(event) = msg {
530            assert_eq!(event.symbol, "BTCUSDT");
531            assert_eq!(event.mark_price, "30000.00");
532            assert_eq!(event.last_funding_rate, "0.0001");
533            assert_eq!(event.interest_rate, "0.000050");
534            assert_eq!(event.next_funding_time, 1625198400000);
535        }
536    }
537
538    #[test]
539    fn book_ticker_deserializes_correctly() {
540        let json = r#"{"u":12345,"s":"BTCUSDT","b":"30000","B":"1.5","a":"30001","A":"0.5"}"#;
541        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
542        assert!(matches!(msg, WebSocketMessage::BookTicker(ref e) if e.symbol == "BTCUSDT"));
543        if let WebSocketMessage::BookTicker(event) = msg {
544            assert_eq!(event.best_bid_price, "30000");
545            assert_eq!(event.best_ask_price, "30001");
546        }
547    }
548
549    #[test]
550    fn diff_depth_deserializes_correctly() {
551        let json = r#"{
552            "e": "depthUpdate",
553            "E": 1625184000000,
554            "s": "BTCUSDT",
555            "U": 100,
556            "u": 200,
557            "b": [["29999","1.0"],["29998","2.0"]],
558            "a": [["30001","0.5"]]
559        }"#;
560        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
561        assert!(matches!(msg, WebSocketMessage::DiffDepth(ref e) if e.symbol == "BTCUSDT"));
562        if let WebSocketMessage::DiffDepth(event) = msg {
563            assert_eq!(event.first_update_id, 100);
564            assert_eq!(event.final_update_id, 200);
565            assert_eq!(event.bids[0][0], "29999");
566        }
567    }
568
569    #[test]
570    fn mini_ticker_deserializes_correctly() {
571        let json = r#"{
572            "e": "24hrMiniTicker",
573            "E": 1625184000000,
574            "s": "ETHUSDT",
575            "c": "2000.00",
576            "o": "1950.00",
577            "h": "2050.00",
578            "l": "1900.00",
579            "v": "5000",
580            "q": "9750000"
581        }"#;
582        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
583        assert!(matches!(msg, WebSocketMessage::MiniTicker(ref e) if e.symbol == "ETHUSDT"));
584        if let WebSocketMessage::MiniTicker(event) = msg {
585            assert_eq!(event.close_price, "2000.00");
586            assert_eq!(event.volume, "5000");
587        }
588    }
589}
asterdex_sdk/futures/models/websocket.rs

asterdex_sdk/futures/models/
websocket.rs
