Implements account updates

examples/account_summary.rs

@@ -15,7 +15,7 @@ fn main() {
     for update in &subscription {
         match update {
             AccountSummaries::Summary(summary) => println!("{summary:?}"),
-            AccountSummaries::End => subscription.cancel().expect("cancel failed"),
+            AccountSummaries::End => subscription.cancel(),
         }
     }
 }

examples/account_updates.rs

@@ -0,0 +1,20 @@
+use ibapi::accounts::AccountUpdate;
+use ibapi::Client;
+
+fn main() {
+    env_logger::init();
+
+    let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
+
+    let account = "DU1234567";
+
+    let subscription = client.account_updates(account).expect("error requesting account updates");
+    for update in &subscription {
+        println!("{update:?}");
+
+        // stop after full initial update
+        if let AccountUpdate::End = update {
+            subscription.cancel();
+        }
+    }
+}

examples/account_updates_multi.rs

@@ -0,0 +1,22 @@
+use ibapi::accounts::AccountUpdateMulti;
+use ibapi::Client;
+
+fn main() {
+    env_logger::init();
+
+    let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
+
+    let account = Some("DU1234567");
+
+    let subscription = client
+        .account_updates_multi(account, None)
+        .expect("error requesting account updates multi");
+    for update in &subscription {
+        println!("{update:?}");
+
+        // stop after full initial update
+        if let AccountUpdateMulti::End = update {
+            subscription.cancel();
+        }
+    }
+}

examples/readme_realtime_data_1.rs

@@ -19,6 +19,6 @@ fn main() {
         println!("bar: {bar:?}");
 
         // when your algorithm is done, cancel subscription
-        subscription.cancel().expect("cancel failed");
+        subscription.cancel();
     }
 }

examples/readme_realtime_data_2.rs

@@ -22,7 +22,7 @@ fn main() {
         println!("NVDA {}, AAPL {}", bar_nvda.close, bar_aapl.close);
 
         // when your algorithm is done, cancel subscription
-        subscription_aapl.cancel().expect("cancel failed");
-        subscription_nvda.cancel().expect("cancel failed");
+        subscription_aapl.cancel();
+        subscription_nvda.cancel();
     }
 }

src/accounts.rs

