rustrade_execution/client/ibkr/
account.rs

1use crate::{
2    balance::{AssetBalance, Balance},
3    error::UnindexedClientError,
4    position::{Position, PositionReport},
5};
6use chrono::{DateTime, Utc};
7use fnv::FnvHashMap;
8use ibapi::contracts::SecurityType;
9use rust_decimal::Decimal;
10use rustrade_instrument::{
11    asset::name::AssetNameExchange, instrument::name::InstrumentNameExchange,
12};
13use smol_str::SmolStr;
14use std::str::FromStr;
15use tracing::warn;
16
17/// Aggregated balance data per currency from AccountSummary events.
18#[derive(Debug, Default)]
19pub struct BalanceAggregator {
20    // SmolStr avoids heap allocation for short currency codes (USD, EUR, etc.)
21    balances: FnvHashMap<SmolStr, CurrencyBalance>,
22}
23
24#[derive(Debug, Default, Clone)]
25struct CurrencyBalance {
26    total_cash: Option<Decimal>,
27    available_funds: Option<Decimal>,
28}
29
30impl BalanceAggregator {
31    pub fn new() -> Self {
32        Self::default()
33    }
34
35    /// Process an AccountSummary event.
36    pub fn process(&mut self, summary: &ibapi::accounts::AccountSummary) {
37        let currency = SmolStr::new(&summary.currency);
38        let entry = self.balances.entry(currency).or_default();
39
40        match summary.tag.as_str() {
41            "TotalCashValue" => {
42                // Parse directly to Decimal to preserve precision
43                match Decimal::from_str(&summary.value) {
44                    Ok(val) => entry.total_cash = Some(val),
45                    Err(e) => {
46                        warn!(tag = %summary.tag, value = %summary.value, error = %e, "Failed to parse balance")
47                    }
48                }
49            }
50            "AvailableFunds" => match Decimal::from_str(&summary.value) {
51                Ok(val) => entry.available_funds = Some(val),
52                Err(e) => {
53                    warn!(tag = %summary.tag, value = %summary.value, error = %e, "Failed to parse balance")
54                }
55            },
56            _ => {}
57        }
58    }
59
60    /// Convert aggregated data to rustrade AssetBalance list.
61    pub fn to_balances(&self) -> Vec<AssetBalance<AssetNameExchange>> {
62        let now = Utc::now();
63        self.balances
64            .iter()
65            .filter_map(|(currency, bal)| {
66                let total = bal.total_cash?;
67                let free = bal.available_funds.unwrap_or(total);
68
69                Some(AssetBalance {
70                    asset: AssetNameExchange::from(currency.as_str()),
71                    balance: Balance::new(total, free),
72                    time_exchange: now,
73                })
74            })
75            .collect()
76    }
77
78    /// Clear all aggregated data.
79    pub fn clear(&mut self) {
80        self.balances.clear();
81    }
82}
83
84/// Collects IB position reports into at most one [`Position`] per instrument.
85///
86/// IB reports each position per account. For each account and instrument, the latest report
87/// wins: IB streams a new report whenever a position changes, and on `ibapi` 4.2.0 the read can
88/// start with stale replies to earlier failed calls (see `POSITION_STREAM_TIMEOUT`).
89///
90/// When more than one account holds the same instrument, the first account to report a non-zero
91/// quantity, in IB's reporting order, is kept and the others are dropped with a warning. Summing them would report a
92/// position that no single account holds.
93#[derive(Debug, Default)]
94pub(crate) struct PositionAggregator {
95    /// Latest report per account and instrument, in the order each pair was first reported.
96    reports: Vec<(InstrumentNameExchange, ibapi::accounts::Position)>,
97    index: FnvHashMap<(String, InstrumentNameExchange), usize>,
98    /// Whether the last thing IB sent was the end of a listing (`PositionEnd`), so every
99    /// position it holds has been reported.
100    listed_all: bool,
101}
102
103impl PositionAggregator {
104    /// Note that IB sent a position report, whether or not it is for a registered instrument.
105    ///
106    /// A listing is then in progress, and an earlier end marker no longer vouches for it: after
107    /// a connection drop the read can start with a stale listing, complete with its end marker,
108    /// followed by the current one.
109    pub(crate) fn report_seen(&mut self) {
110        self.listed_all = false;
111    }
112
113    /// Note that IB marked the end of its listing (`PositionEnd`).
114    pub(crate) fn listing_ended(&mut self) {
115        self.listed_all = true;
116    }
117
118    /// Record a report for `instrument`, replacing any earlier one from the same account.
119    pub(crate) fn process(
120        &mut self,
121        instrument: InstrumentNameExchange,
122        position: ibapi::accounts::Position,
123    ) {
124        let key = (position.account.clone(), instrument);
125        match self.index.get(&key) {
126            Some(&i) => self.reports[i].1 = position,
127            None => {
128                self.index.insert(key.clone(), self.reports.len());
129                self.reports.push((key.1, position));
130            }
131        }
132    }
133
134    /// The [`PositionReport`] of each instrument in an `account_snapshot`.
135    ///
136    /// Each reported instrument is [`PositionReport::Open`], or [`PositionReport::Flat`] when
137    /// every account reported zero, as IB does for a position closed today. `requested` are the
138    /// requested instruments that IB's reports can be attributed to. Those IB did not report are
139    /// [`PositionReport::Flat`] when the read ended on a complete listing (see
140    /// [`Self::listing_ended`]). Otherwise the listing may be incomplete, so they are left out,
141    /// with a warning.
142    ///
143    /// # Errors
144    ///
145    /// As [`Self::into_positions`].
146    pub(crate) fn into_reports<'a>(
147        self,
148        now: DateTime<Utc>,
149        requested: impl IntoIterator<Item = &'a InstrumentNameExchange>,
150    ) -> Result<Vec<(InstrumentNameExchange, PositionReport)>, UnindexedClientError> {
151        let listed_all = self.listed_all;
152        let mut reports: Vec<_> = self
153            .into_positions(now)?
154            .into_iter()
155            .map(|(instrument, position)| {
156                let report = position.map_or(PositionReport::Flat, PositionReport::Open);
157                (instrument, report)
158            })
159            .collect();
160        let unlisted: Vec<_> = requested
161            .into_iter()
162            .filter(|instrument| !reports.iter().any(|(listed, _)| listed == *instrument))
163            .cloned()
164            .collect();
165        if listed_all {
166            reports.extend(
167                unlisted
168                    .into_iter()
169                    .map(|instrument| (instrument, PositionReport::Flat)),
170            );
171        } else if !unlisted.is_empty() {
172            warn!(
173                unlisted = unlisted.len(),
174                "IB did not finish listing positions during the read; requested instruments it \
175                 did not list are left out rather than reported flat"
176            );
177        }
178        Ok(reports)
179    }
180
181    /// One entry per reported instrument, in the order instruments were first reported.
182    ///
183    /// The [`Position`] is `None` when every account reported a zero quantity.
184    ///
185    /// # Errors
186    ///
187    /// [`UnindexedClientError::Internal`] if a quantity does not convert to a [`Decimal`].
188    pub(crate) fn into_positions(
189        self,
190        now: DateTime<Utc>,
191    ) -> Result<Vec<(InstrumentNameExchange, Option<Position>)>, UnindexedClientError> {
192        let mut positions: Vec<(InstrumentNameExchange, Option<Position>)> = Vec::new();
193        let mut by_instrument = FnvHashMap::default();
194        for (instrument, report) in self.reports {
195            let position = convert_position(&report, now)?;
196            match by_instrument.get(&instrument) {
197                None => {
198                    by_instrument.insert(instrument.clone(), positions.len());
199                    positions.push((instrument, position));
200                }
201                Some(&i) => {
202                    let kept = &mut positions[i].1;
203                    match (kept.is_some(), position) {
204                        // A zero quantity in another account adds nothing.
205                        (_, None) => {}
206                        (false, position) => *kept = position,
207                        (true, Some(_)) => warn!(
208                            instrument = %instrument,
209                            account = %report.account,
210                            "IB reports a position in this instrument in more than one account; \
211                             keeping the first account's and dropping this one"
212                        ),
213                    }
214                }
215            }
216        }
217        Ok(positions)
218    }
219}
220
221/// Convert one IB position report to a [`Position`], or `None` if its quantity is zero.
222///
223/// - `quantity`: IB's signed position, negative when short. `ibapi` 4.2.0 hands it over as an
224///   `f64`, converted with `Decimal::try_from`, which rounds to the float's precision of about 15
225///   significant digits (0.1 stays 0.1) rather than keeping its exact binary expansion.
226/// - `entry_price`: see [`entry_price`].
227/// - `time_exchange`: `now`, since IB does not timestamp positions.
228fn convert_position(
229    report: &ibapi::accounts::Position,
230    now: DateTime<Utc>,
231) -> Result<Option<Position>, UnindexedClientError> {
232    let quantity = Decimal::try_from(report.position).map_err(|e| {
233        UnindexedClientError::Internal(format!(
234            "IB position quantity {} for contract {} is not a decimal: {e}",
235            report.position, report.contract.contract_id
236        ))
237    })?;
238    if quantity.is_zero() {
239        return Ok(None);
240    }
241    Ok(Some(Position::new(
242        quantity,
243        entry_price(report),
244        None,
245        None,
246        None,
247        None,
248        now,
249    )))
250}
251
252/// IB's average cost, quoted as orders are priced.
253///
254/// IB's average cost is per contract, so for a future or an option it is the price multiplied by
255/// the contract multiplier; it is divided back out here. It also includes commissions.
256///
257/// `None`, logged unless the cost is simply absent, when:
258/// - the average cost is zero, which is how `ibapi` 4.2.0 decodes an unset one;
259/// - it does not convert to a [`Decimal`];
260/// - IB sends no positive multiplier for a contract other than a stock or a forex pair, whose cost
261///   is already per unit.
262fn entry_price(report: &ibapi::accounts::Position) -> Option<Decimal> {
263    if report.average_cost == 0.0 {
264        return None;
265    }
266    let contract = &report.contract;
267    let average_cost = Decimal::try_from(report.average_cost)
268        .inspect_err(|e| {
269            warn!(
270                contract_id = contract.contract_id,
271                average_cost = report.average_cost,
272                error = %e,
273                "IB average cost is not a decimal; leaving the entry price unset"
274            )
275        })
276        .ok()?;
277    let multiplier = Decimal::from_str(&contract.multiplier)
278        .ok()
279        .filter(|m| m.is_sign_positive() && !m.is_zero());
280    match (multiplier, &contract.security_type) {
281        (Some(multiplier), _) => average_cost.checked_div(multiplier),
282        (None, SecurityType::Stock | SecurityType::ForexPair) => Some(average_cost),
283        (None, security_type) => {
284            warn!(
285                contract_id = contract.contract_id,
286                %security_type,
287                multiplier = %contract.multiplier,
288                "IB sent no contract multiplier; leaving the entry price unset"
289            );
290            None
291        }
292    }
293}
294
295#[cfg(test)]
296#[allow(clippy::unwrap_used)] // Test code: panics are the correct failure mode
297mod tests {
298    use super::*;
299    use ibapi::accounts::AccountSummary;
300    use rust_decimal_macros::dec;
301    use std::str::FromStr;
302
303    fn mock_account_summary(tag: &str, value: &str, currency: &str) -> AccountSummary {
304        AccountSummary {
305            account: "DU123456".to_string(),
306            tag: tag.to_string(),
307            value: value.to_string(),
308            currency: currency.to_string(),
309        }
310    }
311
312    #[test]
313    fn test_balance_aggregator() {
314        let mut agg = BalanceAggregator::new();
315
316        agg.process(&mock_account_summary("TotalCashValue", "10000.50", "USD"));
317        agg.process(&mock_account_summary("AvailableFunds", "8000.25", "USD"));
318        agg.process(&mock_account_summary("TotalCashValue", "5000.00", "EUR"));
319
320        let balances = agg.to_balances();
321        assert_eq!(balances.len(), 2);
322
323        let usd = balances.iter().find(|b| b.asset.as_ref() == "USD").unwrap();
324        assert_eq!(usd.balance.total, Decimal::from_str("10000.50").unwrap());
325        assert_eq!(usd.balance.free, Decimal::from_str("8000.25").unwrap());
326
327        let eur = balances.iter().find(|b| b.asset.as_ref() == "EUR").unwrap();
328        assert_eq!(eur.balance.total, Decimal::from_str("5000.00").unwrap());
329        assert_eq!(eur.balance.free, Decimal::from_str("5000.00").unwrap());
330    }
331
332    fn report(
333        account: &str,
334        security_type: SecurityType,
335        multiplier: &str,
336        quantity: f64,
337        average_cost: f64,
338    ) -> ibapi::accounts::Position {
339        ibapi::accounts::Position {
340            account: account.to_string(),
341            contract: ibapi::contracts::Contract {
342                contract_id: 42,
343                security_type,
344                multiplier: multiplier.to_string(),
345                ..Default::default()
346            },
347            position: quantity,
348            average_cost,
349        }
350    }
351
352    fn instrument(name: &str) -> InstrumentNameExchange {
353        InstrumentNameExchange::from(name)
354    }
355
356    fn now() -> DateTime<Utc> {
357        DateTime::from_timestamp(1_700_000_000, 0).unwrap()
358    }
359
360    #[test]
361    fn test_convert_position_stock_long() {
362        let position =
363            convert_position(&report("DU1", SecurityType::Stock, "", 10.0, 150.25), now())
364                .unwrap()
365                .unwrap();
366        assert_eq!(
367            position,
368            Position::new(dec!(10), Some(dec!(150.25)), None, None, None, None, now())
369        );
370    }
371
372    #[test]
373    fn test_convert_position_short_keeps_sign() {
374        let position = convert_position(&report("DU1", SecurityType::Stock, "", -5.0, 20.0), now())
375            .unwrap()
376            .unwrap();
377        assert_eq!(position.quantity, dec!(-5));
378        assert_eq!(position.entry_price, Some(dec!(20)));
379    }
380
381    #[test]
382    fn test_convert_position_fractional_quantity_is_rounded_to_float_precision() {
383        // 0.3 is not exact in binary; the conversion must not keep the float's expansion.
384        let position = convert_position(&report("DU1", SecurityType::Stock, "", 0.3, 100.0), now())
385            .unwrap()
386            .unwrap();
387        assert_eq!(position.quantity, dec!(0.3));
388    }
389
390    #[test]
391    fn test_convert_position_zero_quantity_is_none() {
392        assert_eq!(
393            convert_position(&report("DU1", SecurityType::Stock, "", 0.0, 150.0), now()).unwrap(),
394            None
395        );
396    }
397
398    #[test]
399    fn test_convert_position_non_finite_quantity_is_internal_error() {
400        let result = convert_position(
401            &report("DU1", SecurityType::Stock, "", f64::NAN, 1.0),
402            now(),
403        );
404        assert!(matches!(result, Err(UnindexedClientError::Internal(_))));
405    }
406
407    #[test]
408    fn test_entry_price_divides_by_multiplier() {
409        // Option: 2.35 premium x 100 multiplier.
410        let option = report("DU1", SecurityType::Option, "100", -2.0, 235.0);
411        assert_eq!(entry_price(&option), Some(dec!(2.35)));
412        // Future: 5000.25 x 50 multiplier.
413        let future = report("DU1", SecurityType::Future, "50", 1.0, 250_012.5);
414        assert_eq!(entry_price(&future), Some(dec!(5000.25)));
415    }
416
417    #[test]
418    fn test_entry_price_without_multiplier() {
419        // A stock or forex pair's average cost is already per unit.
420        assert_eq!(
421            entry_price(&report("DU1", SecurityType::Stock, "", 1.0, 150.0)),
422            Some(dec!(150))
423        );
424        assert_eq!(
425            entry_price(&report("DU1", SecurityType::ForexPair, "0", 1.0, 1.085)),
426            Some(dec!(1.085))
427        );
428        // A derivative's cannot be scaled without one.
429        assert_eq!(
430            entry_price(&report("DU1", SecurityType::Future, "", 1.0, 250_000.0)),
431            None
432        );
433        assert_eq!(
434            entry_price(&report("DU1", SecurityType::Option, "0", 1.0, 235.0)),
435            None
436        );
437    }
438
439    #[test]
440    fn test_entry_price_unset_or_invalid_average_cost_is_none() {
441        assert_eq!(
442            entry_price(&report("DU1", SecurityType::Stock, "", 1.0, 0.0)),
443            None
444        );
445        assert_eq!(
446            entry_price(&report("DU1", SecurityType::Stock, "", 1.0, f64::MAX)),
447            None
448        );
449        assert_eq!(
450            entry_price(&report("DU1", SecurityType::Stock, "", 1.0, f64::NAN)),
451            None
452        );
453    }
454
455    #[test]
456    fn test_position_aggregator_latest_report_per_account_wins() {
457        let mut agg = PositionAggregator::default();
458        agg.process(
459            instrument("AAPL"),
460            report("DU1", SecurityType::Stock, "", 10.0, 150.0),
461        );
462        agg.process(
463            instrument("MSFT"),
464            report("DU1", SecurityType::Stock, "", 3.0, 400.0),
465        );
466        agg.process(
467            instrument("AAPL"),
468            report("DU1", SecurityType::Stock, "", 12.0, 151.0),
469        );
470
471        let positions = agg.into_positions(now()).unwrap();
472        assert_eq!(positions.len(), 2);
473        assert_eq!(positions[0].0, instrument("AAPL"));
474        assert_eq!(positions[0].1.as_ref().unwrap().quantity, dec!(12));
475        assert_eq!(positions[1].0, instrument("MSFT"));
476    }
477
478    #[test]
479    fn test_position_aggregator_closed_position_is_listed_without_position() {
480        let mut agg = PositionAggregator::default();
481        agg.process(
482            instrument("AAPL"),
483            report("DU1", SecurityType::Stock, "", 10.0, 150.0),
484        );
485        agg.process(
486            instrument("AAPL"),
487            report("DU1", SecurityType::Stock, "", 0.0, 0.0),
488        );
489
490        assert_eq!(
491            agg.into_positions(now()).unwrap(),
492            vec![(instrument("AAPL"), None)]
493        );
494    }
495
496    #[test]
497    fn test_position_aggregator_later_flat_account_and_update_keep_latest() {
498        let mut agg = PositionAggregator::default();
499        agg.process(
500            instrument("AAPL"),
501            report("DU2", SecurityType::Stock, "", 7.0, 150.0),
502        );
503        // A flat account after a held one changes nothing.
504        agg.process(
505            instrument("AAPL"),
506            report("DU1", SecurityType::Stock, "", 0.0, 0.0),
507        );
508        // DU2's newer report replaces its earlier one.
509        agg.process(
510            instrument("AAPL"),
511            report("DU2", SecurityType::Stock, "", 9.0, 151.0),
512        );
513
514        let positions = agg.into_positions(now()).unwrap();
515        assert_eq!(positions.len(), 1);
516        assert_eq!(positions[0].1.as_ref().unwrap().quantity, dec!(9));
517    }
518
519    #[test]
520    fn test_position_aggregator_non_finite_quantity_is_internal_error() {
521        let mut agg = PositionAggregator::default();
522        agg.process(
523            instrument("AAPL"),
524            report("DU1", SecurityType::Stock, "", f64::NAN, 150.0),
525        );
526
527        assert!(matches!(
528            agg.into_positions(now()),
529            Err(UnindexedClientError::Internal(_))
530        ));
531    }
532
533    #[test]
534    fn test_position_aggregator_multi_account_keeps_first_non_zero() {
535        let mut agg = PositionAggregator::default();
536        // DU1 is flat, so it must not hide DU2's position.
537        agg.process(
538            instrument("AAPL"),
539            report("DU1", SecurityType::Stock, "", 0.0, 0.0),
540        );
541        agg.process(
542            instrument("AAPL"),
543            report("DU2", SecurityType::Stock, "", 7.0, 150.0),
544        );
545        // A third account's position is dropped, not summed.
546        agg.process(
547            instrument("AAPL"),
548            report("DU3", SecurityType::Stock, "", 4.0, 149.0),
549        );
550
551        let positions = agg.into_positions(now()).unwrap();
552        assert_eq!(positions.len(), 1);
553        assert_eq!(positions[0].1.as_ref().unwrap().quantity, dec!(7));
554    }
555
556    #[test]
557    fn test_balance_aggregator_clear() {
558        let mut agg = BalanceAggregator::new();
559        agg.process(&mock_account_summary("TotalCashValue", "1000", "USD"));
560        assert_eq!(agg.to_balances().len(), 1);
561
562        agg.clear();
563        assert!(agg.to_balances().is_empty());
564    }
565
566    #[test]
567    fn reports_are_open_or_flat_and_add_unlisted_instruments_only_after_a_complete_listing() {
568        let requested = [instrument("AAPL"), instrument("MSFT")];
569        let reports = |events: &[&str]| {
570            let mut agg = PositionAggregator::default();
571            for event in events {
572                match *event {
573                    "end" => agg.listing_ended(),
574                    // A report IB sends for an instrument that is not registered.
575                    "other" => agg.report_seen(),
576                    name => {
577                        agg.report_seen();
578                        let quantity = if name == "F" { 0.0 } else { 3.0 };
579                        agg.process(
580                            instrument(name),
581                            report("DU1", SecurityType::Stock, "", quantity, 10.0),
582                        );
583                    }
584                }
585            }
586            agg.into_reports(now(), &requested).unwrap()
587        };
588        let aapl = || {
589            (
590                instrument("AAPL"),
591                PositionReport::from_position(
592                    convert_position(&report("DU1", SecurityType::Stock, "", 3.0, 10.0), now())
593                        .unwrap()
594                        .unwrap(),
595                ),
596            )
597        };
598
599        assert_eq!(
600            reports(&["AAPL", "F", "end"]),
601            vec![
602                aapl(),
603                (instrument("F"), PositionReport::Flat),
604                (instrument("MSFT"), PositionReport::Flat),
605            ]
606        );
607        // No end marker: MSFT's absence establishes nothing.
608        assert_eq!(
609            reports(&["AAPL", "F"]),
610            vec![aapl(), (instrument("F"), PositionReport::Flat)]
611        );
612        // A stale listing's end marker, then a fresh listing that stalls: not complete.
613        assert_eq!(reports(&["AAPL", "end", "other"]), vec![aapl()]);
614        // A stale listing, then the fresh one to its end: complete.
615        assert_eq!(
616            reports(&["other", "end", "AAPL", "end"]),
617            vec![aapl(), (instrument("MSFT"), PositionReport::Flat)]
618        );
619    }
620}
rustrade_execution/client/ibkr/account.rs

rustrade_execution/client/ibkr/
account.rs
