Trait ExecutionClient 

pub trait ExecutionClient
where
    Self: Clone,
{
    type Config: Clone;
    type AccountStream: Stream<Item = UnindexedAccountEvent> + Send;

    const EXCHANGE: ExchangeId;
    const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant];

    // Required methods
    fn new(config: Self::Config) -> Self;
    fn account_snapshot(
        &self,
        assets: &[AssetNameExchange],
        instruments: &[InstrumentNameExchange],
    ) -> impl Future<Output = Result<UnindexedAccountSnapshot, UnindexedClientError>> + Send;
    fn account_stream(
        &self,
        assets: &[AssetNameExchange],
        instruments: &[InstrumentNameExchange],
    ) -> impl Future<Output = Result<Self::AccountStream, UnindexedClientError>> + Send;
    fn cancel_order(
        &self,
        request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
    ) -> impl Future<Output = UnindexedOrderResponseCancel> + Send;
    fn open_order(
        &self,
        request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
    ) -> impl Future<Output = Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> + Send;
    fn fetch_balances(
        &self,
        assets: &[AssetNameExchange],
    ) -> impl Future<Output = Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError>> + Send;
    fn fetch_open_orders(
        &self,
        instruments: &[InstrumentNameExchange],
    ) -> impl Future<Output = Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError>> + Send;
    fn fetch_trades(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        instruments: &[InstrumentNameExchange],
    ) -> impl Future<Output = Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError>> + Send;

    // Provided methods
    fn validate_config(
        config: &Self::Config,
        instruments: &[ClientInstrument<'_>],
    ) -> Result<(), String> { ... }
    fn cancel_orders<'a>(
        &self,
        requests: impl IntoIterator<Item = OrderRequestCancel<ExchangeId, &'a InstrumentNameExchange>>,
    ) -> impl Stream<Item = UnindexedOrderResponseCancel> + Send { ... }
    fn open_orders<'a>(
        &self,
        requests: impl IntoIterator<Item = OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange>>,
    ) -> impl Stream<Item = Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> + Send { ... }
}

A venue’s account and order API, as the engine’s ExecutionManager drives it.

Wrapping a client

To add behaviour around a venue client without reimplementing it, wrap it in a generic decorator that implements ExecutionClient itself. Typical uses are choosing which fills on a shared account reach the engine, logging, and metrics.

• Take the inner client’s identity. Set const EXCHANGE: ExchangeId = C::EXCHANGE and SUPPORTED_KINDS = C::SUPPORTED_KINDS. The ExecutionManager refuses a venue client whose EXCHANGE differs from its instrument map’s exchange (only the Mock and Simulated ids are exempt), so one generic wrapper then serves every venue client.
• Forward every method, the provided ones included. That is open_orders, cancel_orders and validate_config: left to the trait’s defaults, they would bypass any batching or config check the inner client has. Implement the extension traits the inner client implements as well, with impls bounded on it, forwarding their provided methods too:
  • OrderStatusClient, which the ExecutionManager requires of every client it runs;
  • BracketOrderClient
  • ShortabilityClient
  • BorrowCapacityClient
• Withhold a fill from account_stream and fetch_trades alike, by the same rule. One fill can reach the engine either way: live on the stream, or in a read made after a disconnect. A fill withheld from only one of them still reaches the engine once.
  • Choose a key both paths carry. Trade::client_order_id marks the orders a consumer placed, but not every venue reports it on fetch_trades (its rustdoc lists which do). Binance’s is one that does not. A rule keyed on it has to decide what to do with a fill it cannot classify. Keeping it, as the example below does, delivers too much rather than too little: on such a venue every fill a fetch_trades read returns is kept, and only the stream is filtered. To withhold those fills as well, match them another way, such as by Trade::order_id against the orders the wrapper placed.
  • Keep fills the venue forced (TradeOrigin::is_forced). They moved the position, whoever placed the order.
  • Leave resume alone. Filtering a TradesRead keeps its resume as it was: it marks where the venue’s read stopped, not which fills the wrapper kept.