@@ -273,31 +273,60 @@ pub struct FamilyCode {
 
 /// Account's information, portfolio and last update time
 #[allow(clippy::large_enum_variant)]
-pub enum AccountUpdates {
+#[derive(Debug)]
+pub enum AccountUpdate {
     /// Receives the subscribed account's information.
-    Value(AccountValue),
+    AccountValue(AccountValue),
     /// Receives the subscribed account's portfolio.
-    Portfolio(AccountPortfolio),
+    PortfolioValue(AccountPortfolioValue),
     /// Receives the last time on which the account was updated.
-    Time(AccountTime),
+    UpdateTime(AccountUpdateTime),
     /// Notifies when all the account’s information has finished.
     End,
 }
 
+impl Subscribable<AccountUpdate> for AccountUpdate {
+    const RESPONSE_MESSAGE_IDS: &[IncomingMessages] = &[
+        IncomingMessages::AccountValue,
+        IncomingMessages::PortfolioValue,
+        IncomingMessages::AccountUpdateTime,
+        IncomingMessages::AccountDownloadEnd,
+    ];
+
+    fn decode(server_version: i32, message: &mut ResponseMessage) -> Result<Self, Error> {
+        match message.message_type() {
+            IncomingMessages::AccountValue => Ok(AccountUpdate::AccountValue(decoders::decode_account_value(message)?)),
+            IncomingMessages::PortfolioValue => Ok(AccountUpdate::PortfolioValue(decoders::decode_account_portfolio_value(
+                server_version,
+                message,
+            )?)),
+            IncomingMessages::AccountUpdateTime => Ok(AccountUpdate::UpdateTime(decoders::decode_account_update_time(message)?)),
+            IncomingMessages::AccountDownloadEnd => Ok(AccountUpdate::End),
+            message => Err(Error::Simple(format!("unexpected message: {message:?}"))),
+        }
+    }
+
+    fn cancel_message(server_version: i32, _request_id: Option<i32>) -> Result<RequestMessage, Error> {
+        encoders::encode_cancel_account_updates(server_version)
+    }
+}
+
 /// A value of subscribed account's information.
+#[derive(Debug, Default)]
 pub struct AccountValue {
     /// The value being updated.
     pub key: String,
     /// Current value
     pub value: String,
-    /// The currency inn which the value is expressed.
+    /// The currency in which the value is expressed.
     pub currency: String,
     /// The account identifier.
-    pub account: String,
+    pub account: Option<String>,
 }
 
 /// Subscribed account's portfolio.
-pub struct AccountPortfolio {
+#[derive(Debug, Default)]
+pub struct AccountPortfolioValue {
     /// The Contract for which a position is held.
     pub contract: Contract,
     /// The number of positions held.
@@ -310,18 +339,61 @@ pub struct AccountPortfolio {
     pub average_cost: f64,
     /// Daily unrealized profit and loss on the position.
     pub unrealized_pnl: f64,
-    /// Daily realized profit and loss on the position.   
+    /// Daily realized profit and loss on the position.
     pub realized_pnl: f64,
     /// Account identifier for the update.
-    pub account: String,
+    pub account: Option<String>,
 }
 
 /// Last time at which the account was updated.
-pub struct AccountTime {
+#[derive(Debug, Default)]
+pub struct AccountUpdateTime {
     /// The last update system time.
     pub timestamp: String,
 }
 
+/// Account's information, portfolio and last update time
+#[allow(clippy::large_enum_variant)]
+#[derive(Debug)]
+pub enum AccountUpdateMulti {
+    /// Receives the subscribed account's information.
+    AccountMultiValue(AccountMultiValue),
+    /// Notifies when all the account’s information has finished.
+    End,
+}
+
+// Provides the account updates.
+#[derive(Debug, Default)]
+pub struct AccountMultiValue {
+    /// he account with updates.
+    pub account: String,
+    /// The model code with updates.
+    pub model_code: String,
+    /// The name of parameter.
+    pub key: String,
+    /// The value of parameter.
+    pub value: String,
+    /// The currency of parameter.
+    pub currency: String,
+}
+
+impl Subscribable<AccountUpdateMulti> for AccountUpdateMulti {
+    const RESPONSE_MESSAGE_IDS: &[IncomingMessages] = &[IncomingMessages::AccountUpdateMulti, IncomingMessages::AccountUpdateMultiEnd];
+
+    fn decode(_server_version: i32, message: &mut ResponseMessage) -> Result<Self, Error> {
+        match message.message_type() {
+            IncomingMessages::AccountUpdateMulti => Ok(AccountUpdateMulti::AccountMultiValue(decoders::decode_account_multi_value(message)?)),
+            IncomingMessages::AccountUpdateMultiEnd => Ok(AccountUpdateMulti::End),
+            message => Err(Error::Simple(format!("unexpected message: {message:?}"))),
+        }
+    }
+
+    fn cancel_message(server_version: i32, request_id: Option<i32>) -> Result<RequestMessage, Error> {
+        let request_id = request_id.expect("Request ID required to encode cancel account updates multi");
+        encoders::encode_cancel_account_updates_multi(server_version, request_id)
+    }
+}
+
 // Subscribes to position updates for all accessible accounts.
 // All positions sent initially, and then only updates as positions change.
 pub(crate) fn positions(client: &Client) -> Result<Subscription<PositionUpdate>, Error> {
@@ -412,6 +484,27 @@ pub fn account_summary<'a>(client: &'a Client, group: &str, tags: &[&str]) -> Re
     Ok(Subscription::new(client, subscription))
 }
 
+pub fn account_updates<'a>(client: &'a Client, account: &str) -> Result<Subscription<'a, AccountUpdate>, Error> {
+    let request = encoders::encode_request_account_updates(client.server_version(), account)?;
+    let subscription = client.send_shared_request(OutgoingMessages::RequestAccountData, request)?;
+
+    Ok(Subscription::new(client, subscription))
+}
+
+pub fn account_updates_multi<'a>(
+    client: &'a Client,
+    account: Option<&str>,
+    model_code: Option<&str>,
+) -> Result<Subscription<'a, AccountUpdateMulti>, Error> {
+    client.check_server_version(server_versions::MODELS_SUPPORT, "It does not support account updates multi requests.")?;
+
+    let request_id = client.next_request_id();
+    let request = encoders::encode_request_account_updates_multi(request_id, account, model_code)?;
+    let subscription = client.send_request(request_id, request)?;
+
+    Ok(Subscription::new(client, subscription))
+}
+
 pub fn managed_accounts(client: &Client) -> Result<Vec<String>, Error> {
     let request = encoders::encode_request_managed_accounts()?;
     let subscription = client.send_shared_request(OutgoingMessages::RequestManagedAccounts, request)?;

src/accounts/decoders.rs

@@ -1,8 +1,10 @@
-use crate::contracts::SecurityType;
+use crate::contracts::{Contract, SecurityType};
 use crate::messages::ResponseMessage;
 use crate::{server_versions, Error};
 
-use super::{AccountSummary, FamilyCode, PnL, PnLSingle, Position, PositionMulti};
+use super::{
+    AccountMultiValue, AccountPortfolioValue, AccountSummary, AccountUpdateTime, AccountValue, FamilyCode, PnL, PnLSingle, Position, PositionMulti,
+};
 
 pub(crate) fn decode_position(message: &mut ResponseMessage) -> Result<Position, Error> {
     message.skip(); // message type
@@ -144,5 +146,98 @@ pub(crate) fn decode_account_summary(_server_version: i32, message: &mut Respons
     })
 }
 
