polyoxide_binance/usdm/
venue.rs

1//! USDⓈ-M through the venue-neutral market-data traits.
2//!
3//! [`Usdm`] implements [`MarketData`], [`Instruments`], [`Trades`], [`Candles`]
4//! and [`Funding`] over its own routes, keyed `binance.usdm:<SYMBOL>` by the
5//! constructors in `usdm::key`.
6//! What Binance alone publishes rides in this module's `<Record>Ext` types,
7//! one per record, in each record's [`Extensions`](polyoxide_venue::Extensions).
8//!
9//! `instruments` skips a contract whose `exchangeInfo` row does not convert,
10//! with a WARN naming the symbol and the error, rather than failing the whole
11//! listing.
12//!
13//! | Method | Route | Weight |
14//! | --- | --- | --- |
15//! | `instrument`, `instruments` | `exchangeInfo` | 1 |
16//! | `quote` | `depth`, 5 levels | 2 |
17//! | `book` | `depth`, no limit (500 levels) | 1 |
18//! | `trades` | `aggTrades` (the last 1000) | 20 |
19//! | `candles` | `klines`, from `start` (up to 500) | 2 |
20//! | `funding_rate` | `premiumIndex`, and `fundingInfo` for the interval | 1, and the funding bucket |
21//! | `funding_history` | `fundingRate`, from `start` (up to 1000), and `fundingInfo` for the interval | funding bucket, twice |
22//!
23//! ```no_run
24//! use std::sync::Arc;
25//!
26//! use polyoxide_binance::usdm::{self, types::Symbol, Usdm};
27//! use polyoxide_venue::{DynMarketData, MarketData};
28//!
29//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
30//! let venue: Arc<DynMarketData<'static>> = DynMarketData::new_arc(Usdm::new()?);
31//! let quote = venue.quote(&usdm::key(&Symbol::new("BTCUSDT")?)).await?;
32//! println!("bid {:?} ask {:?}", quote.bid, quote.ask);
33//! # Ok(())
34//! # }
35//! ```
36
37use std::time::Duration;
38
39use polyoxide_venue::{
40    Book, Candle, CandleInterval, CandleQuery, Candles, Class, ClassifiedError, Funding,
41    FundingRate, Instrument, Instruments, Level, MarketData, MarketKey, NotListed, Quote, Side,
42    SizeUnit, Trade, TradeId, Trades, UnixMillis,
43};
44use rust_decimal::Decimal;
45
46use crate::usdm::{
47    key,
48    types::{self, ContractType, DepthLimit, Filter, Interval, Symbol, SymbolInfo, SymbolStatus},
49    Usdm,
50};
51
52/// Binance's facts about a contract, beside [`Instrument`]'s fields.
53#[derive(Debug, Clone, PartialEq, Default)]
54#[non_exhaustive]
55pub struct InstrumentExt {
56    /// Perpetual, TradFi perpetual or a delivery contract.
57    pub contract_type: Option<ContractType>,
58    /// Trading, settling (delisted) and so on.
59    pub status: Option<SymbolStatus>,
60    /// The underlying pair, `BTCUSDT` for `BTCUSDT_261225`.
61    pub pair: Option<String>,
62    /// The margin asset.
63    pub margin_asset: Option<String>,
64    /// Delivery time; far in the future for a perpetual.
65    pub delivery_date: Option<UnixMillis>,
66    /// Listing time.
67    pub onboard_date: Option<UnixMillis>,
68    /// Every order filter, the ones [`Instrument`]'s fields come from
69    /// included.
70    pub filters: Option<Vec<Filter>>,
71}
72
73/// Binance's facts about a quote, beside [`Quote`]'s fields.
74#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
75#[non_exhaustive]
76pub struct QuoteExt {
77    /// The book update id of the snapshot.
78    pub last_update_id: Option<u64>,
79    /// When the message was produced.
80    pub event_time: Option<UnixMillis>,
81}
82
83/// Binance's facts about a book, beside [`Book`]'s fields.
84#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
85#[non_exhaustive]
86pub struct BookExt {
87    /// The book update id of the snapshot, to align it with the depth stream.
88    pub last_update_id: Option<u64>,
89    /// When the message was produced.
90    pub event_time: Option<UnixMillis>,
91}
92
93/// Binance's facts about an aggregate trade, beside [`Trade`]'s fields.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
95#[non_exhaustive]
96pub struct TradeExt {
97    /// The first trade merged into it.
98    pub first_trade_id: Option<u64>,
99    /// The last trade merged into it.
100    pub last_trade_id: Option<u64>,
101    /// The size without trades against RPI (Retail Price Improvement) orders.
102    pub normal_size: Option<Decimal>,
103}
104
105/// Binance's facts about a candle, beside [`Candle`]'s fields.
106#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
107#[non_exhaustive]
108pub struct CandleExt {
109    /// When the candle closes, inclusive.
110    pub close_time: Option<UnixMillis>,
111    /// The size takers bought.
112    pub taker_buy_volume: Option<Decimal>,
113    /// The quote-currency amount takers bought.
114    pub taker_buy_notional: Option<Decimal>,
115}
116
117/// Binance's facts about a funding rate, beside [`FundingRate`]'s fields.
118#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
119#[non_exhaustive]
120pub struct FundingExt {
121    /// The mark price: now, for the rate accruing, and at settlement for a
122    /// settled one (absent before about 2022).
123    pub mark_price: Option<Decimal>,
124    /// The index price now, for the rate accruing.
125    pub index_price: Option<Decimal>,
126    /// The estimated settle price, meaningful only in the hour before a
127    /// settlement.
128    pub estimated_settle_price: Option<Decimal>,
129    /// The interest rate, for the rate accruing.
130    pub interest_rate: Option<Decimal>,
131}
132
133fn level(level: &types::Level) -> Level {
134    Level::new(level.price, level.quantity)
135}
136
137/// The instrument a contract's row describes.
138fn instrument(info: &SymbolInfo) -> Result<Instrument, ClassifiedError> {
139    // A row whose symbol no key can carry is a response this client cannot read.
140    let symbol = Symbol::new(info.symbol.as_str())
141        .map_err(|err| ClassifiedError::new(Class::Decode, err))?;
142    let mut instrument = Instrument::new(key::key(&symbol), info.symbol.clone(), SizeUnit::Base);
143    instrument.base_asset = Some(info.base_asset.clone());
144    instrument.quote_asset = Some(info.quote_asset.clone());
145    instrument.tradable = Some(info.status == SymbolStatus::Trading);
146    for filter in &info.filters {
147        match filter {
148            Filter::PriceFilter { tick_size, .. } => instrument.tick_size = Some(*tick_size),
149            Filter::LotSize {
150                min_qty, step_size, ..
151            } => {
152                instrument.min_size = Some(*min_qty);
153                instrument.size_increment = Some(*step_size);
154            }
155            Filter::MinNotional { notional } => instrument.min_notional = Some(*notional),
156            _ => {}
157        }
158    }
159    instrument.extensions = polyoxide_venue::Extensions::of(InstrumentExt {
160        contract_type: Some(info.contract_type.clone()),
161        status: Some(info.status.clone()),
162        pair: Some(info.pair.clone()),
163        margin_asset: Some(info.margin_asset.clone()),
164        delivery_date: Some(UnixMillis(info.delivery_date)),
165        onboard_date: Some(UnixMillis(info.onboard_date)),
166        filters: Some(info.filters.clone()),
167    });
168    Ok(instrument)
169}
170
171/// The kline width a candle interval is.
172fn interval(interval: CandleInterval) -> Interval {
173    match interval {
174        CandleInterval::Minute1 => Interval::M1,
175        CandleInterval::Minute5 => Interval::M5,
176        CandleInterval::Minute15 => Interval::M15,
177        CandleInterval::Hour1 => Interval::H1,
178        CandleInterval::Hour4 => Interval::H4,
179        CandleInterval::Day1 => Interval::D1,
180        CandleInterval::Week1 => Interval::W1,
181    }
182}
183
184impl MarketData for Usdm {
185    async fn instrument(&self, key: &MarketKey) -> Result<Instrument, ClassifiedError> {
186        let symbol = key::symbol(key)?;
187        let info = self.exchange().exchange_info().send().await?;
188        match info
189            .symbols
190            .iter()
191            .find(|row| row.symbol == symbol.as_str())
192        {
193            Some(row) => instrument(row),
194            None => Err(NotListed::new(key.clone()).into()),
195        }
196    }
197
198    async fn quote(&self, key: &MarketKey) -> Result<Quote, ClassifiedError> {
199        let symbol = key::symbol(key)?;
200        let depth = self
201            .market()
202            .depth(&symbol)
203            .limit(DepthLimit::Five)
204            .send()
205            .await?;
206        let mut quote = Quote::new(key.clone());
207        quote.bid = depth.bids.first().map(level);
208        quote.ask = depth.asks.first().map(level);
209        quote.time = Some(UnixMillis(depth.transaction_time));
210        quote.extensions = polyoxide_venue::Extensions::of(QuoteExt {
211            last_update_id: Some(depth.last_update_id),
212            event_time: Some(UnixMillis(depth.event_time)),
213        });
214        Ok(quote)
215    }
216
217    async fn book(&self, key: &MarketKey) -> Result<Book, ClassifiedError> {
218        let symbol = key::symbol(key)?;
219        let depth = self.market().depth(&symbol).send().await?;
220        let mut book = Book::new(key.clone());
221        book.bids = depth.bids.iter().map(level).collect();
222        book.asks = depth.asks.iter().map(level).collect();
223        book.time = Some(UnixMillis(depth.transaction_time));
224        book.extensions = polyoxide_venue::Extensions::of(BookExt {
225            last_update_id: Some(depth.last_update_id),
226            event_time: Some(UnixMillis(depth.event_time)),
227        });
228        Ok(book)
229    }
230}
231
232impl Instruments for Usdm {
233    /// Every contract `exchangeInfo` lists. A row that does not convert (a
234    /// symbol no key can carry) is logged at WARN with its symbol and the
235    /// error, and skipped: the rest are returned.
236    async fn instruments(&self) -> Result<Vec<Instrument>, ClassifiedError> {
237        let info = self.exchange().exchange_info().send().await?;
238        Ok(info
239            .symbols
240            .iter()
241            .filter_map(|row| match instrument(row) {
242                Ok(instrument) => Some(instrument),
243                Err(err) => {
244                    tracing::warn!(symbol = %row.symbol, error = %err, "skipping an exchangeInfo row that does not convert");
245                    None
246                }
247            })
248            .collect())
249    }
250}
251
252impl Trades for Usdm {
253    async fn trades(&self, key: &MarketKey) -> Result<Vec<Trade>, ClassifiedError> {
254        let symbol = key::symbol(key)?;
255        let rows = self.market().agg_trades(&symbol).limit(1000).send().await?;
256        Ok(rows
257            .iter()
258            .map(|row| {
259                let mut trade =
260                    Trade::new(key.clone(), row.price, row.quantity, UnixMillis(row.time));
261                trade.id = Some(TradeId::new(row.id.to_string()));
262                // The buyer made the market, so the taker sold.
263                trade.taker_side = Some(if row.is_buyer_maker {
264                    Side::Sell
265                } else {
266                    Side::Buy
267                });
268                trade.extensions = polyoxide_venue::Extensions::of(TradeExt {
269                    first_trade_id: Some(row.first_trade_id),
270                    last_trade_id: Some(row.last_trade_id),
271                    normal_size: Some(row.normal_quantity),
272                });
273                trade
274            })
275            .collect())
276    }
277}
278
279impl Candles for Usdm {
280    async fn candles(
281        &self,
282        key: &MarketKey,
283        query: &CandleQuery,
284    ) -> Result<Vec<Candle>, ClassifiedError> {
285        let symbol = key::symbol(key)?;
286        let mut request = self
287            .market()
288            .klines(&symbol, interval(query.interval))
289            .start_time(query.start.0);
290        if let Some(end) = query.end {
291            request = request.end_time(end.0);
292        }
293        // 500 candles cost weight 2; no limit would cost 5.
294        let rows = request.limit(500).send().await?;
295        Ok(rows
296            .iter()
297            .map(|row| {
298                let mut candle = Candle::new(
299                    key.clone(),
300                    query.interval,
301                    UnixMillis(row.open_time),
302                    row.close,
303                );
304                candle.open = Some(row.open);
305                candle.high = Some(row.high);
306                candle.low = Some(row.low);
307                candle.volume = Some(row.volume);
308                candle.notional = Some(row.quote_volume);
309                candle.trades = Some(row.trade_count);
310                candle.extensions = polyoxide_venue::Extensions::of(CandleExt {
311                    close_time: Some(UnixMillis(row.close_time)),
312                    taker_buy_volume: Some(row.taker_buy_base_volume),
313                    taker_buy_notional: Some(row.taker_buy_quote_volume),
314                });
315                candle
316            })
317            .collect())
318    }
319}
320
321/// Binance's funding interval for a contract `fundingInfo` does not list.
322const DEFAULT_FUNDING_INTERVAL: Duration = Duration::from_secs(8 * 60 * 60);
323
324impl Usdm {
325    /// The length of the funding period `symbol` settles on now: its
326    /// `fundingIntervalHours` in `fundingInfo`, or Binance's default of 8
327    /// hours for a contract the list does not carry.
328    ///
329    /// One request on the funding bucket, which `fundingInfo` shares with
330    /// `fundingRate`.
331    async fn funding_interval(&self, symbol: &Symbol) -> Result<Duration, ClassifiedError> {
332        let rows = self.exchange().funding_info().send().await?;
333        match rows.iter().find(|row| row.symbol == symbol.as_ref()) {
334            None => Ok(DEFAULT_FUNDING_INTERVAL),
335            Some(row) if row.funding_interval_hours == 0 => Err(ClassifiedError::new(
336                Class::Decode,
337                format!("fundingInfo gives {symbol} a funding interval of 0 hours"),
338            )),
339            Some(row) => Ok(Duration::from_secs(
340                u64::from(row.funding_interval_hours) * 60 * 60,
341            )),
342        }
343    }
344}
345
346impl Funding for Usdm {
347    async fn funding_rate(&self, key: &MarketKey) -> Result<FundingRate, ClassifiedError> {
348        let symbol = key::symbol(key)?;
349        let index = self.market().premium_index(&symbol).send().await?;
350        let interval = self.funding_interval(&symbol).await?;
351        let mut rate =
352            FundingRate::new(key.clone(), index.last_funding_rate, UnixMillis(index.time));
353        rate.interval = Some(interval);
354        // `0` is a contract with no funding scheduled.
355        rate.next_funding =
356            (index.next_funding_time > 0).then_some(UnixMillis(index.next_funding_time));
357        rate.extensions = polyoxide_venue::Extensions::of(FundingExt {
358            mark_price: Some(index.mark_price),
359            index_price: Some(index.index_price),
360            estimated_settle_price: Some(index.estimated_settle_price),
361            interest_rate: Some(index.interest_rate),
362        });
363        Ok(rate)
364    }
365
366    async fn funding_history(
367        &self,
368        key: &MarketKey,
369        start: UnixMillis,
370    ) -> Result<Vec<FundingRate>, ClassifiedError> {
371        let symbol = key::symbol(key)?;
372        let rows = self
373            .market()
374            .funding_rate()
375            .symbol(&symbol)
376            .start_time(start.0)
377            .limit(1000)
378            .send()
379            .await?;
380        // The symbol's current interval, for every row: `fundingInfo` keeps no
381        // history of it.
382        let interval = self.funding_interval(&symbol).await?;
383        Ok(rows
384            .iter()
385            .map(|row| {
386                let mut rate =
387                    FundingRate::new(key.clone(), row.funding_rate, UnixMillis(row.funding_time));
388                rate.interval = Some(interval);
389                rate.extensions = polyoxide_venue::Extensions::of(FundingExt {
390                    mark_price: row.mark_price,
391                    ..FundingExt::default()
392                });
393                rate
394            })
395            .collect())
396    }
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    #[test]
404    fn every_candle_interval_is_a_kline_width() {
405        let widths: Vec<&str> = CandleInterval::ALL
406            .iter()
407            .map(|each| interval(*each).as_str())
408            .collect();
409        assert_eq!(widths, ["1m", "5m", "15m", "1h", "4h", "1d", "1w"]);
410    }
411
412    #[test]
413    fn a_row_whose_symbol_no_key_carries_is_a_decode_failure() {
414        let info: types::ExchangeInfo = serde_json::from_str(include_str!(
415            "../../tests/fixtures/usdm/rest/exchange_info.json"
416        ))
417        .unwrap();
418        let mut row = info.symbols[0].clone();
419        row.symbol = "BTC/USDT".to_owned();
420        let err = instrument(&row).unwrap_err();
421        assert_eq!(err.class, Class::Decode);
422    }
423}
polyoxide_binance/usdm/venue.rs

polyoxide_binance/usdm/
venue.rs