• One inner client per engine. A wrapper that owns one client counts as that one client. Two clients on one account whose account streams both reach one engine deliver every fill twice.
• Withhold here, not later. The engine applies a fill that matches none of its orders too. In OmsMode::Netting it moves the netting position like any other fill, and the engine cannot tell it from a fill of its own. In OmsMode::Hedging an ordinary fill (TradeOrigin::Order or TradeOrigin::External) opens a position of its own, and a settlement fill is handled as InstrumentState::update_from_trade describes.

use chrono::{DateTime, Utc};
use futures::{Stream, StreamExt, stream::BoxStream};
use rustrade_execution::{
    AccountEventKind, UnindexedAccountEvent, UnindexedAccountSnapshot,
    balance::AssetBalance,
    client::{ClientInstrument, ExecutionClient, OrderStatusClient, VenueOrderStatus},
    error::UnindexedClientError,
    order::{
        Order, UnindexedInactiveOrder, UnindexedOrderKey,
        request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
        state::{Open, UnindexedOrderState},
    },
    trade::{Trade, TradesRead},
};
use rustrade_instrument::{
    asset::name::AssetNameExchange,
    exchange::ExchangeId,
    instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
};

/// Delivers the fills of orders whose client order id starts with `prefix`, fills whose
/// client order id the venue did not report, and every fill the venue forced.
///
/// Keeping a fill without a client order id is a choice: on a venue whose `fetch_trades`
/// reports none, this filters the account stream only.
#[derive(Debug, Clone)]
pub struct OwnFills<C> {
    inner: C,
    prefix: String,
}

#[derive(Debug, Clone)]
pub struct OwnFillsConfig<Config> {
    pub inner: Config,
    pub prefix: String,
}

fn keeps<AssetKey, InstrumentKey>(prefix: &str, trade: &Trade<AssetKey, InstrumentKey>) -> bool {
    trade.origin.is_forced()
        || trade
            .client_order_id
            .as_ref()
            .is_none_or(|cid| cid.0.starts_with(prefix))
}

// `Sync` makes the wrapper's method futures `Send`, as the trait requires. `'static` lets the
// filtered account stream be boxed.
impl<C> ExecutionClient for OwnFills<C>
where
    C: ExecutionClient + Sync,
    C::AccountStream: 'static,
{
    const EXCHANGE: ExchangeId = C::EXCHANGE;
    const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] = C::SUPPORTED_KINDS;
    type Config = OwnFillsConfig<C::Config>;
    type AccountStream = BoxStream<'static, UnindexedAccountEvent>;

    fn validate_config(
        config: &Self::Config,
        instruments: &[ClientInstrument<'_>],
    ) -> Result<(), String> {
        C::validate_config(&config.inner, instruments)
    }

    fn new(config: Self::Config) -> Self {
        Self {
            inner: C::new(config.inner),
            prefix: config.prefix,
        }
    }

    async fn account_stream(
        &self,
        assets: &[AssetNameExchange],
        instruments: &[InstrumentNameExchange],
    ) -> Result<Self::AccountStream, UnindexedClientError> {
        let prefix = self.prefix.clone();
        let stream = self.inner.account_stream(assets, instruments).await?;
        Ok(stream
            .filter(move |event| {
                let keep = match &event.kind {
                    AccountEventKind::Trade(trade) => keeps(&prefix, trade),
                    _ => true,
                };
                std::future::ready(keep)
            })
            .boxed())
    }

    async fn fetch_trades(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        instruments: &[InstrumentNameExchange],
    ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
        let mut read = self.inner.fetch_trades(start, end, instruments).await?;
        read.trades.retain(|trade| keeps(&self.prefix, trade));
        Ok(read)
    }

    // Everything else is forwarded unchanged.

    async fn account_snapshot(
        &self,
        assets: &[AssetNameExchange],
        instruments: &[InstrumentNameExchange],
    ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
        self.inner.account_snapshot(assets, instruments).await
    }

    async fn cancel_order(
        &self,
        request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
    ) -> UnindexedOrderResponseCancel {
        self.inner.cancel_order(request).await
    }

    fn cancel_orders<'a>(
        &self,
        requests: impl IntoIterator<Item = OrderRequestCancel<ExchangeId, &'a InstrumentNameExchange>>,
    ) -> impl Stream<Item = UnindexedOrderResponseCancel> + Send {
        self.inner.cancel_orders(requests)
    }

    async fn open_order(
        &self,
        request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
    ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
        self.inner.open_order(request).await
    }

    fn open_orders<'a>(
        &self,
        requests: impl IntoIterator<Item = OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange>>,
    ) -> impl Stream<Item = Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> + Send
    {
        self.inner.open_orders(requests)
    }

    async fn fetch_balances(
        &self,
        assets: &[AssetNameExchange],
    ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
        self.inner.fetch_balances(assets).await
    }

    async fn fetch_open_orders(
        &self,
        instruments: &[InstrumentNameExchange],
    ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
        self.inner.fetch_open_orders(instruments).await
    }
}

