Ws account events connected and initial position implemention

.env.sample

@@ -1,3 +1,11 @@
 ETH_NODE_URL=
 ETH_ACCOUNT_ADDRESS=
 ACCOUNT_PRIVATE_KEY=
+
+BINANCE_API_KEY=
+BINANCE_SECRET=
+
+BINANCE_TEST_API_KEY= 
+BINANCE_TEST_SECRET= 
+
+RUST_LOG="info"

barter-data-rs/examples/uniswapx_trades.rs

@@ -1,6 +1,5 @@
 use barter_data::dex::uniswapx;
 use dotenv::dotenv;
-use tokio::time::{sleep, Duration};
 
 #[tokio::main]
 async fn main() {

barter-data-rs/src/exchange/binance/futures/mod.rs

@@ -14,9 +14,6 @@ pub mod l2;
 /// Liquidation types.
 pub mod liquidation;
 
-/// Account types.
-pub mod account;
-
 /// [`BinanceFuturesUsd`] WebSocket server base url.
 ///
 /// See docs: <https://binance-docs.github.io/apidocs/futures/en/#websocket-market-streams>

barter-execution-rs/src/error.rs

@@ -25,3 +25,27 @@ pub enum ExecutionError {
     #[error("SocketError: {0}")]
     Socket(#[from] SocketError),
 }
+
+/// All errors generated in the barter::portfolio module.
+#[derive(Error, Clone, Copy, Debug)]
+pub enum PositionError {
+    #[error("Failed to build struct due to missing attributes: {0}")]
+    BuilderIncomplete(&'static str),
+
+    #[error("Failed to parse Position entry Side due to ambiguous fill quantity & Decision.")]
+    ParseEntrySide,
+
+    #[error("Cannot exit Position with an entry decision FillEvent.")]
+    CannotEnterPositionWithExitFill,
+
+    #[error("Cannot exit Position with an entry decision FillEvent.")]
+    CannotExitPositionWithEntryFill,
+
+    #[error("Cannot generate PositionExit from Position that has not been exited")]
+    PositionExit,
+
+    #[error("Negative Trade Quantity")]
+    NegativeTradeQuantity,
+    // #[error("Failed to interact with repository")]
+    // RepositoryInteraction(#[from] RepositoryError),
+}

barter-execution-rs/src/execution/binance/connection.rs

@@ -11,15 +11,13 @@ use barter_integration::{
     },
 };
 use chrono::Utc;
-use dotenv::dotenv;
 use hmac::Hmac;
 use reqwest::{RequestBuilder, StatusCode};
 use serde::Deserialize;
 use tokio::sync::mpsc;
 
 use crate::{
     error::ExecutionError,
-    fill::Decision,
     model::order::{Order, OrderKind, RequestOpen},
 };
 
@@ -258,15 +256,6 @@ impl BinanceClient {
     }
 }
 
-pub(super) fn get_order_side(side: Decision) -> &'static str {
-    match side {
-        Decision::Long => "BUY",
-        Decision::Short => "SELL",
-        Decision::CloseLong => "SELL",
-        Decision::CloseShort => "BUY",
-    }
-}
-
 #[derive(Debug, Clone)]
 pub struct BinanceSigner {
     pub api_key: String,
@@ -371,10 +360,7 @@ impl HttpParser for BinanceParser {
 mod tests {
     use super::*;
     use crate::{
-        execution::binance::requests::FutOrderResponse,
-        fill::MarketMeta,
-        model::{order_event::OrderEventBuilder, ClientOrderId},
-        ExecutionId,
+        execution::binance::requests::FutOrderResponse, model::ClientOrderId, ExecutionId,
     };
     use barter_integration::model::{
         instrument::{kind::InstrumentKind, symbol::Symbol, Instrument},

barter-execution-rs/src/execution/binance/mod.rs

@@ -1,15 +1,17 @@
 use async_trait::async_trait;
+use barter_data::exchange::ExchangeId;
 use barter_integration::model::{instrument::symbol::Symbol, Exchange};
-use futures::future::join_all;
-use tracing::{error, info};
+use futures::{future::join_all, stream::BoxStream};
+use tracing::error;
 
 use crate::{
     error::ExecutionError,
     model::{
         balance::SymbolBalance,
-        order::{self, Cancelled, Open, Order, OrderId, RequestCancel, RequestOpen},
+        order::{Cancelled, Open, Order, OrderId, RequestCancel, RequestOpen},
+        AccountEventKind,
     },
-    ExecutionClient, ExecutionId,
+    ExecutionClient,
 };
 
 use self::{
@@ -20,13 +22,14 @@ use self::{
 
 pub mod connection;
 pub mod requests;
+pub mod types;
 pub mod websocket;
 
 /// Binance [`ExecutionClient`] implementation that integrates with the Barter
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct BinanceExecution {
-    client: BinanceClient,
-    // client_type: BinanceApi,
+    pub client: BinanceClient,
+    client_type: BinanceApi,
 }
 
 /// Config for initializing a [`BinanceExecution`] instance.
@@ -35,25 +38,31 @@ pub struct BinanceConfig {
     pub client_type: BinanceApi,
 }
 
+/// Binance Execution Client
+impl BinanceExecution {}
+
 #[async_trait]
 impl ExecutionClient for BinanceExecution {
     type Config = BinanceConfig;
 
     fn exchange(&self) -> Exchange {
-        Exchange::from(ExecutionId::Simulated)
+        Exchange::from(ExchangeId::BinanceFuturesUsd)
     }
 
     async fn init(config: Self::Config) -> Self {
         let client = BinanceClient::new(config.client_type);
-        let url = BinanceClient::get_url(config.client_type);
-        let (api_key, _) = BinanceClient::get_key_secret(config.client_type);
-        init_listener(&api_key, url).await;
         Self {
             client,
-            // client_type: config.client_type,
+            client_type: config.client_type,
         }
     }
 
+    async fn init_stream(&self) -> Option<BoxStream<'static, AccountEventKind>> {
+        let url = BinanceClient::get_url(self.client_type);
+        let (api_key, _) = BinanceClient::get_key_secret(self.client_type);
+        Some(init_listener(&api_key, url).await)
+    }
+
     async fn fetch_orders_open(&self) -> Result<Vec<Order<Open>>, ExecutionError> {
         todo!()
     }

barter-data-rs/src/exchange/binance/futures/account.rs → barter-execution-rs/src/execution/binance/types/account_update.rs

@@ -1,23 +1,22 @@
-use super::super::BinanceChannel;
-use crate::{
-    event::{MarketEvent, MarketIter},
-    exchange::ExchangeId,
-    subscription::account_update::{AccountUpdate, BalanceUpdate, PositionUpdate},
-    Identifier,
-};
+use super::BinanceFuturesEventType;
 use barter_integration::model::{
-    instrument::{symbol::Symbol, Instrument},
-    Exchange, PerpSide, SubscriptionId,
+    instrument::{kind::InstrumentKind, symbol::Symbol, Instrument},
+    PerpSide, Side,
 };
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
+use crate::model::{
+    balance::{Balance, SymbolBalance},
+    position::{Position, PositionMeta},
+    AccountEventKind,
+};
+
 /// [`BinanceFuturesUsd`](super::BinanceFuturesUsd) AccountUpdate messages.
 ///
 /// ### Raw Payload Examples
 /// See docs: <https://binance-docs.github.io/apidocs/futures/en/#event-balance-and-position-update>
 /// ```json
-//// ```
 /// {
 ///     "e": "ACCOUNT_UPDATE",                // Event Type
 ///     "E": 1564745798939,                   // Event Time
@@ -76,10 +75,12 @@ use serde::{Deserialize, Serialize};
 ///         ]
 ///       }
 ///   }
+/// ```
+
 #[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize, Serialize)]
 pub struct BinanceAccountUpdate {
     #[serde(alias = "e")]
-    pub event_type: String,
+    pub event_type: BinanceFuturesEventType,
     #[serde(
         alias = "E",
         deserialize_with = "barter_integration::de::de_u64_epoch_ms_as_datetime_utc"
@@ -138,6 +139,7 @@ pub struct BinanceBalanceUpdate {
 
 /// [`BinanceFuturesUsd`](super::BinanceFuturesUsd) BinancePositionUpdate.
 /// ### Raw Payload Examples
+/// ```json
 ///           {
 ///             "s":"BTCUSDT",            // Symbol
 ///             "pa":"0",                 // Position Amount
@@ -148,7 +150,7 @@ pub struct BinanceBalanceUpdate {
 ///             "mt":"isolated",          // Margin Type
 ///             "iw":"0.00000000",        // Isolated Wallet (if isolated position)
 ///             "ps":"BOTH"               // Position Side
-///           }，
+///           }
 /// ```
 
 #[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize, Serialize)]
@@ -173,73 +175,80 @@ pub struct BinancePositionUpdate {
     pub position_side: PerpSide,
 }
 
-impl Identifier<Option<SubscriptionId>> for BinanceAccountUpdate {
-    fn id(&self) -> Option<SubscriptionId> {
-        Some(SubscriptionId::from(BinanceChannel::ACCOUNT_UPDATE.0))
-    }
-}
-
-impl From<BinanceBalanceUpdate> for BalanceUpdate {
+impl From<BinanceBalanceUpdate> for SymbolBalance {
     fn from(balance_update: BinanceBalanceUpdate) -> Self {
         Self {
-            asset: balance_update.asset,
-            wallet_balance: balance_update.wallet_balance,
-            cross_wallet_balance: balance_update.cross_wallet_balance,
-            balance_change: balance_update.balance_change,
+            symbol: balance_update.asset,
+            balance: Balance {
+                // TODO not totally clear if these are total or available
+                total: balance_update.wallet_balance + balance_update.cross_wallet_balance,
+                available: balance_update.wallet_balance + balance_update.cross_wallet_balance,
+            },
         }
     }
 }
 
-impl From<BinancePositionUpdate> for PositionUpdate {
-    fn from(position_update: BinancePositionUpdate) -> Self {
+impl From<BinancePositionUpdate> for Position {
+    fn from(update: BinancePositionUpdate) -> Self {
+        let mut side = Side::Sell;
+        if update.position_amount > 0.0 {
+            side = Side::Buy;
+        }
+
         Self {
-            symbol: position_update.symbol,
-            position_amount: position_update.position_amount,
-            position_side: position_update.position_side,
-            unrealized_pnl: position_update.unrealized_pnl,
-            entry_price: position_update.entry_price,
-            breakeven_price: position_update.breakeven_price,
+            position_id: update.symbol.clone().to_string(),
+            meta: PositionMeta {
+                update_time: Utc::now(),
+                exit_balance: None,
+            },
+            instrument: Instrument::from((
+                update.symbol.clone(),
+                update.symbol.clone(),
+                InstrumentKind::Perpetual,
+            )),
+            side,
+            quantity: update.position_amount,
+            enter_avg_price_gross: update.entry_price,
+            enter_value_gross: update.entry_price * update.position_amount,
+            unrealised_profit_loss: update.unrealized_pnl,
+            realised_profit_loss: update.accumulated_realized,
+
+            // Fees
+            exit_avg_price_gross: 0.0,
+            exit_value_gross: 0.0,
+            current_symbol_price: 0.0,
+            enter_fees: Default::default(),
+            exit_fees: Default::default(),
+            enter_fees_total: 0.0,
+            exit_fees_total: 0.0,
+            current_value_gross: 0.0,
         }
     }
 }
 
-impl From<(ExchangeId, Instrument, BinanceAccountUpdate)> for MarketIter<AccountUpdate> {
-    fn from(
-        (exchange_id, instrument, account_update): (ExchangeId, Instrument, BinanceAccountUpdate),
-    ) -> Self {
-        Self(vec![Ok(MarketEvent {
-            exchange_time: account_update.event_time,
-            received_time: Utc::now(),
-            exchange: Exchange::from(exchange_id),
-            instrument,
-            kind: AccountUpdate {
-                time: account_update.event_time,
-                balance_updates: account_update
+impl From<BinanceAccountUpdate> for (AccountEventKind, AccountEventKind) {
+    fn from(update: BinanceAccountUpdate) -> Self {
+        (
+            AccountEventKind::Balances(
+                update
                     .update_data
                     .balance_updates
                     .into_iter()
-                    .map(BalanceUpdate::from)
+                    .map(SymbolBalance::from)
                     .collect(),
-                position_updates: account_update
+            ),
+            AccountEventKind::Positions(
+                update
                     .update_data
                     .position_updates
                     .into_iter()
-                    .map(PositionUpdate::from)
+                    .map(Position::from)
                     .collect(),
-            },
-        })])
+            ),
+        )
     }
 }
 
-/// Deserialize a [`BinanceAccountUpdate`] "s"  as the associated [`SubscriptionId`].
-pub fn de_liquidation_subscription_id<'de, D>(deserializer: D) -> Result<SubscriptionId, D::Error>
-where
-    D: serde::de::Deserializer<'de>,
-{
-    Deserialize::deserialize(deserializer)
-        .map(|_market: String| SubscriptionId::from(BinanceChannel::ACCOUNT_UPDATE.0))
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -310,7 +319,7 @@ mod tests {
             }"#;
 
             let expected = BinanceAccountUpdate {
-                event_type: "ACCOUNT_UPDATE".to_string(),
+                event_type: BinanceFuturesEventType::AccountUpdate,
                 event_time: datetime_utc_from_epoch_duration(Duration::from_millis(1564745798939)),
                 transaction_time: datetime_utc_from_epoch_duration(Duration::from_millis(
                     1564745798938,