Trait OrderExecutor 

pub trait OrderExecutor: Send + Sync {
    // Required methods
    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;
    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;
    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;
    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;
    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;
    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;
    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;
}

Orders: put, cancel, replace, look at them, follow them, and reconcile with your ledger.

The rules that matter:

• YOU make the ClientOrderId of every order, a fresh UUID, and keep it in your own ledger.
• Trading commands (place, cancel, replace) are refused until the first successful reconcile, so start by calling reconcile (with an empty slice if you have no ledger).
• A command that got no answer is Error::Unknown and is NEVER repeated by the library, because a repeated place could create a second order. Look at the order first.
• Shutdown does not cancel orders at the exchange.

Examples

The whole path against the BCS connector on its mock: reconcile, follow the order stream, place, read the state, cancel.

use std::time::Duration;
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{UserDataConfig, UserDataConnector};
use brokapi_core::{
    ClientOrderId, Event, InstrumentId, InstrumentKey, NativeId, OrderExecutor, OrderState,
    OrderType, PlaceOrderRequest, Price, Qty, ShutdownMode, Side,
};
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?;

// 1. Commands wait for this. We have no ledger yet, so we pass an empty one.
let report = ud.reconcile(&[]).await?;
assert!(report.confirmed.is_empty() && report.foreign.is_empty());

// 2. Subscribe BEFORE placing, so no event of our order is missed.
let _sub = ud.subscribe_order_state().await?; // returns when the stream is open

// 3. Place a limit order far below the market, so it stays working.
let cid = ClientOrderId::new_v4();
let ack = ud
    .place(PlaceOrderRequest {
        cid,
        instrument: InstrumentId::new(
            InstrumentKey::new("SBER", "TQBR"),
            NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
        ),
        side: Side::Buy,
        order_type: OrderType::Limit,
        qty: Qty::from(1),
        price: Some(Price::from(100)),
    })
    .await?;
assert_eq!(ack.cid, cid);

// 4. The stream tells about it, and a read agrees.
let seen = tokio::time::timeout(Duration::from_secs(8), async {
    while let Some(env) = events.recv().await {
        if let Event::Order(o) = env.payload {
            if o.cid == cid {
                return Some(o);
            }
        }
    }
    None
})
.await?
.expect("an event for our order");
assert!(!matches!(seen.state, OrderState::Rejected { .. }));
let now = ud.order_state(cid).await?;
assert!(matches!(now.state, OrderState::Acked { .. }));

// 5. Cancel it. Ok(()) means the broker says it is cancelled.
ud.cancel(cid).await?;

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

Required Methods

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,

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

Ok means the broker accepted the order (it is working, not necessarily filled). The OrderAck has the broker’s id. Follow the order with order_state or the stream of subscribe_order_state.

The cid in req must be new. A connector sends a cid at most once: a second place with the same cid is refused with Error::Rejected (BCS: code duplicate_place) without being sent. This is on purpose. After a timeout the first one may have reached the broker, and a retry could make a second order. Use a new cid for a new order, after you know what happened to the first. The price is checked against the price step and the quantity against the lot when the connector knows them (see price_grid in the config of the connector).

Dropping the future does not cancel the order. Once the cid is marked and the request is on its way, the connector (T-Bank, Finam) finishes the command in a task of its own: the request, the answer and the update of the order’s state happen even if you stop waiting (a timeout around the call, a lost select!). Before that moment (the wait for the quota or for earlier commands on the same instrument) dropping the future sends nothing. So after you drop a place look at the order, as after Error::Unknown.

Contract

• When: after the first successful reconcile, and not while the connector is degraded or shutting down. Write req.cid to your journal before the call.
• Guarantee: at most one request with this cid ever leaves the process, whatever happens.
• Nothing sent: the gate errors, Config, a local RateLimited and the input checks (missing_price, price_off_step, a wrong quantity) are raised before the request leaves. No order exists and the cid is unused. A new cid is always the safe choice for a retry.
• Maybe sent: after Error::Unknown (and, to be safe, Error::Transport) the order may exist. Ask order_state with the same cid.
• Ok is the broker’s acceptance, not an execution. Fills come as events.

Errors

