rustrade-data 0.11.0

#![forbid(unsafe_code)]
#![deny(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    // A public item linking to a private one renders as dead, non-hyperlinked code on docs.rs — the
    // reader is pointed at something they cannot open. Denied rather than warned because these
    // accumulate silently: `cargo doc` is not part of the normal build or the pre-commit gate, so a
    // warning here is one nobody sees. Fix by inlining the fact or linking a public item instead.
    rustdoc::private_intra_doc_links
)]
#![warn(
    unused,
    clippy::cognitive_complexity,
    unused_crate_dependencies,
    unused_extern_crates,
    clippy::unused_self,
    clippy::useless_let_if_seq,
    missing_debug_implementations,
    rust_2018_idioms
)]
// Inherited from the barter-rs upstream: the generic, trait-heavy public API legitimately produces
// complex nested types and many-argument constructors, and `type_alias_bounds` fires on
// documentation-bearing alias bounds kept intentionally. Denying these adds churn without improving
// the API.
#![allow(clippy::type_complexity, clippy::too_many_arguments, type_alias_bounds)]
// Clippy 1.99 checks each field's span rather than the whole expression's, so the
// `Self { field: field }` that `derive_more::Constructor` expands to is reported at our own
// field declarations. Remove once rust-lang/rust-clippy#17525 is fixed in a stable release.
#![allow(clippy::redundant_field_names)]

//! # Barter-Data
//! A high-performance WebSocket integration library for streaming public market data from leading cryptocurrency
//! exchanges - batteries included. It is:
//! * **Easy**: Barter-Data's simple [`StreamBuilder`](streams::builder::StreamBuilder) and [`DynamicStreams`](streams::builder::dynamic::DynamicStreams) interface allows for easy & quick setup (see example below and /examples!).
//! * **Normalised**: Barter-Data's unified interface for consuming public WebSocket data means every Exchange returns a normalised data model.
//! * **Real-Time**: Barter-Data utilises real-time WebSocket integrations enabling the consumption of normalised tick-by-tick data.

