Struct UserDataConnector 

pub struct UserDataConnector { /* private fields */ }

BCS user data connector: orders, stop orders, portfolio and the history of fills for one account.

It implements [OrderExecutor], [StopOrderExecutor], [ExecutionSource] and [PortfolioSource]. The account is the one the refresh token in the config belongs to. One connector per account. Start it with start. Commands return their answer directly, and subscriptions deliver [Envelope]s into the channel given to start.

Rules that protect the account

• Write access. The config must be built with with_write(true) (the environment switch BCS_IS_WRITE=true does the same for BcsConfig::from_env). Without it the connector asks for a read-only token. Whether the broker then refuses the orders with a definite error is not verified against the live API. Never rely on it as a safety lock, use a read-only config only for reading.
• Reconcile gate. Trading commands (place, replace, cancel, place_stop, cancel_stop) are refused with [Error::NotReconciled] until reconcile has succeeded once. This forces the bot to compare its own ledger of orders with the broker before it sends a new order, so a restart cannot double an order that was placed before the crash. Reads (order_state, active_orders, portfolio, executions) are not gated.
• Degraded. If a lossless subscription overflows (the bot reads too slowly), the gate closes and commands return [Error::Degraded] until the next successful reconcile.
• No retry of place. A cid that has been sent is spent. If place returns [Error::Unknown] (no answer, a break, HTTP 5xx), the order may exist. Ask order_state or run reconcile. Calling place again with the same cid returns [Error::Rejected] with code duplicate_place, because a retry could create a second order.
• The gate is checked twice. Once at the start, and again when the command’s turn in the queue of its instrument comes: a gate that closed while it waited stops it, and nothing is sent.
• A cut of the order events socket is told. Event::Reconnected on the order state and executions subscriptions, the gate closes until reconcile.
• Quantity is in pieces, a whole number, never rounded: 0.4 or 2.5 is Rejected with code invalid_qty.
• Rate limit. Quota is taken from the shared limiter before the request. A refusal is [Error::RateLimited] and nothing is sent, so the same cid can be used again.

Examples

The whole path against the BCS imitation: reconcile, place, cancel, and two kinds of refusal.

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_conformance::REJECT_PRICE;
use brokapi_core::{
    ClientOrderId, Error, InstrumentId, InstrumentKey, NativeId, OrderExecutor, OrderState, OrderType,
    PlaceOrderRequest, Price, Qty, ShutdownMode, Side,
};
use brokapi_mocks::bcs::REJECT_IN_BODY_PRICE;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

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

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);
let limit_buy = |price: Price| PlaceOrderRequest {
    cid: ClientOrderId::new_v4(),
    instrument: sber.clone(),
    side: Side::Buy,
    order_type: OrderType::Limit,
    qty: Qty::from(1),
    price: Some(price),
};

// Before the first reconcile the gate is closed.
assert!(matches!(ud.place(limit_buy(Price::from(100))).await, Err(Error::NotReconciled)));
// An empty ledger: the bot has no orders of its own to compare. Opens the gate.
ud.reconcile(&[]).await?;

// Place, then cancel.
let req = limit_buy(Price::from(100));
let cid = req.cid;
let ack = ud.place(req).await?;
println!("broker order id: {}", ack.broker_id.as_str());
ud.cancel(cid).await?;
assert!(matches!(ud.order_state(cid).await?.state, OrderState::Cancelled { .. }));

// The broker says no, way one: HTTP 400. Nothing was created.
match ud.place(limit_buy(REJECT_PRICE)).await {
    Err(Error::Rejected { code, .. }) => assert_eq!(code, "validation"),
    other => panic!("expected a refusal, got {other:?}"),
}
// Way two: HTTP 200, but the report in the answer has orderStatus = 8.
match ud.place(limit_buy(Price::new(REJECT_IN_BODY_PRICE))).await {
    Err(Error::Rejected { code, message }) => {
        assert_eq!(code, "rejected");
        println!("reason from the broker: {message}");
    }
    other => panic!("expected a refusal, got {other:?}"),
}

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

Implementations

impl UserDataConnector

pub async fn start( cfg: UserDataConfig, limiter: Arc<BrokerLimiter>, out: Sender<Envelope>, ) -> Result<Self, Error>

Starts the connector: gets the first access token and opens the order events and portfolio WebSockets.

The trading gate starts closed: call reconcile before the first command, or place returns [Error::NotReconciled]. The token refresh does not spend rate limit quota. The two WebSocket sessions run in the background and reconnect on their own. Order status changes and fills arrive on one channel, orders/events/ws.

limiter should be the one shared by every connector of this BCS account, because the limits belong to the token. out receives the envelopes of all subscriptions and must be read.

