rustrade-data 0.5.0

High performance & normalised WebSocket intergration for leading cryptocurrency exchanges - batteries included.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

#![allow(clippy::unwrap_used, clippy::expect_used)] // Example code: panics acceptable for demonstration

//! Live Binance klines (candles) over WebSocket, on the typed `Streams` API.
//!
//! Demonstrates:
//! - **Spot** klines via `BinanceSpot` (`@kline_<interval>` on `stream.binance.com:9443/ws`).
//! - **Futures** klines via `BinanceFuturesUsdMarket` (`@continuousKline_<interval>` on the
//!   `/market` tier `fstream.binance.com/market/ws`) — note the dedicated market-tier exchange
//!   type; routing futures klines through `BinanceFuturesUsd` would be a compile error.
//!
//! Only **closed** candles are emitted (no repaint/lookahead); an in-progress kline yields no
//! event. `close_time` is the exact period-end boundary (`open + interval`), computed
//! library-side. All endpoints are public/unauthenticated — no secrets required.

use futures_util::StreamExt;
use rustrade_data::{
    exchange::binance::{futures::BinanceFuturesUsdMarket, spot::BinanceSpot},
    streams::{Streams, reconnect::stream::ReconnectingStream},
    subscriber::WebSocketSubscriber,
    subscription::candle::{CandleInterval, Candles},
};
use rustrade_instrument::instrument::market_data::kind::MarketDataInstrumentKind;
use tracing::{info, warn};

#[rustfmt::skip]
#[tokio::main]
async fn main() {
    init_logging();

    let streams = Streams::<Candles>::builder()
        // Spot 1s candles on BinanceSpot's `/ws` tier.
        .subscribe(WebSocketSubscriber, [
            (BinanceSpot::default(), "btc", "usdt", MarketDataInstrumentKind::Spot, Candles { interval: CandleInterval::Sec1 }),
            (BinanceSpot::default(), "eth", "usdt", MarketDataInstrumentKind::Spot, Candles { interval: CandleInterval::Min1 }),
        ])
        // Futures continuous (perpetual) 1s candles on the `/market` tier.
        .subscribe(WebSocketSubscriber, [
            (BinanceFuturesUsdMarket::default(), "btc", "usdt", MarketDataInstrumentKind::Perpetual, Candles { interval: CandleInterval::Sec1 }),
        ])
        .init()
        .await
        .unwrap();

    let mut joined_stream = streams
        .select_all()
        .with_error_handler(|error| warn!(?error, "MarketStream generated error"));

    while let Some(event) = joined_stream.next().await {
        info!("{event:?}");
    }
}

// Initialise an INFO `Subscriber` for `Tracing` Json logs and install it as the global default.
fn init_logging() {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::filter::EnvFilter::builder()
                .with_default_directive(tracing_subscriber::filter::LevelFilter::INFO.into())
                .from_env_lossy(),
        )
        .with_ansi(cfg!(debug_assertions))
        .json()
        .init()
}