• Error::NotReconciled, Error::Degraded, Error::ShuttingDown: the gate is closed, nothing was sent. See reconcile.
• Error::Config when req.instrument.native is of a kind this broker does not use.
• Error::RateLimited when the local limiter or the broker refuses for now. The command was not executed. Wait retry_after. The safe rule for the retry is a new cid.
• Error::Rejected when the broker (or the connector before sending) says no: a bad price or quantity, a limit order without a price (BCS: missing_price), a repeated cid (duplicate_place), a refusal inside an accepted request (BCS: rejected). The order does not exist. Change the request.
• Error::Unknown when the request may have reached the broker and there is no answer (timeout, cut connection, HTTP 5xx). Do NOT place again. Ask the broker with order_state or active_orders.
• Error::Transport when the failure happened before the request left, for example authentication. It is not a sign that a repeat is safe: ask Error::retryable_for_command, which says true for Error::RateLimited only, and not Error::class, which is the class of a read.

Examples

A limit order that the mock broker refuses, and what the caller sees:

use brokapi_core::{
    ClientOrderId, Error, ErrorClass, InstrumentId, InstrumentKey, NativeId, OrderExecutor,
    OrderType, PlaceOrderRequest, Price, Qty, Side,
};
let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);
let req = PlaceOrderRequest {
    cid: ClientOrderId::new_v4(),
    instrument: sber,
    side: Side::Buy,
    order_type: OrderType::Limit,
    qty: Qty::from(1),
    price: Some(refusing_price), // the mock broker refuses this price
};
match ud.place(req.clone()).await {
    Ok(ack) => println!("accepted as {}", ack.broker_id.as_str()),
    // The broker said no. The order does not exist. Fix the request and use a new cid.
    Err(e @ Error::Rejected { .. }) => assert_eq!(e.class(), ErrorClass::Fatal),
    // Maybe it exists. Never place again: ask with order_state / active_orders.
    Err(Error::Unknown) => panic!("look at the order before anything else"),
    Err(other) => panic!("unexpected: {other}"),
}

// Sending the same cid again is refused locally: a cid is used once.
let again = ud.place(req).await.unwrap_err();
assert!(matches!(again, Error::Rejected { ref code, .. } if code == "duplicate_place"));

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,

Cancels a working order and waits for the broker to confirm.

Ok(()) means the broker says the order is cancelled. It cancels orders placed through place and replace. For a stop order use StopOrderExecutor::cancel_stop. A part that was filled before stays filled.

Contract

• When: while the gate is open. The order may come from an earlier run of the process: the cid is enough. An order this process did not place is not queued behind other commands on its instrument (the instrument is not known).
• Guarantee: Ok(()) is the broker’s confirmation. A repeated cancel creates no order, so it is the one command that may be repeated by hand after Error::Unknown, and only after order_state shows the order still works.

Errors

• Error::NotReconciled, Error::Degraded, Error::ShuttingDown: the gate is closed, nothing was sent.
• Error::RateLimited: not executed, wait retry_after and repeat.
• Error::Rejected when the broker refuses: the order is unknown to it, it was filled first, or the request was refused. The code depends on how the broker answers (BCS codes include not_found, already_filled, cancel_rejected, validation). Read the order with order_state to see where it stands.
• Error::Unknown when the answer is missing, or the answer leaves the order alive. The order may still be working. Ask order_state and cancel again only if it is. A repeated cancel is harmless for the order, but it uses rate limit.
• Error::Transport when the failure happened before the request left.

Examples

use brokapi_core::{ClientOrderId, Error, OrderExecutor, OrderType, PlaceOrderRequest, Price, Qty, Side};
let req = order(Price::from(100));
let cid = req.cid;
ud.place(req).await?;
ud.cancel(cid).await?;

// An order that was filled before the cancel cannot be cancelled. The broker refuses, and the
// order is not working any more.
let req = order(Price::from(100));
let cid = req.cid;
ud.place(req).await?;
match ud.cancel(cid).await {
    Err(Error::Rejected { code, message }) => println!("refused: {code} - {message}"),
    other => panic!("expected a refusal, got {other:?}"),
}

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,

Changes the quantity or price of a working order.

The broker cancels the original and puts a new order under req.cid, so the answer is a new order with a new broker id. The original is not active afterwards. Only where UserDataCaps::supports_replace is true.

Contract

• When: while the gate is open and the broker has replace (UserDataCaps::supports_replace). req.cid is new (write it to your journal first), req.original_cid is the working order.
• Guarantee: the same cid rules as place. After Error::Unknown ask order_state for both cids before any new command on them.

