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 = "b")]
254    pub bids: Vec<[String; 2]>,
255    #[serde(rename = "a")]
256    pub asks: Vec<[String; 2]>,
257}
258
259// ---------------------------------------------------------------------------
260// WebSocketMessage enum — manual Deserialize impl
261// ---------------------------------------------------------------------------
262
263/// Typed WebSocket message.
264///
265/// Deserialization dispatches on the `e` (event_type) field for object messages.
266/// Array JSON (all-symbol streams such as `!markPrice@arr`) is detected by the
267/// top-level value being a JSON array and dispatched on the first element's `e`.
268///
269/// Special cases without an `"e"` field:
270/// - `PartialDepth` — detected by the presence of `"lastUpdateId"`.
271/// - `BookTicker` — detected by `"u"` key with no `"e"` field.
272///
273/// Any unrecognised or malformed message is mapped to `Unknown(serde_json::Value)`
274/// and a `tracing::warn!` is emitted — the SDK never panics on unknown events.
275#[derive(Debug)]
276pub enum WebSocketMessage {
277    /// Single aggregate trade event (`"e": "aggTrade"`).
278    AggTrade(AggTradeEvent),
279    /// Single mark price update (`"e": "markPriceUpdate"`).
280    MarkPrice(MarkPriceEvent),
281    /// All-symbol mark price array (`!markPrice@arr`).
282    AllMarkPrices(Vec<MarkPriceEvent>),
283    /// Kline/candlestick event (`"e": "kline"`).
284    Kline(KlineEvent),
285    /// 24-hour mini ticker for a single symbol.
286    MiniTicker(MiniTickerEvent),
287    /// All-symbol 24-hour mini tickers (`!miniTicker@arr`).
288    AllMiniTickers(Vec<MiniTickerEvent>),
289    /// Full 24-hour ticker for a single symbol.
290    Ticker(TickerEvent),
291    /// All-symbol full 24-hour tickers.
292    AllTickers(Vec<TickerEvent>),
293    /// Best bid/ask book ticker (detected by `"u"` key without `"e"`).
294    BookTicker(BookTickerEvent),
295    /// Liquidation force order event (`"e": "forceOrder"`).
296    ForceOrder(ForceOrderEvent),
297    /// All liquidation force orders.
298    AllForceOrders(Vec<ForceOrderEvent>),
299    /// Partial depth snapshot (detected by `"lastUpdateId"` key).
300    PartialDepth(PartialDepthEvent),
301    /// Differential depth update (`"e": "depthUpdate"`).
302    DiffDepth(DiffDepthEvent),
303    /// Unknown or unrecognised message — captured as raw JSON for forward compatibility.
304    Unknown(serde_json::Value),
305}
306
307// US-011: manual Deserialize for WebSocketMessage — dispatches on "e" field.
308impl<'de> serde::Deserialize<'de> for WebSocketMessage {
309    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
310        let v = serde_json::Value::deserialize(deserializer)?;
311
312        // ---------------------------------------------------------------
313        // Array = all-symbol stream (e.g. !markPrice@arr, !miniTicker@arr)
314        // ---------------------------------------------------------------
315        if let Some(arr) = v.as_array() {
316            if arr.is_empty() {
317                tracing::warn!("Received empty WebSocket array message");
318                return Ok(WebSocketMessage::Unknown(v));
319            }
320            let first = &arr[0];
321            match first.get("e").and_then(|e| e.as_str()).unwrap_or("") {
322                "markPriceUpdate" => {
323                    match serde_json::from_value::<Vec<MarkPriceEvent>>(v.clone()) {
324                        Ok(events) => return Ok(WebSocketMessage::AllMarkPrices(events)),
325                        Err(err) => {
326                            tracing::warn!(error = ?err, "Failed to parse AllMarkPrices array");
327                            return Ok(WebSocketMessage::Unknown(v));
328                        }
329                    }
330                }
331                "24hrMiniTicker" => {
332                    match serde_json::from_value::<Vec<MiniTickerEvent>>(v.clone()) {
333                        Ok(events) => return Ok(WebSocketMessage::AllMiniTickers(events)),
334                        Err(err) => {
335                            tracing::warn!(error = ?err, "Failed to parse AllMiniTickers array");
336                            return Ok(WebSocketMessage::Unknown(v));
337                        }
338                    }
339                }
340                "24hrTicker" => {
341                    match serde_json::from_value::<Vec<TickerEvent>>(v.clone()) {
342                        Ok(events) => return Ok(WebSocketMessage::AllTickers(events)),
343                        Err(err) => {
344                            tracing::warn!(error = ?err, "Failed to parse AllTickers array");
345                            return Ok(WebSocketMessage::Unknown(v));
346                        }
347                    }
348                }
349                "forceOrder" => {
350                    match serde_json::from_value::<Vec<ForceOrderEvent>>(v.clone()) {
351                        Ok(events) => return Ok(WebSocketMessage::AllForceOrders(events)),
352                        Err(err) => {
353                            tracing::warn!(error = ?err, "Failed to parse AllForceOrders array");
354                            return Ok(WebSocketMessage::Unknown(v));
355                        }
356                    }
357                }
358                other => {
359                    tracing::warn!(event_type = other, "Unknown WebSocket array event type");
360                    return Ok(WebSocketMessage::Unknown(v));
361                }
362            }
363        }
364
365        // ---------------------------------------------------------------
366        // Object messages — check for special no-"e" cases first
367        // ---------------------------------------------------------------
368
369        // PartialDepth: has "lastUpdateId" but no "e" field.
370        if v.get("e").is_none() && v.get("lastUpdateId").is_some() {
371            match serde_json::from_value::<PartialDepthEvent>(v.clone()) {
372                Ok(event) => return Ok(WebSocketMessage::PartialDepth(event)),
373                Err(err) => {
374                    tracing::warn!(error = ?err, "Failed to parse PartialDepth event");
375                    return Ok(WebSocketMessage::Unknown(v));
376                }
377            }
378        }
379
380        // BookTicker: has "u" (update_id) but no "e" field.
381        if v.get("e").is_none() && v.get("u").is_some() {
382            match serde_json::from_value::<BookTickerEvent>(v.clone()) {
383                Ok(event) => return Ok(WebSocketMessage::BookTicker(event)),
384                Err(err) => {
385                    tracing::warn!(error = ?err, "Failed to parse BookTicker event");
386                    return Ok(WebSocketMessage::Unknown(v));
387                }
388            }
389        }
390
391        // ---------------------------------------------------------------
392        // Dispatch by "e" field value
393        // ---------------------------------------------------------------
394        let event_type = v.get("e").and_then(|e| e.as_str()).unwrap_or("");
395
396        /// Helper: deserialize or fall back to Unknown with a warning.
397        macro_rules! parse_event {
398            ($variant:ident, $type:ty, $label:literal) => {
399                match serde_json::from_value::<$type>(v.clone()) {
400                    Ok(event) => Ok(WebSocketMessage::$variant(event)),
401                    Err(err) => {
402                        tracing::warn!(
403                            error = ?err,
404                            event_type = $label,
405                            "Failed to parse known WebSocket event; falling back to Unknown"
406                        );
407                        Ok(WebSocketMessage::Unknown(v))
408                    }
409                }
410            };
411        }
412
413        match event_type {
414            "aggTrade" => parse_event!(AggTrade, AggTradeEvent, "aggTrade"),
415            "markPriceUpdate" => parse_event!(MarkPrice, MarkPriceEvent, "markPriceUpdate"),
416            "kline" => parse_event!(Kline, KlineEvent, "kline"),
417            "24hrMiniTicker" => parse_event!(MiniTicker, MiniTickerEvent, "24hrMiniTicker"),
418            "24hrTicker" => parse_event!(Ticker, TickerEvent, "24hrTicker"),
419            "forceOrder" => parse_event!(ForceOrder, ForceOrderEvent, "forceOrder"),
420            "depthUpdate" => parse_event!(DiffDepth, DiffDepthEvent, "depthUpdate"),
421            other => {
422                tracing::warn!(event_type = other, "Unknown WebSocket message event type");
423                Ok(WebSocketMessage::Unknown(v))
424            }
425        }
426    }
427}
428
429// ---------------------------------------------------------------------------
430// Unit tests
431// ---------------------------------------------------------------------------
432
433#[cfg(test)]
434mod tests {
435    use super::*;
436
437    #[test]
438    fn agg_trade_deserializes_correctly() {
439        let json = r#"{
440            "e": "aggTrade",
441            "E": 1625184000000,
442            "s": "BTCUSDT",
443            "a": 12345,
444            "p": "30000.00",
445            "q": "0.001",
446            "f": 100,
447            "l": 105,
448            "T": 1625184000000,
449            "m": false
450        }"#;
451        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
452        assert!(matches!(msg, WebSocketMessage::AggTrade(ref e) if e.symbol == "BTCUSDT"));
453        if let WebSocketMessage::AggTrade(event) = msg {
454            assert_eq!(event.price, "30000.00");
455            assert_eq!(event.quantity, "0.001");
456            assert!(!event.is_buyer_maker);
457        }
458    }
459
460    #[test]
461    fn unknown_event_type_maps_to_unknown_variant() {
462        let json = r#"{"e":"futureData","x":"someValue"}"#;
463        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
464        assert!(matches!(msg, WebSocketMessage::Unknown(_)));
465    }
466
467    #[test]
468    fn kline_event_deserializes_correctly() {
469        let json = r#"{
470            "e": "kline",
471            "E": 1625184000000,
472            "s": "BTCUSDT",
473            "k": {
474                "t": 1625184000000, "T": 1625184059999, "s": "BTCUSDT",
475                "i": "1m", "f": 1, "L": 10,
476                "o": "30000", "c": "30100", "h": "30200", "l": "29900",
477                "v": "100", "n": 10, "x": false,
478                "q": "3010000", "V": "50", "Q": "1505000"
479            }
480        }"#;
481        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
482        assert!(matches!(msg, WebSocketMessage::Kline(ref e) if e.kline.interval == "1m"));
483        if let WebSocketMessage::Kline(event) = msg {
484            assert!(!event.kline.is_closed);
485            assert_eq!(event.kline.open, "30000");
486            assert_eq!(event.kline.close, "30100");
487        }
488    }
489
490    #[test]
491    fn partial_depth_deserializes_correctly() {
492        let json = r#"{"lastUpdateId":123456,"bids":[["30000","1.5"],["29999","2.0"]],"asks":[["30001","0.5"]]}"#;
493        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
494        assert!(matches!(msg, WebSocketMessage::PartialDepth(_)));
495        if let WebSocketMessage::PartialDepth(event) = msg {
496            assert_eq!(event.bids[0][0], "30000");
497            assert_eq!(event.bids[0][1], "1.5");
498        }
499    }
500
501    #[test]
502    fn all_mark_prices_array_deserializes_correctly() {
503        let json = r#"[{"e":"markPriceUpdate","E":1625184000000,"s":"BTCUSDT","p":"30000","i":"30001","P":"29999","r":"0.0001","T":1625198400000}]"#;
504        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
505        assert!(matches!(msg, WebSocketMessage::AllMarkPrices(_)));
506        if let WebSocketMessage::AllMarkPrices(events) = msg {
507            assert_eq!(events.len(), 1);
508            assert_eq!(events[0].symbol, "BTCUSDT");
509        }
510    }
511
512    #[test]
513    fn mark_price_single_deserializes_correctly() {
514        let json = r#"{
515            "e": "markPriceUpdate",
516            "E": 1625184000000,
517            "s": "BTCUSDT",
518            "p": "30000.00",
519            "i": "30001.00",
520            "P": "29999.00",
521            "r": "0.0001",
522            "R": "0.000050",
523            "T": 1625198400000
524        }"#;
525        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
526        assert!(matches!(msg, WebSocketMessage::MarkPrice(_)));
527        if let WebSocketMessage::MarkPrice(event) = msg {
528            assert_eq!(event.symbol, "BTCUSDT");
529            assert_eq!(event.mark_price, "30000.00");
530            assert_eq!(event.last_funding_rate, "0.0001");
531            assert_eq!(event.interest_rate, "0.000050");
532            assert_eq!(event.next_funding_time, 1625198400000);
533        }
534    }
535
536    #[test]
537    fn book_ticker_deserializes_correctly() {
538        let json = r#"{"u":12345,"s":"BTCUSDT","b":"30000","B":"1.5","a":"30001","A":"0.5"}"#;
539        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
540        assert!(matches!(msg, WebSocketMessage::BookTicker(ref e) if e.symbol == "BTCUSDT"));
541        if let WebSocketMessage::BookTicker(event) = msg {
542            assert_eq!(event.best_bid_price, "30000");
543            assert_eq!(event.best_ask_price, "30001");
544        }
545    }
546
547    #[test]
548    fn diff_depth_deserializes_correctly() {
549        let json = r#"{
550            "e": "depthUpdate",
551            "E": 1625184000000,
552            "s": "BTCUSDT",
553            "U": 100,
554            "u": 200,
555            "b": [["29999","1.0"],["29998","2.0"]],
556            "a": [["30001","0.5"]]
557        }"#;
558        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
559        assert!(matches!(msg, WebSocketMessage::DiffDepth(ref e) if e.symbol == "BTCUSDT"));
560        if let WebSocketMessage::DiffDepth(event) = msg {
561            assert_eq!(event.first_update_id, 100);
562            assert_eq!(event.final_update_id, 200);
563            assert_eq!(event.bids[0][0], "29999");
564        }
565    }
566
567    #[test]
568    fn mini_ticker_deserializes_correctly() {
569        let json = r#"{
570            "e": "24hrMiniTicker",
571            "E": 1625184000000,
572            "s": "ETHUSDT",
573            "c": "2000.00",
574            "o": "1950.00",
575            "h": "2050.00",
576            "l": "1900.00",
577            "v": "5000",
578            "q": "9750000"
579        }"#;
580        let msg: WebSocketMessage = serde_json::from_str(json).unwrap();
581        assert!(matches!(msg, WebSocketMessage::MiniTicker(ref e) if e.symbol == "ETHUSDT"));
582        if let WebSocketMessage::MiniTicker(event) = msg {
583            assert_eq!(event.close_price, "2000.00");
584            assert_eq!(event.volume, "5000");
585        }
586    }
587}
asterdex_sdk/futures/models/websocket.rs

asterdex_sdk/futures/models/
websocket.rs