impl<C> OrderStatusClient for OwnFills<C>
where
    C: OrderStatusClient + Sync,
    C::AccountStream: 'static,
{
    async fn fetch_ended_orders(
        &self,
        orders: &[UnindexedOrderKey],
    ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
        self.inner.fetch_ended_orders(orders).await
    }

    async fn fetch_order_status(
        &self,
        key: &UnindexedOrderKey,
    ) -> Result<VenueOrderStatus, UnindexedClientError> {
        self.inner.fetch_order_status(key).await
    }
}

Required Associated Constants

const EXCHANGE: ExchangeId

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

The InstrumentKindDiscriminants this client can trade.

A client works off InstrumentNameExchange strings and never inspects an instrument’s kind, so nothing downstream can infer this from the implementation — routing a Future through a spot-only venue produces a rejected or, worse, misinterpreted order rather than a type error. Declaring the set here lets ExecutionBuilder reject an unsupported instrument set at build time, before any order is sent.

Declare what the venue and this implementation together can actually route, not what the exchange’s product catalogue advertises. A kind the exchange offers but this client builds no request for does not belong in the set.

There is deliberately no default. A new client must state its capabilities rather than inherit a permissive set that would silently admit every kind.

It is a const and not a fn supported_kinds(&self) because ExecutionBuilder validates the instrument set before Self::new is ever called — there is no instance to ask at the point the answer is needed, and deferring validation until after construction would mean standing up a client for a venue the engine is about to reject. A capability set that varies with Self::Config therefore cannot be expressed here; it would need a separate pre-construction hook on the config.

Required Associated Types

type Config: Clone

type AccountStream: Stream<Item = UnindexedAccountEvent> + Send

Required Methods

fn new(config: Self::Config) -> Self

fn account_snapshot( &self, assets: &[AssetNameExchange], instruments: &[InstrumentNameExchange], ) -> impl Future<Output = Result<UnindexedAccountSnapshot, UnindexedClientError>> + Send

Reads the account’s balances, and its open orders and positions for instruments, from the venue.

Besides startup, the engine’s ExecutionManager calls this again on every account-stream reconnect, and emits the result ahead of the new stream’s updates. It is how the engine catches up on what changed while the stream was down.

Assets

assets are the ones the caller registered. Return balances for those only when the venue lets you ask, which saves bandwidth. A balance for another asset is not an error: the ExecutionManager drops it and logs the asset names once per snapshot.

Order completeness

Each InstrumentAccountSnapshot states through orders_complete whether its orders list is every order open at the venue for that instrument. Set it only when it is true: the engine retires a tracked order that a complete list omits, so a list that can miss an open order (a capped page, an order dropped on conversion, an order reported under an id other than the one it was placed with) must say false. false is always safe; it only forgoes that reconciliation.

fn account_stream( &self, assets: &[AssetNameExchange], instruments: &[InstrumentNameExchange], ) -> impl Future<Output = Result<Self::AccountStream, UnindexedClientError>> + Send