//! * **Extensible**: Barter-Data is highly extensible, and therefore easy to contribute to with coding new integrations!
//!
//! ## User API
//! - [`StreamBuilder`](streams::builder::StreamBuilder) for initialising [`MarketStream`]s of specific data kinds.
//! - [`DynamicStreams`](streams::builder::dynamic::DynamicStreams) for initialising [`MarketStream`]s of every supported data kind at once.
//! - Define what exchange market data you want to stream using the [`Subscription`] type.
//! - Pass [`Subscription`]s to the [`StreamBuilder::subscribe`](streams::builder::StreamBuilder::subscribe) or [`DynamicStreams::init`](streams::builder::dynamic::DynamicStreams::init) methods.
//! - Each call to the [`StreamBuilder::subscribe`](streams::builder::StreamBuilder::subscribe) (or each batch passed to the [`DynamicStreams::init`](streams::builder::dynamic::DynamicStreams::init))
//!   method opens a new WebSocket connection to the exchange - giving you full control - unless its
//!   subscriber shares one connection among its clones (see [`Subscriber::Transport`]).
//!
//! ## Examples
//! For a comprehensive collection of examples, see the /examples directory.
//!
//! ### Multi Exchange Public Trades
//! ```rust,no_run
//! use rustrade_data::{
//!     exchange::{
//!         gateio::spot::GateioSpot,
//!         binance::{futures::BinanceFuturesUsd, spot::BinanceSpot},
//!         coinbase::Coinbase,
//!         okx::Okx,
//!     },
//!     streams::{Streams, reconnect::stream::ReconnectingStream},
//!     subscriber::WebSocketSubscriber,
//!     subscription::trade::PublicTrades,
//! };
//! use rustrade_instrument::instrument::market_data::kind::MarketDataInstrumentKind;
//! use futures::StreamExt;
//! use tracing::warn;
//!
//! #[tokio::main]
//! async fn main() {
//!     // Initialise PublicTrades Streams for various exchanges
//!     // '--> each call to StreamBuilder::subscribe() initialises a separate WebSocket connection
//!
//!     let streams = Streams::<PublicTrades>::builder()
//!         .subscribe(WebSocketSubscriber, [
//!             (BinanceSpot::default(), "btc", "usdt", MarketDataInstrumentKind::Spot, PublicTrades),
//!             (BinanceSpot::default(), "eth", "usdt", MarketDataInstrumentKind::Spot, PublicTrades),
//!         ])
//!         .subscribe(WebSocketSubscriber, [
//!             (BinanceFuturesUsd::default(), "btc", "usdt", MarketDataInstrumentKind::Perpetual, PublicTrades),
//!             (BinanceFuturesUsd::default(), "eth", "usdt", MarketDataInstrumentKind::Perpetual, PublicTrades),
//!         ])
//!         .subscribe(WebSocketSubscriber, [
//!             (Coinbase, "btc", "usd", MarketDataInstrumentKind::Spot, PublicTrades),
//!             (Coinbase, "eth", "usd", MarketDataInstrumentKind::Spot, PublicTrades),
//!         ])
//!         .subscribe(WebSocketSubscriber, [
//!             (GateioSpot::default(), "btc", "usdt", MarketDataInstrumentKind::Spot, PublicTrades),
//!             (GateioSpot::default(), "eth", "usdt", MarketDataInstrumentKind::Spot, PublicTrades),
//!         ])
//!         .subscribe(WebSocketSubscriber, [
//!             (Okx, "btc", "usdt", MarketDataInstrumentKind::Spot, PublicTrades),
//!             (Okx, "eth", "usdt", MarketDataInstrumentKind::Spot, PublicTrades),
//!             (Okx, "btc", "usdt", MarketDataInstrumentKind::Perpetual, PublicTrades),
//!             (Okx, "eth", "usdt", MarketDataInstrumentKind::Perpetual, PublicTrades),
//!        ])
//!         .init()
//!         .await
//!         .unwrap();
//!
//!     // Select and merge every exchange Stream using futures_util::stream::select_all
//!     // Note: use `Streams.select(ExchangeId)` to interact with individual exchange streams!
//!     let mut joined_stream = streams
//!         .select_all()
//!         .with_error_handler(|error| warn!(?error, "MarketStream generated error"));
//!
//!     while let Some(event) = joined_stream.next().await {
//!         println!("{event:?}");
//!     }
//! }
//! ```

// Silence unused_crate_dependencies for dev-dependencies used only in tests
#[cfg(test)]
use tracing_subscriber as _;
// serial_test is only referenced by integration tests under the `massive` feature
// (tests/massive_integration.rs), which compile as separate units.
#[cfg(test)]
use serial_test as _;
// wiremock is referenced by the pagination integration tests under the `massive` feature
// (tests/massive_pagination.rs, a separate compilation unit) and by `exchange::http`'s unit tests,
// which are no longer feature-gated and so reference it in every `cfg(test)` build. This stub is
// therefore redundant today; it is kept alongside its siblings so re-gating those tests cannot
// silently reintroduce the warning.
#[cfg(test)]
use wiremock as _;
// time is only referenced by the integration tests under the `databento`/`ibkr` features
// (tests/databento_integration.rs, tests/ibkr_integration.rs) and by a databento example, which
// compile as separate units.
#[cfg(test)]
use time as _;
// temp_env is only referenced by the in-tree unit tests under the `massive` feature
// (exchange::massive::rest), so it is unused when that feature is off.
#[cfg(test)]
use temp_env as _;
// http is only referenced by the in-tree unit tests under the `ibkr` feature
// (exchange::ibkr::flex), which build synthetic `reqwest::Response`s without a socket.
#[cfg(test)]
use http as _;
// tempfile, sha2 and hex are only referenced by the export-job integration tests under the `lse`
// feature (tests/lse_export.rs), which compiles as a separate unit: tempfile supplies a scratch
// directory for the download/resume cases, and sha2/hex compute the digest the synthetic job
// records carry. (sha2 and hex are also non-dev dependencies under `lse`, so their stubs matter
// only when that feature is off.)
#[cfg(test)]
use {hex as _, sha2 as _, tempfile as _};
// tokio_tungstenite is a dev-dependency, for the tests that speak the server half of a WebSocket
// protocol in-process: the shared-connection actor's unit tests, and the integration tests
// (tests/lse_ws_handshake.rs, tests/alpaca_ws_shared.rs, tests/massive_ws_shared.rs), which compile
// as separate units. The `cfg(test)` gate is load-bearing: it is not a dependency of a non-test
// build, so an unconditional `use` would fail to resolve.
#[cfg(test)]
use tokio_tungstenite as _;

