diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b3f882d1..c39ccd47f 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 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 eccd8eea6..cad40f942 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::*; @@ -192,6 +196,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. @@ -224,6 +235,62 @@ 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 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 + /// * `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. @@ -599,6 +666,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..ac5dfff20 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,128 @@ 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_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(); + tx.send(summary_frame("BuyingPower", "400.0", "USD")).unwrap(); + let first = snapshots.next().await.unwrap().unwrap(); + + tx.send(end_frame()).unwrap(); + tx.send(summary_frame("NetLiquidation", "100.0", "USD")).unwrap(); + let repeated = tokio::time::timeout(Duration::from_millis(300), snapshots.next()).await; + + assert_eq!(values(&first), vec![("BuyingPower", "USD", "400.0"), ("NetLiquidation", "USD", "100.0")]); + assert!(repeated.is_err()); +} + +#[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()); +} + +#[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/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..142a59bfe --- /dev/null +++ b/src/accounts/common/snapshots.rs @@ -0,0 +1,52 @@ +//! 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 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 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); + false + } + AccountSummaryResult::End => self.pending || !self.emitted, + } + } + + /// 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 + } + + /// 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()); + let changed = self.snapshot.rows.get(&key) != Some(&summary); + + if changed { + self.snapshot.rows.insert(key, summary); + self.pending = true; + } + } +} + +#[cfg(test)] +#[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 212ba07e7..878830966 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. @@ -144,11 +145,67 @@ 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. See + /// [`Client::account_summary_snapshots`](crate::Client::account_summary_snapshots). 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. + /// + /// # 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() + } + + /// 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)] @@ -448,7 +505,13 @@ pub struct VerificationResult { // Feature-specific implementations #[cfg(feature = "sync")] -mod sync; +pub(crate) mod sync; + +#[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 5254208f2..4e97390a7 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::*; @@ -171,6 +174,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. @@ -197,6 +207,57 @@ 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 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 + /// * `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. @@ -497,6 +558,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..f300d2199 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,124 @@ 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()); +} + +#[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};