feat: use order book in quotes server

GaloyMoney · Oct 3, 2023 · 990ea90 · 990ea90
1 parent 8121681
commit 990ea90
Show file tree

Hide file tree

Showing 6 changed files with 477 additions and 21 deletions.
diff --git a/quotes-server/src/app/mod.rs b/quotes-server/src/app/mod.rs
@@ -2,7 +2,7 @@ mod config;
 
 use futures::stream::StreamExt;
 use rust_decimal::Decimal;
-use tracing::{info_span, instrument, Instrument};
+use tracing::{info_span, Instrument};
 
 use shared::{
     health::HealthCheckTrigger,
@@ -41,9 +41,10 @@ impl QuotesApp {
 
         if let Some(weight) = exchange_weights.okex {
             if weight > Decimal::ZERO {
-                let okex_price_cache = ExchangeTickCache::new(price_cache_config.clone());
-                Self::subscribe_okex(subscriber.resubscribe(), okex_price_cache.clone()).await?;
-                price_mixer.add_provider(OKEX_EXCHANGE_ID, okex_price_cache, weight);
+                let okex_order_book_cache = OrderBookCache::new(price_cache_config.clone());
+                Self::subscribe_okex(subscriber.resubscribe(), okex_order_book_cache.clone())
+                    .await?;
+                price_mixer.add_provider(OKEX_EXCHANGE_ID, okex_order_book_cache, weight);
             }
         }
 
@@ -66,11 +67,11 @@ impl QuotesApp {
         sats: Decimal,
         immediate_execution: bool,
     ) -> Result<Quote, QuotesAppError> {
-        let usd_amount = self
+        let _usd_amount = self
             .price_calculator
             .cents_from_sats_for_buy(Satoshis::from(sats), immediate_execution)
             .await?;
-        let quote = NewQuote::builder()
+        let _quote = NewQuote::builder()
             .direction(Direction::BuyCents)
             .immediate_execution(immediate_execution)
             .build()
@@ -80,21 +81,19 @@ impl QuotesApp {
 
     async fn subscribe_okex(
         mut subscriber: memory::Subscriber<PriceStreamPayload>,
-        price_cache: ExchangeTickCache,
+        order_book_cache: OrderBookCache,
     ) -> Result<(), QuotesAppError> {
         tokio::spawn(async move {
             while let Some(msg) = subscriber.next().await {
-                if let PriceStreamPayload::OkexBtcSwapPricePayload(price_msg) = msg.payload {
+                if let PriceStreamPayload::OkexBtcUsdSwapOrderBookPayload(price_msg) = msg.payload {
                     let span = info_span!(
-                        "price_server.okex_price_tick_received",
+                        "price_server.okex_order_book_received",
                         message_type = %msg.payload_type,
                         correlation_id = %msg.meta.correlation_id
                     );
                     shared::tracing::inject_tracing_data(&span, &msg.meta.tracing_data);
                     async {
-                        price_cache
-                            .apply_update(price_msg, msg.meta.correlation_id)
-                            .await;
+                        order_book_cache.apply_update(price_msg).await;
                     }
                     .instrument(span)
                     .await;

diff --git a/quotes-server/src/cache/mod.rs b/quotes-server/src/cache/mod.rs
@@ -1,4 +1,5 @@
 mod config;
+mod order_book_cache;
 
 use chrono::Duration;
 use opentelemetry::trace::{SpanContext, TraceContextExt};
@@ -12,6 +13,7 @@ use crate::{currency::*, price::*};
 use shared::{payload::*, pubsub::CorrelationId, time::*};
 
 pub use config::*;
+pub use order_book_cache::*;
 
 #[derive(Clone)]
 pub struct ExchangeTickCache {
@@ -38,15 +40,7 @@ impl ExchangeTickCache {
 impl PriceProvider for ExchangeTickCache {
     async fn latest(&self) -> Result<Box<dyn SidePicker>, ExchangePriceCacheError> {
         if let Some(mock_price) = self.config.dev_mock_price_btc_in_usd {
-            let price = PriceRatioRaw::from_one_btc_in_usd_price(mock_price);
-            let cent_price = UsdCents::try_from(price).expect("couldn't create mack UsdCents");
-            return Ok(Box::new(BtcSatTick {
-                timestamp: TimeStamp::now(),
-                correlation_id: CorrelationId::new(),
-                span_context: Span::current().context().span().span_context().clone(),
-                ask_price_of_one_sat: cent_price.clone(),
-                bid_price_of_one_sat: cent_price,
-            }));
+            return Ok(Box::new(mock_price_tick(mock_price)));
         }
         let inner = self.inner.read().await;
         let tick = inner.latest_tick()?;
@@ -61,6 +55,18 @@ impl PriceProvider for ExchangeTickCache {
     }
 }
 
+pub fn mock_price_tick(mock_price: rust_decimal::Decimal) -> BtcSatTick {
+    let price = PriceRatioRaw::from_one_btc_in_usd_price(mock_price);
+    let cent_price = UsdCents::try_from(price).expect("couldn't create mock UsdCents");
+    BtcSatTick {
+        timestamp: TimeStamp::now(),
+        correlation_id: CorrelationId::new(),
+        span_context: Span::current().context().span().span_context().clone(),
+        ask_price_of_one_sat: cent_price.clone(),
+        bid_price_of_one_sat: cent_price,
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct BtcSatTick {
     timestamp: TimeStamp,