Restore liquidity-aware weighting for centralized exchange prices

src/sources/binance.rs

@@ -85,12 +85,13 @@ impl BinanceSource {
         };
         let token = &stream.token;
         let value = Decimal::from_str(&message.data.price)?;
+        let volume = Decimal::from_str(&message.data.volume_base)?;
 
         sink.send(PriceInfo {
             token: token.to_string(),
             unit: stream.unit.to_string(),
             value,
-            reliability: Decimal::ONE,
+            reliability: volume,
         })?;
 
         Ok(())
@@ -106,4 +107,6 @@ struct BinanceMarkPriceMessage {
 struct BinanceMarkPriceMessageData {
     #[serde(rename(deserialize = "c"))]
     price: String,
+    #[serde(rename(deserialize = "v"))]
+    volume_base: String,
 }

src/sources/bybit.rs

@@ -21,13 +21,15 @@ struct ByBitPriceInfo {
     token: String,
     unit: String,
     last_value: Option<Decimal>,
+    last_volume: Option<Decimal>,
 }
 impl ByBitPriceInfo {
     fn new(token: &str, unit: &str) -> Self {
         Self {
             token: token.to_string(),
             unit: unit.to_string(),
             last_value: None,
+            last_volume: None,
         }
     }
 }
@@ -129,11 +131,20 @@ impl ByBitSource {
                 };
                 info.last_value = Some(value);
 
+                let Some(volume) = data
+                    .volume_24h
+                    .and_then(|x| Decimal::from_str(&x).ok())
+                    .or(info.last_volume)
+                else {
+                    continue;
+                };
+                info.last_volume = Some(volume);
+
                 let price_info = PriceInfo {
                     token: info.token.clone(),
                     unit: info.unit.clone(),
                     value,
-                    reliability: Decimal::ONE,
+                    reliability: volume,
                 };
                 sink.send(price_info)?;
             }
@@ -174,4 +185,5 @@ enum ByBitResponse {
 struct TickerSnapshotData {
     symbol: String,
     mark_price: Option<String>,
+    volume_24h: Option<String>,
 }

src/sources/coinbase.rs

@@ -83,18 +83,24 @@ impl CoinbaseSource {
 
     fn parse_message(&self, message: Message) -> Result<PriceInfo> {
         let response: CoinbaseResponse = message.clone().try_into()?;
-        let CoinbaseResponse::Ticker { product_id, price } = response else {
+        let CoinbaseResponse::Ticker {
+            product_id,
+            price,
+            volume_24h,
+        } = response
+        else {
             return Err(anyhow!("Unexpected response from coinbase: {:?}", response));
         };
         let Some(product) = self.products.get(&product_id) else {
             return Err(anyhow!("Unrecognized price to match: {}", product_id));
         };
         let value = Decimal::from_str(&price)?;
+        let volume = Decimal::from_str(&volume_24h)?;
         Ok(PriceInfo {
             token: product.token.clone(),
             unit: product.unit.clone(),
             value,
-            reliability: Decimal::ONE,
+            reliability: volume,
         })
     }
 }
@@ -127,6 +133,7 @@ enum CoinbaseResponse {
     Ticker {
         product_id: String,
         price: String,
+        volume_24h: String,
     },
 }
 

src/sources/crypto_com.rs

@@ -107,11 +107,12 @@ impl CryptoComSource {
             }
             let data = &result.data[0];
             let value = Decimal::from_str(&data.best_bid_price)?;
+            let volume = Decimal::from_str(&data.volume_24h)?;
             sink.send(PriceInfo {
                 token: stream.token.clone(),
                 unit: stream.unit.clone(),
                 value,
-                reliability: Decimal::ONE,
+                reliability: volume,
             })?;
         }
 
@@ -158,4 +159,6 @@ struct CryptoComResponseResult {
 struct CryptoComResponseData {
     #[serde(rename = "b")]
     best_bid_price: String,
+    #[serde(rename = "v")]
+    volume_24h: String,
 }

src/sources/kucoin.rs

@@ -3,7 +3,6 @@ use std::{collections::HashMap, time::Duration};
 use anyhow::{Context, Result, bail};
 use futures::{FutureExt as _, SinkExt, StreamExt, future::BoxFuture};
 use rand::{RngCore, thread_rng};
-use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use tokio::{
     net::TcpStream,
@@ -97,7 +96,7 @@ impl KucoinSource {
                     token: symbol.token.clone(),
                     unit: symbol.unit.clone(),
                     value: data.data.buy.try_into()?,
-                    reliability: Decimal::ONE,
+                    reliability: data.data.vol.try_into()?,
                 })?;
             }
             bail!("Kucoin stream has closed")
@@ -163,6 +162,7 @@ struct KucoinMessage {
 struct KucoinResponseData {
     symbol: String,
     buy: f64,
+    vol: f64,
 }
 
 #[derive(Deserialize)]

src/sources/maestro.rs

@@ -1,7 +1,6 @@
 use anyhow::{Result, anyhow};
 use futures::{FutureExt, future::BoxFuture};
 use reqwest::Client;
-use rust_decimal::Decimal;
 use serde::Deserialize;
 use std::{env, sync::Arc, time::Duration};
 use tokio::{task::JoinSet, time::sleep};
@@ -100,12 +99,13 @@ impl MaestroSource {
         let contents = response.text().await?;
         let messages: [MaestroOHLCMessage; 1] = serde_json::from_str(&contents)?;
         let res = (messages[0].coin_a_open + messages[0].coin_a_close) / 2.;
+        let volume = messages[0].coin_a_volume;
 
         sink.send(PriceInfo {
             token: config.token.to_string(),
             unit: config.unit.to_string(),
             value: res.try_into()?,
-            reliability: Decimal::ONE,
+            reliability: volume.try_into()?,
         })?;
         Ok(())
     }
@@ -115,4 +115,5 @@ impl MaestroSource {
 struct MaestroOHLCMessage {
     coin_a_open: f64,
     coin_a_close: f64,
+    coin_a_volume: f64,
 }