From a9ade6b8a6e032b969c07ad44078ae81bd2d2372 Mon Sep 17 00:00:00 2001 From: Faysal Aberkane Date: Mon, 5 Oct 2026 07:51:13 +0100 Subject: [PATCH 1/3] accounts: document that account_summary stays open after End The docs called End the end marker for a batch of summaries. TWS sends End once the initial snapshot is complete, then keeps pushing changed values on the same subscription as Summary rows without another End. A consumer that treats End as a per-update boundary stops refreshing after the first snapshot, or drops the subscription and cancels the request. Document this on AccountSummaryResult and on both account_summary methods. --- src/accounts/async.rs | 7 +++++++ src/accounts/mod.rs | 7 ++++++- src/accounts/sync.rs | 7 +++++++ 3 files changed, 20 insertions(+), 1 deletion(-) diff --git a/src/accounts/async.rs b/src/accounts/async.rs index eccd8eea6..439ba989d 100644 --- a/src/accounts/async.rs +++ b/src/accounts/async.rs @@ -192,6 +192,13 @@ impl Client { /// Subscribe to account summary updates for a group of accounts. /// + /// # Subscription lifetime + /// + /// The subscription stays open until it is dropped or cancelled. TWS first sends the requested + /// tags followed by [`AccountSummaryResult::End`], then pushes changed values as further + /// [`AccountSummaryResult::Summary`] rows without another `End`. Dropping the subscription cancels + /// the request. + /// /// # Arguments /// * `group` - Set to "All" to return account summary data for all accounts, or set to a specific Advisor Account Group name. /// * `tags` - List of the desired tags. diff --git a/src/accounts/mod.rs b/src/accounts/mod.rs index 212ba07e7..729a94762 100644 --- a/src/accounts/mod.rs +++ b/src/accounts/mod.rs @@ -144,8 +144,13 @@ impl AccountSummaryTags { #[derive(Debug)] pub enum AccountSummaryResult { /// Summary of account details such as net liquidation, cash balance, etc. + /// + /// After [`End`](Self::End), TWS keeps sending rows on the same subscription when values change. Summary(AccountSummary), - /// End marker for a batch of account summaries + /// End marker for the initial snapshot of account summaries. + /// + /// An `End` does not follow the rows TWS pushes later, so a consumer that wants complete + /// updates must decide for itself when a pushed batch is finished. End, } diff --git a/src/accounts/sync.rs b/src/accounts/sync.rs index 5254208f2..f8118c737 100644 --- a/src/accounts/sync.rs +++ b/src/accounts/sync.rs @@ -171,6 +171,13 @@ impl Client { /// Requests a specific account's summary. Subscribes to the account summary as presented in the TWS' Account Summary tab. Data received is specified by using a specific tags value. /// + /// # Subscription lifetime + /// + /// The subscription stays open until it is dropped or cancelled. TWS first sends the requested + /// tags followed by [`AccountSummaryResult::End`], then pushes changed values as further + /// [`AccountSummaryResult::Summary`] rows without another `End`. Dropping the subscription cancels + /// the request. + /// /// # Arguments /// * `group` - Set to "All" to return account summary data for all accounts, or set to a specific Advisor Account Group name that has already been created in TWS Global Configuration. /// * `tags` - List of the desired tags. From 55ca4a0ae5e8b6986c8a1563f2e0c4ed6f15c4b5 Mon Sep 17 00:00:00 2001 From: Faysal Aberkane Date: Mon, 5 Oct 2026 08:05:04 +0100 Subject: [PATCH 2/3] accounts: add account_summary_snapshots for complete summary snapshots account_summary yields one row at a time. TWS sends End once, after the initial snapshot, and then pushes only the rows whose values changed, with no further End. A consumer that wants the account as a whole has to keep the latest value of every row itself and decide when a pushed batch is finished. account_summary_snapshots wraps account_summary on both clients and yields AccountSummarySnapshot, the latest value of every row by account, tag and currency. A snapshot completes at an End, or once no row has arrived for a caller-chosen quiet period, because the rows of one push arrive within milliseconds of each other. Rows received before the subscription ends are returned as a final snapshot. Dropping the wrapper cancels the subscription. The row folding and the pending-row tracking are shared by both clients. When both features are enabled the blocking type is at accounts::blocking. AccountSummary now implements Clone and PartialEq. --- CHANGELOG.md | 2 + src/accounts/async.rs | 102 +++++++++++++++++++++++++++- src/accounts/async_tests.rs | 93 ++++++++++++++++++++++++++ src/accounts/common/mod.rs | 1 + src/accounts/common/snapshots.rs | 110 +++++++++++++++++++++++++++++++ src/accounts/mod.rs | 47 ++++++++++++- src/accounts/sync.rs | 99 ++++++++++++++++++++++++++++ src/accounts/sync_tests.rs | 79 ++++++++++++++++++++++ 8 files changed, 531 insertions(+), 2 deletions(-) create mode 100644 src/accounts/common/snapshots.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b3f882d1..6cef43c6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `Client::account_summary_snapshots(&group, tags, quiet)` (async and blocking) wraps `account_summary` and yields `AccountSummarySnapshot`, the latest value of every row by account, tag and currency. TWS sends no `End` after the first snapshot and then pushes only changed rows, so a snapshot completes at an `End` or once no row has arrived for `quiet`. Blocking types are at `accounts::blocking` when both features are enabled. +- `AccountSummary` implements `Clone` and `PartialEq`. - `generic_tick::ETF_FROZEN_NAV_LAST` (`"623"`, frozen-data ETF NAV last price, tick 97) (#946). ### Deprecated diff --git a/src/accounts/async.rs b/src/accounts/async.rs index 439ba989d..2c3f05664 100644 --- a/src/accounts/async.rs +++ b/src/accounts/async.rs @@ -1,14 +1,18 @@ //! Asynchronous implementation of account management functionality +use std::time::Duration; + +use futures::StreamExt; use time::OffsetDateTime; use crate::client::ClientRequestBuilders; use crate::common::request_helpers::{self, empty_on_end_of_stream, expect_proto}; use crate::messages::OutgoingMessages; use crate::protocol::{check_version, Features}; -use crate::subscriptions::Subscription; +use crate::subscriptions::{Subscription, SubscriptionItem}; use crate::{Client, Error}; +use super::common::snapshots::SnapshotBuilder; use super::common::{decoders, encoders}; use super::types::{AccountGroup, AccountId, ContractId, ModelCode}; use super::*; @@ -231,6 +235,57 @@ impl Client { .await } + /// Subscribe to account summary updates as complete snapshots. + /// + /// Wraps [`account_summary`](Self::account_summary) and keeps the latest value of every row, so + /// each [`AccountSummarySnapshot`] holds the whole account rather than only the rows TWS pushed + /// last. TWS sends no `End` marker after the initial snapshot, so a snapshot is emitted at an + /// `End` and once no row has arrived for `quiet`. Choose `quiet` longer than the gap between the + /// rows of one push, which arrive within milliseconds of each other. + /// + /// Dropping the returned stream cancels the subscription. + /// + /// # Arguments + /// * `group` - Set to "All" to return account summary data for all accounts, or set to a specific Advisor Account Group name. + /// * `tags` - List of the desired tags. + /// * `quiet` - How long without a row completes a pushed update. + /// + /// # Examples + /// + /// ```no_run + /// use ibapi::Client; + /// use ibapi::accounts::AccountSummaryTags; + /// use ibapi::accounts::types::AccountGroup; + /// use std::time::Duration; + /// + /// #[tokio::main] + /// async fn main() { + /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed"); + /// + /// let group = AccountGroup("All".to_string()); + /// let tags = &[AccountSummaryTags::NET_LIQUIDATION]; + /// + /// let mut snapshots = client + /// .account_summary_snapshots(&group, tags, Duration::from_secs(1)) + /// .await + /// .expect("error requesting account summary"); + /// + /// while let Some(snapshot) = snapshots.next().await { + /// let snapshot = snapshot.expect("account summary error"); + /// println!("{} rows", snapshot.len()); + /// } + /// } + /// ``` + pub async fn account_summary_snapshots(&self, group: &AccountGroup, tags: &[&str], quiet: Duration) -> Result { + let subscription = self.account_summary(group, tags).await?; + + Ok(AccountSummarySnapshots { + subscription, + builder: SnapshotBuilder::default(), + quiet, + }) + } + /// Subscribe to detailed account updates for a specific account. /// /// All account values and positions will be returned initially, and then there will only be updates when there is a change in a position, or to an account value every 3 minutes if it has changed. @@ -606,6 +661,51 @@ impl Client { } } +/// Complete account summary snapshots from [`Client::account_summary_snapshots`]. +pub struct AccountSummarySnapshots { + subscription: Subscription, + builder: SnapshotBuilder, + quiet: Duration, +} + +impl std::fmt::Debug for AccountSummarySnapshots { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AccountSummarySnapshots") + .field("quiet", &self.quiet) + .finish_non_exhaustive() + } +} + +impl AccountSummarySnapshots { + /// Returns the next snapshot, or `None` once the subscription has ended. + /// + /// A snapshot completes at an `End` marker or after the `quiet` period without a row. Rows + /// received before the subscription ends are returned as a final snapshot. + pub async fn next(&mut self) -> Option> { + loop { + let item = if self.builder.has_pending() { + match tokio::time::timeout(self.quiet, self.subscription.next()).await { + Ok(item) => item, + Err(_) => return Some(Ok(self.builder.take())), + } + } else { + self.subscription.next().await + }; + + match item { + Some(Ok(SubscriptionItem::Data(result))) => { + if self.builder.apply(result) { + return Some(Ok(self.builder.take())); + } + } + Some(Ok(SubscriptionItem::Notice(_))) => {} + Some(Err(e)) => return Some(Err(e)), + None => return self.builder.has_pending().then(|| Ok(self.builder.take())), + } + } + } +} + #[cfg(test)] #[path = "async_tests.rs"] mod tests; diff --git a/src/accounts/async_tests.rs b/src/accounts/async_tests.rs index 5890b4330..c4fe3e786 100644 --- a/src/accounts/async_tests.rs +++ b/src/accounts/async_tests.rs @@ -2,6 +2,7 @@ use super::*; use crate::common::test_utils::helpers::*; use crate::messages::IncomingMessages; use crate::stubs::MessageBusStub; +use crate::subscriptions::common::{DecoderContext, RoutedItem}; use crate::subscriptions::SubscriptionItem; use crate::testdata::builders::accounts::{ account_download_end, account_summary, account_summary_end, account_update_multi, account_update_multi_end, account_value, @@ -15,9 +16,11 @@ use crate::testdata::builders::positions::{ cancel_positions, cancel_positions_multi, position, position_end, position_multi, position_multi_end, request_positions, request_positions_multi, }; use crate::testdata::builders::ResponseProtoEncoder; +use crate::transport::r#async::AsyncInternalSubscription; use crate::{server_versions, Client}; use futures::StreamExt; use std::sync::Arc; +use std::time::Duration; #[tokio::test] async fn test_positions() { @@ -671,3 +674,93 @@ async fn test_verify_message_version_error() { assert!(matches!(result, Err(Error::ServerVersion(_, _, _))), "got {result:?}"); assert_eq!(request_message_count(&message_bus), 0); } + +fn summary_frame(tag: &str, value: &str, currency: &str) -> RoutedItem { + RoutedItem::Response(proto_response( + IncomingMessages::AccountSummary, + account_summary().tag(tag).value(value).currency(currency).encode_proto(), + )) +} + +fn end_frame() -> RoutedItem { + RoutedItem::Response(proto_response(IncomingMessages::AccountSummaryEnd, account_summary_end().encode_proto())) +} + +fn snapshots_over_open_channel(quiet: Duration) -> (AccountSummarySnapshots, tokio::sync::broadcast::Sender) { + let message_bus = Arc::new(MessageBusStub::default()); + let (tx, rx) = tokio::sync::broadcast::channel::(16); + let subscription = Subscription::new_from_internal( + AsyncInternalSubscription::new(rx), + message_bus, + Some(TEST_REQ_ID_FIRST), + None, + DecoderContext::default(), + ); + + let snapshots = AccountSummarySnapshots { + subscription, + builder: SnapshotBuilder::default(), + quiet, + }; + + (snapshots, tx) +} + +fn values(snapshot: &AccountSummarySnapshot) -> Vec<(&str, &str, &str)> { + snapshot + .iter() + .map(|row| (row.tag.as_str(), row.currency.as_str(), row.value.as_str())) + .collect() +} + +#[tokio::test] +async fn test_account_summary_snapshots_emit_at_end_then_after_quiet_period() { + let (mut snapshots, tx) = snapshots_over_open_channel(Duration::from_millis(50)); + + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + tx.send(summary_frame("BuyingPower", "400.0", "USD")).unwrap(); + tx.send(end_frame()).unwrap(); + let initial = snapshots.next().await.unwrap().unwrap(); + + tx.send(summary_frame("NetLiquidation", "101.5", "USD")).unwrap(); + let refreshed = snapshots.next().await.unwrap().unwrap(); + let still_open = tokio::time::timeout(Duration::from_millis(300), snapshots.next()).await; + + assert_eq!( + values(&initial), + vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "100.0")] + ); + assert_eq!( + values(&refreshed), + vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "101.5")] + ); + assert!(still_open.is_err()); +} + +#[tokio::test] +async fn test_account_summary_snapshots_group_one_push_and_emit_on_end_without_new_rows() { + let (mut snapshots, tx) = snapshots_over_open_channel(Duration::from_millis(50)); + + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + tx.send(summary_frame("BuyingPower", "400.0", "USD")).unwrap(); + let first = snapshots.next().await.unwrap().unwrap(); + + tx.send(end_frame()).unwrap(); + let on_end = snapshots.next().await.unwrap().unwrap(); + + assert_eq!(first.len(), 2); + assert_eq!(on_end, first); +} + +#[tokio::test] +async fn test_account_summary_snapshots_return_pending_rows_when_subscription_ends() { + let (mut snapshots, tx) = snapshots_over_open_channel(Duration::from_secs(60)); + + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + drop(tx); + let last = snapshots.next().await.unwrap().unwrap(); + let after = snapshots.next().await; + + assert_eq!(values(&last), vec![("NetLiquidation", "USD", "100.0")]); + assert!(after.is_none()); +} diff --git a/src/accounts/common/mod.rs b/src/accounts/common/mod.rs index e3bf5c0d8..318017d6b 100644 --- a/src/accounts/common/mod.rs +++ b/src/accounts/common/mod.rs @@ -1,6 +1,7 @@ pub(super) mod constants; pub(super) mod decoders; pub(crate) mod encoders; +pub(super) mod snapshots; pub(super) mod stream_decoders; #[cfg(test)] diff --git a/src/accounts/common/snapshots.rs b/src/accounts/common/snapshots.rs new file mode 100644 index 000000000..496097ef4 --- /dev/null +++ b/src/accounts/common/snapshots.rs @@ -0,0 +1,110 @@ +//! Folds account summary rows into snapshots, shared by the sync and async clients. + +use super::super::{AccountSummary, AccountSummaryResult, AccountSummarySnapshot}; + +/// Accumulates rows into a snapshot and tracks whether rows arrived since the last one was taken. +#[derive(Debug, Default)] +pub(in crate::accounts) struct SnapshotBuilder { + snapshot: AccountSummarySnapshot, + pending: bool, +} + +impl SnapshotBuilder { + /// Applies one subscription item. Returns `true` when it was an `End` marker, which completes + /// a snapshot. + pub(in crate::accounts) fn apply(&mut self, result: AccountSummaryResult) -> bool { + match result { + AccountSummaryResult::Summary(summary) => { + self.insert(summary); + self.pending = true; + false + } + AccountSummaryResult::End => true, + } + } + + /// Returns `true` when rows arrived since the last snapshot was taken. + pub(in crate::accounts) fn has_pending(&self) -> bool { + self.pending + } + + /// Returns a copy of the current snapshot and clears the pending flag. + pub(in crate::accounts) fn take(&mut self) -> AccountSummarySnapshot { + self.pending = false; + self.snapshot.clone() + } + + fn insert(&mut self, summary: AccountSummary) { + let key = (summary.account.clone(), summary.tag.clone(), summary.currency.clone()); + self.snapshot.rows.insert(key, summary); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn row(account: &str, tag: &str, value: &str, currency: &str) -> AccountSummaryResult { + AccountSummaryResult::Summary(AccountSummary { + account: account.to_string(), + tag: tag.to_string(), + value: value.to_string(), + currency: currency.to_string(), + }) + } + + #[test] + fn test_later_row_replaces_earlier_value_for_same_key() { + let mut builder = SnapshotBuilder::default(); + + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + builder.apply(row("DU1", "NetLiquidation", "101.5", "USD")); + let snapshot = builder.take(); + + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot.get("DU1", "NetLiquidation", "USD").unwrap().value, "101.5"); + } + + #[test] + fn test_snapshot_keeps_rows_not_resent_and_separates_keys() { + let mut builder = SnapshotBuilder::default(); + + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + builder.apply(row("DU1", "NetLiquidation", "90.0", "EUR")); + builder.apply(row("DU2", "NetLiquidation", "50.0", "USD")); + builder.apply(row("DU1", "BuyingPower", "400.0", "USD")); + builder.take(); + builder.apply(row("DU1", "NetLiquidation", "102.0", "USD")); + let snapshot = builder.take(); + + let observed: Vec<_> = snapshot + .iter() + .map(|row| (row.account.as_str(), row.tag.as_str(), row.currency.as_str(), row.value.as_str())) + .collect(); + assert_eq!( + observed, + vec![ + ("DU1", "BuyingPower", "USD", "400.0"), + ("DU1", "NetLiquidation", "EUR", "90.0"), + ("DU1", "NetLiquidation", "USD", "102.0"), + ("DU2", "NetLiquidation", "USD", "50.0"), + ] + ); + } + + #[test] + fn test_pending_follows_rows_and_take() { + let mut builder = SnapshotBuilder::default(); + assert!(!builder.has_pending()); + + assert!(!builder.apply(row("DU1", "NetLiquidation", "100.0", "USD"))); + let pending_after_row = builder.has_pending(); + assert!(builder.apply(AccountSummaryResult::End)); + let pending_after_end = builder.has_pending(); + builder.take(); + + assert!(pending_after_row); + assert!(pending_after_end); + assert!(!builder.has_pending()); + } +} diff --git a/src/accounts/mod.rs b/src/accounts/mod.rs index 729a94762..76fd5d946 100644 --- a/src/accounts/mod.rs +++ b/src/accounts/mod.rs @@ -13,11 +13,12 @@ pub mod types; use crate::contracts::Contract; use crate::Error; use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; // Public types - always available regardless of feature flags #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))] -#[derive(Debug, Default, Serialize, Deserialize)] +#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)] /// Account information as it appears in the TWS' Account Summary Window pub struct AccountSummary { /// The account identifier. @@ -154,6 +155,38 @@ pub enum AccountSummaryResult { End, } +/// The latest account summary values, keyed by account, tag and currency. +/// +/// Built from the rows of an account summary subscription. After the initial snapshot TWS pushes +/// only the values that changed, so each snapshot holds the most recent value of every row seen +/// so far, not just the latest batch. +#[derive(Debug, Default, Clone, PartialEq)] +pub struct AccountSummarySnapshot { + rows: BTreeMap<(String, String, String), AccountSummary>, +} + +impl AccountSummarySnapshot { + /// Returns the latest row for an account, tag and currency. + pub fn get(&self, account: &str, tag: &str, currency: &str) -> Option<&AccountSummary> { + self.rows.get(&(account.to_string(), tag.to_string(), currency.to_string())) + } + + /// Iterates over the latest rows, ordered by account, tag and currency. + pub fn iter(&self) -> impl Iterator { + self.rows.values() + } + + /// Returns the number of rows. + pub fn len(&self) -> usize { + self.rows.len() + } + + /// Returns `true` when no row has been received. + pub fn is_empty(&self) -> bool { + self.rows.is_empty() + } +} + /// Aggregated profit and loss metrics for the entire account. #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))] #[derive(Debug, Default, Serialize, Deserialize)] @@ -455,5 +488,17 @@ pub struct VerificationResult { #[cfg(feature = "sync")] mod sync; +#[cfg(feature = "sync")] +pub mod blocking { + //! Blocking account types, for builds that enable both `sync` and `async`. + pub use super::sync::AccountSummarySnapshots; +} + +#[cfg(all(feature = "sync", not(feature = "async")))] +pub use sync::AccountSummarySnapshots; + +#[cfg(feature = "async")] +pub use r#async::AccountSummarySnapshots; + #[cfg(feature = "async")] mod r#async; diff --git a/src/accounts/sync.rs b/src/accounts/sync.rs index f8118c737..61ff600f4 100644 --- a/src/accounts/sync.rs +++ b/src/accounts/sync.rs @@ -1,6 +1,7 @@ //! Synchronous implementation of account management functionality use std::sync::Arc; +use std::time::Duration; use time::OffsetDateTime; @@ -8,8 +9,10 @@ use crate::client::blocking::{ClientRequestBuilders, Subscription}; use crate::common::request_helpers::{self, empty_on_end_of_stream, expect_proto}; use crate::messages::OutgoingMessages; use crate::protocol::{check_version, Features}; +use crate::subscriptions::SubscriptionItem; use crate::{client::sync::Client, Error}; +use super::common::snapshots::SnapshotBuilder; use super::common::{decoders, encoders}; use super::types::{AccountGroup, AccountId, ContractId, ModelCode}; use super::*; @@ -204,6 +207,52 @@ impl Client { }) } + /// Subscribe to account summary updates as complete snapshots. + /// + /// Wraps [`account_summary`](Self::account_summary) and keeps the latest value of every row, so + /// each [`AccountSummarySnapshot`] holds the whole account rather than only the rows TWS pushed + /// last. TWS sends no `End` marker after the initial snapshot, so a snapshot is emitted at an + /// `End` and once no row has arrived for `quiet`. Choose `quiet` longer than the gap between the + /// rows of one push, which arrive within milliseconds of each other. + /// + /// Dropping the returned iterator cancels the subscription. + /// + /// # Arguments + /// * `group` - Set to "All" to return account summary data for all accounts, or set to a specific Advisor Account Group name that has already been created in TWS Global Configuration. + /// * `tags` - List of the desired tags. + /// * `quiet` - How long without a row completes a pushed update. + /// + /// # Examples + /// + /// ```no_run + /// use ibapi::client::blocking::Client; + /// use ibapi::accounts::AccountSummaryTags; + /// use ibapi::accounts::types::AccountGroup; + /// use std::time::Duration; + /// + /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed"); + /// + /// let group = AccountGroup("All".to_string()); + /// let tags = &[AccountSummaryTags::NET_LIQUIDATION]; + /// + /// let snapshots = client + /// .account_summary_snapshots(&group, tags, Duration::from_secs(1)) + /// .expect("error requesting account summary"); + /// for snapshot in snapshots { + /// let snapshot = snapshot.expect("account summary error"); + /// println!("{} rows", snapshot.len()); + /// } + /// ``` + pub fn account_summary_snapshots(&self, group: &AccountGroup, tags: &[&str], quiet: Duration) -> Result { + let subscription = self.account_summary(group, tags)?; + + Ok(AccountSummarySnapshots { + subscription, + builder: SnapshotBuilder::default(), + quiet, + }) + } + /// Subscribes to a specific account's information and portfolio. /// /// All account values and positions will be returned initially, and then there will only be updates when there is a change in a position, or to an account value every 3 minutes if it has changed. @@ -504,6 +553,56 @@ impl Client { } } +/// Complete account summary snapshots from [`Client::account_summary_snapshots`]. +pub struct AccountSummarySnapshots { + subscription: Subscription, + builder: SnapshotBuilder, + quiet: Duration, +} + +impl std::fmt::Debug for AccountSummarySnapshots { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AccountSummarySnapshots") + .field("quiet", &self.quiet) + .finish_non_exhaustive() + } +} + +impl Iterator for AccountSummarySnapshots { + type Item = Result; + + /// Returns the next snapshot, or `None` once the subscription has ended. + /// + /// A snapshot completes at an `End` marker or after the `quiet` period without a row. Rows + /// received before the subscription ends are returned as a final snapshot. + fn next(&mut self) -> Option { + loop { + // `next_timeout` returns `None` for a timeout and for an ended subscription alike. Either + // way the pending rows are complete, and an ended subscription then returns `None` from + // the blocking `next` below. + let item = if self.builder.has_pending() { + match self.subscription.next_timeout(self.quiet) { + Some(item) => Some(item), + None => return Some(Ok(self.builder.take())), + } + } else { + self.subscription.next() + }; + + match item { + Some(Ok(SubscriptionItem::Data(result))) => { + if self.builder.apply(result) { + return Some(Ok(self.builder.take())); + } + } + Some(Ok(SubscriptionItem::Notice(_))) => {} + Some(Err(e)) => return Some(Err(e)), + None => return None, + } + } + } +} + #[cfg(test)] #[path = "sync_tests.rs"] mod tests; diff --git a/src/accounts/sync_tests.rs b/src/accounts/sync_tests.rs index 6a0c221d1..71229fc46 100644 --- a/src/accounts/sync_tests.rs +++ b/src/accounts/sync_tests.rs @@ -1,7 +1,10 @@ +use super::{AccountSummarySnapshot, AccountSummarySnapshots, SnapshotBuilder, Subscription}; use crate::accounts::types::{AccountGroup, AccountId, ContractId, ModelCode}; use crate::accounts::{AccountSummaryTags, AccountUpdateMulti}; +use crate::client::ids::RequestId; use crate::common::test_utils::helpers::*; use crate::messages::IncomingMessages; +use crate::subscriptions::common::{DecoderContext, RoutedItem}; use crate::testdata::builders::accounts::{ account_download_end, account_summary, account_summary_end, account_update_multi, account_update_multi_end, account_value, cancel_account_summary, cancel_account_updates, cancel_account_updates_multi, cancel_pnl, cancel_pnl_single, current_time, @@ -14,8 +17,11 @@ use crate::testdata::builders::positions::{ cancel_positions, cancel_positions_multi, position, position_end, position_multi, position_multi_end, request_positions, request_positions_multi, }; use crate::testdata::builders::ResponseProtoEncoder; +use crate::transport::common::Lease; +use crate::transport::SubscriptionBuilder; use crate::{client::blocking::Client, server_versions, stubs::MessageBusStub, Error}; use std::sync::Arc; +use std::time::Duration; #[test] fn test_pnl() { @@ -891,3 +897,76 @@ fn test_verify_message_version_error() { assert!(matches!(result, Err(Error::ServerVersion(_, _, _))), "got {result:?}"); assert_eq!(request_message_count(&message_bus), 0); } + +fn summary_frame(tag: &str, value: &str, currency: &str) -> RoutedItem { + RoutedItem::Response(proto_response( + IncomingMessages::AccountSummary, + account_summary().tag(tag).value(value).currency(currency).encode_proto(), + )) +} + +fn end_frame() -> RoutedItem { + RoutedItem::Response(proto_response(IncomingMessages::AccountSummaryEnd, account_summary_end().encode_proto())) +} + +fn snapshots_over_channel(quiet: Duration) -> (AccountSummarySnapshots, crossbeam::channel::Sender) { + let (sender, receiver) = crossbeam::channel::unbounded::(); + let (signaler, _signaler_rx) = crossbeam::channel::unbounded(); + let internal = SubscriptionBuilder::new() + .receiver(receiver) + .signaler(signaler) + .lease(Lease::new()) + .request_id(RequestId::nth(1)) + .build(); + let subscription = Subscription::new(Arc::new(MessageBusStub::default()), internal, DecoderContext::default()); + + let snapshots = AccountSummarySnapshots { + subscription, + builder: SnapshotBuilder::default(), + quiet, + }; + + (snapshots, sender) +} + +fn values(snapshot: &AccountSummarySnapshot) -> Vec<(&str, &str, &str)> { + snapshot + .iter() + .map(|row| (row.tag.as_str(), row.currency.as_str(), row.value.as_str())) + .collect() +} + +#[test] +fn test_account_summary_snapshots_emit_at_end_then_after_quiet_period() { + let (mut snapshots, tx) = snapshots_over_channel(Duration::from_millis(50)); + + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + tx.send(summary_frame("BuyingPower", "400.0", "USD")).unwrap(); + tx.send(end_frame()).unwrap(); + let initial = snapshots.next().unwrap().unwrap(); + + tx.send(summary_frame("NetLiquidation", "101.5", "USD")).unwrap(); + let refreshed = snapshots.next().unwrap().unwrap(); + + assert_eq!( + values(&initial), + vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "100.0")] + ); + assert_eq!( + values(&refreshed), + vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "101.5")] + ); +} + +#[test] +fn test_account_summary_snapshots_end_when_subscription_closes() { + let (mut snapshots, tx) = snapshots_over_channel(Duration::from_millis(50)); + + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + drop(tx); + let last = snapshots.next().unwrap().unwrap(); + let after = snapshots.next(); + + assert_eq!(values(&last), vec![("NetLiquidation", "USD", "100.0")]); + assert!(after.is_none()); +} From 017fb89ff98aca7c081a8cf70cd05491c6f8c52d Mon Sep 17 00:00:00 2001 From: Faysal Aberkane Date: Mon, 5 Oct 2026 10:24:33 +0100 Subject: [PATCH 3/3] accounts: address review of account_summary_snapshots Emit a snapshot only when a row changed a value since the previous one. An End with nothing changed is skipped once a snapshot has gone out, and the first End still emits, possibly empty, so a consumer always gets an initial snapshot. A repeated row with an unchanged value no longer counts as a change. Move the blocking type to client::blocking, as the dual-feature type rule asks, and drop the accounts::blocking module. Move the snapshot builder tests to a sibling test file. Add tests that go through Client::account_summary_snapshots on both clients with the message bus stub, and the grouping test on the blocking side. Document that notices are dropped and what IB documents about the push cadence, add examples to AccountSummarySnapshot::get and iter, and end the changelog bullets with the PR number. --- CHANGELOG.md | 4 +- src/accounts/async.rs | 9 ++- src/accounts/async_tests.rs | 43 ++++++++++-- src/accounts/common/snapshots.rs | 90 +++++------------------- src/accounts/common/snapshots_tests.rs | 95 ++++++++++++++++++++++++++ src/accounts/mod.rs | 29 +++++--- src/accounts/sync.rs | 9 ++- src/accounts/sync_tests.rs | 48 +++++++++++++ src/client/mod.rs | 1 + 9 files changed, 236 insertions(+), 92 deletions(-) create mode 100644 src/accounts/common/snapshots_tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cef43c6e..c39ccd47f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- `Client::account_summary_snapshots(&group, tags, quiet)` (async and blocking) wraps `account_summary` and yields `AccountSummarySnapshot`, the latest value of every row by account, tag and currency. TWS sends no `End` after the first snapshot and then pushes only changed rows, so a snapshot completes at an `End` or once no row has arrived for `quiet`. Blocking types are at `accounts::blocking` when both features are enabled. -- `AccountSummary` implements `Clone` and `PartialEq`. +- `Client::account_summary_snapshots(&group, tags, quiet)` (async and blocking) wraps `account_summary` and yields `AccountSummarySnapshot`, the latest value of every row by account, tag and currency. TWS sends one `End` after the first snapshot and none after the rows it pushes later, so a snapshot completes at an `End` or once no row has arrived for `quiet`, and only when a row changed a value. The blocking type is at `client::blocking::AccountSummarySnapshots` when both features are enabled (#957). +- `AccountSummary` implements `Clone` and `PartialEq` (#957). - `generic_tick::ETF_FROZEN_NAV_LAST` (`"623"`, frozen-data ETF NAV last price, tick 97) (#946). ### Deprecated diff --git a/src/accounts/async.rs b/src/accounts/async.rs index 2c3f05664..cad40f942 100644 --- a/src/accounts/async.rs +++ b/src/accounts/async.rs @@ -239,10 +239,15 @@ impl Client { /// /// Wraps [`account_summary`](Self::account_summary) and keeps the latest value of every row, so /// each [`AccountSummarySnapshot`] holds the whole account rather than only the rows TWS pushed - /// last. TWS sends no `End` marker after the initial snapshot, so a snapshot is emitted at an - /// `End` and once no row has arrived for `quiet`. Choose `quiet` longer than the gap between the + /// last. TWS sends one `End` marker after the initial snapshot and none after the rows it pushes + /// later, so a snapshot is emitted at an `End` and once no row has arrived for `quiet`. IB + /// documents the pushes as every three minutes for the values that changed. A snapshot is + /// emitted only when a row changed a value since the previous one, except the first, which is + /// emitted at the first `End` even when empty. Choose `quiet` longer than the gap between the /// rows of one push, which arrive within milliseconds of each other. /// + /// Notices that arrive on the subscription are dropped. + /// /// Dropping the returned stream cancels the subscription. /// /// # Arguments diff --git a/src/accounts/async_tests.rs b/src/accounts/async_tests.rs index c4fe3e786..ac5dfff20 100644 --- a/src/accounts/async_tests.rs +++ b/src/accounts/async_tests.rs @@ -738,7 +738,7 @@ async fn test_account_summary_snapshots_emit_at_end_then_after_quiet_period() { } #[tokio::test] -async fn test_account_summary_snapshots_group_one_push_and_emit_on_end_without_new_rows() { +async fn test_account_summary_snapshots_group_one_push_and_skip_end_without_changes() { let (mut snapshots, tx) = snapshots_over_open_channel(Duration::from_millis(50)); tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); @@ -746,10 +746,11 @@ async fn test_account_summary_snapshots_group_one_push_and_emit_on_end_without_n let first = snapshots.next().await.unwrap().unwrap(); tx.send(end_frame()).unwrap(); - let on_end = snapshots.next().await.unwrap().unwrap(); + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + let repeated = tokio::time::timeout(Duration::from_millis(300), snapshots.next()).await; - assert_eq!(first.len(), 2); - assert_eq!(on_end, first); + assert_eq!(values(&first), vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "100.0")]); + assert!(repeated.is_err()); } #[tokio::test] @@ -764,3 +765,37 @@ async fn test_account_summary_snapshots_return_pending_rows_when_subscription_en assert_eq!(values(&last), vec![("NetLiquidation", "USD", "100.0")]); assert!(after.is_none()); } + +#[tokio::test] +async fn test_client_account_summary_snapshots_sends_request_and_yields_snapshot_at_end() { + let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![ + proto_response( + IncomingMessages::AccountSummary, + account_summary().tag("NetLiquidation").value("100.0").currency("USD").encode_proto(), + ), + proto_response(IncomingMessages::AccountSummaryEnd, account_summary_end().encode_proto()), + ])); + let client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES); + let group = AccountGroup("All".to_string()); + let tags = &[AccountSummaryTags::NET_LIQUIDATION]; + + let mut snapshots = client + .account_summary_snapshots(&group, tags, Duration::from_secs(1)) + .await + .expect("request account_summary_snapshots failed"); + let snapshot = snapshots.next().await.unwrap().unwrap(); + drop(snapshots); + tokio::time::sleep(Duration::from_millis(50)).await; + + assert_eq!(values(&snapshot), vec![("NetLiquidation", "USD", "100.0")]); + assert_eq!(request_message_count(&message_bus), 2); + assert_request( + &message_bus, + 0, + &request_account_summary() + .request_id(TEST_REQ_ID_FIRST) + .group("All") + .tags([AccountSummaryTags::NET_LIQUIDATION]), + ); + assert_request(&message_bus, 1, &cancel_account_summary().request_id(TEST_REQ_ID_FIRST)); +} diff --git a/src/accounts/common/snapshots.rs b/src/accounts/common/snapshots.rs index 496097ef4..142a59bfe 100644 --- a/src/accounts/common/snapshots.rs +++ b/src/accounts/common/snapshots.rs @@ -2,28 +2,29 @@ use super::super::{AccountSummary, AccountSummaryResult, AccountSummarySnapshot}; -/// Accumulates rows into a snapshot and tracks whether rows arrived since the last one was taken. +/// Accumulates rows into a snapshot and tracks whether a row changed a value since the last +/// snapshot was taken. #[derive(Debug, Default)] pub(in crate::accounts) struct SnapshotBuilder { snapshot: AccountSummarySnapshot, pending: bool, + emitted: bool, } impl SnapshotBuilder { - /// Applies one subscription item. Returns `true` when it was an `End` marker, which completes - /// a snapshot. + /// Applies one subscription item. Returns `true` when it was an `End` marker that completes a + /// snapshot: one with changed rows since the last snapshot, or the first one, which may be empty. pub(in crate::accounts) fn apply(&mut self, result: AccountSummaryResult) -> bool { match result { AccountSummaryResult::Summary(summary) => { self.insert(summary); - self.pending = true; false } - AccountSummaryResult::End => true, + AccountSummaryResult::End => self.pending || !self.emitted, } } - /// Returns `true` when rows arrived since the last snapshot was taken. + /// Returns `true` when a row changed a value since the last snapshot was taken. pub(in crate::accounts) fn has_pending(&self) -> bool { self.pending } @@ -31,80 +32,21 @@ impl SnapshotBuilder { /// Returns a copy of the current snapshot and clears the pending flag. pub(in crate::accounts) fn take(&mut self) -> AccountSummarySnapshot { self.pending = false; + self.emitted = true; self.snapshot.clone() } fn insert(&mut self, summary: AccountSummary) { let key = (summary.account.clone(), summary.tag.clone(), summary.currency.clone()); - self.snapshot.rows.insert(key, summary); + let changed = self.snapshot.rows.get(&key) != Some(&summary); + + if changed { + self.snapshot.rows.insert(key, summary); + self.pending = true; + } } } #[cfg(test)] -mod tests { - use super::*; - - fn row(account: &str, tag: &str, value: &str, currency: &str) -> AccountSummaryResult { - AccountSummaryResult::Summary(AccountSummary { - account: account.to_string(), - tag: tag.to_string(), - value: value.to_string(), - currency: currency.to_string(), - }) - } - - #[test] - fn test_later_row_replaces_earlier_value_for_same_key() { - let mut builder = SnapshotBuilder::default(); - - builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); - builder.apply(row("DU1", "NetLiquidation", "101.5", "USD")); - let snapshot = builder.take(); - - assert_eq!(snapshot.len(), 1); - assert_eq!(snapshot.get("DU1", "NetLiquidation", "USD").unwrap().value, "101.5"); - } - - #[test] - fn test_snapshot_keeps_rows_not_resent_and_separates_keys() { - let mut builder = SnapshotBuilder::default(); - - builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); - builder.apply(row("DU1", "NetLiquidation", "90.0", "EUR")); - builder.apply(row("DU2", "NetLiquidation", "50.0", "USD")); - builder.apply(row("DU1", "BuyingPower", "400.0", "USD")); - builder.take(); - builder.apply(row("DU1", "NetLiquidation", "102.0", "USD")); - let snapshot = builder.take(); - - let observed: Vec<_> = snapshot - .iter() - .map(|row| (row.account.as_str(), row.tag.as_str(), row.currency.as_str(), row.value.as_str())) - .collect(); - assert_eq!( - observed, - vec![ - ("DU1", "BuyingPower", "USD", "400.0"), - ("DU1", "NetLiquidation", "EUR", "90.0"), - ("DU1", "NetLiquidation", "USD", "102.0"), - ("DU2", "NetLiquidation", "USD", "50.0"), - ] - ); - } - - #[test] - fn test_pending_follows_rows_and_take() { - let mut builder = SnapshotBuilder::default(); - assert!(!builder.has_pending()); - - assert!(!builder.apply(row("DU1", "NetLiquidation", "100.0", "USD"))); - let pending_after_row = builder.has_pending(); - assert!(builder.apply(AccountSummaryResult::End)); - let pending_after_end = builder.has_pending(); - builder.take(); - - assert!(pending_after_row); - assert!(pending_after_end); - assert!(!builder.has_pending()); - } -} +#[path = "snapshots_tests.rs"] +mod tests; diff --git a/src/accounts/common/snapshots_tests.rs b/src/accounts/common/snapshots_tests.rs new file mode 100644 index 000000000..ca55a4cfe --- /dev/null +++ b/src/accounts/common/snapshots_tests.rs @@ -0,0 +1,95 @@ +use super::*; + +fn row(account: &str, tag: &str, value: &str, currency: &str) -> AccountSummaryResult { + AccountSummaryResult::Summary(AccountSummary { + account: account.to_string(), + tag: tag.to_string(), + value: value.to_string(), + currency: currency.to_string(), + }) +} + +#[test] +fn test_later_row_replaces_earlier_value_for_same_key() { + let mut builder = SnapshotBuilder::default(); + + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + builder.apply(row("DU1", "NetLiquidation", "101.5", "USD")); + let snapshot = builder.take(); + + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot.get("DU1", "NetLiquidation", "USD").unwrap().value, "101.5"); +} + +#[test] +fn test_snapshot_keeps_rows_not_resent_and_separates_keys() { + let mut builder = SnapshotBuilder::default(); + + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + builder.apply(row("DU1", "NetLiquidation", "90.0", "EUR")); + builder.apply(row("DU2", "NetLiquidation", "50.0", "USD")); + builder.apply(row("DU1", "BuyingPower", "400.0", "USD")); + builder.take(); + builder.apply(row("DU1", "NetLiquidation", "102.0", "USD")); + let snapshot = builder.take(); + + let observed: Vec<_> = snapshot + .iter() + .map(|row| (row.account.as_str(), row.tag.as_str(), row.currency.as_str(), row.value.as_str())) + .collect(); + assert_eq!( + observed, + vec![ + ("DU1", "BuyingPower", "USD", "400.0"), + ("DU1", "NetLiquidation", "EUR", "90.0"), + ("DU1", "NetLiquidation", "USD", "102.0"), + ("DU2", "NetLiquidation", "USD", "50.0"), + ] + ); +} + +#[test] +fn test_pending_follows_rows_and_take() { + let mut builder = SnapshotBuilder::default(); + assert!(!builder.has_pending()); + + assert!(!builder.apply(row("DU1", "NetLiquidation", "100.0", "USD"))); + let pending_after_row = builder.has_pending(); + assert!(builder.apply(AccountSummaryResult::End)); + let pending_after_end = builder.has_pending(); + builder.take(); + + assert!(pending_after_row); + assert!(pending_after_end); + assert!(!builder.has_pending()); +} + +#[test] +fn test_row_with_unchanged_value_is_not_pending() { + let mut builder = SnapshotBuilder::default(); + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + builder.take(); + + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + let pending_after_repeat = builder.has_pending(); + builder.apply(row("DU1", "NetLiquidation", "100.5", "USD")); + let pending_after_change = builder.has_pending(); + + assert!(!pending_after_repeat); + assert!(pending_after_change); +} + +#[test] +fn test_end_completes_first_snapshot_even_when_empty_then_only_with_changes() { + let mut builder = SnapshotBuilder::default(); + + let first_end = builder.apply(AccountSummaryResult::End); + builder.take(); + let end_without_change = builder.apply(AccountSummaryResult::End); + builder.apply(row("DU1", "NetLiquidation", "100.0", "USD")); + let end_with_change = builder.apply(AccountSummaryResult::End); + + assert!(first_end); + assert!(!end_without_change); + assert!(end_with_change); +} diff --git a/src/accounts/mod.rs b/src/accounts/mod.rs index 76fd5d946..878830966 100644 --- a/src/accounts/mod.rs +++ b/src/accounts/mod.rs @@ -151,7 +151,8 @@ pub enum AccountSummaryResult { /// End marker for the initial snapshot of account summaries. /// /// An `End` does not follow the rows TWS pushes later, so a consumer that wants complete - /// updates must decide for itself when a pushed batch is finished. + /// updates must decide for itself when a pushed batch is finished. See + /// [`Client::account_summary_snapshots`](crate::Client::account_summary_snapshots). End, } @@ -167,11 +168,29 @@ pub struct AccountSummarySnapshot { impl AccountSummarySnapshot { /// Returns the latest row for an account, tag and currency. + /// + /// # Examples + /// + /// ``` + /// use ibapi::accounts::AccountSummarySnapshot; + /// + /// let snapshot = AccountSummarySnapshot::default(); + /// assert!(snapshot.get("DU1234567", "NetLiquidation", "USD").is_none()); + /// ``` pub fn get(&self, account: &str, tag: &str, currency: &str) -> Option<&AccountSummary> { self.rows.get(&(account.to_string(), tag.to_string(), currency.to_string())) } /// Iterates over the latest rows, ordered by account, tag and currency. + /// + /// # Examples + /// + /// ``` + /// use ibapi::accounts::AccountSummarySnapshot; + /// + /// let snapshot = AccountSummarySnapshot::default(); + /// assert_eq!(snapshot.iter().count(), 0); + /// ``` pub fn iter(&self) -> impl Iterator { self.rows.values() } @@ -486,13 +505,7 @@ pub struct VerificationResult { // Feature-specific implementations #[cfg(feature = "sync")] -mod sync; - -#[cfg(feature = "sync")] -pub mod blocking { - //! Blocking account types, for builds that enable both `sync` and `async`. - pub use super::sync::AccountSummarySnapshots; -} +pub(crate) mod sync; #[cfg(all(feature = "sync", not(feature = "async")))] pub use sync::AccountSummarySnapshots; diff --git a/src/accounts/sync.rs b/src/accounts/sync.rs index 61ff600f4..4e97390a7 100644 --- a/src/accounts/sync.rs +++ b/src/accounts/sync.rs @@ -211,10 +211,15 @@ impl Client { /// /// Wraps [`account_summary`](Self::account_summary) and keeps the latest value of every row, so /// each [`AccountSummarySnapshot`] holds the whole account rather than only the rows TWS pushed - /// last. TWS sends no `End` marker after the initial snapshot, so a snapshot is emitted at an - /// `End` and once no row has arrived for `quiet`. Choose `quiet` longer than the gap between the + /// last. TWS sends one `End` marker after the initial snapshot and none after the rows it pushes + /// later, so a snapshot is emitted at an `End` and once no row has arrived for `quiet`. IB + /// documents the pushes as every three minutes for the values that changed. A snapshot is + /// emitted only when a row changed a value since the previous one, except the first, which is + /// emitted at the first `End` even when empty. Choose `quiet` longer than the gap between the /// rows of one push, which arrive within milliseconds of each other. /// + /// Notices that arrive on the subscription are dropped. + /// /// Dropping the returned iterator cancels the subscription. /// /// # Arguments diff --git a/src/accounts/sync_tests.rs b/src/accounts/sync_tests.rs index 71229fc46..f300d2199 100644 --- a/src/accounts/sync_tests.rs +++ b/src/accounts/sync_tests.rs @@ -970,3 +970,51 @@ fn test_account_summary_snapshots_end_when_subscription_closes() { assert_eq!(values(&last), vec![("NetLiquidation", "USD", "100.0")]); assert!(after.is_none()); } + +#[test] +fn test_account_summary_snapshots_group_one_push_and_skip_end_without_changes() { + let (mut snapshots, tx) = snapshots_over_channel(Duration::from_millis(50)); + + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + tx.send(summary_frame("BuyingPower", "400.0", "USD")).unwrap(); + let first = snapshots.next().unwrap().unwrap(); + + tx.send(end_frame()).unwrap(); + tx.send(summary_frame("NetLiquidation", "101.0", "USD")).unwrap(); + let second = snapshots.next().unwrap().unwrap(); + + assert_eq!(values(&first), vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "100.0")]); + assert_eq!(values(&second), vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "101.0")]); +} + +#[test] +fn test_client_account_summary_snapshots_sends_request_and_yields_snapshot_at_end() { + let message_bus = Arc::new(MessageBusStub::with_ordered_responses(vec![ + proto_response( + IncomingMessages::AccountSummary, + account_summary().tag("NetLiquidation").value("100.0").currency("USD").encode_proto(), + ), + proto_response(IncomingMessages::AccountSummaryEnd, account_summary_end().encode_proto()), + ])); + let client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES); + let group = AccountGroup("All".to_string()); + let tags = &[AccountSummaryTags::NET_LIQUIDATION]; + + let mut snapshots = client + .account_summary_snapshots(&group, tags, Duration::from_secs(1)) + .expect("request account_summary_snapshots failed"); + let snapshot = snapshots.next().unwrap().unwrap(); + drop(snapshots); + + assert_eq!(values(&snapshot), vec![("NetLiquidation", "USD", "100.0")]); + assert_eq!(request_message_count(&message_bus), 2); + assert_request( + &message_bus, + 0, + &request_account_summary() + .request_id(TEST_REQ_ID_FIRST) + .group("All") + .tags([AccountSummaryTags::NET_LIQUIDATION]), + ); + assert_request(&message_bus, 1, &cancel_account_summary().request_id(TEST_REQ_ID_FIRST)); +} diff --git a/src/client/mod.rs b/src/client/mod.rs index c57ce3c47..2aab6f570 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -37,6 +37,7 @@ pub mod r#async; #[cfg(feature = "sync")] pub mod blocking { pub use super::sync::Client; + pub use crate::accounts::sync::AccountSummarySnapshots; pub(crate) use crate::client::builders::blocking::{ClientRequestBuilders, SubscriptionBuilderExt}; pub use crate::client::builders::client_builder::sync_impl::ClientBuilder; pub use crate::subscriptions::notice_stream::sync_impl::{NoticeStream, NoticeStreamIter};