Errors

• [Error::Config] when the HTTP or WebSocket client cannot be built from cfg.
• [Error::Transport] when the token refresh fails: no connection, a timeout, or the auth server refuses the refresh token. Exact auth server answers: not verified against the live API.
• [Error::RateLimited] when the auth server answers HTTP 429.

Examples

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::{OrderExecutor, ShutdownMode};
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await?;
// The gate is closed until the first reconcile.
ud.reconcile(&[]).await?;

A broker that cannot be reached is a [Error::Transport]:

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::Error;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;
use brokapi_bcs_protocol::config::BcsConfig;

// Port 1 of the local host: nothing listens there.
let cfg = BcsConfig::for_mock("any-token", "http://127.0.0.1:1").with_write(true);
let (events_tx, _events) = mpsc::channel(16);
let limiter = BrokerLimiter::new(&default_limits());
let started = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await;
assert!(matches!(started, Err(Error::Transport(_))));

pub fn raw(&self) -> BcsRaw

Returns a handle for any other method of the BCS API, with this connector’s token, endpoint and rate limiter.

Orders sent through the handle bypass this connector’s order state: there are no events for them, they are not in the ledger and reconcile does not know them. Use [OrderExecutor] for orders, and the handle for reads and for methods the traits do not wrap. Use call_command for anything that changes the account. See BcsRaw for the rules of a call.

Examples

A raw read, spending the quota of the name GetQuotes:

use brokapi_bcs_protocol::http::models::{QuoteInstrument, QuotesRequest};
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::ShutdownMode;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await?;
let quotes = ud
    .raw()
    .call("GetQuotes", |http| async move {
        http.get_quotes(&QuotesRequest { instruments: vec![QuoteInstrument::new("SBER", "TQBR")] }).await
    })
    .await?;
assert!(!quotes.records.is_empty());
ud.shutdown(ShutdownMode::Immediate).await;

pub fn price_grid(&self) -> &PriceGrid

The price steps this connector checks order prices against. Fill it from TradingParams (grid.learn(&params)) to have a price off the step refused before anything is sent. See [PriceGrid].

pub fn caps(&self) -> UserDataCaps

Describes what this connector can do, without any request.

The answer is fixed for BCS: a place request is deduplicated by label only ([PlaceIdempotency::LabelOnly], so never repeat a place), replace and stop orders are supported, fills are streamed, there is no command channel over WebSocket and no sandbox.

Examples

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::ShutdownMode;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await?;
let caps = ud.caps();
assert!(caps.supports_replace && caps.supports_stop_orders && caps.streams_executions);
assert!(!caps.sandbox);

pub async fn shutdown(self, mode: ShutdownMode) -> ShutdownReport

Stops the connector and reports what was left unfinished. Consumes the connector.

The command gate closes first, so nothing new is sent. Orders that were placed but have no answer yet are listed in inflight_abandoned and marked unknown locally. The broker is not asked to cancel anything: orders that exist on the exchange stay alive, and a later reconcile finds them. With [ShutdownMode::Graceful] the events that wait in the subscription buffers first get time to reach the bot’s channel, until deadline. The WebSocket sessions are not unsubscribed message by message, so unsubscribed is 0.

Examples

use std::time::Duration;
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::ShutdownMode;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter, events_tx).await?;
let report = ud.shutdown(ShutdownMode::Graceful { deadline: Duration::from_secs(1) }).await;
assert!(report.inflight_abandoned.is_empty());

Trait Implementations

impl Connector for UserDataConnector

type Caps = UserDataCaps

The type that describes what this connector can do.

fn caps(&self) -> UserDataCaps

What this connector can do. Read it before you use a feature that not every broker has.

fn shutdown<'async_trait>( self, mode: ShutdownMode, ) -> Pin<Box<dyn Future<Output = ShutdownReport> + Send + 'async_trait>>

where Self: 'async_trait,

Stops the connector: closes its streams, gives up commands that have no answer yet (they become [Error::Unknown], see [ShutdownReport]) and ends its tasks. Takes the connector by value, so it cannot be used afterwards.

impl ExecutionSource for UserDataConnector

