diff --git a/.gitignore b/.gitignore index ddb88654..ad00c32f 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,6 @@ Cargo.lock *.fmt *.iml -.env \ No newline at end of file +.env +.idea +.envrc \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index 1792ca93..f719015f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,34 +23,34 @@ name = "binance_async" path = "src/lib.rs" [dependencies] -failure = "0.1" -tracing = "0.1" - -tungstenite = "0.10" -tokio-tungstenite = { version = "0.10", features = ["tls"] } - -url = "2" -futures = "0.3" -headers = "0.3" -http = "0.2" -maplit = "1" -once_cell = "1" -reqwest = { version = "0.10", features = ["json"] } -reqwest-ext = { git = "https://github.com/vorot93/reqwest-ext" } -snafu = "0.6" -streamunordered = "0.5" -tokio = { version = "0.2", features = ["tcp"] } - -chrono = { version = "0.4", features = ["serde"] } - -serde = { version = "1", features = ["derive"] } -serde_json = "1" - -hex = "0.4" +failure = "0.1.8" +tracing = "0.1.25" + +tungstenite = "0.13.0" +tokio-tungstenite = { version = "0.13.0", features = ["tls"] } + +url = "2.2.1" +futures = "0.3.13" +headers = "0.3.4" +http = "0.2.4" +maplit = "1.0.2" +once_cell = "1.7.2" +reqwest = { version = "0.11.2", features = ["json"] } +reqwest-ext = { git = "https://github.com/utx0/reqwest-ext" } +snafu = "0.6.10" +streamunordered = "0.5.2" +tokio = { version = "1.4.0", features = ["full"] } + +chrono = { version = "0.4.19", features = ["serde"] } + +serde = { version = "1.0.125", features = ["derive"] } +serde_json = "1.0.64" + +hex = "0.4.3" sha2 = "0.8" hmac = "0.7" [dev-dependencies] -csv = "1" -tokio = { version = "0.2", features = ["full"] } -tracing-subscriber = "0.2" +csv = "1.1.6" +tokio = { version = "1.4.0", features = ["full"] } +tracing-subscriber = "0.2.17" diff --git a/examples/endpoints.rs b/examples/endpoints.rs index e52ab5f8..64f62784 100644 --- a/examples/endpoints.rs +++ b/examples/endpoints.rs @@ -13,63 +13,63 @@ async fn main() -> Fallible<()> { let bn = Binance::with_credential(&api_key, &secret_key); // General - match bn.ping()?.await { + match bn.ping().await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.get_server_time()?.await { + match bn.get_server_time().await { Ok(answer) => println!("Server Time: {}", answer.server_time), Err(e) => println!("Error: {}", e), } // Account - match bn.get_account()?.await { + match bn.get_account().await { Ok(answer) => println!("{:?}", answer.balances), Err(e) => println!("Error: {}", e), } - match bn.get_open_orders("WTCETH")?.await { + match bn.get_open_orders("BTCUSDT").await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.limit_buy("ETHBTC", 1., 0.1)?.await { + match bn.limit_buy("BTCUSDT", 1., 0.1).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.market_buy("WTCETH", 5.)?.await { + match bn.market_buy("BTCUSDT", 5.).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.limit_sell("WTCETH", 10., 0.035_000)?.await { + match bn.limit_sell("BTCUSDT", 10., 0.035_000).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.market_sell("WTCETH", 5.)?.await { + match bn.market_sell("BTCUSDT", 5.).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.order_status("WTCETH", 1_957_528)?.await { + match bn.order_status("BTCUSDT", 1_957_528).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.cancel_order("WTCETH", 1_957_528)?.await { + match bn.cancel_order("BTCUSDT", 1_957_528).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.get_balance("KNC")?.await { + match bn.get_balance("BTC").await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } - match bn.trade_history("WTCETH")?.await { + match bn.trade_history("BTCUSDT").await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } @@ -77,31 +77,31 @@ async fn main() -> Fallible<()> { // Market // Order book - match bn.get_depth("BNBETH", None)?.await { + match bn.get_depth("BTCUSDT", None).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } // Latest price for ALL symbols - match bn.get_all_prices()?.await { + match bn.get_all_prices().await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } // Latest price for ONE symbol - match bn.get_price("KNCETH")?.await { + match bn.get_price("BTCUSDT").await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } // Best price/qty on the order book for ALL symbols - match bn.get_all_book_tickers()?.await { + match bn.get_all_book_tickers().await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } // Best price/qty on the order book for ONE symbol - match bn.get_book_ticker("BNBETH")?.await { + match bn.get_book_ticker("BTCUSDT").await { Ok(answer) => println!( "Bid Price: {}, Ask Price: {}", answer.bid_price, answer.ask_price @@ -110,7 +110,7 @@ async fn main() -> Fallible<()> { } // 24hr ticker price change statistics - match bn.get_24h_price_stats("BNBETH")?.await { + match bn.get_24h_price_stats("BTCUSDT").await { Ok(answer) => println!( "Open Price: {}, Higher Price: {}, Lower Price: {:?}", answer.open_price, answer.high_price, answer.low_price @@ -119,7 +119,7 @@ async fn main() -> Fallible<()> { } // last 10 5min klines (candlesticks) for a symbol: - match bn.get_klines("BNBETH", "5m", 10, None, None)?.await { + match bn.get_klines("BTCUSDT", "5m", 10, None, None).await { Ok(answer) => println!("{:?}", answer), Err(e) => println!("Error: {}", e), } diff --git a/examples/new_endpoints.rs b/examples/new_endpoints.rs new file mode 100644 index 00000000..cbba4716 --- /dev/null +++ b/examples/new_endpoints.rs @@ -0,0 +1,21 @@ +use crate::binance::Binance; +use binance_async as binance; +use failure::Fallible; +use std::env::var; + +#[tokio::main] +async fn main() -> Fallible<()> { + tracing::subscriber::set_global_default(tracing_subscriber::FmtSubscriber::new()).unwrap(); + + let api_key = var("BINANCE_KEY")?; + let secret_key = var("BINANCE_SECRET")?; + + let bn = Binance::with_credential(&api_key, &secret_key); + + match bn.get_historical_trades("BTCUSDT", 10, 963563573).await { + Ok(answer) => println!("{:?}", answer), + Err(e) => println!("Error: {}", e), + } + + Ok(()) +} diff --git a/examples/ws.rs b/examples/ws.rs index 0d3e0f4b..887963b1 100644 --- a/examples/ws.rs +++ b/examples/ws.rs @@ -1,8 +1,8 @@ use crate::binance::{model::websocket::Subscription, Binance, BinanceWebsocket}; use binance_async as binance; use failure::Fallible; +use futures::TryStreamExt; use std::env::var; -use tokio::stream::StreamExt; #[tokio::main] async fn main() -> Fallible<()> { @@ -12,30 +12,29 @@ async fn main() -> Fallible<()> { let api_secret_user = var("BINANCE_SECRET")?; let bn = Binance::with_credential(&api_key_user, &api_secret_user); - match bn.user_stream_start()?.await { - Ok(answer) => { + match bn.user_stream_start().await { + Ok(_) => { println!("Data Stream Started ..."); - let listen_key = answer.listen_key; let mut ws = BinanceWebsocket::default(); for sub in vec![ - Subscription::Ticker("ethbtc".to_string()), - Subscription::AggregateTrade("eosbtc".to_string()), - Subscription::Candlestick("ethbtc".to_string(), "1m".to_string()), - Subscription::Depth("xrpbtc".to_string()), - Subscription::MiniTicker("zrxbtc".to_string()), - Subscription::OrderBook("trxbtc".to_string(), 5), - Subscription::Trade("adabtc".to_string()), - Subscription::UserData(listen_key), - Subscription::MiniTickerAll, - Subscription::TickerAll, + // Subscription::Ticker("btcusdt.to_string()), + // Subscription::AggregateTrade("btcusdt.to_string()), + // Subscription::Candlestick("btcusdt".to_string(), "1m".to_string()), + // Subscription::Depth("btcusdt".to_string()), + // Subscription::MiniTicker("btcusdt".to_string()), + // Subscription::OrderBook("btcusdt".to_string(), 10), + Subscription::Trade("btcusdt".to_string()), + // Subscription::UserData(listen_key), + // Subscription::MiniTickerAll, + // Subscription::TickerAll, ] { ws.subscribe(sub).await?; } while let Some(msg) = ws.try_next().await? { - println!("{:?}", msg) + println!("\n\n{:#?}", msg) } } Err(e) => println!("Error obtaining userstream: {}", e), diff --git a/src/client/account.rs b/src/client/account.rs index 628289f4..e6d70398 100644 --- a/src/client/account.rs +++ b/src/client/account.rs @@ -1,6 +1,6 @@ +use crate::error::Error; use crate::{ client::Binance, - error::Error, model::{ AccountInformation, AssetDetail, Balance, DepositAddressData, DepositHistory, Order, OrderCanceled, TradeHistory, Transaction, @@ -8,7 +8,6 @@ use crate::{ }; use chrono::prelude::*; use failure::Fallible; -use futures::prelude::*; use serde_json::json; use std::collections::HashMap; @@ -31,67 +30,57 @@ struct OrderRequest { impl Binance { // Account Information - pub fn get_account(&self) -> Fallible>> { + pub async fn get_account(&self) -> Fallible { let account_info = self .transport - .signed_get::<_, ()>("/api/v3/account", None)?; + .signed_get::<_, ()>("/api/v3/account", None) + .await?; Ok(account_info) } // Balance for ONE Asset - pub fn get_balance(&self, asset: &str) -> Fallible>> { + pub async fn get_balance(&self, asset: &str) -> Fallible { let asset = asset.to_string(); - let search = move |account: AccountInformation| { - let balance = account - .balances - .into_iter() - .find(|balance| balance.asset == asset); - future::ready(balance.ok_or_else(|| Error::AssetsNotFound.into())) - }; - - let balance = self.get_account()?.and_then(search); - Ok(balance) + self.get_account() + .await? + .balances + .into_iter() + .find(|balance| balance.asset == asset) + .ok_or_else(|| Error::AssetsNotFound.into()) } // Current open orders for ONE symbol - pub fn get_open_orders( - &self, - symbol: &str, - ) -> Fallible>>> { + pub async fn get_open_orders(&self, symbol: &str) -> Fallible> { let params = json! {{"symbol": symbol}}; let orders = self .transport - .signed_get("/api/v3/openOrders", Some(params))?; + .signed_get("/api/v3/openOrders", Some(params)) + .await?; Ok(orders) } // All current open orders - pub fn get_all_open_orders(&self) -> Fallible>>> { + pub async fn get_all_open_orders(&self) -> Fallible> { let orders = self .transport - .signed_get::<_, ()>("/api/v3/openOrders", None)?; + .signed_get::<_, ()>("/api/v3/openOrders", None) + .await?; Ok(orders) } // Check an order's status - pub fn order_status( - &self, - symbol: &str, - order_id: u64, - ) -> Fallible>> { + pub async fn order_status(&self, symbol: &str, order_id: u64) -> Fallible { let params = json! {{"symbol": symbol, "orderId": order_id}}; - let order = self.transport.signed_get(API_V3_ORDER, Some(params))?; + let order = self + .transport + .signed_get(API_V3_ORDER, Some(params)) + .await?; Ok(order) } // Place a LIMIT order - BUY - pub fn limit_buy( - &self, - symbol: &str, - qty: f64, - price: f64, - ) -> Fallible>> { + pub async fn limit_buy(&self, symbol: &str, qty: f64, price: f64) -> Fallible { let order = OrderRequest { symbol: symbol.into(), qty, @@ -102,18 +91,16 @@ impl Binance { }; let params = Self::build_order(order); - let transaction = self.transport.signed_post(API_V3_ORDER, Some(params))?; + let transaction = self + .transport + .signed_post(API_V3_ORDER, Some(params)) + .await?; Ok(transaction) } // Place a LIMIT order - SELL - pub fn limit_sell( - &self, - symbol: &str, - qty: f64, - price: f64, - ) -> Fallible>> { + pub async fn limit_sell(&self, symbol: &str, qty: f64, price: f64) -> Fallible { let order = OrderRequest { symbol: symbol.into(), qty, @@ -123,17 +110,16 @@ impl Binance { time_in_force: TIME_IN_FORCE_GTC.to_string(), }; let params = Self::build_order(order); - let transaction = self.transport.signed_post(API_V3_ORDER, Some(params))?; + let transaction = self + .transport + .signed_post(API_V3_ORDER, Some(params)) + .await?; Ok(transaction) } // Place a MARKET order - BUY - pub fn market_buy( - &self, - symbol: &str, - qty: f64, - ) -> Fallible>> { + pub async fn market_buy(&self, symbol: &str, qty: f64) -> Fallible { let order = OrderRequest { symbol: symbol.into(), qty, @@ -143,17 +129,16 @@ impl Binance { time_in_force: TIME_IN_FORCE_GTC.to_string(), }; let params = Self::build_order(order); - let transaction = self.transport.signed_post(API_V3_ORDER, Some(params))?; + let transaction = self + .transport + .signed_post(API_V3_ORDER, Some(params)) + .await?; Ok(transaction) } // Place a MARKET order - SELL - pub fn market_sell( - &self, - symbol: &str, - qty: f64, - ) -> Fallible>> { + pub async fn market_sell(&self, symbol: &str, qty: f64) -> Fallible { let order = OrderRequest { symbol: symbol.into(), qty, @@ -163,65 +148,61 @@ impl Binance { time_in_force: TIME_IN_FORCE_GTC.to_string(), }; let params = Self::build_order(order); - let transaction = self.transport.signed_post(API_V3_ORDER, Some(params))?; + let transaction = self + .transport + .signed_post(API_V3_ORDER, Some(params)) + .await?; Ok(transaction) } // Check an order's status - pub fn cancel_order( - &self, - symbol: &str, - order_id: u64, - ) -> Fallible>> { + pub async fn cancel_order(&self, symbol: &str, order_id: u64) -> Fallible { let params = json! {{"symbol":symbol, "orderId":order_id}}; - let order_canceled = self.transport.signed_delete(API_V3_ORDER, Some(params))?; + let order_canceled = self + .transport + .signed_delete(API_V3_ORDER, Some(params)) + .await?; Ok(order_canceled) } // Trade history - pub fn trade_history( - &self, - symbol: &str, - ) -> Fallible>>> { + pub async fn trade_history(&self, symbol: &str) -> Fallible> { let params = json! {{"symbol":symbol}}; let trade_history = self .transport - .signed_get("/api/v3/myTrades", Some(params))?; - + .signed_get("/api/v3/myTrades", Some(params)) + .await?; Ok(trade_history) } - pub fn get_deposit_address( - &self, - symbol: &str, - ) -> Fallible>> { + pub async fn get_deposit_address(&self, symbol: &str) -> Fallible { let params = json! {{"asset":symbol}}; let deposit_address = self .transport - .signed_get("/wapi/v3/depositAddress.html", Some(params))?; - + .signed_get("/wapi/v3/depositAddress.html", Some(params)) + .await?; Ok(deposit_address) } - pub fn get_deposit_history( + pub async fn get_deposit_history( &self, symbol: Option<&str>, start_time: Option>, end_time: Option>, - ) -> Fallible>> { + ) -> Fallible { let params = json! {{"asset":symbol, "startTime":start_time.map(|t| t.timestamp_millis()), "endTime":end_time.map(|t| t.timestamp_millis())}}; let deposit_history = self .transport - .signed_get("/wapi/v3/depositHistory.html", Some(params))?; - + .signed_get("/wapi/v3/depositHistory.html", Some(params)) + .await?; Ok(deposit_history) } - pub fn asset_detail(&self) -> Fallible>> { + pub async fn asset_detail(&self) -> Fallible { let asset_detail = self .transport - .signed_get::<_, ()>("/wapi/v3/assetDetail.html", None)?; - + .signed_get::<_, ()>("/wapi/v3/assetDetail.html", None) + .await?; Ok(asset_detail) } @@ -237,7 +218,6 @@ impl Binance { params.insert("price", order.price.to_string()); params.insert("timeInForce", order.time_in_force.to_string()); } - params } } diff --git a/src/client/general.rs b/src/client/general.rs index d5fe6947..45eb6768 100644 --- a/src/client/general.rs +++ b/src/client/general.rs @@ -3,30 +3,30 @@ use crate::{ model::{ExchangeInfo, ExchangeInformation, ServerTime}, }; use failure::Fallible; -use futures::prelude::*; -use serde_json::Value; impl Binance { // Test connectivity - pub fn ping(&self) -> Fallible>> { - Ok(self - .transport - .get::<_, ()>("/api/v1/ping", None)? - .map_ok(|_: Value| "pong".into())) + pub async fn ping(&self) -> Fallible { + Ok(self.transport.get::<_, ()>("/api/v1/ping", None).await?) } // Check server time - pub fn get_server_time(&self) -> Fallible>> { - Ok(self.transport.get::<_, ()>("/api/v1/time", None)?) + pub async fn get_server_time(&self) -> Fallible { + Ok(self.transport.get::<_, ()>("/api/v1/time", None).await?) } - pub fn get_exchange_info(&self) -> Fallible>> { - Ok(self.transport.get::<_, ()>("/api/v1/exchangeInfo", None)?) + pub async fn get_exchange_info(&self) -> Fallible { + Ok(self + .transport + .get::<_, ()>("/api/v1/exchangeInfo", None) + .await?) } // Obtain exchange information (rate limits, symbol metadata etc) - pub fn exchange_info(&self) -> Fallible>> { - let info = self.transport.get::<_, ()>("/api/v1/exchangeInfo", None)?; - Ok(info) + pub async fn exchange_info(&self) -> Fallible { + Ok(self + .transport + .get::<_, ()>("/api/v1/exchangeInfo", None) + .await?) } } diff --git a/src/client/market.rs b/src/client/market.rs index 2b281eb5..a1d51633 100644 --- a/src/client/market.rs +++ b/src/client/market.rs @@ -1,101 +1,107 @@ use super::Binance; use crate::{ error::Error, - model::{BookTickers, KlineSummaries, KlineSummary, OrderBook, PriceStats, Prices, Ticker}, + model::{ + BookTickers, HistoricalTrade, KlineSummaries, KlineSummary, OrderBook, PriceStats, Prices, + Ticker, + }, }; use failure::Fallible; -use futures::prelude::*; use serde_json::{json, Value}; use std::{collections::HashMap, iter::FromIterator}; // Market Data endpoints impl Binance { // Order book (Default 100; max 100) - pub fn get_depth( - &self, - symbol: &str, - limit: I, - ) -> Fallible>> + pub async fn get_depth(&self, symbol: &str, limit: I) -> Fallible where I: Into>, { let limit = limit.into().unwrap_or(100); let params = json! {{"symbol": symbol, "limit": limit}}; - Ok(self.transport.get("/api/v1/depth", Some(params))?) + Ok(self.transport.get("/api/v1/depth", Some(params)).await?) } // Latest price for ALL symbols. - pub fn get_all_prices(&self) -> Fallible>> { + pub async fn get_all_prices(&self) -> Fallible { Ok(self .transport - .get::<_, ()>("/api/v1/ticker/allPrices", None)?) + .get::<_, ()>("/api/v1/ticker/allPrices", None) + .await?) } // Latest price for ONE symbol. - pub fn get_price(&self, symbol: &str) -> Fallible>> { + pub async fn get_price(&self, symbol: &str) -> Fallible { let symbol = symbol.to_string(); - let all_prices = self.get_all_prices()?; - Ok(async move { - let Prices::AllPrices(prices) = all_prices.await?; - Ok(prices - .into_iter() - .find_map(|obj| { - if obj.symbol == symbol { - Some(obj.price) - } else { - None - } - }) - .ok_or_else(|| Error::SymbolNotFound)?) - }) + let all_prices = self.get_all_prices(); + let Prices::AllPrices(prices) = all_prices.await?; + Ok(prices + .into_iter() + .find_map(|obj| { + if obj.symbol == symbol { + Some(obj.price) + } else { + None + } + }) + .ok_or_else(|| Error::SymbolNotFound)?) + } + + pub async fn get_historical_trades( + &self, + symbol: &str, + limit: u16, + from_id: u64, + ) -> Fallible> { + let params = json! {{"symbol":symbol, "limit": limit, "fromId": from_id}}; + Ok(self + .transport + .get("/api/v3/historicalTrades", Some(params)) + .await?) } // Symbols order book ticker // -> Best price/qty on the order book for ALL symbols. - pub fn get_all_book_tickers(&self) -> Fallible>> { + pub async fn get_all_book_tickers(&self) -> Fallible { Ok(self .transport - .get::<_, ()>("/api/v1/ticker/allBookTickers", None)?) + .get::<_, ()>("/api/v1/ticker/allBookTickers", None) + .await?) } // -> Best price/qty on the order book for ONE symbol - pub fn get_book_ticker( - &self, - symbol: &str, - ) -> Fallible>> { + pub async fn get_book_ticker(&self, symbol: &str) -> Fallible { let symbol = symbol.to_string(); - let all_book_tickers = self.get_all_book_tickers()?; + let all_book_tickers = self.get_all_book_tickers(); - Ok(async move { - let BookTickers::AllBookTickers(book_tickers) = all_book_tickers.await?; + let BookTickers::AllBookTickers(book_tickers) = all_book_tickers.await?; - Ok(book_tickers - .into_iter() - .find(|obj| obj.symbol == symbol) - .ok_or_else(|| Error::SymbolNotFound)?) - }) + Ok(book_tickers + .into_iter() + .find(|obj| obj.symbol == symbol) + .ok_or_else(|| Error::SymbolNotFound)?) } // 24hr ticker price change statistics - pub fn get_24h_price_stats( - &self, - symbol: &str, - ) -> Fallible>> { + pub async fn get_24h_price_stats(&self, symbol: &str) -> Fallible { let params = json! {{"symbol": symbol}}; - Ok(self.transport.get("/api/v1/ticker/24hr", Some(params))?) + Ok(self + .transport + .get("/api/v1/ticker/24hr", Some(params)) + .await?) } // Returns up to 'limit' klines for given symbol and interval ("1m", "5m", ...) // https://github.com/binance-exchange/binance-official-api-docs/blob/master/rest-api.md#klinecandlestick-data - pub fn get_klines( + pub async fn get_klines( &self, symbol: &str, interval: &str, limit: S3, start_time: S4, end_time: S5, - ) -> Fallible>> + ) -> Fallible where S3: Into>, S4: Into>, @@ -118,38 +124,35 @@ impl Binance { } let params: HashMap<&str, String> = HashMap::from_iter(params); - let f = self.transport.get("/api/v1/klines", Some(params))?; - - Ok({ - async move { - let data: Vec> = f.await?; - - Ok(KlineSummaries::AllKlineSummaries( - data.iter() - .map(|row| KlineSummary { - open_time: to_i64(&row[0]), - open: to_f64(&row[1]), - high: to_f64(&row[2]), - low: to_f64(&row[3]), - close: to_f64(&row[4]), - volume: to_f64(&row[5]), - close_time: to_i64(&row[6]), - quote_asset_volume: to_f64(&row[7]), - number_of_trades: to_i64(&row[8]), - taker_buy_base_asset_volume: to_f64(&row[9]), - taker_buy_quote_asset_volume: to_f64(&row[10]), - }) - .collect(), - )) - } - }) + let f = self.transport.get("/api/v1/klines", Some(params)); + + let data: Vec> = f.await?; + + Ok(KlineSummaries::AllKlineSummaries( + data.iter() + .map(|row| KlineSummary { + open_time: to_i64(&row[0]), + open: to_f64(&row[1]), + high: to_f64(&row[2]), + low: to_f64(&row[3]), + close: to_f64(&row[4]), + volume: to_f64(&row[5]), + close_time: to_i64(&row[6]), + quote_asset_volume: to_f64(&row[7]), + number_of_trades: to_i64(&row[8]), + taker_buy_base_asset_volume: to_f64(&row[9]), + taker_buy_quote_asset_volume: to_f64(&row[10]), + }) + .collect(), + )) } // 24hr ticker price change statistics - pub fn get_24h_price_stats_all( - &self, - ) -> Fallible>>> { - Ok(self.transport.get::<_, ()>("/api/v1/ticker/24hr", None)?) + pub async fn get_24h_price_stats_all(&self) -> Fallible> { + Ok(self + .transport + .get::<_, ()>("/api/v1/ticker/24hr", None) + .await?) } } diff --git a/src/client/mod.rs b/src/client/mod.rs index 16df90a1..eda184ec 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -6,7 +6,7 @@ pub mod websocket; use crate::transport::Transport; -#[derive(Clone, Default)] +#[derive(Clone, Default, Debug)] pub struct Binance { pub transport: Transport, } diff --git a/src/client/userstream.rs b/src/client/userstream.rs index 3000732f..7b8ca6ef 100644 --- a/src/client/userstream.rs +++ b/src/client/userstream.rs @@ -3,37 +3,34 @@ use crate::{ model::{Success, UserDataStream}, }; use failure::Fallible; -use futures::prelude::*; const USER_DATA_STREAM: &str = "/api/v1/userDataStream"; impl Binance { // User Stream - pub fn user_stream_start(&self) -> Fallible>> { - let user_data_stream = self.transport.post::<_, ()>(USER_DATA_STREAM, None)?; - Ok(user_data_stream) + pub async fn user_stream_start(&self) -> Fallible { + Ok(self.transport.post::<_, ()>(USER_DATA_STREAM, None).await?) } // Current open orders on a symbol - pub fn user_stream_keep_alive( - &self, - listen_key: &str, - ) -> Fallible>> { - let success = self.transport.put( - USER_DATA_STREAM, - Some(vec![("listen_key", listen_key.to_string())]), - )?; - Ok(success) + pub async fn user_stream_keep_alive(&self, listen_key: &str) -> Fallible { + Ok(self + .transport + .put( + USER_DATA_STREAM, + Some(vec![("listen_key", listen_key.to_string())]), + ) + .await?) } - pub fn user_stream_close( - &self, - listen_key: &str, - ) -> Fallible>> { - let success = self.transport.delete( - USER_DATA_STREAM, - Some(vec![("listen_key", listen_key.to_string())]), - )?; + pub async fn user_stream_close(&self, listen_key: &str) -> Fallible { + let success = self + .transport + .delete( + USER_DATA_STREAM, + Some(vec![("listen_key", listen_key.to_string())]), + ) + .await?; Ok(success) } } diff --git a/src/client/websocket.rs b/src/client/websocket.rs index 8610b91f..e9476012 100644 --- a/src/client/websocket.rs +++ b/src/client/websocket.rs @@ -13,9 +13,8 @@ use std::{ }; use streamunordered::{StreamUnordered, StreamYield}; use tokio::net::TcpStream; -use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; +use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; use tracing::*; -use tungstenite::Message; use url::Url; const WS_URL: &str = "wss://stream.binance.com:9443/ws"; @@ -56,7 +55,7 @@ impl BinanceWebsocket { let token = self .streams - .push(connect_async(endpoint).await?.0.split().1); + .insert(connect_async(endpoint).await?.0.split().1); self.subscriptions.insert(subscription.clone(), token); self.tokens.insert(token, subscription); diff --git a/src/model/mod.rs b/src/model/mod.rs index 3d7d2036..8c862475 100644 --- a/src/model/mod.rs +++ b/src/model/mod.rs @@ -87,7 +87,7 @@ pub struct Bids { pub qty: f64, // Never serialized. - #[serde(skip_serializing)] + #[serde(skip)] ignore: Vec, } @@ -99,7 +99,7 @@ pub struct Asks { pub qty: f64, // Never serialized. - #[serde(skip_serializing)] + #[serde(skip)] ignore: Vec, } @@ -171,6 +171,21 @@ pub struct TradeHistory { pub is_best_match: bool, } +#[derive(Debug, Serialize, Deserialize, Clone)] +#[serde(rename_all = "camelCase")] +pub struct HistoricalTrade { + pub id: i64, + #[serde(with = "string_or_float")] + pub price: f64, + #[serde(with = "string_or_float")] + pub qty: f64, + #[serde(with = "string_or_float")] + pub quote_qty: f64, + pub time: i64, + pub is_buyer_maker: bool, + pub is_best_match: bool, +} + #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct PriceStats { diff --git a/src/model/websocket.rs b/src/model/websocket.rs index c6fdb1b7..4bf0922d 100644 --- a/src/model/websocket.rs +++ b/src/model/websocket.rs @@ -43,21 +43,21 @@ pub struct TradeMessage { #[serde(rename = "e")] pub event_type: String, #[serde(rename = "E")] - pub event_time: u64, + pub event_time: i64, #[serde(rename = "s")] pub symbol: String, #[serde(rename = "t")] - pub trade_id: u64, + pub trade_id: i64, #[serde(rename = "p", with = "string_or_float")] pub price: f64, #[serde(rename = "q", with = "string_or_float")] pub qty: f64, #[serde(rename = "b")] - pub buyer_order_id: u64, + pub buyer_order_id: i64, #[serde(rename = "a")] - pub seller_order_id: u64, + pub seller_order_id: i64, #[serde(rename = "T")] - pub trade_order_time: u64, + pub trade_order_time: i64, #[serde(rename = "m")] pub is_buyer_maker: bool, #[serde(skip_serializing, rename = "M")] diff --git a/src/transport.rs b/src/transport.rs index bca1c6c4..8787d670 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -1,7 +1,6 @@ use crate::error::{BinanceResponse, Error}; use chrono::Utc; use failure::Fallible; -use futures::prelude::*; use headers::*; use hex::encode as hexify; use hmac::{Hmac, Mac}; @@ -44,7 +43,7 @@ impl headers::Header for BinanceApiKey { } } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct Transport { credential: Option<(String, String)>, client: reqwest::Client, @@ -74,109 +73,85 @@ impl Transport { } } - pub fn get( - &self, - endpoint: &str, - params: Option, - ) -> Fallible>> + pub async fn get(&self, endpoint: &str, params: Option) -> Fallible where O: DeserializeOwned, Q: Serialize, { self.request::<_, _, ()>(Method::GET, endpoint, params, None) + .await } - pub fn post( - &self, - endpoint: &str, - data: Option, - ) -> Fallible>> + pub async fn post(&self, endpoint: &str, data: Option) -> Fallible where O: DeserializeOwned, D: Serialize, { self.request::<_, (), _>(Method::POST, endpoint, None, data) + .await } - pub fn put( - &self, - endpoint: &str, - data: Option, - ) -> Fallible>> + pub async fn put(&self, endpoint: &str, data: Option) -> Fallible where O: DeserializeOwned, D: Serialize, { self.request::<_, (), _>(Method::PUT, endpoint, None, data) + .await } - pub fn delete( - &self, - endpoint: &str, - params: Option, - ) -> Fallible>> + pub async fn delete(&self, endpoint: &str, params: Option) -> Fallible where O: DeserializeOwned, Q: Serialize, { self.request::<_, _, ()>(Method::DELETE, endpoint, params, None) + .await } - pub fn signed_get( - &self, - endpoint: &str, - params: Option, - ) -> Fallible>> + pub async fn signed_get(&self, endpoint: &str, params: Option) -> Fallible where O: DeserializeOwned, Q: Serialize, { self.signed_request::<_, _, ()>(Method::GET, endpoint, params, None) + .await } - pub fn signed_post( - &self, - endpoint: &str, - data: Option, - ) -> Fallible>> + pub async fn signed_post(&self, endpoint: &str, data: Option) -> Fallible where O: DeserializeOwned, D: Serialize, { self.signed_request::<_, (), _>(Method::POST, endpoint, None, data) + .await } - pub fn signed_put( - &self, - endpoint: &str, - params: Option, - ) -> Fallible>> + pub async fn signed_put(&self, endpoint: &str, params: Option) -> Fallible where O: DeserializeOwned, Q: Serialize, { self.signed_request::<_, _, ()>(Method::PUT, endpoint, params, None) + .await } - pub fn signed_delete( - &self, - endpoint: &str, - params: Option, - ) -> Fallible>> + pub async fn signed_delete(&self, endpoint: &str, params: Option) -> Fallible where O: DeserializeOwned, Q: Serialize, { self.signed_request::<_, _, ()>(Method::DELETE, endpoint, params, None) + .await } - pub fn request( + pub async fn request( &self, method: Method, endpoint: &str, params: Option, data: Option, - ) -> Fallible>> + ) -> Fallible where O: DeserializeOwned, Q: Serialize, @@ -206,23 +181,21 @@ impl Transport { let req = req.body(body); - Ok(async move { - Ok(req - .send() - .await? - .json::>() - .await? - .into_result()?) - }) + Ok(req + .send() + .await? + .json::>() + .await? + .into_result()?) } - pub fn signed_request( + pub async fn signed_request( &self, method: Method, endpoint: &str, params: Option, data: Option, - ) -> Fallible>> + ) -> Fallible where O: DeserializeOwned, Q: Serialize, @@ -249,14 +222,12 @@ impl Transport { .typed_header(BinanceApiKey(key.to_string())) .body(body); - Ok(async move { - Ok(req - .send() - .await? - .json::>() - .await? - .into_result()?) - }) + Ok(req + .send() + .await? + .json::>() + .await? + .into_result()?) } fn check_key(&self) -> Fallible<(&str, &str)> { diff --git a/tests/ping.rs b/tests/ping.rs index 689baa5e..18af0806 100644 --- a/tests/ping.rs +++ b/tests/ping.rs @@ -10,7 +10,7 @@ async fn ping() -> Fallible<()> { let binance = Binance::new(); - binance.ping()?.await?; + binance.ping().await?; Ok(()) }