Returns a live stream of account events (fills, order updates, balance changes).

Startup race window

There is an unavoidable gap between the WebSocket subscribe response and the first event being delivered: fills arriving in this window (typically milliseconds, no sub-millisecond guarantee) are silently dropped. account_snapshot reconciles open-order state, but TRADE fills in this window are not recoverable from the stream alone. Callers that require fill completeness at startup must call ExecutionClient::fetch_trades with at least a 1-second lookback after this method returns.

Backpressure

Implementations use unbounded internal channels. If the consumer cannot keep up, events queue in memory rather than being dropped — per library philosophy, OOM crashes are preferable to silent data loss. Consumers requiring backpressure should implement it at their boundary (e.g., bounded channel with overflow policy).

fn cancel_order( &self, request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>, ) -> impl Future<Output = UnindexedOrderResponseCancel> + Send

Cancel an order at the venue.

Return value

Every request gets exactly one answer, which the engine’s ExecutionManager forwards as the order’s response. An implementation must not swallow a request: a failure to send it, or to read the venue’s answer, is an Err.

• Ok(Cancelled): the venue took the cancel. Whether the order has ended by then depends on the venue; see below.
• Err: the venue refused the cancel, or it could not be sent or answered. The order may still be open, or may have ended some other way, such as by filling. The account stream reports which.

A taken cancel is not always an ended order

Some venues answer a cancel before carrying it out. Until they do, the order is live and can still fill. From those venues Ok(Cancelled) means only that the cancel was accepted, and the account stream reports how the order actually ended, including any fill in between. The engine stops tracking an order on Ok(Cancelled) from any venue.

Binance, Hyperliquid and the mock venue answer once the order is cancelled. Alpaca answers once it has accepted the cancel, and IBKR once the cancel is submitted. The connector comparison lists the same for each connector; the mock venue is not one, so it has no row there.

filled_quantity and time_exchange

Cancelled::filled_quantity is the quantity the venue reported filled where the answer carries one, which Binance’s and the mock venue’s do. Alpaca, IBKR and Hyperliquid answer without it and report zero, which does not mean nothing filled. The fill total comes from the account stream, or from Self::fetch_trades. Likewise time_exchange is the venue’s time where the answer carries one, and the local time the answer arrived otherwise.

fn open_order( &self, request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>, ) -> impl Future<Output = Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> + Send

Place an order on the exchange.

Return value

Every request gets exactly one answer, which the engine’s ExecutionManager forwards as the order’s response. An implementation must not swallow a request: a failure to send it, or to read the venue’s answer, is OpenFailed.

Returns OrderState directly rather than Result<Open, OrderError>:

• OrderState::Active(Open) - order is resting on the order book
• OrderState::Inactive(FullyFilled) - order was immediately filled (includes avg_price when available)
• OrderState::Inactive(Expired) - order ended in the response itself without filling completely, carrying what filled: an IOC or FOK order that found too little liquidity, or one expired by self-trade prevention
• OrderState::Inactive(Cancelled) - order the venue reports cancelled in the response itself, carrying what filled
• OrderState::Inactive(OpenFailed) - order placement failed (API error, connectivity, a rejection the venue reports in the response, etc.)

An implementation returns Expired or Cancelled only when its venue reports the order’s status in the placement response (currently Binance Spot and Margin). The others return Open for such an order, and their account stream reports how it ended.

This design allows immediate fills to carry metadata (e.g., avg_price) that would be lost if we had to infer terminal state from Open::filled_quantity.

fn fetch_balances( &self, assets: &[AssetNameExchange], ) -> impl Future<Output = Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError>> + Send

Fetch current balances for the specified assets.

An empty assets slice is the “return all” sentinel: implementations must return balances for every asset held. When non-empty, only the listed assets are returned.

fn fetch_open_orders( &self, instruments: &[InstrumentNameExchange], ) -> impl Future<Output = Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError>> + Send

Fetch currently open orders, optionally filtered by instrument.

An empty instruments slice is the “return all” sentinel: implementations must return open orders across all instruments. When non-empty, only orders for the listed instruments are returned.

