accounts: add account_summary_snapshots and document that account_summary stays open after End

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

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<AccountSummarySnapshots, Error> {
+        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<AccountSummaryResult>,
+    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<Result<AccountSummarySnapshot, Error>> {
+        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;

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<RoutedItem>) {
+    let message_bus = Arc::new(MessageBusStub::default());
+    let (tx, rx) = tokio::sync::broadcast::channel::<RoutedItem>(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));
+}

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)]

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;