Errors

• Error::Unsupported when the broker has no replace (Finam). Nothing was sent, the order is untouched.
• Error::NotReconciled, Error::Degraded, Error::ShuttingDown: the gate is closed.
• Error::RateLimited: not executed, wait retry_after.
• Error::Rejected when the broker refuses, for example the original is unknown or filled, or (BCS) the original is partly filled and qty is None (qty_required).
• Error::Unknown when the outcome is not known. The original may be replaced, still working or gone. Ask order_state for both cids. Do not replace again until you know.
• Error::Transport when the failure happened before the request left.

Examples

use brokapi_core::{ClientOrderId, OrderExecutor, OrderType, PlaceOrderRequest, Price, Qty, ReplaceOrderRequest, Side};
assert!(ud.caps().supports_replace);
let first = ClientOrderId::new_v4();
ud.place(PlaceOrderRequest {
    cid: first, instrument: sber, side: Side::Buy, order_type: OrderType::Limit,
    qty: Qty::from(2), price: Some(Price::from(100)),
}).await?;

// Move the price. The new order has a NEW cid, the old cid is not active any more.
let second = ClientOrderId::new_v4();
let ack = ud.replace(ReplaceOrderRequest {
    cid: second, original_cid: first, qty: None, price: Some(Price::from(101)),
}).await?;
assert_eq!(ack.cid, second);
let active = ud.active_orders().await?;
assert!(active.iter().any(|o| o.cid == second));
assert!(!active.iter().any(|o| o.cid == first));
ud.cancel(second).await?;

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,

Reads one order by its cid, as the broker shows it now.

This is how you leave the doubt of Error::Unknown. A read is not behind the trading gate, it works before the first reconcile. It also finds orders that are no longer active (filled, cancelled), unlike active_orders.

Contract

• When: at any time, also before the first reconcile and while the trading gate is closed: a read is not gated.
• Guarantee: reads only, so it may be repeated as often as the rate limit allows. It answers for final orders too. A cid the broker does not know is an error (see below), and right after an Error::Unknown that is not final: the request may still be on its way.

Errors

• Error::RateLimited, Error::Transport: wait or repeat later.
• Error::Rejected with code not_found (T-Bank: NOT_FOUND, compare without regard to case) when the broker does not know the cid. The same at all brokers. After an Error::Unknown this can mean that the order never reached the broker, or has not reached it yet. There is no separate variant for it.

Examples

use brokapi_core::{ClientOrderId, OrderExecutor, OrderState, OrderType, PlaceOrderRequest, Price, Qty, Side};
let cid = ClientOrderId::new_v4();
ud.place(PlaceOrderRequest {
    cid, instrument: sber, side: Side::Buy, order_type: OrderType::Limit,
    qty: Qty::from(2), price: Some(Price::from(100)),
}).await?;

let snapshot = ud.order_state(cid).await?;
match snapshot.state {
    OrderState::Acked { .. } => println!("working, nothing filled"),
    OrderState::PartiallyFilled { filled, left, .. } => println!("{} done, {} to go", filled.decimal(), left.decimal()),
    OrderState::Filled { .. } | OrderState::Cancelled { .. } | OrderState::Rejected { .. } => println!("final"),
    // InFlight, Unknown, and a state that a later version adds: ask again a bit later.
    _ => println!("ask again a bit later"),
}
assert_eq!(snapshot.qty, Qty::from(2));

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,

Lists the working orders of the account.

Plain orders only: stop orders are in StopOrderExecutor::active_stop_orders. The list includes orders that were not placed through this connector (by hand, by another program). Their cid is not yours: where the broker does not give a cid back, the BCS connector makes a stable one from the broker id, the same for the same order every time.

Contract

• When: at any time, the trading gate does not matter. Reads only.
• Guarantee: the list is what the broker shows now. Orders of others are in it too: do not touch what is not in your journal.

Errors

• Error::RateLimited, Error::Transport: wait or repeat later.

Examples

use brokapi_core::{ClientOrderId, OrderExecutor, OrderType, PlaceOrderRequest, Price, Qty, Side};
let cid = ClientOrderId::new_v4();
ud.place(PlaceOrderRequest {
    cid, instrument: sber, side: Side::Buy, order_type: OrderType::Limit,
    qty: Qty::from(3), price: Some(Price::from(100)),
}).await?;

