use super::MarketSummary;
use super::analytics::order_book_from_analytics;
use super::exchange::ExchangeClient;
use super::health::HealthThresholds;
use super::pair::Pair;
use super::types::{OrderBook, Trade};
use super::venue::{DexVenue, MarketVenue};
use crate::chains::{DexPair, TokenAnalytics};
use crate::error::{Result, ScopeError};
use async_trait::async_trait;
#[async_trait]
pub trait DexAnalyticsSource: Send + Sync {
async fn token_analytics(&self, token: &str, chain: &str) -> Result<TokenAnalytics>;
}
struct KnownAnalytics<'a>(&'a TokenAnalytics);
#[async_trait]
impl DexAnalyticsSource for KnownAnalytics<'_> {
async fn token_analytics(&self, _token: &str, _chain: &str) -> Result<TokenAnalytics> {
Ok(self.0.clone())
}
}
pub async fn fetch_book_and_volume(
venue: &MarketVenue,
pair: &Pair,
thresholds: &HealthThresholds,
dex: &dyn DexAnalyticsSource,
) -> Result<(OrderBook, Option<f64>)> {
match venue {
MarketVenue::Dex(dex_venue) => {
let analytics = dex.token_analytics(&pair.base, dex_venue.chain()).await?;
dex_book_and_volume(*dex_venue, &analytics, thresholds)
}
MarketVenue::Cex(exchange) => cex_book_and_volume(exchange, pair).await,
}
}
pub async fn fetch_summary(
venue: &MarketVenue,
pair: &Pair,
thresholds: &HealthThresholds,
dex: &dyn DexAnalyticsSource,
) -> Result<MarketSummary> {
let (book, volume_24h) = fetch_book_and_volume(venue, pair, thresholds, dex).await?;
let mut summary = MarketSummary::from_order_book(&book, thresholds, volume_24h);
let trades = fetch_pattern_trades(venue, pair).await;
summary.check_bot_patterns(trades.as_deref().map_err(String::as_str));
Ok(summary)
}
pub async fn fetch_pattern_trades(
venue: &MarketVenue,
pair: &Pair,
) -> std::result::Result<Vec<Trade>, String> {
match venue {
MarketVenue::Dex(_) => Err("a DEX pool has no trade tape here".to_string()),
MarketVenue::Cex(exchange) if !exchange.has_trade_history() => {
Err("the venue gives no trade feed".to_string())
}
MarketVenue::Cex(exchange) => exchange
.fetch_recent_trades(&exchange.venue_symbol(pair), super::patterns::TRADE_SAMPLE)
.await
.map_err(|e| format!("the trade feed failed: {e}")),
}
}
pub async fn fetch_token_summary(
venue: &MarketVenue,
analytics: &TokenAnalytics,
thresholds: &HealthThresholds,
) -> Result<MarketSummary> {
let pair = Pair::base_only(&analytics.token.symbol);
fetch_summary(venue, &pair, thresholds, &KnownAnalytics(analytics)).await
}
#[deprecated(
since = "0.8.0",
note = "pass the transport: use `fetch_token_summary_by_name_with_http`"
)]
pub async fn fetch_token_summary_by_name(
venue: &str,
analytics: &TokenAnalytics,
thresholds: &HealthThresholds,
) -> Result<MarketSummary> {
fetch_token_summary_by_name_with_http(
venue,
analytics,
thresholds,
crate::http::shared::fallback(),
)
.await
}
pub async fn fetch_token_summary_by_name_with_http(
venue: &str,
analytics: &TokenAnalytics,
thresholds: &HealthThresholds,
http: std::sync::Arc<dyn crate::http::HttpClient>,
) -> Result<MarketSummary> {
let venue = MarketVenue::from_name_with_http(venue, http)?;
fetch_token_summary(&venue, analytics, thresholds).await
}
fn dex_book_and_volume(
venue: DexVenue,
analytics: &TokenAnalytics,
thresholds: &HealthThresholds,
) -> Result<(OrderBook, Option<f64>)> {
venue.require_chain(&analytics.chain)?;
let best = most_liquid(&analytics.dex_pairs).ok_or_else(|| {
ScopeError::NotFound(format!(
"No DEX pairs found for {} on {}",
analytics.token.symbol,
venue.chain()
))
})?;
let book = order_book_from_analytics(venue.chain(), best, &analytics.token.symbol, thresholds);
Ok((book, Some(best.volume_24h)))
}
pub fn most_liquid(pairs: &[DexPair]) -> Option<&DexPair> {
pairs
.iter()
.max_by(|a, b| a.liquidity_usd.total_cmp(&b.liquidity_usd))
}
async fn cex_book_and_volume(
exchange: &ExchangeClient,
pair: &Pair,
) -> Result<(OrderBook, Option<f64>)> {
let symbol = exchange.venue_symbol(pair);
let book = exchange.fetch_order_book(&symbol).await?;
let volume = if exchange.has_ticker() {
exchange
.fetch_ticker(&symbol)
.await
.ok()
.and_then(|t| t.quote_volume(book.mid_price()))
} else {
None
};
Ok((book, volume))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::chains::Token;
use crate::market::types::{BookSource, OrderBookLevel};
async fn binance_with_bot_tape(n: usize) -> (mockito::ServerGuard, MarketVenue) {
let mut server = mockito::Server::new_async().await;
let trades: Vec<String> = (0..n)
.map(|i| {
format!(
r#"{{"id":{i},"price":"1.0001","qty":"1000.00","quoteQty":"1000.1","time":{},"isBuyerMaker":{}}}"#,
1_700_000_000_000u64 + i as u64 * 2_000,
i % 2 == 0
)
})
.collect();
server
.mock("GET", mockito::Matcher::Regex(r"^/api/v3/trades.*".into()))
.with_header("content-type", "application/json")
.with_body(format!("[{}]", trades.join(",")))
.create_async()
.await;
let venue = MarketVenue::Cex(Box::new(crate::market::testing::venue_at(
"binance",
&server.url(),
)));
(server, venue)
}
#[tokio::test]
async fn test_a_venue_tape_feeds_the_bot_pattern_advisory() {
let (_server, venue) = binance_with_bot_tape(60).await;
let trades = fetch_pattern_trades(&venue, &Pair::parse("USDC"))
.await
.unwrap();
assert_eq!(trades.len(), 60);
let book = OrderBook {
pair: "USDC/USDT".into(),
source: BookSource::Exchange,
bids: vec![OrderBookLevel {
price: 0.9999,
quantity: 100.0,
}],
asks: vec![OrderBookLevel {
price: 1.0001,
quantity: 120.0,
}],
};
let mut summary = MarketSummary::from_order_book(&book, &HealthThresholds::default(), None);
let healthy = summary.healthy;
summary.check_bot_patterns(Ok(&trades));
assert_eq!(
summary.healthy, healthy,
"an advisory never changes healthy"
);
let last = summary.checks.last().unwrap();
assert_eq!(last.status(), "warn", "{}", last.message());
assert!(
last.message()
.contains("100% of 60 trades are exactly 1000"),
"{}",
last.message()
);
assert!(last.message().contains("every 2.0s"), "{}", last.message());
let json = summary.to_json();
assert_eq!(json["bot_patterns"]["trades_sampled"], 60);
assert_eq!(json["bot_patterns"]["findings"][0]["kind"], "repeated_size");
}
#[tokio::test]
async fn test_a_failed_trade_feed_is_a_reason_not_an_error() {
let mut server = mockito::Server::new_async().await;
server
.mock("GET", mockito::Matcher::Regex(r"^/api/v3/trades.*".into()))
.with_status(500)
.create_async()
.await;
let venue = MarketVenue::Cex(Box::new(crate::market::testing::venue_at(
"binance",
&server.url(),
)));
let why = fetch_pattern_trades(&venue, &Pair::parse("USDC"))
.await
.unwrap_err();
assert!(why.starts_with("the trade feed failed:"), "{why}");
}
fn dex_pair(liquidity_usd: f64, volume_24h: f64) -> DexPair {
DexPair {
dex_name: "DEX".into(),
pair_address: "0xpair".into(),
base_token: "USDC".into(),
quote_token: "USDT".into(),
price_usd: 1.0,
volume_24h,
liquidity_usd,
liquidity_base: None,
price_change_24h: 0.0,
buys_24h: 0,
sells_24h: 0,
buys_6h: 0,
sells_6h: 0,
buys_1h: 0,
sells_1h: 0,
pair_created_at: None,
url: None,
}
}
fn analytics(chain: &str, pairs: Vec<DexPair>) -> TokenAnalytics {
TokenAnalytics {
token: Token {
contract_address: "0x".into(),
symbol: "USDC".into(),
name: "USD Coin".into(),
decimals: 6,
},
chain: chain.into(),
holders: vec![],
total_holders: 0,
volume_24h: 0.0,
volume_7d: 0.0,
price_usd: 1.0,
price_change_24h: 0.0,
price_change_7d: 0.0,
liquidity_usd: 0.0,
market_cap: None,
fdv: None,
total_supply: None,
circulating_supply: None,
price_history: vec![],
volume_history: vec![],
holder_history: vec![],
dex_pairs: pairs,
fetched_at: 0,
top_10_concentration: None,
top_50_concentration: None,
top_100_concentration: None,
price_change_6h: 0.0,
price_change_1h: 0.0,
total_buys_24h: 0,
total_sells_24h: 0,
total_buys_6h: 0,
total_sells_6h: 0,
total_buys_1h: 0,
total_sells_1h: 0,
token_age_hours: None,
image_url: None,
websites: vec![],
socials: vec![],
dexscreener_url: None,
}
}
#[tokio::test]
async fn dex_book_comes_from_the_most_liquid_pool() {
let a = analytics(
"ethereum",
vec![
dex_pair(1_000.0, 10.0),
dex_pair(50_000.0, 777.0),
dex_pair(2_000.0, 20.0),
],
);
let (book, volume) = fetch_book_and_volume(
&MarketVenue::Dex(DexVenue::Ethereum),
&Pair::parse("USDC"),
&HealthThresholds::default(),
&KnownAnalytics(&a),
)
.await
.unwrap();
assert_eq!(volume, Some(777.0));
assert_eq!(book.pair, "USDC/USD");
}
#[tokio::test]
async fn dex_without_pools_is_an_error() {
let a = analytics("solana", vec![]);
let err = fetch_summary(
&MarketVenue::Dex(DexVenue::Solana),
&Pair::parse("USDC"),
&HealthThresholds::default(),
&KnownAnalytics(&a),
)
.await
.unwrap_err();
assert!(
err.to_string()
.contains("No DEX pairs found for USDC on solana"),
"{err}"
);
assert!(matches!(err, ScopeError::NotFound(_)), "{err:?}");
}
#[tokio::test]
async fn fetch_token_summary_by_name_resolves_then_fetches() {
let a = analytics("ethereum", vec![dex_pair(1_000.0, 10.0)]);
let t = HealthThresholds::default();
assert!(
fetch_token_summary_by_name_with_http("eth", &a, &t, crate::http::test_transport())
.await
.is_ok()
);
let err =
fetch_token_summary_by_name_with_http("polygon", &a, &t, crate::http::test_transport())
.await
.unwrap_err();
assert!(matches!(err, ScopeError::InvalidInput(_)), "{err:?}");
assert!(err.to_string().contains("Unknown venue"), "{err}");
}
#[tokio::test]
async fn dex_analytics_from_another_chain_are_an_error() {
let a = analytics("Ethereum", vec![dex_pair(1_000.0, 1.0)]);
let venue = MarketVenue::Dex(DexVenue::Solana);
let pair = Pair::parse("USDC");
let t = HealthThresholds::default();
let err = fetch_summary(&venue, &pair, &t, &KnownAnalytics(&a))
.await
.unwrap_err();
assert!(err.to_string().contains("reads chain solana"), "{err}");
let same_chain = MarketVenue::Dex(DexVenue::Ethereum);
assert!(
fetch_summary(&same_chain, &pair, &t, &KnownAnalytics(&a))
.await
.is_ok()
);
}
}