use crate::{
    error::DataError,
    event::MarketEvent,
    exchange::{Connector, PingInterval},
    instrument::InstrumentData,
    subscriber::{Subscribed, Subscriber},
    subscription::{Subscription, SubscriptionKind},
    transformer::ExchangeTransformer,
};
use futures::{SinkExt, Stream, StreamExt};
use rustrade_instrument::exchange::ExchangeId;
use rustrade_integration::{
    Transformer,
    error::SocketError,
    protocol::{
        StreamParser,
        websocket::{WebSocket, WsError, WsMessage, WsSink, WsStream},
    },
};

use rustrade_integration::stream::ExchangeStream;
use std::{collections::VecDeque, future::Future};
use tokio::sync::mpsc;
use tracing::{debug, error, warn};

/// All [`Error`](std::error::Error)s generated in Barter-Data.
pub mod error;

/// Defines the generic [`MarketEvent<T>`](MarketEvent) used in every [`MarketStream`].
pub mod event;

/// [`Connector`] implementations for each exchange.
pub mod exchange;

/// High-level API types used for building [`MarketStream`]s from collections
/// of Barter [`Subscription`]s.
pub mod streams;

/// [`Subscriber`], [`SubscriptionMapper`](subscriber::mapper::SubscriptionMapper) and
/// [`SubscriptionValidator`](subscriber::validator::SubscriptionValidator)  traits that define how a
/// [`Connector`] will subscribe to exchange [`MarketStream`]s.
///
/// Standard implementations for subscribing to WebSocket [`MarketStream`]s are included.
pub mod subscriber;

/// Types that communicate the type of each [`MarketStream`] to initialise, and what normalised
/// Barter output type the exchange will be transformed into.
pub mod subscription;

/// [`InstrumentData`] trait for instrument describing data.
pub mod instrument;

/// [`OrderBook`](books::OrderBook) related types, and utilities for initialising and maintaining
/// a collection of sorted local Instrument [`OrderBook`](books::OrderBook)s
pub mod books;

/// Generic [`ExchangeTransformer`] implementations used by [`MarketStream`]s to translate exchange
/// specific types to normalised Barter types.
///
/// A standard [`StatelessTransformer`](transformer::stateless::StatelessTransformer) implementation
/// that works for most `Exchange`-`SubscriptionKind` combinations is included.
///
/// Cases that need custom logic, such as fetching initial [`OrderBooksL2`](subscription::book::OrderBooksL2)
/// and [`OrderBooksL3`](subscription::book::OrderBooksL3) snapshots on startup, may require custom
/// [`ExchangeTransformer`] implementations.
/// For examples, see [`Binance`](exchange::binance::Binance) [`OrderBooksL2`](subscription::book::OrderBooksL2)
/// [`ExchangeTransformer`] implementations for
/// [`spot`](exchange::binance::spot::l2::BinanceSpotOrderBooksL2Transformer) and
/// [`futures_usd`](exchange::binance::futures::l2::BinanceFuturesUsdOrderBooksL2Transformer).
pub mod transformer;

/// Convenient type alias for an [`ExchangeStream`] utilizing a tungstenite [`WebSocket`].
pub type ExchangeWsStream<Parser, Transformer> = ExchangeStream<Parser, WsStream, Transformer>;

/// Defines a generic identification type for the implementor.
pub trait Identifier<T> {
    fn id(&self) -> T;
}

