Streaming binance L2Books example

.github/workflows/rust-ci.yml

@@ -1,9 +1,9 @@
-on: 
-  pull_request
-
-name: 
-  Rust
-
+name: Rust
+permissions:
+  pull-requests: read
+  statuses: write
+on:
+  pull_request:
 env: 
   CARGO_TERM_COLOR: always
   RUSTFLAGS: -D warnings

Cargo.toml

@@ -10,3 +10,9 @@ resolver = "2"
 [workspace.lints.rust]
 unsafe_code = "forbid"
 unused_extern_crates = "warn"
+
+#[patch."https://github.com/CAGS295/trolly.git"]
+#trolly = { path = "../trolly"}
+
+[profile.dev]
+panic = "abort"

atelier/Cargo.toml

@@ -17,8 +17,12 @@ serde = { version = "1.0.203", features = ["derive"] }
 
 rand = {version="0.8.5"}
 rand_distr = "0.4.3"
-
-[dev-dependencies]
+trolly = { version = "*", git = "https://github.com/CAGS295/trolly.git" , default-features = false }
+serde_json = "1.0.128"
+reqwest = { version = "0.12.7", default-features = false }
+anyhow = "1.0.89"
+tokio = { version = "1.40.0", default-features = false }
+chrono = "0.4.38"
 
 [lib]
 name = "atelier"
@@ -31,3 +35,7 @@ workspace = true
 
 [package.metadata.docs.rs]
 rustdoc-args = [ "--html-in-header", "katex-header.html" ]
+
+[dev-dependencies]
+tokio = { version = "*", features = ["rt"]} 
+tracing-subscriber = "0.3.18"

atelier/examples/ob_gen_naive.rs

@@ -7,36 +7,32 @@ fn main() {
     let n_levels = 200;
     let n_orders = 300;
 
-    let i_ob = Orderbook::synthetize(bid_price, ask_price, tick_size, n_levels, n_orders);
+    let ob = Orderbook::synthetize(bid_price, ask_price, tick_size, n_levels, n_orders);
 
-    println!("\nlevel_id {:?}", i_ob.bids[199].level_id);
-    println!("side {:?}", i_ob.bids[199].side);
-    println!("price {:?}", i_ob.bids[199].price);
-    println!("orders[0]{:?}", i_ob.bids[199].orders[0]);
-    println!("orders[1]{:?}", i_ob.bids[199].orders[1]);
+    println!("\nlevel_id {:?}", ob.bids[199].level_id);
+    println!("price {:?}", ob.bids[199].price);
+    println!("orders[0]{:?}", ob.bids[199].orders[0]);
+    println!("orders[1]{:?}", ob.bids[199].orders[1]);
     println!(" ... ");
 
-    println!("\nlevel_id {:?}", i_ob.bids[0].level_id);
-    println!("side {:?}", i_ob.bids[0].side);
-    println!("price {:?}", i_ob.bids[0].price);
-    println!("orders[0]{:?}", i_ob.bids[0].orders[0]);
-    println!("orders[1]{:?}", i_ob.bids[0].orders[1]);
+    println!("\nlevel_id {:?}", ob.bids[0].level_id);
+    println!("price {:?}", ob.bids[0].price);
+    println!("orders[0]{:?}", ob.bids[0].orders[0]);
+    println!("orders[1]{:?}", ob.bids[0].orders[1]);
     println!(" ... ");
 
-    let mid_price = (i_ob.bids[0].price + i_ob.asks[0].price) / 2.0;
+    let mid_price = (ob.bids[0].price + ob.asks[0].price) / 2.0;
     println!("Midprice: {}", mid_price);
 
-    println!("\nlevel_id {:?}", i_ob.asks[0].level_id);
-    println!("side {:?}", i_ob.asks[0].side);
-    println!("price {:?}", i_ob.asks[0].price);
-    println!("orders[0]{:?}", i_ob.asks[0].orders[0]);
-    println!("orders[1]{:?}", i_ob.asks[0].orders[1]);
+    println!("\nlevel_id {:?}", ob.asks[0].level_id);
+    println!("price {:?}", ob.asks[0].price);
+    println!("orders[0]{:?}", ob.asks[0].orders[0]);
+    println!("orders[1]{:?}", ob.asks[0].orders[1]);
     println!(" ... ");
 
-    println!("\nlevel_id {:?}", i_ob.asks[199].level_id);
-    println!("side {:?}", i_ob.asks[199].side);
-    println!("price {:?}", i_ob.asks[199].price);
-    println!("orders[0]{:?}", i_ob.asks[199].orders[0]);
-    println!("orders[1]{:?}", i_ob.asks[199].orders[1]);
+    println!("\nlevel_id {:?}", ob.asks[199].level_id);
+    println!("price {:?}", ob.asks[199].price);
+    println!("orders[0]{:?}", ob.asks[199].orders[0]);
+    println!("orders[1]{:?}", ob.asks[199].orders[1]);
     println!(" ... ");
 }

