1use super::super::book::BinanceLevel;
2use crate::{
3 Identifier, SnapshotFetcher,
4 books::{OrderBook, OrderBookTimes},
5 error::DataError,
6 event::{MarketEvent, MarketIter},
7 exchange::{
8 Connector,
9 binance::{
10 book::l2::{BinanceOrderBookL2Meta, BinanceOrderBookL2Snapshot},
11 market::BinanceMarket,
12 spot::BinanceSpot,
13 },
14 },
15 instrument::InstrumentData,
16 subscription::{
17 Map, Subscription,
18 book::{OrderBookEvent, OrderBooksL2},
19 },
20 transformer::ExchangeTransformer,
21};
22use chrono::{DateTime, Utc};
23use futures_util::future::try_join_all;
24use rustrade_instrument::exchange::ExchangeId;
25use rustrade_integration::{
26 Transformer, error::SocketError, protocol::websocket::WsMessage, subscription::SubscriptionId,
27};
28use serde::{Deserialize, Serialize};
29use std::future::Future;
30use tokio::sync::mpsc::UnboundedSender;
31
32pub const HTTP_BOOK_L2_SNAPSHOT_URL_BINANCE_SPOT: &str = "https://api.binance.com/api/v3/depth";
36
37#[derive(Debug)]
38pub struct BinanceSpotOrderBooksL2SnapshotFetcher;
39
40impl SnapshotFetcher<BinanceSpot, OrderBooksL2> for BinanceSpotOrderBooksL2SnapshotFetcher {
41 fn fetch_snapshots<Instrument>(
42 subscriptions: &[Subscription<BinanceSpot, Instrument, OrderBooksL2>],
43 ) -> impl Future<Output = Result<Vec<MarketEvent<Instrument::Key, OrderBookEvent>>, SocketError>>
44 + Send
45 where
46 Instrument: InstrumentData,
47 Subscription<BinanceSpot, Instrument, OrderBooksL2>: Identifier<BinanceMarket>,
48 {
49 let l2_snapshot_futures = subscriptions.iter().map(|subscription| {
50 let market = subscription.id();
52 let snapshot_url = format!(
53 "{}?symbol={}&limit=100",
54 HTTP_BOOK_L2_SNAPSHOT_URL_BINANCE_SPOT, market.0,
55 );
56
57 async move {
58 let snapshot = reqwest::get(snapshot_url)
60 .await
61 .map_err(SocketError::Http)?
62 .json::<BinanceOrderBookL2Snapshot>()
63 .await
64 .map_err(SocketError::Http)?;
65
66 Ok(MarketEvent::from((
67 ExchangeId::BinanceSpot,
68 subscription.instrument.key().clone(),
69 snapshot,
70 )))
71 }
72 });
73
74 try_join_all(l2_snapshot_futures)
75 }
76}
77
78#[derive(Debug)]
79pub struct BinanceSpotOrderBooksL2Transformer<InstrumentKey> {
80 instrument_map: Map<BinanceOrderBookL2Meta<InstrumentKey, BinanceSpotOrderBookL2Sequencer>>,
81}
82
83impl<InstrumentKey> ExchangeTransformer<BinanceSpot, InstrumentKey, OrderBooksL2>
84 for BinanceSpotOrderBooksL2Transformer<InstrumentKey>
85where
86 InstrumentKey: Clone + PartialEq + Send + Sync,
87{
88 async fn init(
89 instrument_map: Map<InstrumentKey>,
90 initial_snapshots: &[MarketEvent<InstrumentKey, OrderBookEvent>],
91 _: UnboundedSender<WsMessage>,
92 ) -> Result<Self, DataError> {
93 let instrument_map = instrument_map
94 .0
95 .into_iter()
96 .map(|(sub_id, instrument_key)| {
97 let snapshot = initial_snapshots
98 .iter()
99 .find(|snapshot| snapshot.instrument == instrument_key)
100 .ok_or_else(|| DataError::InitialSnapshotMissing(sub_id.clone()))?;
101
102 let OrderBookEvent::Snapshot(snapshot) = &snapshot.kind else {
103 return Err(DataError::InitialSnapshotInvalid(String::from(
104 "expected OrderBookEvent::Snapshot but found OrderBookEvent::Update",
105 )));
106 };
107
108 let book_meta = BinanceOrderBookL2Meta::new(
109 instrument_key,
110 BinanceSpotOrderBookL2Sequencer::new(snapshot.sequence()),
111 );
112
113 Ok((sub_id, book_meta))
114 })
115 .collect::<Result<Map<_>, _>>()?;
116
117 Ok(Self { instrument_map })
118 }
119}
120
121impl<InstrumentKey> Transformer for BinanceSpotOrderBooksL2Transformer<InstrumentKey>
122where
123 InstrumentKey: Clone,
124{
125 type Error = DataError;
126 type Input = BinanceSpotOrderBookL2Update;
127 type Output = MarketEvent<InstrumentKey, OrderBookEvent>;
128 type OutputIter = Vec<Result<Self::Output, Self::Error>>;
129
130 fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
131 let subscription_id = match input.id() {
133 Some(subscription_id) => subscription_id,
134 None => return vec![],
135 };
136
137 let instrument = match self.instrument_map.find_mut(&subscription_id) {
139 Ok(instrument) => instrument,
140 Err(unidentifiable) => return vec![Err(DataError::from(unidentifiable))],
141 };
142
143 let valid_update = match instrument.sequencer.validate_sequence(input) {
145 Ok(Some(valid_update)) => valid_update,
146 Ok(None) => return vec![],
147 Err(error) => return vec![Err(error)],
148 };
149
150 MarketIter::<InstrumentKey, OrderBookEvent>::from((
151 BinanceSpot::ID,
152 instrument.key.clone(),
153 valid_update,
154 ))
155 .0
156 }
157}
158
159#[derive(Debug)]
184pub struct BinanceSpotOrderBookL2Sequencer {
185 pub updates_processed: u64,
186 pub last_update_id: u64,
187 pub prev_last_update_id: u64,
188}
189
190impl BinanceSpotOrderBookL2Sequencer {
191 pub fn new(last_update_id: u64) -> Self {
193 Self {
194 updates_processed: 0,
195 last_update_id,
196 prev_last_update_id: last_update_id,
197 }
198 }
199
200 pub fn validate_sequence(
204 &mut self,
205 update: BinanceSpotOrderBookL2Update,
206 ) -> Result<Option<BinanceSpotOrderBookL2Update>, DataError> {
207 if update.last_update_id <= self.last_update_id {
209 return Ok(None);
210 }
211
212 if self.is_first_update() {
213 self.validate_first_update(&update)?;
215 } else {
216 self.validate_next_update(&update)?;
218 }
219
220 self.updates_processed += 1;
222 self.prev_last_update_id = self.last_update_id;
223 self.last_update_id = update.last_update_id;
224
225 Ok(Some(update))
226 }
227
228 pub fn is_first_update(&self) -> bool {
233 self.updates_processed == 0
234 }
235
236 pub fn validate_first_update(
241 &self,
242 update: &BinanceSpotOrderBookL2Update,
243 ) -> Result<(), DataError> {
244 let expected_next_id = self.last_update_id + 1;
245 if update.first_update_id <= expected_next_id && update.last_update_id >= expected_next_id {
246 Ok(())
247 } else {
248 Err(DataError::InvalidSequence {
249 prev_last_update_id: self.last_update_id,
250 first_update_id: update.first_update_id,
251 })
252 }
253 }
254
255 pub fn validate_next_update(
261 &self,
262 update: &BinanceSpotOrderBookL2Update,
263 ) -> Result<(), DataError> {
264 let expected_next_id = self.last_update_id + 1;
265 if update.first_update_id == expected_next_id {
266 Ok(())
267 } else {
268 Err(DataError::InvalidSequence {
269 prev_last_update_id: self.last_update_id,
270 first_update_id: update.first_update_id,
271 })
272 }
273 }
274}
275
276#[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize, Serialize)]
295pub struct BinanceSpotOrderBookL2Update {
296 #[serde(
297 alias = "s",
298 deserialize_with = "super::super::book::l2::de_ob_l2_subscription_id"
299 )]
300 pub subscription_id: SubscriptionId,
301 #[serde(
302 alias = "E",
303 deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
304 )]
305 pub time_exchange: DateTime<Utc>,
306 #[serde(alias = "U")]
307 pub first_update_id: u64,
308 #[serde(alias = "u")]
309 pub last_update_id: u64,
310 #[serde(alias = "b")]
311 pub bids: Vec<BinanceLevel>,
312 #[serde(alias = "a")]
313 pub asks: Vec<BinanceLevel>,
314}
315
316impl Identifier<Option<SubscriptionId>> for BinanceSpotOrderBookL2Update {
317 fn id(&self) -> Option<SubscriptionId> {
318 Some(self.subscription_id.clone())
319 }
320}
321
322impl<InstrumentKey> From<(ExchangeId, InstrumentKey, BinanceSpotOrderBookL2Update)>
323 for MarketIter<InstrumentKey, OrderBookEvent>
324{
325 fn from(
326 (exchange_id, instrument, update): (
327 ExchangeId,
328 InstrumentKey,
329 BinanceSpotOrderBookL2Update,
330 ),
331 ) -> Self {
332 let time_received = Utc::now();
333 Self(vec![Ok(MarketEvent {
334 time_exchange: update.time_exchange,
335 time_received,
336 exchange: exchange_id,
337 instrument,
338 kind: OrderBookEvent::Update(OrderBook::new(
339 update.last_update_id,
340 OrderBookTimes {
342 time_engine: None,
343 time_exchange: Some(update.time_exchange),
344 time_received,
345 },
346 update.bids,
347 update.asks,
348 )),
349 })])
350 }
351}
352
353#[cfg(test)]
354#[allow(clippy::unwrap_used)] mod tests {
356 use super::*;
357 use crate::books::Level;
358 use rust_decimal_macros::dec;
359
360 #[test]
361 fn test_spot_order_book_times() {
362 let e0 = DateTime::from_timestamp_millis(1671656397761).unwrap();
367 let e1 = DateTime::from_timestamp_millis(1671656397800).unwrap();
368
369 let update = |last_update_id, time_exchange| BinanceSpotOrderBookL2Update {
370 subscription_id: SubscriptionId::from("@depth@100ms|ETHUSDT"),
371 time_exchange,
372 first_update_id: last_update_id,
373 last_update_id,
374 bids: vec![BinanceLevel {
375 price: dec!(100),
376 amount: dec!(1),
377 }],
378 asks: vec![],
379 };
380
381 let MarketIter(mut events) = MarketIter::<&str, OrderBookEvent>::from((
382 ExchangeId::BinanceSpot,
383 "inst",
384 update(1, e0),
385 ));
386 let event = events.remove(0).unwrap();
387 let envelope_time_received = event.time_received;
388 let OrderBookEvent::Update(mut book) = event.kind else {
389 panic!("expected Update");
390 };
391 assert_eq!(book.time_exchange(), Some(e0));
392 assert_eq!(book.time_engine(), None);
393 assert_eq!(book.time_received(), envelope_time_received);
395
396 let MarketIter(mut next) = MarketIter::<&str, OrderBookEvent>::from((
398 ExchangeId::BinanceSpot,
399 "inst",
400 update(2, e1),
401 ));
402 let event = next.remove(0).unwrap().kind;
403 book.update(&event);
404 assert_eq!(book.time_exchange(), Some(e1));
405 assert_eq!(book.time_engine(), None);
406 }
407
408 #[test]
409 fn test_de_binance_spot_order_book_l2_update() {
410 let input = r#"
411 {
412 "e":"depthUpdate",
413 "E":1671656397761,
414 "s":"ETHUSDT",
415 "U":22611425143,
416 "u":22611425151,
417 "b":[
418 ["1209.67000000","85.48210000"],
419 ["1209.66000000","20.68790000"]
420 ],
421 "a":[]
422 }
423 "#;
424
425 assert_eq!(
426 serde_json::from_str::<BinanceSpotOrderBookL2Update>(input).unwrap(),
427 BinanceSpotOrderBookL2Update {
428 subscription_id: SubscriptionId::from("@depth@100ms|ETHUSDT"),
429 time_exchange: DateTime::from_timestamp_millis(1671656397761).unwrap(),
430 first_update_id: 22611425143,
431 last_update_id: 22611425151,
432 bids: vec![
433 BinanceLevel {
434 price: dec!(1209.67000000),
435 amount: dec!(85.48210000)
436 },
437 BinanceLevel {
438 price: dec!(1209.66000000),
439 amount: dec!(20.68790000)
440 },
441 ],
442 asks: vec![]
443 }
444 );
445 }
446
447 #[test]
448 fn test_sequencer_is_first_update() {
449 struct TestCase {
450 input: BinanceSpotOrderBookL2Sequencer,
451 expected: bool,
452 }
453
454 let tests = vec![
455 TestCase {
456 input: BinanceSpotOrderBookL2Sequencer::new(10),
458 expected: true,
459 },
460 TestCase {
461 input: BinanceSpotOrderBookL2Sequencer {
463 updates_processed: 10,
464 last_update_id: 100,
465 prev_last_update_id: 90,
466 },
467 expected: false,
468 },
469 ];
470
471 for (index, test) in tests.into_iter().enumerate() {
472 assert_eq!(
473 test.input.is_first_update(),
474 test.expected,
475 "TC{} failed",
476 index
477 );
478 }
479 }
480
481 #[test]
482 fn test_sequencer_validate_first_update() {
483 struct TestCase {
484 sequencer: BinanceSpotOrderBookL2Sequencer,
485 input: BinanceSpotOrderBookL2Update,
486 expected: Result<(), DataError>,
487 }
488
489 let tests = vec![
490 TestCase {
491 sequencer: BinanceSpotOrderBookL2Sequencer {
493 updates_processed: 0,
494 last_update_id: 100,
495 prev_last_update_id: 90,
496 },
497 input: BinanceSpotOrderBookL2Update {
498 subscription_id: SubscriptionId::from("subscription_id"),
499 time_exchange: Default::default(),
500 first_update_id: 100,
501 last_update_id: 110,
502 bids: vec![],
503 asks: vec![],
504 },
505 expected: Ok(()),
506 },
507 TestCase {
508 sequencer: BinanceSpotOrderBookL2Sequencer {
510 updates_processed: 0,
511 last_update_id: 100,
512 prev_last_update_id: 90,
513 },
514 input: BinanceSpotOrderBookL2Update {
515 subscription_id: SubscriptionId::from("subscription_id"),
516 time_exchange: Default::default(),
517 first_update_id: 102,
518 last_update_id: 90,
519 bids: vec![],
520 asks: vec![],
521 },
522 expected: Err(DataError::InvalidSequence {
523 prev_last_update_id: 100,
524 first_update_id: 102,
525 }),
526 },
527 TestCase {
528 sequencer: BinanceSpotOrderBookL2Sequencer {
530 updates_processed: 0,
531 last_update_id: 100,
532 prev_last_update_id: 90,
533 },
534 input: BinanceSpotOrderBookL2Update {
535 subscription_id: SubscriptionId::from("subscription_id"),
536 time_exchange: Default::default(),
537 first_update_id: 110,
538 last_update_id: 90,
539 bids: vec![],
540 asks: vec![],
541 },
542 expected: Err(DataError::InvalidSequence {
543 prev_last_update_id: 100,
544 first_update_id: 110,
545 }),
546 },
547 TestCase {
548 sequencer: BinanceSpotOrderBookL2Sequencer {
550 updates_processed: 0,
551 last_update_id: 100,
552 prev_last_update_id: 90,
553 },
554 input: BinanceSpotOrderBookL2Update {
555 subscription_id: SubscriptionId::from("subscription_id"),
556 time_exchange: Default::default(),
557 first_update_id: 110,
558 last_update_id: 90,
559 bids: vec![],
560 asks: vec![],
561 },
562 expected: Err(DataError::InvalidSequence {
563 prev_last_update_id: 100,
564 first_update_id: 110,
565 }),
566 },
567 ];
568
569 for (index, test) in tests.into_iter().enumerate() {
570 let actual = test.sequencer.validate_first_update(&test.input);
571 match (actual, test.expected) {
572 (Ok(_), Ok(_)) => {
573 }
575 (Err(actual), Err(expected)) => {
576 assert_eq!(actual, expected, "TC{index} error variant mismatch");
577 }
578 (actual, expected) => {
579 panic!(
581 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
582 );
583 }
584 }
585 }
586 }
587
588 #[test]
589 fn test_sequencer_validate_next_update() {
590 struct TestCase {
591 sequencer: BinanceSpotOrderBookL2Sequencer,
592 input: BinanceSpotOrderBookL2Update,
593 expected: Result<(), DataError>,
594 }
595
596 let tests = vec![
597 TestCase {
598 sequencer: BinanceSpotOrderBookL2Sequencer {
600 updates_processed: 100,
601 last_update_id: 100,
602 prev_last_update_id: 100,
603 },
604 input: BinanceSpotOrderBookL2Update {
605 subscription_id: SubscriptionId::from("subscription_id"),
606 time_exchange: Default::default(),
607 first_update_id: 101,
608 last_update_id: 110,
609 bids: vec![],
610 asks: vec![],
611 },
612 expected: Ok(()),
613 },
614 TestCase {
615 sequencer: BinanceSpotOrderBookL2Sequencer {
617 updates_processed: 100,
618 last_update_id: 100,
619 prev_last_update_id: 90,
620 },
621 input: BinanceSpotOrderBookL2Update {
622 subscription_id: SubscriptionId::from("subscription_id"),
623 time_exchange: Default::default(),
624 first_update_id: 120,
625 last_update_id: 130,
626 bids: vec![],
627 asks: vec![],
628 },
629 expected: Err(DataError::InvalidSequence {
630 prev_last_update_id: 100,
631 first_update_id: 120,
632 }),
633 },
634 ];
635
636 for (index, test) in tests.into_iter().enumerate() {
637 let actual = test.sequencer.validate_next_update(&test.input);
638 match (actual, test.expected) {
639 (Ok(_), Ok(_)) => {
640 }
642 (Err(actual), Err(expected)) => {
643 assert_eq!(actual, expected, "TC{index} error variant mismatch");
644 }
645 (actual, expected) => {
646 panic!(
648 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
649 );
650 }
651 }
652 }
653 }
654
655 #[test]
656 fn test_update_rustrade_order_book_with_sequenced_updates() {
657 struct TestCase {
658 sequencer: BinanceSpotOrderBookL2Sequencer,
659 book: OrderBook,
660 input_update: BinanceSpotOrderBookL2Update,
661 expected: OrderBook,
662 }
663
664 let tests = vec![
665 TestCase {
666 sequencer: BinanceSpotOrderBookL2Sequencer {
668 updates_processed: 100,
669 last_update_id: 100,
670 prev_last_update_id: 0,
671 },
672 book: OrderBook::new(
673 100,
674 OrderBookTimes::default(),
675 vec![Level::new(50, 1)],
676 vec![Level::new(100, 1)],
677 ),
678 input_update: BinanceSpotOrderBookL2Update {
679 subscription_id: SubscriptionId::from("subscription_id"),
680 time_exchange: Default::default(),
681 first_update_id: 0,
682 last_update_id: 100, bids: vec![],
684 asks: vec![],
685 },
686 expected: OrderBook::new(
687 100,
688 OrderBookTimes::default(),
689 vec![Level::new(50, 1)],
690 vec![Level::new(100, 1)],
691 ),
692 },
693 TestCase {
694 sequencer: BinanceSpotOrderBookL2Sequencer {
696 updates_processed: 100,
697 last_update_id: 100,
698 prev_last_update_id: 100,
699 },
700 book: OrderBook::new(
701 100,
702 OrderBookTimes::default(),
703 vec![Level::new(80, 1), Level::new(100, 1), Level::new(90, 1)],
704 vec![Level::new(150, 1), Level::new(110, 1), Level::new(120, 1)],
705 ),
706 input_update: BinanceSpotOrderBookL2Update {
707 subscription_id: SubscriptionId::from("subscription_id"),
708 time_exchange: Default::default(),
709 first_update_id: 101,
710 last_update_id: 110,
711 bids: vec![
712 BinanceLevel {
714 price: dec!(80),
715 amount: dec!(0),
716 },
717 BinanceLevel {
719 price: dec!(90),
720 amount: dec!(10),
721 },
722 ],
723 asks: vec![
724 BinanceLevel {
726 price: dec!(200),
727 amount: dec!(1),
728 },
729 BinanceLevel {
731 price: dec!(500),
732 amount: dec!(0),
733 },
734 ],
735 },
736 expected: OrderBook::new(
737 110,
738 OrderBookTimes::default(),
739 vec![Level::new(100, 1), Level::new(90, 10)],
740 vec![
741 Level::new(110, 1),
742 Level::new(120, 1),
743 Level::new(150, 1),
744 Level::new(200, 1),
745 ],
746 ),
747 },
748 ];
749
750 for (index, mut test) in tests.into_iter().enumerate() {
751 if let Some(valid_update) = test.sequencer.validate_sequence(test.input_update).unwrap()
752 {
753 let rustrade_update = OrderBookEvent::Update(OrderBook::new(
754 valid_update.last_update_id,
755 OrderBookTimes::default(),
756 valid_update.bids,
757 valid_update.asks,
758 ));
759
760 test.book.update(&rustrade_update);
761 }
762
763 assert_eq!(test.book, test.expected, "TC{index} failed");
764 }
765 }
766}