fn executions<'life0, 'async_trait>( &'life0 self, range: TimeRange, ) -> Pin<Box<dyn Future<Output = Result<Vec<Execution>, Error>> + Send + 'async_trait>>

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

Own trades in range (POST /trades/search), every page.

The connector asks for the biggest page (size = 100) and reads page after page until totalPages. Every page spends quota of SearchTrades. BCS lists at most the 1000 most recent trades since 26.01.2026: when the answer says it holds that many (totalRecords of 1000 or more), older trades of the range may be missing, and the answer is [Error::Rejected] with the code range_too_long instead of a list that looks complete: read a shorter range.

Errors

[Error::RateLimited], [Error::Transport] or [Error::Rejected] (including range_too_long), as for any read.

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

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

Fills pushed by the orders/events/ws channel: messages with lastQuantity > 0, each executionId once. Lossless: a fill is never dropped (an overflow is a visible Overflow).

impl OrderExecutor for UserDataConnector

fn place<'life0, 'async_trait>( &'life0 self, req: PlaceOrderRequest, ) -> Pin<Box<dyn Future<Output = Result<OrderAck, Error>> + Send + 'async_trait>>

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

POST /orders/sync, not POST /orders. The synchronous method waits for the broker’s decision and answers with the order’s state, so (1) a refusal (orderStatus = 8, rejectReason) is known right here as Rejected, not a minute later from the stream, and (2) the ack carries the broker’s real order id (orderId), the same one the order list and the stream use. The plain method only says “request accepted”.

fn cancel<'life0, 'async_trait>( &'life0 self, cid: ClientOrderId, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

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

POST /orders/cancel/sync (the order is named in the body). Synchronous for the same reason as place: “cancelled” means the broker says so, an order that was filled first is Rejected (already_filled), and an answer that leaves the order alive is Unknown, never “ok”.

fn replace<'life0, 'async_trait>( &'life0 self, req: ReplaceOrderRequest, ) -> Pin<Box<dyn Future<Output = Result<OrderAck, Error>> + Send + 'async_trait>>

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

POST /orders/edit/sync: the broker cancels the original and puts a new order under the new cid. The quantity and price the caller leaves out are the order’s current ones (one GET /orders), because the request needs orderQuantity and a limit order needs its price. An order that is partly filled needs an explicit quantity: whether orderQuantity of an edit is the total or the rest is not documented (live question 7).

fn order_state<'life0, 'async_trait>( &'life0 self, cid: ClientOrderId, ) -> Pin<Box<dyn Future<Output = Result<OrderSnapshot, Error>> + Send + 'async_trait>>

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

GET /orders?orderIdType=1&orderId=<cid>. The answer is about the order asked for, so its cid is the one asked.

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

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

Active orders (orderStatus = 3, plain order types) from POST /orders/search. The list carries the broker’s order id only: our cid comes from what this process has seen or from one lookup by the exchange id; an order we cannot name gets a stable foreign cid (foreign_cid).

fn reconcile<'life0, 'life1, 'async_trait>( &'life0 self, ledger: &'life1 [ClientOrderId], ) -> Pin<Box<dyn Future<Output = Result<ReconcileReport, Error>> + Send + 'async_trait>>

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

Compares the ledger with the broker. The decision for each cid of the ledger is the answer of GET /orders for that order, not the list of orders (the list is cut at the 1000 most recent orders and is cut by a time window, so an order that is alive may be missing from it):

• the order is alive (new, partly filled, being replaced, or being cancelled): confirmed, with the snapshot of that answer;
• the order is over (filled, cancelled, rejected), or the broker does not know the cid (not_found): closed_offline;
• the answer has no report, or any other error: the reconcile fails with that error and the gate stays as it was. Nothing is guessed.

foreign are the active orders of the list that are not in the ledger. Success opens the gate and ends Degraded.

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

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

Starts the stream of changes of your orders.

impl PortfolioSource for UserDataConnector

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.

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.

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.

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.

impl StopOrderExecutor for UserDataConnector

Stop orders are BCS conditional orders (orderType = "5"): stopPrice for a stop-loss and stop-limit, takePrice for a take-profit, market or limit execution. How the request is built, and every choice the docs leave open (lifetime, trigger direction, the meaning of price, TP+SL pairs), is in crate::map::stop; each of them is a choice made from the docs. Both commands use the synchronous methods, for the reasons given at place.

There is no stream of stop order statuses: orders/events/ws documents only order types 1 and 2, whether it carries conditional orders is a live question, so subscribe_stop_orders is Unsupported.

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

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

Conditional orders waiting for their trigger: POST /orders/search, orderStatus = 3, types 4..7 and 11. A TP+SL pair (type 7) is shown as its two legs with the same ids. cid is None when the order is not one this connector can name.

fn place_stop<'life0, 'async_trait>( &'life0 self, req: PlaceStopOrderRequest, ) -> Pin<Box<dyn Future<Output = Result<OrderAck, Error>> + Send + 'async_trait>>

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

Puts a stop order and waits for the broker’s decision.

fn cancel_stop<'life0, 'async_trait>( &'life0 self, stop: StopOrderRef, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

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

Cancels a stop order that has not triggered.

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

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

Stream of stop order statuses. Unsupported where the broker has no such stream.