Barter-Integration

High-performance, low-level framework for composing flexible web integrations. The core Socket abstraction provides customisable communication over any asynchronous protocol (WebSocket, FIX, etc.), with built-in translation between server & client data models.

Utilised by other Barter trading ecosystem crates to build robust financial exchange integrations, primarily for public data collection & trade execution. It is:

• Low-Level: Translates raw data streams communicated over the web into any desired data model using arbitrary data transformations.
• Flexible: Compatible with any protocol (WebSocket, FIX, etc.), any input/output model, and any user defined transformations.

See: Barter, Barter-Data & Barter-Execution

API Documentation | Chat

Overview

Barter-Integration is a high-performance, low-level framework for composing flexible web integrations. It presents an
extensible core abstraction called the ExchangeStream. At a high level, an ExchangeStream is made up of a few major components:

• Inner Stream/Sink (eg/ WebSocket, FIX socket, etc).
• ProtocolParser that is capable of parsing input protocol messages (eg/ WebSocket, FIX, etc.) as exchange specific message.
• Transformer that transforms from exchange specific message into an iterator of desired outputs.

Example

Consume Binance Futures tick-by-tick Trades and calculate a rolling sum of volume.

use barter_integration::{
    error::SocketError,
    protocol::websocket::{WebSocket, WebSocketParser, WsMessage},
    ExchangeStream, Transformer,
};
use futures::{SinkExt, StreamExt};
use serde::{de, Deserialize};
use serde_json::json;
use std::str::FromStr;
use tokio_tungstenite::connect_async;
use tracing::debug;

// Convenient type alias for an `ExchangeStream` utilising a tungstenite `WebSocket`
type ExchangeWsStream<Exchange> = ExchangeStream<WebSocketParser, WebSocket, Exchange, VolumeSum>;

// Communicative type alias for what the VolumeSum the Transformer is generating
type VolumeSum = f64;

#[derive(Deserialize)]
#[serde(untagged, rename_all = "camelCase")]
enum BinanceMessage {
    SubResponse {
        result: Option<Vec<String>>,
        id: u32,
    },
    Trade {
        #[serde(rename = "q", deserialize_with = "de_str")]
        quantity: f64,
    },
}

struct StatefulTransformer {
    sum_of_volume: VolumeSum,
}

impl Transformer<VolumeSum> for StatefulTransformer {
    type Input = BinanceMessage;
    type OutputIter = Vec<Result<VolumeSum, SocketError>>;

    fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
        // Add new input Trade quantity to sum
        match input {
            BinanceMessage::SubResponse { result, id } => {
                debug!("Received SubResponse for {}: {:?}", id, result);
                // Don't care about this for the example
            }
            BinanceMessage::Trade { quantity, .. } => {
                // Add new Trade volume to internal state VolumeSum
                self.sum_of_volume += quantity;
            }
        };

        // Return IntoIterator of length 1 containing the running sum of volume
        vec![Ok(self.sum_of_volume)]
    }
}

/// See Barter-Data for a comprehensive real-life example, as well as code you can use out of the
/// box to collect real-time public market data from many exchanges.
#[tokio::main]
async fn main() {
    // Establish Sink/Stream communication with desired WebSocket server
    let mut binance_conn = connect_async("wss://fstream.binance.com/ws/")
        .await
        .map(|(ws_conn, _)| ws_conn)
        .expect("failed to connect");

    // Send something over the socket (eg/ Binance trades subscription)
    binance_conn
        .send(WsMessage::Text(
            json!({"method": "SUBSCRIBE","params": ["btcusdt@aggTrade"],"id": 1}).to_string(),
        ))
        .await
        .expect("failed to send WsMessage over socket");

    // Instantiate some arbitrary Transformer to apply to data parsed from the WebSocket protocol
    let transformer = StatefulTransformer { sum_of_volume: 0.0 };

    // ExchangeWsStream includes pre-defined WebSocket Sink/Stream & WebSocket ProtocolParser
    let mut ws_stream = ExchangeWsStream::new(binance_conn, transformer);

    // Receive a stream of your desired Output data model from the ExchangeStream
    while let Some(volume_result) = ws_stream.next().await {
        match volume_result {
            Ok(cumulative_volume) => {
                // Do something with your data
                println!("{cumulative_volume:?}");
            }
            Err(error) => {
                // React to any errors produced by the internal transformation
                eprintln!("{error}")
            }
        }
    }
}

/// Deserialize a `String` as the desired type.
fn de_str<'de, D, T>(deserializer: D) -> Result<T, D::Error>
where
    D: de::Deserializer<'de>,
    T: FromStr,
    T::Err: std::fmt::Display,
{
    let data: String = Deserialize::deserialize(deserializer)?;
    data.parse::<T>().map_err(de::Error::custom)
}

For a larger, "real world" example, see the Barter-Data repository.

Getting Help

Firstly, see if the answer to your question can be found in the API Documentation. If the answer is not there, I'd be happy to help to Chat and try answer your question via Discord.

Contributing

Thanks for your help in improving the Barter ecosystem! Please do get in touch on the discord to discuss development, new features, and the future roadmap.

Related Projects

In addition to the Barter-Integration crate, the Barter project also maintains:

• Barter: High-performance, extensible & modular trading components with batteries-included. Contains a pre-built trading Engine that can serve as a live-trading or backtesting system.
• Barter-Data: A high-performance WebSocket integration library for streaming public data from leading cryptocurrency exchanges.
• Barter-Execution: Financial exchange integrations for trade execution - yet to be released!

Roadmap

• Add new default ProtocolParser implementations to enable integration with other popular systems such as Kafka.

Licence

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in Barter-Integration by you, shall be licensed as MIT, without any additional terms or conditions.