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

rustrade_execution/client/ibkr/
account.rs