let active = ud.active_orders().await?;
let mine = active.iter().find(|o| o.cid == cid).expect("our order is working");
assert_eq!(mine.price, Some(Price::from(100)));

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.

Every change (accepted, partly filled, filled, cancelled, refused) arrives as Event::Order, tagged with the returned id, and carries the cid.

The stream is Backpressure::Lossless with a buffer size set in the connector config. If you read too slowly and the buffer fills, an Event::Overflow is sent and trading is refused with Error::Degraded until reconcile. Keep draining your channel.

Subscribe BEFORE you place. The call returns when the stream is open (see Contract), so an order placed right after it cannot be missed. Changes that happened before the call are not delivered: to catch up, read order_state or active_orders. There is no way to end this subscription except shutting the connector down.

Contract

• When: once per connector, before reconcile and before the first place. Allowed while the gate is closed.
• Guarantee: the call returns only when the stream is open at the broker. The broker’s own confirmation where the protocol has one (T-Bank, ALOR), else the open connection of the stream (BCS). Everything that happens to your orders after the return reaches you. A stream that is not open within subscribe_ack_timeout (3 seconds by default) is Error::Transport and nothing is left half open, so the call can be repeated. Finam is the exception: it is gRPC and may send the headers of a quiet stream only with its first message, so after the wait the call returns Ok and the stream goes on opening in the background, with no guarantee until it is open (sub_status of the connector shows it).
• Guarantee: every change of your orders arrives, in order, or the loss is announced with Event::Overflow and trading stops until a reconcile. The subscription cannot be ended before shutdown.

Errors

• Error::Unsupported when the connector has no order stream.
• Error::Transport when the subscription could not be made, also when the stream did not open in time.

Examples

See the example of the trait, which follows an order on the stream.

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 your ledger with the broker and opens the trading gate.

Call it once after the connector starts, and again after Error::Degraded. A successful call is what lifts Error::NotReconciled and Error::Degraded.

ledger is the list of cids of orders you believe are working. Pass &[] if you have none. The report sorts the world into three:

• confirmed: in your ledger and active at the broker.
• closed_offline: in your ledger and NOT active at the broker (filled, cancelled, refused, or never received). The report does not say which. Ask order_state for each, and the executions for the fills.
• foreign: active at the broker and not in your ledger. Leave them alone.

The comparison is with active_orders, which lists plain orders. Keep the cids of stop orders out of the ledger and check them with StopOrderExecutor::active_stop_orders (not verified against the live API for every connector).

Contract

• When: after subscribe_order_state at start, again after Error::Degraded (read your channel to the end first), and whenever you doubt the ledger. It only reads, so any number of calls is safe.
• Guarantee: on Ok the gate is open (lifts Error::NotReconciled and Error::Degraded). On an error the gate stays as it was and nothing is known about your orders. Orders are never moved or cancelled.

Errors

• Error::RateLimited, Error::Transport when the broker could not be read. The gate stays closed. Repeat later. A cid the broker answers “not found” for is not an error, it goes to closed_offline.

Examples

use brokapi_core::{ClientOrderId, Error, OrderExecutor, OrderType, PlaceOrderRequest, Price, Qty, Side};
use brokapi_bcs::{UserDataConfig, UserDataConnector};
let (events_tx, _events) = tokio::sync::mpsc::channel(256);
let ud = UserDataConnector::start(UserDataConfig::new(cfg, "bcs-ud"), limiter(), events_tx).await?;

// Before the first reconcile, trading commands are refused and nothing is sent.
assert!(matches!(ud.place(order()).await, Err(Error::NotReconciled)));

// Our ledger has the cid saved by the previous run, and one the broker never heard of.
let lost = ClientOrderId::new_v4();
let ledger = [saved_ledger[0], lost];
let report = ud.reconcile(&ledger).await?;
assert_eq!(report.confirmed.len(), 1); // still working at the broker
assert_eq!(report.confirmed[0].cid, saved_ledger[0]);
assert_eq!(report.closed_offline, vec![lost]); // not active at the broker

// The gate is open now.
let next = order();
let cid = next.cid;
ud.place(next).await?;
ud.cancel(cid).await?;

Dyn Compatibility

This trait is dyn compatible.

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

Implementors