Trait PortfolioSource 

pub trait PortfolioSource: Send + Sync {
    // Required methods
    fn portfolio<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Portfolio, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn positions<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Position>, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn balances<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Balance>, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn subscribe_portfolio<'life0, 'async_trait>(
        &'life0 self,
        opts: SubscribeOpts,
    ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
}

The account: money and positions, as a snapshot and as a stream.

Examples

use std::time::Duration;
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::{Event, PortfolioSource, ShutdownMode, SubscribeOpts};
use brokapi_runtime::BrokerLimiter;

let (events_tx, mut events) = tokio::sync::mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await?;

// A stream: each event is a full snapshot, so the latest one wins if we read slowly.
// Subscribe first: an update that arrives before the subscription belongs to no subscription.
let sub = ud.subscribe_portfolio(SubscribeOpts::conflate()).await?;
let update = tokio::time::timeout(Duration::from_secs(8), async {
    while let Some(env) = events.recv().await {
        // The channel is shared by all subscriptions of the connector: pick ours by `sub`.
        if env.sub == Some(sub) {
            return Some(env.payload);
        }
    }
    None
})
.await?
.expect("an update arrived");
assert!(matches!(
    update,
    Event::Portfolio(_) | Event::Positions(_) | Event::Balances(_)
));

// The whole account at once, or only one part of it.
let portfolio = ud.portfolio().await?;
let balances = ud.balances().await?;
let positions = ud.positions().await?;
println!("{} currencies, {} positions", balances.len(), positions.len());
assert_eq!(portfolio.positions.len(), positions.len());

ud.shutdown(ShutdownMode::Immediate).await;

Required Methods

fn portfolio<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Portfolio, Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Reads money and positions at one moment.

Errors

• Error::RateLimited, Error::Transport: wait or repeat later.
• Error::Rejected when the broker refuses the request.

Examples

use brokapi_core::PortfolioSource;
let p = ud.portfolio().await?;
for b in &p.balances {
    assert!(!b.available.is_negative() && !b.blocked.is_negative());
}

fn positions<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Position>, Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Reads the positions only.

The BCS connector reads the whole portfolio and returns this part, so it costs the same as portfolio.

Errors

The same as portfolio.

Examples

use brokapi_core::PortfolioSource;
for p in ud.positions().await? {
    println!("{}: {}", p.instrument.key.display(), p.qty);
}

fn balances<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Balance>, Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Reads the money balances only.

The BCS connector reads the whole portfolio and returns this part, so it costs the same as portfolio.

Errors

The same as portfolio.

Examples

use brokapi_core::PortfolioSource;
let free_money: Money = ud.balances().await?.iter().map(|b| b.available).sum();
assert!(!free_money.is_negative());

fn subscribe_portfolio<'life0, 'async_trait>( &'life0 self, opts: SubscribeOpts, ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Starts a stream of portfolio updates.

Updates arrive as Event::Portfolio, or as Event::Positions and Event::Balances where the connector sends the parts separately (the T-Bank connector does). Each is a full snapshot of its part and replaces the previous one, so Backpressure::Conflate fits.

The broker may push a first snapshot as soon as its portfolio channel connects, which can be before you subscribe. Such an update has no subscription (Envelope::sub is None). Subscribe right after the connector starts and then read portfolio once for a known starting point.

Errors

• Error::Unsupported when the connector has no portfolio stream.
• Error::Transport when the subscription could not be made, also when the stream did not open in time (the call returns when the stream is open, as subscribe_order_state).

Examples

See the example of the trait, which receives an update.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors