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;
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)]
298pub struct BinanceSpotOrderBookL2Update {
299 #[serde(
300 alias = "s",
301 deserialize_with = "super::super::book::l2::de_ob_l2_subscription_id"
302 )]
303 pub subscription_id: SubscriptionId,
304 #[serde(
305 alias = "E",
306 deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
307 )]
308 pub time_exchange: DateTime<Utc>,
309 #[serde(alias = "U")]
310 pub first_update_id: u64,
311 #[serde(alias = "u")]
312 pub last_update_id: u64,
313 #[serde(alias = "b")]
314 pub bids: Vec<BinanceLevel>,
315 #[serde(alias = "a")]
316 pub asks: Vec<BinanceLevel>,
317}
318
319impl Identifier<Option<SubscriptionId>> for BinanceSpotOrderBookL2Update {
320 fn id(&self) -> Option<SubscriptionId> {
321 Some(self.subscription_id.clone())
322 }
323}
324
325impl<InstrumentKey> From<(ExchangeId, InstrumentKey, BinanceSpotOrderBookL2Update)>
326 for MarketIter<InstrumentKey, OrderBookEvent>
327{
328 fn from(
329 (exchange_id, instrument, update): (
330 ExchangeId,
331 InstrumentKey,
332 BinanceSpotOrderBookL2Update,
333 ),
334 ) -> Self {
335 let time_received = Utc::now();
336 Self(vec![Ok(MarketEvent {
337 time_exchange: update.time_exchange,
338 time_received,
339 exchange: exchange_id,
340 instrument,
341 kind: OrderBookEvent::Update(OrderBook::new(
342 update.last_update_id,
343 OrderBookTimes {
345 time_engine: None,
346 time_exchange: Some(update.time_exchange),
347 time_received,
348 },
349 update.bids,
350 update.asks,
351 )),
352 })])
353 }
354}
355
356#[cfg(test)]
357#[allow(clippy::unwrap_used)] mod tests {
359 use super::*;
360 use crate::books::Level;
361 use rust_decimal_macros::dec;
362
363 #[test]
364 fn test_spot_order_book_times() {
365 let e0 = DateTime::from_timestamp_millis(1671656397761).unwrap();
370 let e1 = DateTime::from_timestamp_millis(1671656397800).unwrap();
371
372 let update = |last_update_id, time_exchange| BinanceSpotOrderBookL2Update {
373 subscription_id: SubscriptionId::from("@depth@100ms|ETHUSDT"),
374 time_exchange,
375 first_update_id: last_update_id,
376 last_update_id,
377 bids: vec![BinanceLevel {
378 price: dec!(100),
379 amount: dec!(1),
380 }],
381 asks: vec![],
382 };
383
384 let MarketIter(mut events) = MarketIter::<&str, OrderBookEvent>::from((
385 ExchangeId::BinanceSpot,
386 "inst",
387 update(1, e0),
388 ));
389 let event = events.remove(0).unwrap();
390 let envelope_time_received = event.time_received;
391 let OrderBookEvent::Update(mut book) = event.kind else {
392 panic!("expected Update");
393 };
394 assert_eq!(book.time_exchange(), Some(e0));
395 assert_eq!(book.time_engine(), None);
396 assert_eq!(book.time_received(), envelope_time_received);
398
399 let MarketIter(mut next) = MarketIter::<&str, OrderBookEvent>::from((
401 ExchangeId::BinanceSpot,
402 "inst",
403 update(2, e1),
404 ));
405 let event = next.remove(0).unwrap().kind;
406 book.update(&event);
407 assert_eq!(book.time_exchange(), Some(e1));
408 assert_eq!(book.time_engine(), None);
409 }
410
411 #[test]
412 fn test_de_binance_spot_order_book_l2_update() {
413 let input = r#"
414 {
415 "e":"depthUpdate",
416 "E":1671656397761,
417 "s":"ETHUSDT",
418 "U":22611425143,
419 "u":22611425151,
420 "b":[
421 ["1209.67000000","85.48210000"],
422 ["1209.66000000","20.68790000"]
423 ],
424 "a":[]
425 }
426 "#;
427
428 assert_eq!(
429 serde_json::from_str::<BinanceSpotOrderBookL2Update>(input).unwrap(),
430 BinanceSpotOrderBookL2Update {
431 subscription_id: SubscriptionId::from("@depth@100ms|ETHUSDT"),
432 time_exchange: DateTime::from_timestamp_millis(1671656397761).unwrap(),
433 first_update_id: 22611425143,
434 last_update_id: 22611425151,
435 bids: vec![
436 BinanceLevel {
437 price: dec!(1209.67000000),
438 amount: dec!(85.48210000)
439 },
440 BinanceLevel {
441 price: dec!(1209.66000000),
442 amount: dec!(20.68790000)
443 },
444 ],
445 asks: vec![]
446 }
447 );
448 }
449
450 #[test]
451 fn test_sequencer_is_first_update() {
452 struct TestCase {
453 input: BinanceSpotOrderBookL2Sequencer,
454 expected: bool,
455 }
456
457 let tests = vec![
458 TestCase {
459 input: BinanceSpotOrderBookL2Sequencer::new(10),
461 expected: true,
462 },
463 TestCase {
464 input: BinanceSpotOrderBookL2Sequencer {
466 updates_processed: 10,
467 last_update_id: 100,
468 prev_last_update_id: 90,
469 },
470 expected: false,
471 },
472 ];
473
474 for (index, test) in tests.into_iter().enumerate() {
475 assert_eq!(
476 test.input.is_first_update(),
477 test.expected,
478 "TC{} failed",
479 index
480 );
481 }
482 }
483
484 #[test]
485 fn test_sequencer_validate_first_update() {
486 struct TestCase {
487 sequencer: BinanceSpotOrderBookL2Sequencer,
488 input: BinanceSpotOrderBookL2Update,
489 expected: Result<(), DataError>,
490 }
491
492 let tests = vec![
493 TestCase {
494 sequencer: BinanceSpotOrderBookL2Sequencer {
496 updates_processed: 0,
497 last_update_id: 100,
498 prev_last_update_id: 90,
499 },
500 input: BinanceSpotOrderBookL2Update {
501 subscription_id: SubscriptionId::from("subscription_id"),
502 time_exchange: Default::default(),
503 first_update_id: 100,
504 last_update_id: 110,
505 bids: vec![],
506 asks: vec![],
507 },
508 expected: Ok(()),
509 },
510 TestCase {
511 sequencer: BinanceSpotOrderBookL2Sequencer {
513 updates_processed: 0,
514 last_update_id: 100,
515 prev_last_update_id: 90,
516 },
517 input: BinanceSpotOrderBookL2Update {
518 subscription_id: SubscriptionId::from("subscription_id"),
519 time_exchange: Default::default(),
520 first_update_id: 102,
521 last_update_id: 90,
522 bids: vec![],
523 asks: vec![],
524 },
525 expected: Err(DataError::InvalidSequence {
526 prev_last_update_id: 100,
527 first_update_id: 102,
528 }),
529 },
530 TestCase {
531 sequencer: BinanceSpotOrderBookL2Sequencer {
533 updates_processed: 0,
534 last_update_id: 100,
535 prev_last_update_id: 90,
536 },
537 input: BinanceSpotOrderBookL2Update {
538 subscription_id: SubscriptionId::from("subscription_id"),
539 time_exchange: Default::default(),
540 first_update_id: 110,
541 last_update_id: 90,
542 bids: vec![],
543 asks: vec![],
544 },
545 expected: Err(DataError::InvalidSequence {
546 prev_last_update_id: 100,
547 first_update_id: 110,
548 }),
549 },
550 TestCase {
551 sequencer: BinanceSpotOrderBookL2Sequencer {
553 updates_processed: 0,
554 last_update_id: 100,
555 prev_last_update_id: 90,
556 },
557 input: BinanceSpotOrderBookL2Update {
558 subscription_id: SubscriptionId::from("subscription_id"),
559 time_exchange: Default::default(),
560 first_update_id: 110,
561 last_update_id: 90,
562 bids: vec![],
563 asks: vec![],
564 },
565 expected: Err(DataError::InvalidSequence {
566 prev_last_update_id: 100,
567 first_update_id: 110,
568 }),
569 },
570 ];
571
572 for (index, test) in tests.into_iter().enumerate() {
573 let actual = test.sequencer.validate_first_update(&test.input);
574 match (actual, test.expected) {
575 (Ok(_), Ok(_)) => {
576 }
578 (Err(actual), Err(expected)) => {
579 assert_eq!(actual, expected, "TC{index} error variant mismatch");
580 }
581 (actual, expected) => {
582 panic!(
584 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
585 );
586 }
587 }
588 }
589 }
590
591 #[test]
592 fn test_sequencer_validate_next_update() {
593 struct TestCase {
594 sequencer: BinanceSpotOrderBookL2Sequencer,
595 input: BinanceSpotOrderBookL2Update,
596 expected: Result<(), DataError>,
597 }
598
599 let tests = vec![
600 TestCase {
601 sequencer: BinanceSpotOrderBookL2Sequencer {
603 updates_processed: 100,
604 last_update_id: 100,
605 prev_last_update_id: 100,
606 },
607 input: BinanceSpotOrderBookL2Update {
608 subscription_id: SubscriptionId::from("subscription_id"),
609 time_exchange: Default::default(),
610 first_update_id: 101,
611 last_update_id: 110,
612 bids: vec![],
613 asks: vec![],
614 },
615 expected: Ok(()),
616 },
617 TestCase {
618 sequencer: BinanceSpotOrderBookL2Sequencer {
620 updates_processed: 100,
621 last_update_id: 100,
622 prev_last_update_id: 90,
623 },
624 input: BinanceSpotOrderBookL2Update {
625 subscription_id: SubscriptionId::from("subscription_id"),
626 time_exchange: Default::default(),
627 first_update_id: 120,
628 last_update_id: 130,
629 bids: vec![],
630 asks: vec![],
631 },
632 expected: Err(DataError::InvalidSequence {
633 prev_last_update_id: 100,
634 first_update_id: 120,
635 }),
636 },
637 ];
638
639 for (index, test) in tests.into_iter().enumerate() {
640 let actual = test.sequencer.validate_next_update(&test.input);
641 match (actual, test.expected) {
642 (Ok(_), Ok(_)) => {
643 }
645 (Err(actual), Err(expected)) => {
646 assert_eq!(actual, expected, "TC{index} error variant mismatch");
647 }
648 (actual, expected) => {
649 panic!(
651 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
652 );
653 }
654 }
655 }
656 }
657
658 #[test]
659 fn test_update_rustrade_order_book_with_sequenced_updates() {
660 struct TestCase {
661 sequencer: BinanceSpotOrderBookL2Sequencer,
662 book: OrderBook,
663 input_update: BinanceSpotOrderBookL2Update,
664 expected: OrderBook,
665 }
666
667 let tests = vec![
668 TestCase {
669 sequencer: BinanceSpotOrderBookL2Sequencer {
671 updates_processed: 100,
672 last_update_id: 100,
673 prev_last_update_id: 0,
674 },
675 book: OrderBook::new(
676 100,
677 OrderBookTimes::default(),
678 vec![Level::new(50, 1)],
679 vec![Level::new(100, 1)],
680 ),
681 input_update: BinanceSpotOrderBookL2Update {
682 subscription_id: SubscriptionId::from("subscription_id"),
683 time_exchange: Default::default(),
684 first_update_id: 0,
685 last_update_id: 100, bids: vec![],
687 asks: vec![],
688 },
689 expected: OrderBook::new(
690 100,
691 OrderBookTimes::default(),
692 vec![Level::new(50, 1)],
693 vec![Level::new(100, 1)],
694 ),
695 },
696 TestCase {
697 sequencer: BinanceSpotOrderBookL2Sequencer {
699 updates_processed: 100,
700 last_update_id: 100,
701 prev_last_update_id: 100,
702 },
703 book: OrderBook::new(
704 100,
705 OrderBookTimes::default(),
706 vec![Level::new(80, 1), Level::new(100, 1), Level::new(90, 1)],
707 vec![Level::new(150, 1), Level::new(110, 1), Level::new(120, 1)],
708 ),
709 input_update: BinanceSpotOrderBookL2Update {
710 subscription_id: SubscriptionId::from("subscription_id"),
711 time_exchange: Default::default(),
712 first_update_id: 101,
713 last_update_id: 110,
714 bids: vec![
715 BinanceLevel {
717 price: dec!(80),
718 amount: dec!(0),
719 },
720 BinanceLevel {
722 price: dec!(90),
723 amount: dec!(10),
724 },
725 ],
726 asks: vec![
727 BinanceLevel {
729 price: dec!(200),
730 amount: dec!(1),
731 },
732 BinanceLevel {
734 price: dec!(500),
735 amount: dec!(0),
736 },
737 ],
738 },
739 expected: OrderBook::new(
740 110,
741 OrderBookTimes::default(),
742 vec![Level::new(100, 1), Level::new(90, 10)],
743 vec![
744 Level::new(110, 1),
745 Level::new(120, 1),
746 Level::new(150, 1),
747 Level::new(200, 1),
748 ],
749 ),
750 },
751 ];
752
753 for (index, mut test) in tests.into_iter().enumerate() {
754 if let Some(valid_update) = test.sequencer.validate_sequence(test.input_update).unwrap()
755 {
756 let rustrade_update = OrderBookEvent::Update(OrderBook::new(
757 valid_update.last_update_id,
758 OrderBookTimes::default(),
759 valid_update.bids,
760 valid_update.asks,
761 ));
762
763 test.book.update(&rustrade_update);
764 }
765
766 assert_eq!(test.book, test.expected, "TC{index} failed");
767 }
768 }
769}