+pub(crate) fn decode_account_value(message: &mut ResponseMessage) -> Result<AccountValue, Error> {
+    message.skip(); // message type
+
+    let message_version = message.next_int()?;
+
+    let mut account_value = AccountValue {
+        key: message.next_string()?,
+        value: message.next_string()?,
+        currency: message.next_string()?,
+        ..Default::default()
+    };
+
+    if message_version >= 2 {
+        account_value.account = Some(message.next_string()?);
+    }
+
+    Ok(account_value)
+}
+
+pub(crate) fn decode_account_portfolio_value(server_version: i32, message: &mut ResponseMessage) -> Result<AccountPortfolioValue, Error> {
+    message.skip(); // message type
+
+    let message_version = message.next_int()?;
+
+    let mut contract = Contract::default();
+    if message_version >= 6 {
+        contract.contract_id = message.next_int()?;
+    }
+    contract.symbol = message.next_string()?;
+    contract.security_type = SecurityType::from(&message.next_string()?);
+    contract.last_trade_date_or_contract_month = message.next_string()?;
+    contract.strike = message.next_double()?;
+    contract.right = message.next_string()?;
+    if message_version >= 7 {
+        contract.multiplier = message.next_string()?;
+        contract.primary_exchange = message.next_string()?;
+    }
+    contract.currency = message.next_string()?;
+    if message_version >= 2 {
+        contract.local_symbol = message.next_string()?;
+    }
+    if message_version >= 8 {
+        contract.trading_class = message.next_string()?;
+    }
+
+    let mut portfolio_value = AccountPortfolioValue {
+        contract,
+        ..Default::default()
+    };
+
+    portfolio_value.position = message.next_double()?;
+    portfolio_value.market_price = message.next_double()?;
+    portfolio_value.market_value = message.next_double()?;
+    if message_version >= 3 {
+        portfolio_value.average_cost = message.next_double()?;
+        portfolio_value.unrealized_pnl = message.next_double()?;
+        portfolio_value.realized_pnl = message.next_double()?;
+    }
+    if message_version >= 4 {
+        portfolio_value.account = Some(message.next_string()?);
+    }
+    if message_version == 6 && server_version == 39 {
+        portfolio_value.contract.primary_exchange = message.next_string()?
+    }
+
+    Ok(portfolio_value)
+}
+
+pub(crate) fn decode_account_update_time(message: &mut ResponseMessage) -> Result<AccountUpdateTime, Error> {
+    message.skip(); // message type
+    message.skip(); // version
+
+    Ok(AccountUpdateTime {
+        timestamp: message.next_string()?,
+    })
+}
+
+pub(crate) fn decode_account_multi_value(message: &mut ResponseMessage) -> Result<AccountMultiValue, Error> {
+    message.skip(); // message type
+    message.skip(); // message version
+    message.skip(); // request id
+
+    let value = AccountMultiValue {
+        account: message.next_string()?,
+        model_code: message.next_string()?,
+        key: message.next_string()?,
+        value: message.next_string()?,
+        currency: message.next_string()?,
+    };
+
+    Ok(value)
+}
+
 #[cfg(test)]
 mod tests;

src/accounts/decoders/tests.rs

@@ -1,4 +1,4 @@
-use crate::{accounts::AccountSummaryTags, server_versions};
+use crate::{accounts::AccountSummaryTags, server_versions, testdata::responses};
 
 #[test]
 fn test_decode_positions() {
@@ -119,3 +119,16 @@ fn test_decode_account_summary() {
     assert_eq!(account_summary.value, "FA", "account_summary.value");
     assert_eq!(account_summary.currency, "", "account_summary.currency");
 }
+
+#[test]
+fn test_decode_account_multi_value() {
+    let mut message = super::ResponseMessage::from_simple(responses::ACCOUNT_UPDATE_MULTI_CURRENCY);
+
+    let value = super::decode_account_multi_value(&mut message).expect("error decoding account multi value");
+
+    assert_eq!(value.account, "DU1234567", "value.account");
+    assert_eq!(value.model_code, "", "value.model_code");
+    assert_eq!(value.key, "Currency", "value.key");
+    assert_eq!(value.value, "USD", "value.value");
+    assert_eq!(value.currency, "USD", "value.currency");
+}