/// [`Stream`] that yields [`Market<Kind>`](MarketEvent) events. The type of [`Market<Kind>`](MarketEvent)
/// depends on the provided [`SubscriptionKind`] of the passed [`Subscription`]s.
pub trait MarketStream<Exchange, Instrument, Kind>
where
    Self: Stream<Item = Result<MarketEvent<Instrument::Key, Kind::Event>, DataError>>
        + Send
        + Sized
        + Unpin,
    Exchange: Connector,
    Instrument: InstrumentData,
    Kind: SubscriptionKind,
{
    fn init<SnapFetcher>(
        subscriber: &Exchange::Subscriber,
        subscriptions: &[Subscription<Exchange, Instrument, Kind>],
    ) -> impl Future<Output = Result<Self, DataError>> + Send
    where
        SnapFetcher: SnapshotFetcher<Exchange, Kind>,
        Subscription<Exchange, Instrument, Kind>:
            Identifier<Exchange::Channel> + Identifier<Exchange::Market>;
}

/// Defines how to fetch market data snapshots for a collection of [`Subscription`]s.
///
/// Useful when a [`MarketStream`] requires an initial snapshot on start-up.
///
/// See examples such as Binance OrderBooksL2: <br>
/// - [`BinanceSpotOrderBooksL2SnapshotFetcher`](exchange::binance::spot::l2::BinanceSpotOrderBooksL2SnapshotFetcher)
/// - [`BinanceFuturesUsdOrderBooksL2SnapshotFetcher`](exchange::binance::futures::l2::BinanceFuturesUsdOrderBooksL2SnapshotFetcher)
pub trait SnapshotFetcher<Exchange, Kind> {
    fn fetch_snapshots<Instrument>(
        subscriptions: &[Subscription<Exchange, Instrument, Kind>],
    ) -> impl Future<Output = Result<Vec<MarketEvent<Instrument::Key, Kind::Event>>, SocketError>> + Send
    where
        Exchange: Connector,
        Instrument: InstrumentData,
        Kind: SubscriptionKind,
        Kind::Event: Send,
        Subscription<Exchange, Instrument, Kind>: Identifier<Exchange::Market>;
}

impl<Exchange, Instrument, Kind, Transformer, Parser> MarketStream<Exchange, Instrument, Kind>
    for ExchangeWsStream<Parser, Transformer>
where
    Exchange: Connector + Send + Sync,
    // This initialisation splits and reads a socket of the stream's own, so it serves only a
    // subscriber that hands one out; see `Subscriber::Transport`.
    Exchange::Subscriber: Subscriber<Transport = WebSocket>,
    Instrument: InstrumentData,
    Kind: SubscriptionKind + Send + Sync,
    Transformer: ExchangeTransformer<Exchange, Instrument::Key, Kind> + Send,
    Kind::Event: Send + Sync,
    Parser: StreamParser<Transformer::Input, Message = WsMessage, Error = WsError> + Send,
{
    async fn init<SnapFetcher>(
        subscriber: &Exchange::Subscriber,
        subscriptions: &[Subscription<Exchange, Instrument, Kind>],
    ) -> Result<Self, DataError>
    where
        SnapFetcher: SnapshotFetcher<Exchange, Kind>,
        Subscription<Exchange, Instrument, Kind>:
            Identifier<Exchange::Channel> + Identifier<Exchange::Market>,
    {
        // Connect & subscribe
        let Subscribed {
            transport: websocket,
            map: instrument_map,
            buffered_websocket_events,
        } = subscriber.subscribe(subscriptions).await?;

        // Fetch any required initial MarketEvent snapshots
        let initial_snapshots = SnapFetcher::fetch_snapshots(subscriptions).await?;

        // Split WebSocket into WsStream & WsSink components
        let (ws_sink, ws_stream) = websocket.split();

        // Spawn task to distribute Transformer messages (eg/ custom pongs) to the exchange
        let (ws_sink_tx, ws_sink_rx) = mpsc::unbounded_channel();
        tokio::spawn(distribute_messages_to_exchange(
            Exchange::ID,
            ws_sink,
            ws_sink_rx,
        ));

        // Spawn optional task to distribute custom application-level pings to the exchange
        if let Some(ping_interval) = Exchange::ping_interval() {
            tokio::spawn(schedule_pings_to_exchange(
                Exchange::ID,
                ws_sink_tx.clone(),
                ping_interval,
            ));
        }

        // Initialise Transformer associated with this Exchange and SubscriptionKind
        let mut transformer =
            Transformer::init(instrument_map, &initial_snapshots, ws_sink_tx).await?;

        // Process any buffered active subscription events received during Subscription validation
        let mut processed = process_buffered_events::<Parser, Transformer>(
            &mut transformer,
            buffered_websocket_events,
        );

        // Extend buffered events with any initial snapshot events
        processed.extend(initial_snapshots.into_iter().map(Ok));

        Ok(ExchangeWsStream::new(ws_stream, transformer, processed))
    }
}

