polyoxide_binance/spot/ws/
event.rs

1//! Stream payloads, and the update that carries one.
2//!
3//! The socket uses one-letter keys and different fields from REST, so each
4//! stream has its own payload type, with long field names over the wire's keys.
5//! Symbols are `String`, as sent. The keys Binance documents as "Ignore", `M`
6//! on trades and aggregate trades and `B` on the kline, are not modelled.
7
8use std::fmt;
9
10use rust_decimal::Decimal;
11use serde::{Deserialize, Serialize, Serializer};
12use serde_json::Value;
13
14use crate::spot::{
15    types::{Interval, Level},
16    ws::{error::SpotWsError, stream::StreamName},
17};
18
19/// `<s>@aggTrade`.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21#[non_exhaustive]
22pub struct AggTradeEvent {
23    /// `aggTrade`.
24    #[serde(rename = "e")]
25    pub event_type: String,
26    /// When the event was produced.
27    #[serde(rename = "E")]
28    pub event_time: u64,
29    /// The symbol, as listed.
30    #[serde(rename = "s")]
31    pub symbol: String,
32    /// Aggregate trade id.
33    #[serde(rename = "a")]
34    pub id: u64,
35    /// Price.
36    #[serde(rename = "p", with = "rust_decimal::serde::str")]
37    pub price: Decimal,
38    /// Quantity.
39    #[serde(rename = "q", with = "rust_decimal::serde::str")]
40    pub quantity: Decimal,
41    /// First trade id merged.
42    #[serde(rename = "f")]
43    pub first_trade_id: u64,
44    /// Last trade id merged.
45    #[serde(rename = "l")]
46    pub last_trade_id: u64,
47    /// Trade time.
48    #[serde(rename = "T")]
49    pub trade_time: u64,
50    /// Whether the buyer was the maker, so the taker sold.
51    #[serde(rename = "m")]
52    pub is_buyer_maker: bool,
53}
54
55/// `<s>@trade`.
56#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
57#[non_exhaustive]
58pub struct TradeEvent {
59    /// `trade`.
60    #[serde(rename = "e")]
61    pub event_type: String,
62    /// When the event was produced.
63    #[serde(rename = "E")]
64    pub event_time: u64,
65    /// The symbol, as listed.
66    #[serde(rename = "s")]
67    pub symbol: String,
68    /// Trade id.
69    #[serde(rename = "t")]
70    pub id: u64,
71    /// Price.
72    #[serde(rename = "p", with = "rust_decimal::serde::str")]
73    pub price: Decimal,
74    /// Quantity.
75    #[serde(rename = "q", with = "rust_decimal::serde::str")]
76    pub quantity: Decimal,
77    /// Trade time.
78    #[serde(rename = "T")]
79    pub trade_time: u64,
80    /// Whether the buyer was the maker, so the taker sold.
81    #[serde(rename = "m")]
82    pub is_buyer_maker: bool,
83}
84
85/// `<s>@kline_<interval>`.
86#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
87#[non_exhaustive]
88pub struct KlineEvent {
89    /// `kline`.
90    #[serde(rename = "e")]
91    pub event_type: String,
92    /// When the event was produced.
93    #[serde(rename = "E")]
94    pub event_time: u64,
95    /// The symbol, as listed.
96    #[serde(rename = "s")]
97    pub symbol: String,
98    /// The candle so far.
99    #[serde(rename = "k")]
100    pub kline: KlineBar,
101}
102
103/// The candle in a [`KlineEvent`]. The wire's `B`, documented as "Ignore", is
104/// dropped.
105#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
106#[non_exhaustive]
107pub struct KlineBar {
108    /// Candle start.
109    #[serde(rename = "t")]
110    pub open_time: u64,
111    /// Candle end, inclusive.
112    #[serde(rename = "T")]
113    pub close_time: u64,
114    /// The symbol, as listed.
115    #[serde(rename = "s")]
116    pub symbol: String,
117    /// The candle's width.
118    #[serde(rename = "i")]
119    pub interval: Interval,
120    /// First trade id; `-1` when the candle holds no trade.
121    #[serde(rename = "f")]
122    pub first_trade_id: i64,
123    /// Last trade id; `-1` when the candle holds no trade.
124    #[serde(rename = "L")]
125    pub last_trade_id: i64,
126    /// Open price.
127    #[serde(rename = "o", with = "rust_decimal::serde::str")]
128    pub open: Decimal,
129    /// Close price so far.
130    #[serde(rename = "c", with = "rust_decimal::serde::str")]
131    pub close: Decimal,
132    /// High price.
133    #[serde(rename = "h", with = "rust_decimal::serde::str")]
134    pub high: Decimal,
135    /// Low price.
136    #[serde(rename = "l", with = "rust_decimal::serde::str")]
137    pub low: Decimal,
138    /// Base-asset volume.
139    #[serde(rename = "v", with = "rust_decimal::serde::str")]
140    pub volume: Decimal,
141    /// Trades.
142    #[serde(rename = "n")]
143    pub trade_count: u64,
144    /// Whether the candle is closed.
145    #[serde(rename = "x")]
146    pub is_closed: bool,
147    /// Quote-asset volume.
148    #[serde(rename = "q", with = "rust_decimal::serde::str")]
149    pub quote_volume: Decimal,
150    /// Base-asset volume bought by takers.
151    #[serde(rename = "V", with = "rust_decimal::serde::str")]
152    pub taker_buy_base_volume: Decimal,
153    /// Quote-asset volume bought by takers.
154    #[serde(rename = "Q", with = "rust_decimal::serde::str")]
155    pub taker_buy_quote_volume: Decimal,
156}
157
158/// `<s>@miniTicker` and each row of `!miniTicker@arr`: the 24-hour window's
159/// close, open, high, low and volumes.
160#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
161#[non_exhaustive]
162pub struct MiniTickerEvent {
163    /// `24hrMiniTicker`.
164    #[serde(rename = "e")]
165    pub event_type: String,
166    /// When the event was produced.
167    #[serde(rename = "E")]
168    pub event_time: u64,
169    /// The symbol, as listed.
170    #[serde(rename = "s")]
171    pub symbol: String,
172    /// Close price: the last price.
173    #[serde(rename = "c", with = "rust_decimal::serde::str")]
174    pub close_price: Decimal,
175    /// Price 24 hours ago.
176    #[serde(rename = "o", with = "rust_decimal::serde::str")]
177    pub open_price: Decimal,
178    /// Highest price.
179    #[serde(rename = "h", with = "rust_decimal::serde::str")]
180    pub high_price: Decimal,
181    /// Lowest price.
182    #[serde(rename = "l", with = "rust_decimal::serde::str")]
183    pub low_price: Decimal,
184    /// Base-asset volume.
185    #[serde(rename = "v", with = "rust_decimal::serde::str")]
186    pub volume: Decimal,
187    /// Quote-asset volume.
188    #[serde(rename = "q", with = "rust_decimal::serde::str")]
189    pub quote_volume: Decimal,
190}
191
192/// `<s>@ticker`: rolling 24-hour statistics.
193#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
194#[non_exhaustive]
195pub struct TickerEvent {
196    /// `24hrTicker`.
197    #[serde(rename = "e")]
198    pub event_type: String,
199    /// When the event was produced.
200    #[serde(rename = "E")]
201    pub event_time: u64,
202    /// The symbol, as listed.
203    #[serde(rename = "s")]
204    pub symbol: String,
205    /// Last price less the open price.
206    #[serde(rename = "p", with = "rust_decimal::serde::str")]
207    pub price_change: Decimal,
208    /// The change as a percentage.
209    #[serde(rename = "P", with = "rust_decimal::serde::str")]
210    pub price_change_percent: Decimal,
211    /// Volume-weighted average price.
212    #[serde(rename = "w", with = "rust_decimal::serde::str")]
213    pub weighted_avg_price: Decimal,
214    /// The price before the window opened.
215    #[serde(rename = "x", with = "rust_decimal::serde::str")]
216    pub prev_close_price: Decimal,
217    /// Last price.
218    #[serde(rename = "c", with = "rust_decimal::serde::str")]
219    pub last_price: Decimal,
220    /// Last quantity.
221    #[serde(rename = "Q", with = "rust_decimal::serde::str")]
222    pub last_quantity: Decimal,
223    /// Best bid price.
224    #[serde(rename = "b", with = "rust_decimal::serde::str")]
225    pub bid_price: Decimal,
226    /// Best bid quantity.
227    #[serde(rename = "B", with = "rust_decimal::serde::str")]
228    pub bid_quantity: Decimal,
229    /// Best ask price.
230    #[serde(rename = "a", with = "rust_decimal::serde::str")]
231    pub ask_price: Decimal,
232    /// Best ask quantity.
233    #[serde(rename = "A", with = "rust_decimal::serde::str")]
234    pub ask_quantity: Decimal,
235    /// Price 24 hours ago.
236    #[serde(rename = "o", with = "rust_decimal::serde::str")]
237    pub open_price: Decimal,
238    /// Highest price.
239    #[serde(rename = "h", with = "rust_decimal::serde::str")]
240    pub high_price: Decimal,
241    /// Lowest price.
242    #[serde(rename = "l", with = "rust_decimal::serde::str")]
243    pub low_price: Decimal,
244    /// Base-asset volume.
245    #[serde(rename = "v", with = "rust_decimal::serde::str")]
246    pub volume: Decimal,
247    /// Quote-asset volume.
248    #[serde(rename = "q", with = "rust_decimal::serde::str")]
249    pub quote_volume: Decimal,
250    /// Window start.
251    #[serde(rename = "O")]
252    pub open_time: u64,
253    /// Window end.
254    #[serde(rename = "C")]
255    pub close_time: u64,
256    /// First trade id in the window.
257    #[serde(rename = "F")]
258    pub first_trade_id: i64,
259    /// Last trade id in the window.
260    #[serde(rename = "L")]
261    pub last_trade_id: i64,
262    /// Trades in the window.
263    #[serde(rename = "n")]
264    pub trade_count: u64,
265}
266
267/// `<s>@bookTicker`: the best bid and ask. The frame carries no event type
268/// and no time.
269#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
270#[non_exhaustive]
271pub struct BookTickerEvent {
272    /// Book update id.
273    #[serde(rename = "u")]
274    pub update_id: u64,
275    /// The symbol, as listed.
276    #[serde(rename = "s")]
277    pub symbol: String,
278    /// Best bid price.
279    #[serde(rename = "b", with = "rust_decimal::serde::str")]
280    pub bid_price: Decimal,
281    /// Best bid quantity.
282    #[serde(rename = "B", with = "rust_decimal::serde::str")]
283    pub bid_quantity: Decimal,
284    /// Best ask price.
285    #[serde(rename = "a", with = "rust_decimal::serde::str")]
286    pub ask_price: Decimal,
287    /// Best ask quantity.
288    #[serde(rename = "A", with = "rust_decimal::serde::str")]
289    pub ask_quantity: Decimal,
290}
291
292/// `<s>@depth<levels>` and `<s>@depth<levels>@100ms`: the top of the book.
293///
294/// The frame carries `lastUpdateId`, `bids` and `asks` and no symbol, so
295/// `symbol` is the envelope's, uppercased as [`Symbol`](crate::spot::types::Symbol)
296/// renders it, and it is not serialised back.
297#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
298#[non_exhaustive]
299pub struct PartialDepthEvent {
300    /// The symbol, from the envelope's stream name.
301    #[serde(skip)]
302    pub symbol: String,
303    /// Book update id of the snapshot.
304    #[serde(rename = "lastUpdateId")]
305    pub last_update_id: u64,
306    /// Bids, best first.
307    pub bids: Vec<Level>,
308    /// Asks, best first.
309    pub asks: Vec<Level>,
310}
311
312/// `<s>@avgPrice`: the average price over `interval`.
313#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
314#[non_exhaustive]
315pub struct AvgPriceEvent {
316    /// `avgPrice`.
317    #[serde(rename = "e")]
318    pub event_type: String,
319    /// When the event was produced.
320    #[serde(rename = "E")]
321    pub event_time: u64,
322    /// The symbol, as listed.
323    #[serde(rename = "s")]
324    pub symbol: String,
325    /// The window the average covers, such as `5m`.
326    #[serde(rename = "i")]
327    pub interval: String,
328    /// The average price.
329    #[serde(rename = "w", with = "rust_decimal::serde::str")]
330    pub price: Decimal,
331    /// Time of the last trade.
332    #[serde(rename = "T")]
333    pub last_trade_time: u64,
334}
335
336/// What a stream carries, by stream kind.
337#[derive(Debug, Clone, PartialEq, Eq)]
338#[non_exhaustive]
339pub enum Payload {
340    /// `<s>@aggTrade`.
341    AggTrade(AggTradeEvent),
342    /// `<s>@trade`.
343    Trade(TradeEvent),
344    /// `<s>@kline_<interval>`.
345    Kline(KlineEvent),
346    /// `<s>@miniTicker`.
347    MiniTicker(MiniTickerEvent),
348    /// `!miniTicker@arr`: only the symbols that changed. A row's `e` is not
349    /// checked, and one row that does not decode fails the whole frame.
350    MiniTickers(Vec<MiniTickerEvent>),
351    /// `<s>@ticker`.
352    Ticker(TickerEvent),
353    /// `<s>@bookTicker`.
354    BookTicker(BookTickerEvent),
355    /// `<s>@depth<levels>` and `<s>@depth<levels>@100ms`.
356    PartialDepth(PartialDepthEvent),
357    /// `<s>@avgPrice`.
358    AvgPrice(AvgPriceEvent),
359    /// A single-symbol stream's object whose event type (`e`) is not the one
360    /// its stream carries, kept whole rather than failing the frame. The live
361    /// suite fails on one, since it means Binance renamed an event.
362    Unknown {
363        /// The payload's `e`, or empty when it has none.
364        event_type: String,
365        /// The payload, re-serialised: its values as sent, its keys sorted.
366        raw: String,
367    },
368}
369
370impl Serialize for Payload {
371    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
372        match self {
373            Self::AggTrade(event) => event.serialize(serializer),
374            Self::Trade(event) => event.serialize(serializer),
375            Self::Kline(event) => event.serialize(serializer),
376            Self::MiniTicker(event) => event.serialize(serializer),
377            Self::MiniTickers(rows) => rows.serialize(serializer),
378            Self::Ticker(event) => event.serialize(serializer),
379            Self::BookTicker(event) => event.serialize(serializer),
380            Self::PartialDepth(event) => event.serialize(serializer),
381            Self::AvgPrice(event) => event.serialize(serializer),
382            Self::Unknown { raw, .. } => serde_json::from_str::<Value>(raw)
383                .map_err(serde::ser::Error::custom)?
384                .serialize(serializer),
385        }
386    }
387}
388
389/// One frame of a combined stream: `{"stream": <name>, "data": <payload>}`.
390///
391/// It serialises back to that envelope, so `serde_json::to_string(&update)`
392/// is the frame as the wire sent it, less the fields this crate drops.
393#[derive(Debug, Clone, PartialEq, Eq)]
394#[non_exhaustive]
395pub struct Update {
396    /// The stream the frame arrived on.
397    pub stream: StreamName,
398    /// What it carries.
399    pub payload: Payload,
400}
401
402impl Serialize for Update {
403    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
404        use serde::ser::SerializeMap;
405        let mut map = serializer.serialize_map(Some(2))?;
406        map.serialize_entry("stream", &self.stream.to_string())?;
407        map.serialize_entry("data", &self.payload)?;
408        map.end()
409    }
410}
411
412#[derive(Deserialize)]
413struct Envelope {
414    stream: String,
415    data: Value,
416}
417
418impl Update {
419    /// Decodes one combined-stream frame.
420    ///
421    /// Fails with [`SpotWsError::Frame`] when the text is not an envelope, its
422    /// stream is not one this crate builds, or its payload does not decode as
423    /// the stream's type.
424    pub fn from_json(text: &str) -> Result<Self, SpotWsError> {
425        let envelope: Envelope =
426            serde_json::from_str(text).map_err(|err| frame_error("", text, err))?;
427        let stream: StreamName = envelope
428            .stream
429            .parse()
430            .map_err(|err| frame_error(&envelope.stream, text, err))?;
431        let payload = decode(&stream, envelope.data)
432            .map_err(|err| frame_error(&envelope.stream, text, err))?;
433        Ok(Self { stream, payload })
434    }
435}
436
437fn frame_error(stream: &str, raw: &str, reason: impl fmt::Display) -> SpotWsError {
438    SpotWsError::Frame {
439        stream: stream.to_owned(),
440        raw: raw.to_owned(),
441        reason: reason.to_string(),
442    }
443}
444
445fn decode(stream: &StreamName, data: Value) -> Result<Payload, serde_json::Error> {
446    // The streams whose frames carry no event type, and the array stream, decode
447    // as they are. A partial-depth frame carries no symbol either: the
448    // envelope's stream name supplies it.
449    let expected = match stream {
450        StreamName::AllMiniTickers => {
451            return serde_json::from_value(data).map(Payload::MiniTickers)
452        }
453        StreamName::BookTicker(_) => return serde_json::from_value(data).map(Payload::BookTicker),
454        StreamName::PartialDepth(symbol, ..) => {
455            let mut event: PartialDepthEvent = serde_json::from_value(data)?;
456            event.symbol = symbol.to_string();
457            return Ok(Payload::PartialDepth(event));
458        }
459        StreamName::AggTrade(_) => "aggTrade",
460        StreamName::Trade(_) => "trade",
461        StreamName::Kline(..) => "kline",
462        StreamName::MiniTicker(_) => "24hrMiniTicker",
463        StreamName::Ticker(_) => "24hrTicker",
464        StreamName::AvgPrice(_) => "avgPrice",
465    };
466    let event_type = data.get("e").and_then(Value::as_str).unwrap_or_default();
467    if event_type != expected {
468        return Ok(Payload::Unknown {
469            event_type: event_type.to_owned(),
470            raw: data.to_string(),
471        });
472    }
473    Ok(match stream {
474        StreamName::AggTrade(_) => Payload::AggTrade(serde_json::from_value(data)?),
475        StreamName::Trade(_) => Payload::Trade(serde_json::from_value(data)?),
476        StreamName::Kline(..) => Payload::Kline(serde_json::from_value(data)?),
477        StreamName::MiniTicker(_) => Payload::MiniTicker(serde_json::from_value(data)?),
478        StreamName::Ticker(_) => Payload::Ticker(serde_json::from_value(data)?),
479        StreamName::AvgPrice(_) => Payload::AvgPrice(serde_json::from_value(data)?),
480        StreamName::AllMiniTickers | StreamName::BookTicker(_) | StreamName::PartialDepth(..) => {
481            unreachable!("returned above")
482        }
483    })
484}
485
486#[cfg(test)]
487mod tests {
488    use super::*;
489    use crate::spot::ws::fixtures;
490    use serde_json::json;
491
492    fn wire(frame: &str) -> Value {
493        serde_json::from_str(frame).unwrap()
494    }
495
496    fn data(frame: &str) -> Value {
497        wire(frame)["data"].clone()
498    }
499
500    fn dec(row: &Value, key: &str) -> Decimal {
501        row[key]
502            .as_str()
503            .unwrap_or_else(|| panic!("{key} is not a string"))
504            .parse()
505            .unwrap()
506    }
507
508    fn int(row: &Value, key: &str) -> u64 {
509        row[key]
510            .as_u64()
511            .unwrap_or_else(|| panic!("{key} is not a u64"))
512    }
513
514    fn signed(row: &Value, key: &str) -> i64 {
515        row[key]
516            .as_i64()
517            .unwrap_or_else(|| panic!("{key} is not an i64"))
518    }
519
520    fn text(row: &Value, key: &str) -> String {
521        row[key]
522            .as_str()
523            .unwrap_or_else(|| panic!("{key} is not a string"))
524            .to_owned()
525    }
526
527    fn payload(frame: &str) -> Payload {
528        Update::from_json(frame).unwrap().payload
529    }
530
531    /// A captured frame with its envelope's stream name replaced.
532    fn on_stream(frame: &str, stream: &str) -> String {
533        let mut frame = wire(frame);
534        frame["stream"] = json!(stream);
535        frame.to_string()
536    }
537
538    #[test]
539    fn every_captured_frame_decodes_as_its_stream() {
540        for (name, frame) in fixtures::ALL {
541            let update = Update::from_json(frame).unwrap_or_else(|e| panic!("{name}: {e}"));
542            assert!(
543                !matches!(update.payload, Payload::Unknown { .. }),
544                "{name} decoded as Unknown"
545            );
546        }
547    }
548
549    #[test]
550    fn each_frame_lands_in_the_payload_its_stream_names() {
551        assert!(matches!(payload(fixtures::AGG_TRADE), Payload::AggTrade(_)));
552        assert!(matches!(payload(fixtures::TRADE), Payload::Trade(_)));
553        assert!(matches!(payload(fixtures::KLINE), Payload::Kline(_)));
554        assert!(matches!(
555            payload(fixtures::MINI_TICKER),
556            Payload::MiniTicker(_)
557        ));
558        assert!(matches!(
559            payload(fixtures::ALL_MINI_TICKERS),
560            Payload::MiniTickers(_)
561        ));
562        assert!(matches!(payload(fixtures::TICKER), Payload::Ticker(_)));
563        assert!(matches!(
564            payload(fixtures::BOOK_TICKER),
565            Payload::BookTicker(_)
566        ));
567        assert!(matches!(
568            payload(fixtures::PARTIAL_DEPTH),
569            Payload::PartialDepth(_)
570        ));
571        assert!(matches!(
572            payload(fixtures::PARTIAL_DEPTH_100MS),
573            Payload::PartialDepth(_)
574        ));
575        assert!(matches!(payload(fixtures::AVG_PRICE), Payload::AvgPrice(_)));
576        assert!(matches!(
577            payload(fixtures::NON_ASCII_MINI_TICKER),
578            Payload::MiniTicker(_)
579        ));
580    }
581
582    /// The round trip proves the key set, not which field a key lands in: two
583    /// fields of one type swapped would pass it. Each field is compared here
584    /// with its own key in the raw frame, as Binance's pages define the keys,
585    /// so this holds for any capture. Where two keys carry equal values on
586    /// every frame, a swap of those two stays invisible.
587    #[test]
588    fn each_field_holds_the_value_of_its_own_key() {
589        fn mini(e: &MiniTickerEvent, row: &Value) {
590            assert_eq!(
591                [
592                    e.close_price,
593                    e.open_price,
594                    e.high_price,
595                    e.low_price,
596                    e.volume,
597                    e.quote_volume,
598                ],
599                ["c", "o", "h", "l", "v", "q"].map(|k| dec(row, k)),
600                "{}",
601                e.symbol
602            );
603            assert_eq!(e.event_time, int(row, "E"), "{}", e.symbol);
604            assert_eq!(e.symbol, text(row, "s"));
605            assert_eq!(e.event_type, text(row, "e"));
606        }
607
608        let Payload::MiniTickers(rows) = payload(fixtures::ALL_MINI_TICKERS) else {
609            panic!("not MiniTickers");
610        };
611        assert_eq!(rows.len(), 2);
612        for (e, row) in rows
613            .iter()
614            .zip(data(fixtures::ALL_MINI_TICKERS).as_array().unwrap())
615        {
616            mini(e, row);
617        }
618        for frame in [fixtures::MINI_TICKER, fixtures::NON_ASCII_MINI_TICKER] {
619            let Payload::MiniTicker(e) = payload(frame) else {
620                panic!("not MiniTicker");
621            };
622            mini(&e, &data(frame));
623        }
624
625        let row = data(fixtures::TICKER);
626        let Payload::Ticker(e) = payload(fixtures::TICKER) else {
627            panic!("not Ticker");
628        };
629        assert_eq!(
630            [
631                e.price_change,
632                e.price_change_percent,
633                e.weighted_avg_price,
634                e.prev_close_price,
635                e.last_price,
636                e.last_quantity,
637                e.bid_price,
638                e.bid_quantity,
639                e.ask_price,
640                e.ask_quantity,
641                e.open_price,
642                e.high_price,
643                e.low_price,
644                e.volume,
645                e.quote_volume,
646            ],
647            ["p", "P", "w", "x", "c", "Q", "b", "B", "a", "A", "o", "h", "l", "v", "q"]
648                .map(|k| dec(&row, k))
649        );
650        assert_eq!(e.symbol, text(&row, "s"));
651        assert_eq!(e.event_type, text(&row, "e"));
652        assert_eq!(
653            [e.event_time, e.open_time, e.close_time, e.trade_count],
654            ["E", "O", "C", "n"].map(|k| int(&row, k))
655        );
656        assert_eq!(
657            [e.first_trade_id, e.last_trade_id],
658            ["F", "L"].map(|k| signed(&row, k))
659        );
660
661        let row = data(fixtures::AGG_TRADE);
662        let Payload::AggTrade(e) = payload(fixtures::AGG_TRADE) else {
663            panic!("not AggTrade");
664        };
665        assert_eq!([e.price, e.quantity], ["p", "q"].map(|k| dec(&row, k)));
666        assert_eq!(
667            [
668                e.event_time,
669                e.id,
670                e.first_trade_id,
671                e.last_trade_id,
672                e.trade_time
673            ],
674            ["E", "a", "f", "l", "T"].map(|k| int(&row, k))
675        );
676        assert_eq!(e.is_buyer_maker, row["m"].as_bool().unwrap());
677        assert_eq!(e.symbol, text(&row, "s"));
678        assert_eq!(e.event_type, text(&row, "e"));
679
680        let row = data(fixtures::TRADE);
681        let Payload::Trade(e) = payload(fixtures::TRADE) else {
682            panic!("not Trade");
683        };
684        assert_eq!([e.price, e.quantity], ["p", "q"].map(|k| dec(&row, k)));
685        assert_eq!(
686            [e.event_time, e.id, e.trade_time],
687            ["E", "t", "T"].map(|k| int(&row, k))
688        );
689        assert_eq!(e.is_buyer_maker, row["m"].as_bool().unwrap());
690        assert_eq!(e.symbol, text(&row, "s"));
691        assert_eq!(e.event_type, text(&row, "e"));
692
693        let event = data(fixtures::KLINE);
694        let row = event["k"].clone();
695        let Payload::Kline(e) = payload(fixtures::KLINE) else {
696            panic!("not Kline");
697        };
698        let k = &e.kline;
699        assert_eq!(
700            [
701                k.open,
702                k.close,
703                k.high,
704                k.low,
705                k.volume,
706                k.quote_volume,
707                k.taker_buy_base_volume,
708                k.taker_buy_quote_volume,
709            ],
710            ["o", "c", "h", "l", "v", "q", "V", "Q"].map(|key| dec(&row, key))
711        );
712        assert_eq!(
713            [k.open_time, k.close_time, k.trade_count],
714            ["t", "T", "n"].map(|key| int(&row, key))
715        );
716        assert_eq!(
717            [k.first_trade_id, k.last_trade_id],
718            ["f", "L"].map(|key| signed(&row, key))
719        );
720        assert_eq!(k.interval.as_str(), text(&row, "i"));
721        assert_eq!(k.is_closed, row["x"].as_bool().unwrap());
722        assert_eq!(k.symbol, text(&row, "s"));
723        assert_eq!(e.symbol, text(&event, "s"));
724        assert_eq!(e.event_type, text(&event, "e"));
725
726        let row = data(fixtures::BOOK_TICKER);
727        let Payload::BookTicker(e) = payload(fixtures::BOOK_TICKER) else {
728            panic!("not BookTicker");
729        };
730        assert_eq!(
731            [e.bid_price, e.bid_quantity, e.ask_price, e.ask_quantity],
732            ["b", "B", "a", "A"].map(|k| dec(&row, k))
733        );
734        assert_eq!(e.update_id, int(&row, "u"));
735        assert_eq!(e.symbol, text(&row, "s"));
736
737        let row = data(fixtures::AVG_PRICE);
738        let Payload::AvgPrice(e) = payload(fixtures::AVG_PRICE) else {
739            panic!("not AvgPrice");
740        };
741        assert_eq!(e.price, dec(&row, "w"));
742        assert_eq!(
743            [e.event_time, e.last_trade_time],
744            ["E", "T"].map(|k| int(&row, k))
745        );
746        assert_eq!(e.interval, text(&row, "i"));
747        assert_eq!(e.symbol, text(&row, "s"));
748        assert_eq!(e.event_type, text(&row, "e"));
749    }
750
751    #[test]
752    fn a_partial_depth_frame_takes_its_symbol_from_the_envelope() {
753        // The frame carries `lastUpdateId`, `bids` and `asks` and nothing else.
754        for (frame, bids) in [
755            (fixtures::PARTIAL_DEPTH, 5),
756            (fixtures::PARTIAL_DEPTH_100MS, 3),
757        ] {
758            let row = data(frame);
759            assert_eq!(row.as_object().unwrap().len(), 3);
760            let update = Update::from_json(frame).unwrap();
761            let Payload::PartialDepth(e) = update.payload else {
762                panic!("not PartialDepth");
763            };
764            assert_eq!(e.symbol, "BTCUSDT");
765            assert_eq!(e.last_update_id, int(&row, "lastUpdateId"));
766            assert_eq!(e.bids.len(), bids);
767            assert_eq!(e.asks.len(), bids);
768            for (side, key) in [(&e.bids, "bids"), (&e.asks, "asks")] {
769                let sent: Vec<(Decimal, Decimal)> = row[key]
770                    .as_array()
771                    .unwrap()
772                    .iter()
773                    .map(|level| {
774                        (
775                            level[0].as_str().unwrap().parse().unwrap(),
776                            level[1].as_str().unwrap().parse().unwrap(),
777                        )
778                    })
779                    .collect();
780                let decoded: Vec<(Decimal, Decimal)> =
781                    side.iter().map(|l| (l.price, l.quantity)).collect();
782                assert_eq!(decoded, sent, "{key}");
783            }
784        }
785
786        // Another stream's name gives another symbol: it is the envelope's, not
787        // the frame's.
788        let ethereum = on_stream(fixtures::PARTIAL_DEPTH, "ethusdt@depth5");
789        let Payload::PartialDepth(e) = Update::from_json(&ethereum).unwrap().payload else {
790            panic!("not PartialDepth");
791        };
792        assert_eq!(e.symbol, "ETHUSDT");
793        let chinese = on_stream(fixtures::PARTIAL_DEPTH, "币安人生usdt@depth5");
794        let Payload::PartialDepth(e) = Update::from_json(&chinese).unwrap().payload else {
795            panic!("not PartialDepth");
796        };
797        assert_eq!(e.symbol, "币安人生USDT");
798    }
799
800    #[test]
801    fn a_book_ticker_frame_has_no_event_type_and_decodes() {
802        assert!(data(fixtures::BOOK_TICKER).get("e").is_none());
803        assert!(matches!(
804            payload(fixtures::BOOK_TICKER),
805            Payload::BookTicker(_)
806        ));
807    }
808
809    #[test]
810    fn the_non_ascii_symbol_keeps_its_characters_in_the_name_and_the_payload() {
811        let update = Update::from_json(fixtures::NON_ASCII_MINI_TICKER).unwrap();
812        let StreamName::MiniTicker(symbol) = &update.stream else {
813            panic!("not MiniTicker: {:?}", update.stream);
814        };
815        assert!(!symbol.as_str().is_ascii());
816        let Payload::MiniTicker(e) = &update.payload else {
817            panic!("not MiniTicker");
818        };
819        assert_eq!(
820            e.symbol.to_ascii_lowercase(),
821            symbol.as_str().to_ascii_lowercase()
822        );
823        assert_eq!(
824            wire(fixtures::NON_ASCII_MINI_TICKER)["stream"],
825            update.stream.to_string()
826        );
827    }
828
829    #[test]
830    fn an_object_whose_event_type_is_not_its_streams_is_kept_whole() {
831        // An aggregate trade's data on a trade stream: `e` is `aggTrade`, not
832        // `trade`.
833        let crossed = on_stream(fixtures::AGG_TRADE, "btcusdt@trade");
834        let update = Update::from_json(&crossed).unwrap();
835        assert_eq!(update.stream.to_string(), "btcusdt@trade");
836        let Payload::Unknown { event_type, raw } = &update.payload else {
837            panic!("not Unknown: {:?}", update.payload);
838        };
839        assert_eq!(event_type, "aggTrade");
840        assert_eq!(
841            serde_json::from_str::<Value>(raw).unwrap(),
842            data(fixtures::AGG_TRADE)
843        );
844
845        // A renamed event, and an object with none.
846        for (e, expected) in [(json!("aggTradeV2"), "aggTradeV2"), (Value::Null, "")] {
847            let mut frame = wire(fixtures::AGG_TRADE);
848            match e {
849                Value::Null => {
850                    frame["data"].as_object_mut().unwrap().remove("e");
851                }
852                e => frame["data"]["e"] = e,
853            }
854            let update = Update::from_json(&frame.to_string()).unwrap();
855            assert!(
856                matches!(&update.payload, Payload::Unknown { event_type, .. } if event_type == expected),
857                "{:?}",
858                update.payload
859            );
860            // It serialises back as it came.
861            assert_eq!(serde_json::to_value(&update).unwrap(), frame);
862        }
863    }
864
865    #[test]
866    fn a_broken_frame_is_a_frame_error_that_names_its_stream() {
867        let mut broken = wire(fixtures::AGG_TRADE);
868        broken["data"]["p"] = json!("x");
869        let err = Update::from_json(&broken.to_string()).unwrap_err();
870        assert!(
871            matches!(&err, SpotWsError::Frame { stream, .. } if stream == "btcusdt@aggTrade"),
872            "{err:?}"
873        );
874
875        // An array stream whose data is an object, and a book ticker with no
876        // `u`.
877        let mut object = wire(fixtures::ALL_MINI_TICKERS);
878        object["data"] = json!({});
879        assert!(matches!(
880            Update::from_json(&object.to_string()),
881            Err(SpotWsError::Frame { .. })
882        ));
883        let mut thin = wire(fixtures::BOOK_TICKER);
884        thin["data"].as_object_mut().unwrap().remove("u");
885        assert!(matches!(
886            Update::from_json(&thin.to_string()),
887            Err(SpotWsError::Frame { .. })
888        ));
889        let mut empty = wire(fixtures::PARTIAL_DEPTH);
890        empty["data"] = json!({});
891        assert!(matches!(
892            Update::from_json(&empty.to_string()),
893            Err(SpotWsError::Frame { .. })
894        ));
895    }
896
897    #[test]
898    fn a_stream_this_crate_does_not_build_is_a_frame_error() {
899        for stream in [
900            "btcusdt@forceOrder",
901            "BTCUSDT@trade",
902            "!ticker@arr",
903            "btcusdt@depth5@250ms",
904        ] {
905            let err = Update::from_json(&on_stream(fixtures::TRADE, stream)).unwrap_err();
906            assert!(
907                matches!(&err, SpotWsError::Frame { stream: s, .. } if s == stream),
908                "{stream}: {err:?}"
909            );
910        }
911        assert!(matches!(
912            Update::from_json("not json"),
913            Err(SpotWsError::Frame { stream, .. }) if stream.is_empty()
914        ));
915        assert!(matches!(
916            Update::from_json(r#"{"result":null,"id":1}"#),
917            Err(SpotWsError::Frame { .. })
918        ));
919    }
920
921    #[test]
922    fn an_update_serialises_back_to_its_envelope() {
923        // Frames whose every key the types model come back whole. The partial
924        // depth's symbol is not serialised: it was never in the frame.
925        for frame in [
926            fixtures::BOOK_TICKER,
927            fixtures::PARTIAL_DEPTH,
928            fixtures::PARTIAL_DEPTH_100MS,
929            fixtures::MINI_TICKER,
930            fixtures::ALL_MINI_TICKERS,
931            fixtures::TICKER,
932            fixtures::AVG_PRICE,
933        ] {
934            let update = Update::from_json(frame).unwrap();
935            assert_eq!(serde_json::to_value(&update).unwrap(), wire(frame));
936        }
937    }
938}
polyoxide_binance/spot/ws/event.rs

polyoxide_binance/spot/ws/
event.rs