atelier/examples/ob_metrics.rs

@@ -25,23 +25,15 @@ fn main() {
     println!("Midprice: {}", midprice_value);
 
     // Compute the Volume Imbalance
-    let iter_bids: Vec<f64> = i_ob.bids.clone().into_iter().map(|x| x.volume).collect();
-    let iter_asks: Vec<f64> = i_ob.asks.clone().into_iter().map(|x| x.volume).collect();
+    let iter_bids: Vec<f64> = i_ob.bids.iter().map(|x| x.volume).collect();
+    let iter_asks: Vec<f64> = i_ob.asks.iter().map(|x| x.volume).collect();
 
     let obimb_value = VolumeImbalance::compute(&iter_bids, &iter_asks, 1);
     println!("Volume Imbalance: {:?}", obimb_value);
 
     // Compute the Volume-Weighted Average Price
-    let iter_bids: Vec<_> = i_ob
-        .bids
-        .into_iter()
-        .map(|x| vec![x.price, x.volume])
-        .collect();
-    let iter_asks: Vec<_> = i_ob
-        .asks
-        .into_iter()
-        .map(|x| vec![x.price, x.volume])
-        .collect();
+    let iter_bids: Vec<_> = i_ob.bids.iter().map(|x| vec![x.price, x.volume]).collect();
+    let iter_asks: Vec<_> = i_ob.asks.iter().map(|x| vec![x.price, x.volume]).collect();
 
     // Compute the VWAP
     let vwap_value = VWAP::compute(&iter_bids.clone(), &iter_asks.clone(), 1);

atelier/examples/stream_orderbook.rs

@@ -0,0 +1,49 @@
+use std::{ops::Deref, time::Duration};
+
+use atelier::net::OrderBook as SyncBook;
+use atelier::net::OrderBook;
+use tokio::{sync::mpsc::unbounded_channel, time::interval};
+use tracing_subscriber::EnvFilter;
+use trolly::{
+    monitor::Depth,
+    net::MultiSymbolStream,
+    providers::{self},
+};
+
+#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
+async fn main() {
+    tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .with_target(true)
+        .init();
+
+    let (tx, mut rx) = unbounded_channel();
+
+    let local = tokio::task::LocalSet::new();
+
+    tokio::spawn(async move {
+        // BTCUSDT book
+        let btc_book: OrderBook = rx.recv().await.unwrap();
+        // ETHUSDT book
+        let eth_book = rx.recv().await.unwrap();
+
+        let mut every = interval(Duration::from_millis(5_000));
+        every.tick().await;
+
+        loop {
+            every.tick().await;
+            let btc_snapshot = btc_book.0.read().unwrap();
+            println!("{:?}", &*btc_snapshot.deref());
+            let eth_snapshot = eth_book.0.read().unwrap();
+            println!("{:?}", &*eth_snapshot.deref());
+        }
+    });
+
+    local
+        .run_until(async move {
+            let symbols = &["BTCUSDT".into(), "ETHUSDT".into()];
+            MultiSymbolStream::stream::<Depth, SyncBook, _, _>(providers::Binance, tx, symbols)
+                .await
+        })
+        .await;
+}