fn fetch_trades( &self, start: DateTime<Utc>, end: DateTime<Utc>, instruments: &[InstrumentNameExchange], ) -> impl Future<Output = Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError>> + Send

Read the trades (fills) from start to end, both inclusive, optionally filtered by instrument.

An empty instruments slice is the “return all” sentinel: implementations must return trades across all instruments. When non-empty, only trades for the listed instruments are returned.

Bounded, resumable reads

One call is bounded by the venue’s limits, so it may stop before end. TradesRead::resume says whether it did:

• None: every fill in the span the venue still holds was read.
• Some(t): call again with start = t and the same end to read on. That read can return again fills the previous one returned, so match them by instrument and TradeId.

The caller owns the loop, and with it how many calls, and how long, it spends on a span. A start after end is an empty span, read as no trades and resume: None.

Venue limits

resume: None covers what the venue still holds, which can be less than the span:

• IBKR lists the last 7 days’ executions at most.
• Hyperliquid keeps only each wallet’s 10,000 most recent fills.

Binance needs instruments: an empty slice reads nothing. Each client documents its limits.

Errors

Truncated when a read cannot advance, because more fills share the span’s first moment than one call can read past.

The fee asset (AssetNameExchange) may be quote, base, or third-party (e.g., BNB). Use fees.fees_quote for quote-equivalent value when available.

Provided Methods

fn validate_config( config: &Self::Config, instruments: &[ClientInstrument<'_>], ) -> Result<(), String>

Checks config against the instruments this client is about to be built to trade, before Self::new is called. Err names what is wrong.

SUPPORTED_KINDS rejects a kind the client cannot route at all. This catches what only the config knows: a per-instrument entry that is invalid, or that disagrees with the instrument it is keyed to, such as a contract registered as a stock for an instrument modelled as an option. Each such entry would otherwise be skipped or acted on silently once the client runs.

The default accepts every config. Override it when Self::Config carries per-instrument detail.

ExecutionBuilder calls it, after checking SUPPORTED_KINDS, with the instruments registered on this client’s exchange. A client constructed by calling new directly is not checked; call this first to get the same check.

fn cancel_orders<'a>( &self, requests: impl IntoIterator<Item = OrderRequestCancel<ExchangeId, &'a InstrumentNameExchange>>, ) -> impl Stream<Item = UnindexedOrderResponseCancel> + Send

fn open_orders<'a>( &self, requests: impl IntoIterator<Item = OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange>>, ) -> impl Stream<Item = Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> + Send

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors

impl ExecutionClient for AlpacaClient

const EXCHANGE: ExchangeId = ExchangeId::AlpacaBroker

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = AlpacaConfig

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>

impl ExecutionClient for BinanceMargin

const EXCHANGE: ExchangeId = ExchangeId::BinanceMargin

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = BinanceMarginConfig

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>

impl ExecutionClient for BinanceSpot

const EXCHANGE: ExchangeId = ExchangeId::BinanceSpot

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = BinanceSpotConfig

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>

impl ExecutionClient for HyperliquidClient

const EXCHANGE: ExchangeId = ExchangeId::HyperliquidPerp

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = HyperliquidConfig

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>

impl ExecutionClient for HyperliquidSpotClient

const EXCHANGE: ExchangeId = ExchangeId::HyperliquidSpot

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = HyperliquidConfig

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>

impl ExecutionClient for IbkrClient

const EXCHANGE: ExchangeId = ExchangeId::Ibkr

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = IbkrConfig

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>

impl<FnTime> ExecutionClient for MockExecution<FnTime>

where FnTime: Fn() -> DateTime<Utc> + Clone + Send + Sync,

const EXCHANGE: ExchangeId = ExchangeId::Mock

const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant]

type Config = MockExecutionClientConfig<FnTime>

type AccountStream = Pin<Box<dyn Stream<Item = AccountEvent<ExchangeId, AssetNameExchange, InstrumentNameExchange>> + Send>>