From f158bc0482a53d8d9aaa264884bd78ba387705f6 Mon Sep 17 00:00:00 2001 From: Micah <66159982+micahkendall@users.noreply.github.com> Date: Mon, 4 May 2026 21:30:26 +1000 Subject: [PATCH] Restore volume-based reliability weighting for CEX feeds --- src/sources/binance.rs | 5 ++++- src/sources/bybit.rs | 14 +++++++++++++- src/sources/coinbase.rs | 11 +++++++++-- src/sources/crypto_com.rs | 5 ++++- src/sources/kucoin.rs | 4 ++-- src/sources/maestro.rs | 5 +++-- 6 files changed, 35 insertions(+), 9 deletions(-) diff --git a/src/sources/binance.rs b/src/sources/binance.rs index 532ebb1..5436603 100644 --- a/src/sources/binance.rs +++ b/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, } diff --git a/src/sources/bybit.rs b/src/sources/bybit.rs index f9206e2..2032066 100644 --- a/src/sources/bybit.rs +++ b/src/sources/bybit.rs @@ -21,6 +21,7 @@ struct ByBitPriceInfo { token: String, unit: String, last_value: Option, + last_volume: Option, } impl ByBitPriceInfo { fn new(token: &str, unit: &str) -> Self { @@ -28,6 +29,7 @@ impl ByBitPriceInfo { 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, + volume_24h: Option, } diff --git a/src/sources/coinbase.rs b/src/sources/coinbase.rs index 9605452..f63ed55 100644 --- a/src/sources/coinbase.rs +++ b/src/sources/coinbase.rs @@ -83,18 +83,24 @@ impl CoinbaseSource { fn parse_message(&self, message: Message) -> Result { 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, }, } diff --git a/src/sources/crypto_com.rs b/src/sources/crypto_com.rs index 1485c09..17c6903 100644 --- a/src/sources/crypto_com.rs +++ b/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, } diff --git a/src/sources/kucoin.rs b/src/sources/kucoin.rs index 16389ca..9c68832 100644 --- a/src/sources/kucoin.rs +++ b/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)] diff --git a/src/sources/maestro.rs b/src/sources/maestro.rs index 2b2d2de..e89f7fb 100644 --- a/src/sources/maestro.rs +++ b/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, }