/// Implementation of [`SnapshotFetcher`] that does not fetch any initial market data snapshots.
/// Often used for stateless [`MarketStream`]s, such as public trades.
#[derive(Debug)]
pub struct NoInitialSnapshots;

impl<Exchange, Kind> SnapshotFetcher<Exchange, Kind> for NoInitialSnapshots {
    fn fetch_snapshots<Instrument>(
        _: &[Subscription<Exchange, Instrument, Kind>],
    ) -> impl Future<Output = Result<Vec<MarketEvent<Instrument::Key, Kind::Event>>, SocketError>> + Send
    where
        Exchange: Connector,
        Instrument: InstrumentData,
        Kind: SubscriptionKind,
        Kind::Event: Send,
        Subscription<Exchange, Instrument, Kind>: Identifier<Exchange::Market>,
    {
        std::future::ready(Ok(vec![]))
    }
}

pub fn process_buffered_events<Parser, StreamTransformer>(
    transformer: &mut StreamTransformer,
    events: Vec<Parser::Message>,
) -> VecDeque<Result<StreamTransformer::Output, StreamTransformer::Error>>
where
    Parser: StreamParser<StreamTransformer::Input>,
    StreamTransformer: Transformer,
{
    events
        .into_iter()
        .filter_map(|event| {
            Parser::parse(Ok(event))?
                .inspect_err(|error| {
                    warn!(
                        ?error,
                        "failed to parse message buffered during Subscription validation"
                    )
                })
                .ok()
        })
        .flat_map(|parsed| transformer.transform(parsed))
        .collect()
}

/// Transmit [`WsMessage`]s sent from the [`ExchangeTransformer`] to the exchange via
/// the [`WsSink`].
///
/// **Note:**
/// [`Transformer`] is a synchronous trait, so we use this separate task to handle
/// async WebSocket writes without requiring the transformer to be async.
pub async fn distribute_messages_to_exchange(
    exchange: ExchangeId,
    mut ws_sink: WsSink,
    mut ws_sink_rx: mpsc::UnboundedReceiver<WsMessage>,
) {
    while let Some(message) = ws_sink_rx.recv().await {
        if let Err(error) = ws_sink.send(message).await {
            if rustrade_integration::protocol::websocket::is_websocket_disconnected(&error) {
                break;
            }

            // Log error only if WsMessage failed to send over a connected WebSocket
            error!(
                %exchange,
                %error,
                "failed to send output message to the exchange via WsSink"
            );
        }
    }
}

/// Schedule the sending of custom application-level ping [`WsMessage`]s to the exchange using
/// the provided [`PingInterval`].
///
/// **Notes:**
///  - This is only used for those exchanges that require custom application-level pings.
///  - This is additional to the protocol-level pings already handled by `tokio_tungstenite`.
pub async fn schedule_pings_to_exchange(
    exchange: ExchangeId,
    ws_sink_tx: mpsc::UnboundedSender<WsMessage>,
    PingInterval { mut interval, ping }: PingInterval,
) {
    loop {
        // Wait for next scheduled ping
        interval.tick().await;

        // Construct exchange custom application-level ping payload
        let payload = ping();
        debug!(%exchange, %payload, "sending custom application-level ping to exchange");

        if ws_sink_tx.send(payload).is_err() {
            break;
        }
    }
}