Skip to main content

rustrade_execution/
indexer.rs

1use crate::{
2    AccountEvent, AccountEventKind, AccountSnapshot, InstrumentAccountSnapshot,
3    InstrumentBalanceUpdate, IsolatedInstrumentState, UnindexedAccountEvent,
4    UnindexedAccountSnapshot,
5    balance::{AssetBalance, AssetBalanceUpdate},
6    cash_flow::CashFlow,
7    error::{
8        AccountReinitFailure, AccountStreamInitError, ApiError, ClientError, KeyError, OrderError,
9        UnindexedAccountStreamInitError, UnindexedApiError, UnindexedClientError,
10        UnindexedOrderError,
11    },
12    fill_recovery::{FillRecoveryGap, FillRecoveryScope},
13    map::ExecutionInstrumentMap,
14    margin_debt::MarginDebtUpdate,
15    notice::AccountNotice,
16    order::{
17        Order, OrderEvent, OrderKey, OrderSnapshot, UnindexedOrderKey, UnindexedOrderSnapshot,
18        request::OrderResponseCancel,
19        state::{InactiveOrderState, OrderState, UnindexedOrderState},
20    },
21    trade::{AssetFees, Trade, TradeAmendment, TradeAmendmentKind},
22};
23use derive_more::Constructor;
24use rust_decimal::Decimal;
25use rustrade_instrument::{
26    asset::{AssetIndex, name::AssetNameExchange},
27    exchange::{ExchangeId, ExchangeIndex},
28    index::error::IndexError,
29    instrument::{InstrumentIndex, name::InstrumentNameExchange},
30};
31use rustrade_integration::collection::snapshot::Snapshot;
32use std::sync::Arc;
33use tracing::warn;
34
35#[derive(Debug, Clone, Constructor)]
36pub struct AccountEventIndexer {
37    pub map: Arc<ExecutionInstrumentMap>,
38}
39
40/// Why [`AccountEventIndexer::index_account_event`] did not index an event.
41///
42/// [`AccountEventIndexer::account_event`] folds both causes into one [`IndexError`]; this tells
43/// them apart for a caller that treats them differently.
44#[derive(Debug, Clone, PartialEq)]
45pub enum AccountIndexError {
46    /// A balance snapshot or update named an asset the [`ExecutionInstrumentMap`] does not
47    /// register.
48    ///
49    /// Expected rather than a fault: a venue such as Binance pushes balances for every asset on
50    /// the account, including ones the caller never asked to track (BNB held for fees, say). The
51    /// balance is not tracked. An unregistered asset on any other event is still an
52    /// [`Self::Rejected`] error, because that drops ledger data.
53    UntrackedBalance(AssetNameExchange),
54    /// The event named an exchange, asset or instrument the map does not register, and is lost.
55    Rejected(IndexError),
56}
57
58/// An indexed [`AccountSnapshot`] and the assets whose balances it left out. See
59/// [`AccountEventIndexer::snapshot_partitioned`], whose caller must report `untracked`.
60#[derive(Debug, Clone, PartialEq)]
61#[non_exhaustive]
62pub struct PartitionedSnapshot {
63    /// The snapshot, without the balances of the untracked assets.
64    pub snapshot: AccountSnapshot,
65    /// The assets of the skipped balances, in the order the venue reported them.
66    pub untracked: Vec<AssetNameExchange>,
67}
68
69impl AccountEventIndexer {
70    /// Index `event`, telling a balance for an unregistered asset apart from a lost event.
71    ///
72    /// See [`AccountIndexError`]. [`Self::account_event`] is the same indexing with one error.
73    pub fn index_account_event(
74        &self,
75        event: UnindexedAccountEvent,
76    ) -> Result<AccountEvent, AccountIndexError> {
77        let balance_asset = match &event.kind {
78            AccountEventKind::BalanceSnapshot(snapshot) => Some(snapshot.0.asset.clone()),
79            AccountEventKind::BalanceStreamUpdate(update) => Some(update.0.asset.clone()),
80            _ => None,
81        };
82
83        self.account_event(event)
84            .map_err(|error| match (error, balance_asset) {
85                // A balance event reaches no other lookup that can fail on an asset.
86                (IndexError::AssetIndex(_), Some(asset)) => {
87                    AccountIndexError::UntrackedBalance(asset)
88                }
89                (error, _) => AccountIndexError::Rejected(error),
90            })
91    }
92
93    pub fn account_event(&self, event: UnindexedAccountEvent) -> Result<AccountEvent, IndexError> {
94        let UnindexedAccountEvent { exchange, kind } = event;
95
96        let exchange = self.map.find_exchange_index(exchange)?;
97
98        let kind = match kind {
99            AccountEventKind::Snapshot(snapshot) => {
100                AccountEventKind::Snapshot(self.snapshot(snapshot)?)
101            }
102            AccountEventKind::BalanceSnapshot(snapshot) => {
103                AccountEventKind::BalanceSnapshot(self.asset_balance(snapshot.0).map(Snapshot)?)
104            }
105            AccountEventKind::BalanceStreamUpdate(snapshot) => {
106                AccountEventKind::BalanceStreamUpdate(
107                    self.asset_balance_update(snapshot.0).map(Snapshot)?,
108                )
109            }
110            AccountEventKind::InstrumentBalanceUpdate(update) => {
111                AccountEventKind::InstrumentBalanceUpdate(self.instrument_balance_update(update)?)
112            }
113            AccountEventKind::OrderSnapshot(snapshot) => {
114                AccountEventKind::OrderSnapshot(self.order_snapshot(snapshot.0).map(Snapshot)?)
115            }
116            AccountEventKind::OrderCancelled(response) => {
117                AccountEventKind::OrderCancelled(self.order_response_cancel(response)?)
118            }
119            AccountEventKind::Trade(trade) => AccountEventKind::Trade(self.trade(trade)?),
120            // Termination reason carries no exchange/asset/instrument keys — pass through verbatim.
121            AccountEventKind::StreamTerminated(reason) => {
122                AccountEventKind::StreamTerminated(reason)
123            }
124            AccountEventKind::FillRecoveryGaveUp(gap) => {
125                AccountEventKind::FillRecoveryGaveUp(self.fill_recovery_gap(gap)?)
126            }
127            AccountEventKind::TradeAmended(amendment) => {
128                AccountEventKind::TradeAmended(self.trade_amendment(amendment)?)
129            }
130            AccountEventKind::ReinitFailed(failure) => {
131                AccountEventKind::ReinitFailed(self.reinit_failure(failure))
132            }
133            AccountEventKind::CashFlow(flow) => AccountEventKind::CashFlow(self.cash_flow(flow)?),
134            AccountEventKind::Notice(notice) => {
135                AccountEventKind::Notice(self.notice(exchange, notice))
136            }
137            AccountEventKind::MarginDebt(update) => {
138                AccountEventKind::MarginDebt(self.margin_debt(update)?)
139            }
140        };
141
142        Ok(AccountEvent { exchange, kind })
143    }
144
145    /// Index `snapshot`, failing on any asset the [`ExecutionInstrumentMap`] does not register.
146    ///
147    /// Use [`Self::snapshot_partitioned`] to skip an unregistered balance instead. This is also
148    /// the path of an [`AccountEventKind::Snapshot`] event on an account stream, which stays
149    /// strict: it replaces the engine's whole view of the account, so losing part of it is not
150    /// tolerated.
151    ///
152    /// # Errors
153    ///
154    /// [`IndexError`] for an exchange, asset, instrument or order the map does not register.
155    pub fn snapshot(
156        &self,
157        snapshot: UnindexedAccountSnapshot,
158    ) -> Result<AccountSnapshot, IndexError> {
159        let PartitionedSnapshot {
160            snapshot,
161            untracked,
162        } = self.snapshot_partitioned(snapshot)?;
163        match untracked.first() {
164            None => Ok(snapshot),
165            Some(asset) => Err(self
166                .map
167                .find_asset_index(asset)
168                .err()
169                .unwrap_or_else(|| IndexError::AssetIndex(format!("not registered: {asset}")))),
170        }
171    }
172
173    /// Index `snapshot`, skipping the top-level balance of an asset the
174    /// [`ExecutionInstrumentMap`] does not register, and returning those assets beside it.
175    ///
176    /// A venue may report more assets than the caller registered (see
177    /// [`AccountIndexError::UntrackedBalance`]), so such a balance is not a fault. Only top-level
178    /// balances are skipped: an unregistered exchange, instrument, order, or isolated-margin
179    /// base or quote asset still fails the snapshot, because each registered instrument brings
180    /// its own assets and the loss would be a mapping defect.
181    ///
182    /// The caller must report [`PartitionedSnapshot::untracked`], or the drop is silent. An empty
183    /// [`PartitionedSnapshot::snapshot`] balance list beside a non-empty `untracked` usually means
184    /// the registered asset names do not match the venue's; it can also be an account that holds
185    /// only unregistered assets.
186    ///
187    /// # Errors
188    ///
189    /// [`IndexError`] for an exchange, instrument, order or isolated-margin asset the map does
190    /// not register.
191    pub fn snapshot_partitioned(
192        &self,
193        snapshot: UnindexedAccountSnapshot,
194    ) -> Result<PartitionedSnapshot, IndexError> {
195        let UnindexedAccountSnapshot {
196            exchange,
197            balances,
198            instruments,
199        } = snapshot;
200
201        let exchange = self.map.find_exchange_index(exchange)?;
202
203        let mut untracked = Vec::new();
204        let mut indexed_balances = Vec::with_capacity(balances.len());
205        for balance in balances {
206            match self.map.find_asset_index(&balance.asset) {
207                Ok(asset) => indexed_balances.push(AssetBalance {
208                    asset,
209                    balance: balance.balance,
210                    time_exchange: balance.time_exchange,
211                }),
212                Err(IndexError::AssetIndex(_)) => untracked.push(balance.asset),
213                Err(error) => return Err(error),
214            }
215        }
216
217        let instruments = instruments
218            .into_iter()
219            .map(|snapshot| {
220                let InstrumentAccountSnapshot {
221                    instrument,
222                    orders,
223                    orders_complete,
224                    position,
225                    isolated,
226                } = snapshot;
227
228                let instrument = self.map.find_instrument_index(&instrument)?;
229
230                let orders = orders
231                    .into_iter()
232                    .map(|order| self.order_snapshot(order))
233                    .collect::<Result<Vec<_>, _>>()?;
234
235                // Per-pair isolated balances are generic over `AssetKey`, so (unlike `position`)
236                // their base/quote asset names must be mapped to indices. A pair whose asset is
237                // unregistered fails the snapshot index, matching top-level-balance behaviour.
238                let isolated = isolated
239                    .map(|state| self.isolated_instrument_state(state))
240                    .transpose()?;
241
242                Ok(InstrumentAccountSnapshot {
243                    instrument,
244                    orders,
245                    orders_complete,
246                    position,
247                    isolated,
248                })
249            })
250            .collect::<Result<Vec<_>, _>>()?;
251
252        Ok(PartitionedSnapshot {
253            snapshot: AccountSnapshot {
254                exchange,
255                balances: indexed_balances,
256                instruments,
257            },
258            untracked,
259        })
260    }
261
262    pub fn asset_balance(
263        &self,
264        balance: AssetBalance<AssetNameExchange>,
265    ) -> Result<AssetBalance<AssetIndex>, IndexError> {
266        let AssetBalance {
267            asset,
268            balance,
269            time_exchange,
270        } = balance;
271        let asset = self.map.find_asset_index(&asset)?;
272
273        Ok(AssetBalance {
274            asset,
275            balance,
276            time_exchange,
277        })
278    }
279
280    pub fn asset_balance_update(
281        &self,
282        update: AssetBalanceUpdate<AssetNameExchange>,
283    ) -> Result<AssetBalanceUpdate<AssetIndex>, IndexError> {
284        let AssetBalanceUpdate {
285            asset,
286            update,
287            time_exchange,
288        } = update;
289        let asset = self.map.find_asset_index(&asset)?;
290
291        Ok(AssetBalanceUpdate {
292            asset,
293            update,
294            time_exchange,
295        })
296    }
297
298    /// Index the per-pair isolated state's `base`/`quote` asset names to indices.
299    ///
300    /// # Errors
301    /// Returns `IndexError` if either the base or quote asset is not registered in the map —
302    /// matching the fail-fast behaviour of top-level [`Self::asset_balance`].
303    pub fn isolated_instrument_state(
304        &self,
305        state: IsolatedInstrumentState<AssetNameExchange>,
306    ) -> Result<IsolatedInstrumentState<AssetIndex>, IndexError> {
307        let IsolatedInstrumentState { base, quote, risk } = state;
308
309        Ok(IsolatedInstrumentState {
310            base: self.asset_balance(base)?,
311            quote: self.asset_balance(quote)?,
312            risk,
313        })
314    }
315
316    /// Index a [`FillRecoveryGap`]'s instruments.
317    ///
318    /// # Errors
319    /// Returns `IndexError` if any instrument is not registered in the map. The whole gap fails
320    /// rather than dropping that instrument, which would understate what is missing. A venue names
321    /// only the instruments its stream was opened with, so this fails only for a stream opened
322    /// with an instrument the map does not hold.
323    pub fn fill_recovery_gap(
324        &self,
325        gap: FillRecoveryGap<InstrumentNameExchange>,
326    ) -> Result<FillRecoveryGap<InstrumentIndex>, IndexError> {
327        let FillRecoveryGap {
328            scope,
329            start,
330            end,
331            attempts,
332            reason,
333        } = gap;
334
335        let scope = match scope {
336            FillRecoveryScope::Instruments(instruments) => FillRecoveryScope::Instruments(
337                instruments
338                    .iter()
339                    .map(|instrument| self.map.find_instrument_index(instrument))
340                    .collect::<Result<_, _>>()?,
341            ),
342            FillRecoveryScope::AllInstruments => FillRecoveryScope::AllInstruments,
343        };
344
345        Ok(FillRecoveryGap {
346            scope,
347            start,
348            end,
349            attempts,
350            reason,
351        })
352    }
353
354    /// Index a [`CashFlow`]'s asset and, when it names one, instrument.
355    ///
356    /// # Errors
357    /// Returns `IndexError` if the asset, or a named instrument, is not registered in the map.
358    pub fn cash_flow(
359        &self,
360        flow: CashFlow<AssetNameExchange, InstrumentNameExchange>,
361    ) -> Result<CashFlow<AssetIndex, InstrumentIndex>, IndexError> {
362        let CashFlow {
363            kind,
364            asset,
365            amount,
366            instrument,
367            time_exchange,
368            id,
369        } = flow;
370
371        Ok(CashFlow {
372            kind,
373            asset: self.map.find_asset_index(&asset)?,
374            amount,
375            instrument: instrument
376                .map(|instrument| self.map.find_instrument_index(&instrument))
377                .transpose()?,
378            time_exchange,
379            id,
380        })
381    }
382
383    /// Index a [`MarginDebtUpdate`]'s asset and, when it names one, instrument.
384    ///
385    /// Unlike a notice's, an unknown instrument is an error rather than dropped: an update without
386    /// one is a debt of the account as a whole, which the engine applies to its asset balances, so
387    /// an isolated pair's debt delivered unattributed would overwrite the account's.
388    ///
389    /// # Errors
390    /// Returns `IndexError` if the asset, or a named instrument, is not registered in the map.
391    pub fn margin_debt(
392        &self,
393        update: MarginDebtUpdate<AssetNameExchange, InstrumentNameExchange>,
394    ) -> Result<MarginDebtUpdate<AssetIndex, InstrumentIndex>, IndexError> {
395        let MarginDebtUpdate {
396            asset,
397            instrument,
398            debt,
399            kind,
400            venue_kind,
401            time_exchange,
402        } = update;
403
404        Ok(MarginDebtUpdate {
405            asset: self.map.find_asset_index(&asset)?,
406            instrument: instrument
407                .map(|instrument| self.map.find_instrument_index(&instrument))
408                .transpose()?,
409            debt,
410            kind,
411            venue_kind,
412            time_exchange,
413        })
414    }
415
416    /// Index an [`AccountNotice`]'s instrument, when it names one.
417    ///
418    /// Never fails. A notice naming an instrument that is not in the map is kept, with its
419    /// instrument set to `None` and a `warn`: a notice changes no state, so dropping one about a
420    /// margin call or liquidation would cost more than losing its attribution.
421    pub fn notice(
422        &self,
423        exchange: ExchangeIndex,
424        notice: AccountNotice<InstrumentNameExchange>,
425    ) -> AccountNotice<InstrumentIndex> {
426        let AccountNotice {
427            kind,
428            instrument,
429            time_exchange,
430            margin_level,
431            status,
432        } = notice;
433
434        let instrument = instrument.and_then(|instrument| {
435            self.map
436                .find_instrument_index(&instrument)
437                .inspect_err(|error| {
438                    warn!(
439                        ?exchange,
440                        %instrument,
441                        %kind,
442                        %status,
443                        %error,
444                        "account notice names an instrument that is not in the map, \
445                         delivering it unattributed",
446                    );
447                })
448                .ok()
449        });
450
451        AccountNotice {
452            kind,
453            instrument,
454            time_exchange,
455            margin_level,
456            status,
457        }
458    }
459
460    /// Index an [`InstrumentBalanceUpdate`]'s instrument and asset keys.
461    ///
462    /// # Errors
463    /// Returns `IndexError` if the instrument or the asset is not registered in the map.
464    pub fn instrument_balance_update(
465        &self,
466        update: InstrumentBalanceUpdate<AssetNameExchange, InstrumentNameExchange>,
467    ) -> Result<InstrumentBalanceUpdate, IndexError> {
468        let InstrumentBalanceUpdate {
469            instrument,
470            balance,
471        } = update;
472
473        Ok(InstrumentBalanceUpdate {
474            instrument: self.map.find_instrument_index(&instrument)?,
475            balance: self.asset_balance_update(balance)?,
476        })
477    }
478
479    pub fn order_snapshot(
480        &self,
481        order: UnindexedOrderSnapshot,
482    ) -> Result<OrderSnapshot, IndexError> {
483        let Order {
484            key,
485            side,
486            price,
487            quantity,
488            kind,
489            time_in_force,
490            state,
491        } = order;
492
493        let key = self.order_key(key)?;
494        let state = self.order_state(state);
495
496        Ok(Order {
497            key,
498            side,
499            price,
500            quantity,
501            kind,
502            time_in_force,
503            state,
504        })
505    }
506
507    pub fn order_response_cancel(
508        &self,
509        response: OrderResponseCancel<ExchangeId, AssetNameExchange, InstrumentNameExchange>,
510    ) -> Result<OrderResponseCancel, IndexError> {
511        let OrderResponseCancel { key, state } = response;
512
513        Ok(OrderResponseCancel {
514            key: self.order_key(key)?,
515            state: match state {
516                Ok(cancelled) => Ok(cancelled),
517                Err(error) => Err(self.order_error(error)),
518            },
519        })
520    }
521
522    pub fn order_key(&self, key: UnindexedOrderKey) -> Result<OrderKey, IndexError> {
523        let UnindexedOrderKey {
524            exchange,
525            instrument,
526            strategy,
527            cid,
528        } = key;
529
530        Ok(OrderKey {
531            exchange: self.map.find_exchange_index(exchange)?,
532            instrument: self.map.find_instrument_index(&instrument)?,
533            strategy,
534            cid,
535        })
536    }
537
538    /// Index an [`UnindexedOrderState`] to an [`OrderState`].
539    ///
540    /// Used by `ExecutionManager` to index `open_order` responses. Never fails: an order state
541    /// carries no key, and the error of a failed open is indexed by [`Self::order_error`].
542    pub fn order_state(&self, state: UnindexedOrderState) -> OrderState {
543        match state {
544            UnindexedOrderState::Active(active) => OrderState::Active(active),
545            UnindexedOrderState::Inactive(inactive) => match inactive {
546                InactiveOrderState::OpenFailed(failed) => {
547                    OrderState::inactive(self.order_error(failed))
548                }
549                InactiveOrderState::Cancelled(cancelled) => OrderState::inactive(cancelled),
550                InactiveOrderState::FullyFilled(filled) => OrderState::fully_filled(filled),
551                InactiveOrderState::Expired(expired) => OrderState::expired(expired),
552            },
553        }
554    }
555
556    /// Index an [`UnindexedApiError`] returned for an order request (open or cancel).
557    ///
558    /// Never fails. An error carries the venue's answer to a request, and the caller must
559    /// receive that answer, so a name the [`ExecutionInstrumentMap`] does not hold degrades the
560    /// error instead of discarding it (and the order response or client error carrying it). Each
561    /// degrade logs a `warn!` naming what was dropped:
562    /// - [`ApiError::BalanceInsufficient`] keeps its variant, with the asset `None`.
563    /// - [`ApiError::AssetInvalid`] and [`ApiError::InstrumentInvalid`] have no index to carry,
564    ///   so they become [`ApiError::OrderRejected`] holding their own message, which keeps the
565    ///   name and the venue's text.
566    ///
567    /// For an error from any other request, use [`Self::client_error`], which degrades to
568    /// [`ApiError::RequestRejected`] instead.
569    pub fn api_error(&self, error: UnindexedApiError) -> ApiError {
570        self.index_api_error(error, ApiError::OrderRejected)
571    }
572
573    /// Index an [`UnindexedApiError`], degrading a name the map does not hold as
574    /// [`Self::api_error`] describes. `rejected` builds the rejection that an unresolvable
575    /// [`ApiError::AssetInvalid`] or [`ApiError::InstrumentInvalid`] becomes, which depends on
576    /// whether the request was an order.
577    fn index_api_error(
578        &self,
579        error: UnindexedApiError,
580        rejected: fn(String) -> ApiError,
581    ) -> ApiError {
582        match error {
583            UnindexedApiError::RateLimit { retry_after } => ApiError::rate_limited(retry_after),
584            UnindexedApiError::CapacityExceeded(msg) => ApiError::CapacityExceeded(msg),
585            UnindexedApiError::Unauthenticated(msg) => ApiError::Unauthenticated(msg),
586            UnindexedApiError::AssetInvalid(asset, value) => {
587                match self.map.find_asset_index(&asset) {
588                    Ok(asset) => ApiError::AssetInvalid(asset, value),
589                    Err(_) => self.degrade_unresolvable(
590                        UnindexedApiError::AssetInvalid(asset, value),
591                        rejected,
592                    ),
593                }
594            }
595            UnindexedApiError::InstrumentInvalid(instrument, value) => {
596                match self.map.find_instrument_index(&instrument) {
597                    Ok(instrument) => ApiError::InstrumentInvalid(instrument, value),
598                    Err(_) => self.degrade_unresolvable(
599                        UnindexedApiError::InstrumentInvalid(instrument, value),
600                        rejected,
601                    ),
602                }
603            }
604            UnindexedApiError::BalanceInsufficient(asset, value) => {
605                let asset = asset.and_then(|asset| {
606                    self.map
607                        .find_asset_index(&asset)
608                        .inspect_err(|_| {
609                            warn!(
610                                exchange = %self.map.exchange.value,
611                                %asset,
612                                message = %value,
613                                "AccountEventIndexer dropping the asset of a BalanceInsufficient \
614                                 error: the instrument map does not hold it"
615                            )
616                        })
617                        .ok()
618                });
619                ApiError::BalanceInsufficient(asset, value)
620            }
621            UnindexedApiError::OrderRejected(reason) => ApiError::OrderRejected(reason),
622            UnindexedApiError::OrderAlreadyCancelled => ApiError::OrderAlreadyCancelled,
623            UnindexedApiError::OrderAlreadyFullyFilled => ApiError::OrderAlreadyFullyFilled,
624            UnindexedApiError::OrderAlreadyExpired => ApiError::OrderAlreadyExpired,
625            UnindexedApiError::DuplicateClientOrderId(message) => {
626                ApiError::DuplicateClientOrderId(message)
627            }
628            UnindexedApiError::RequestRejected(reason) => ApiError::RequestRejected(reason),
629            UnindexedApiError::BorrowRejected(reject) => ApiError::BorrowRejected(reject),
630        }
631    }
632
633    /// Degrade an error naming an asset or instrument the map does not hold to the rejection
634    /// `rejected` builds, carrying the error's own message so neither the name nor the venue's
635    /// text is lost.
636    fn degrade_unresolvable(
637        &self,
638        error: UnindexedApiError,
639        rejected: fn(String) -> ApiError,
640    ) -> ApiError {
641        let error = error.to_string();
642        warn!(
643            exchange = %self.map.exchange.value,
644            %error,
645            "AccountEventIndexer degrading an API error to a rejection: the instrument map does \
646             not hold the asset or instrument it names"
647        );
648        rejected(error)
649    }
650
651    /// Index an order request for this exchange, resolving its instrument to the name this
652    /// exchange knows it by.
653    ///
654    /// # Errors
655    /// Returns [`KeyError`] when the order's exchange or instrument is not in the map, and when
656    /// the instrument is indexed but listed on a different exchange: a request keyed to another
657    /// exchange's instrument is not this exchange's to send.
658    pub fn order_request<Kind>(
659        &self,
660        order: &OrderEvent<Kind, ExchangeIndex, InstrumentIndex>,
661    ) -> Result<OrderEvent<Kind, ExchangeId, &InstrumentNameExchange>, KeyError>
662    where
663        Kind: Clone,
664    {
665        let OrderEvent {
666            key:
667                OrderKey {
668                    exchange,
669                    instrument,
670                    strategy,
671                    cid,
672                },
673            state,
674        } = order;
675
676        let exchange = self.map.find_exchange_id(*exchange)?;
677        let instrument = self
678            .map
679            .find_own_instrument_name_exchange(*instrument)
680            .ok_or_else(|| {
681                KeyError::InstrumentKey(format!(
682                    "ExecutionInstrumentMap for {} does not contain: {instrument}",
683                    self.map.exchange.value
684                ))
685            })?;
686
687        Ok(OrderEvent {
688            key: OrderKey {
689                exchange,
690                instrument,
691                strategy: strategy.clone(),
692                cid: cid.clone(),
693            },
694            state: state.clone(),
695        })
696    }
697
698    /// Index an [`UnindexedOrderError`]. Never fails: see [`Self::api_error`].
699    pub fn order_error(&self, error: UnindexedOrderError) -> OrderError {
700        match error {
701            UnindexedOrderError::Connectivity(error) => OrderError::Connectivity(error),
702            UnindexedOrderError::NotPlaced(reason) => OrderError::NotPlaced(reason),
703            UnindexedOrderError::Rejected(error) => OrderError::Rejected(self.api_error(error)),
704            UnindexedOrderError::UnsupportedOrderType(msg) => OrderError::UnsupportedOrderType(msg),
705            UnindexedOrderError::InvalidPrecision(violation) => {
706                OrderError::InvalidPrecision(violation)
707            }
708            UnindexedOrderError::OutOfBounds(violation) => OrderError::OutOfBounds(violation),
709            UnindexedOrderError::UnsupportedMarginEffect(effect) => {
710                OrderError::UnsupportedMarginEffect(effect)
711            }
712        }
713    }
714
715    /// Index an [`UnindexedClientError`]. Never fails: like [`Self::api_error`], but an
716    /// unresolvable [`ApiError::AssetInvalid`] or [`ApiError::InstrumentInvalid`] becomes
717    /// [`ApiError::RequestRejected`], since the request need not have been an order.
718    pub fn client_error(&self, error: UnindexedClientError) -> ClientError {
719        match error {
720            UnindexedClientError::Connectivity(error) => ClientError::Connectivity(error),
721            UnindexedClientError::Api(error) => {
722                ClientError::Api(self.index_api_error(error, ApiError::RequestRejected))
723            }
724            UnindexedClientError::TaskFailed(value) => ClientError::TaskFailed(value),
725            UnindexedClientError::Internal(value) => ClientError::Internal(value),
726            UnindexedClientError::Truncated { fills_read } => ClientError::Truncated { fills_read },
727            UnindexedClientError::TruncatedSnapshot { limit } => {
728                ClientError::TruncatedSnapshot { limit }
729            }
730        }
731    }
732
733    /// Index an [`UnindexedAccountStreamInitError`]. Never fails, as [`Self::client_error`].
734    pub fn account_stream_init_error(
735        &self,
736        error: UnindexedAccountStreamInitError,
737    ) -> AccountStreamInitError {
738        match error {
739            AccountStreamInitError::Client(error) => {
740                AccountStreamInitError::Client(self.client_error(error))
741            }
742            AccountStreamInitError::Index(error) => AccountStreamInitError::Index(error),
743        }
744    }
745
746    /// Index an [`AccountReinitFailure`]. Never fails, as [`Self::client_error`].
747    pub fn reinit_failure(
748        &self,
749        failure: AccountReinitFailure<AssetNameExchange, InstrumentNameExchange>,
750    ) -> AccountReinitFailure {
751        let AccountReinitFailure { attempt, error } = failure;
752        AccountReinitFailure {
753            attempt,
754            error: self.account_stream_init_error(error),
755        }
756    }
757
758    /// Index a [`TradeAmendment`]. A replacement trade is indexed as [`trade`](Self::trade) indexes
759    /// one, so an amendment fails where its trade would.
760    pub fn trade_amendment(
761        &self,
762        amendment: TradeAmendment<AssetNameExchange, InstrumentNameExchange>,
763    ) -> Result<TradeAmendment<AssetIndex, InstrumentIndex>, IndexError> {
764        let TradeAmendment {
765            instrument,
766            order_id,
767            time_exchange,
768            original,
769            kind,
770        } = amendment;
771
772        let instrument = self.map.find_instrument_index(&instrument)?;
773        let kind = match kind {
774            TradeAmendmentKind::Busted { quantity } => TradeAmendmentKind::Busted { quantity },
775            TradeAmendmentKind::Corrected { replacement } => TradeAmendmentKind::Corrected {
776                replacement: self.trade(replacement)?,
777            },
778            TradeAmendmentKind::CorrectedUnresolved {
779                id,
780                price,
781                quantity,
782            } => TradeAmendmentKind::CorrectedUnresolved {
783                id,
784                price,
785                quantity,
786            },
787        };
788
789        Ok(TradeAmendment {
790            instrument,
791            order_id,
792            time_exchange,
793            original,
794            kind,
795        })
796    }
797
798    /// Index a trade, converting fee asset and computing `fees_quote`.
799    ///
800    /// Computes `fees_quote` based on fee asset relationship to instrument:
801    /// - Fee in quote asset: `fees_quote = Some(fees)`
802    /// - Fee in base asset: `fees_quote = Some(fees * price)`
803    /// - Fee in third-party asset (e.g., BNB): `fees_quote = None`, since the indexer holds no
804    ///   prices. The `rustrade` engine values it when it prices that asset.
805    ///
806    /// # A fee asset that is not in the map
807    /// The trade is kept, with its fee zeroed: `fees` is `0` in the instrument's quote asset and
808    /// `fees_quote` is `None`. A `warn` names the fee's asset and amount. Dropping the trade would
809    /// leave the engine's position different from the venue's, which costs far more than one
810    /// unknown fee. The `warn` is per trade, so an account that pays every fee in an unregistered
811    /// asset logs one per fill until the asset is registered. Two cases reach this:
812    /// - An asset the account pays fees in but no instrument implies, such as BNB on Binance spot
813    ///   with "pay fees with BNB" turned on. Register it with
814    ///   [`IndexedInstrumentsBuilder::add_asset`](rustrade_instrument::index::builder::IndexedInstrumentsBuilder::add_asset)
815    ///   to keep its fee.
816    /// - The `"UNKNOWN"` placeholder some integrations use when the venue gave no fee asset: IBKR
817    ///   `fetch_trades` for an execution IB sent no commission report for, and Binance when a
818    ///   report omits `commissionAsset`.
819    ///
820    /// # Errors
821    /// Returns `IndexError` if the trade's instrument is not in the map.
822    pub fn trade(
823        &self,
824        trade: Trade<AssetNameExchange, InstrumentNameExchange>,
825    ) -> Result<Trade<AssetIndex, InstrumentIndex>, IndexError> {
826        let Trade {
827            id,
828            order_id,
829            instrument,
830            strategy,
831            time_exchange,
832            side,
833            price: trade_price,
834            quantity,
835            order_filled_quantity,
836            fees,
837            origin,
838            position_side,
839            client_order_id,
840        } = trade;
841
842        let instrument_index = self.map.find_instrument_index(&instrument)?;
843        // Invariant check: `find_instrument_index` resolves only names built from `instruments`,
844        // so the index is always present. Failing loudly beats indexing a fee against nothing.
845        let underlying = &self
846            .map
847            .instruments
848            .get_index(instrument_index.index())
849            .ok_or_else(|| {
850                IndexError::InstrumentIndex(format!(
851                    "ExecutionInstrumentMap has no instrument at {instrument_index} for {instrument}"
852                ))
853            })?
854            .underlying;
855
856        let fees = match self.map.find_asset_index(&fees.asset) {
857            Ok(fee_asset) => {
858                let fees_quote = if fee_asset == underlying.quote {
859                    // Fee is in quote asset — no conversion needed
860                    Some(fees.fees)
861                } else if fee_asset == underlying.base {
862                    // Fee is in base asset — convert using trade price
863                    Some(fees.fees * trade_price)
864                } else {
865                    // Fee is in third-party asset (e.g., BNB) — needs external price
866                    None
867                };
868                AssetFees::new(fee_asset, fees.fees, fees_quote)
869            }
870            Err(error) => {
871                warn!(
872                    exchange = %self.map.exchange.value,
873                    %instrument,
874                    trade_id = %id,
875                    fee_asset = %fees.asset,
876                    fee = %fees.fees,
877                    %error,
878                    "trade's fee asset is not in the map, keeping the trade with its fee zeroed; \
879                     register the asset with IndexedInstrumentsBuilder::add_asset to keep its fee",
880                );
881                AssetFees::new(underlying.quote, Decimal::ZERO, None)
882            }
883        };
884
885        Ok(Trade {
886            id,
887            order_id,
888            instrument: instrument_index,
889            strategy,
890            time_exchange,
891            side,
892            price: trade_price,
893            quantity,
894            order_filled_quantity,
895            fees,
896            origin,
897            position_side,
898            client_order_id,
899        })
900    }
901}
902
903#[cfg(test)]
904#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
905mod tests {
906    use super::*;
907    use crate::{
908        error::StreamTerminationReason, fill_recovery::FillRecoveryFailure,
909        map::generate_execution_instrument_map,
910    };
911    use chrono::{DateTime, Utc};
912    use rustrade_instrument::{index::IndexedInstruments, test_utils};
913
914    #[test]
915    fn a_partitioned_snapshot_skips_an_unregistered_balance_and_the_strict_one_fails() {
916        let indexer = binance_indexer();
917        let snapshot = || {
918            UnindexedAccountSnapshot::new(
919                ExchangeId::BinanceSpot,
920                ["usdt", "bnb", "btc", "sol"]
921                    .map(|asset| {
922                        AssetBalance::new(
923                            AssetNameExchange::new(asset),
924                            crate::balance::Balance::new(Decimal::ONE, Decimal::ONE),
925                            DateTime::<Utc>::MIN_UTC,
926                        )
927                    })
928                    .to_vec(),
929                vec![],
930            )
931        };
932
933        let partitioned = indexer.snapshot_partitioned(snapshot()).unwrap();
934        assert_eq!(partitioned.snapshot.balances.len(), 2);
935        assert_eq!(
936            partitioned.untracked,
937            vec![AssetNameExchange::new("bnb"), AssetNameExchange::new("sol")]
938        );
939        assert!(matches!(
940            indexer.snapshot(snapshot()),
941            Err(IndexError::AssetIndex(_))
942        ));
943    }
944
945    #[test]
946    fn a_partitioned_snapshot_still_fails_on_an_unregistered_instrument_or_exchange() {
947        let indexer = binance_indexer();
948        let instrument = |name: &str| {
949            InstrumentAccountSnapshot::new(
950                InstrumentNameExchange::new(name),
951                vec![],
952                false,
953                crate::position::PositionReport::Unreported,
954                None,
955            )
956        };
957        let unknown_instrument = UnindexedAccountSnapshot::new(
958            ExchangeId::BinanceSpot,
959            vec![],
960            vec![instrument("ETHUSDT")],
961        );
962        assert!(matches!(
963            indexer.snapshot_partitioned(unknown_instrument),
964            Err(IndexError::InstrumentIndex(_))
965        ));
966        // A skipped balance does not hide a failure elsewhere in the same snapshot.
967        let skipped_and_unknown_instrument = UnindexedAccountSnapshot::new(
968            ExchangeId::BinanceSpot,
969            vec![AssetBalance::new(
970                AssetNameExchange::new("bnb"),
971                crate::balance::Balance::new(Decimal::ONE, Decimal::ONE),
972                DateTime::<Utc>::MIN_UTC,
973            )],
974            vec![instrument("ETHUSDT")],
975        );
976        assert!(matches!(
977            indexer.snapshot_partitioned(skipped_and_unknown_instrument),
978            Err(IndexError::InstrumentIndex(_))
979        ));
980        let other_exchange = UnindexedAccountSnapshot::new(ExchangeId::Ibkr, vec![], vec![]);
981        assert!(matches!(
982            indexer.snapshot_partitioned(other_exchange),
983            Err(IndexError::ExchangeIndex(_))
984        ));
985    }
986
987    /// An order naming an instrument listed on another exchange does not resolve, though the plain
988    /// instrument lookup would find it.
989    #[test]
990    fn an_order_request_for_another_exchanges_instrument_fails() {
991        use crate::order::id::{ClientOrderId, StrategyId};
992        let instruments = IndexedInstruments::new(vec![
993            test_utils::instrument(ExchangeId::Kraken, "BTC", "USDT"),
994            test_utils::instrument(ExchangeId::BinanceSpot, "BTC", "USDT"),
995        ]);
996        let binance = generate_execution_instrument_map(&instruments, ExchangeId::BinanceSpot)
997            .map(Arc::new)
998            .map(AccountEventIndexer::new)
999            .unwrap();
1000        let instrument_on = |exchange| {
1001            instruments
1002                .find_instrument_index(
1003                    exchange,
1004                    &test_utils::instrument(exchange, "BTC", "USDT").name_internal,
1005                )
1006                .unwrap()
1007        };
1008        let order = |exchange, instrument| OrderEvent {
1009            key: OrderKey {
1010                exchange,
1011                instrument,
1012                strategy: StrategyId::new("test"),
1013                cid: ClientOrderId::random(),
1014            },
1015            state: (),
1016        };
1017        let binance_index = instruments
1018            .find_exchange_index(ExchangeId::BinanceSpot)
1019            .unwrap();
1020
1021        let own = binance
1022            .order_request(&order(
1023                binance_index,
1024                instrument_on(ExchangeId::BinanceSpot),
1025            ))
1026            .unwrap();
1027        assert_eq!(own.key.exchange, ExchangeId::BinanceSpot);
1028        assert_eq!(own.key.instrument.as_ref(), "BTC_USDT");
1029
1030        let foreign =
1031            binance.order_request(&order(binance_index, instrument_on(ExchangeId::Kraken)));
1032        assert!(
1033            matches!(foreign, Err(KeyError::InstrumentKey(_))),
1034            "{foreign:?}"
1035        );
1036    }
1037
1038    fn binance_indexer() -> AccountEventIndexer {
1039        let instruments = IndexedInstruments::new(vec![test_utils::instrument(
1040            ExchangeId::BinanceSpot,
1041            "BTC",
1042            "USDT",
1043        )]);
1044        let map = generate_execution_instrument_map(&instruments, ExchangeId::BinanceSpot).unwrap();
1045        AccountEventIndexer::new(Arc::new(map))
1046    }
1047
1048    fn balance_update(asset: &str) -> UnindexedAccountEvent {
1049        UnindexedAccountEvent::new(
1050            ExchangeId::BinanceSpot,
1051            AccountEventKind::BalanceStreamUpdate(Snapshot(AssetBalanceUpdate::new(
1052                AssetNameExchange::new(asset),
1053                crate::balance::BalanceUpdate::new(rust_decimal::Decimal::ONE, Default::default()),
1054                DateTime::<Utc>::MIN_UTC,
1055            ))),
1056        )
1057    }
1058
1059    #[test]
1060    fn a_balance_for_an_unregistered_asset_is_an_untracked_balance_not_a_rejection() {
1061        let indexer = binance_indexer();
1062
1063        assert!(indexer.index_account_event(balance_update("usdt")).is_ok());
1064        assert_eq!(
1065            indexer.index_account_event(balance_update("bnb")),
1066            Err(AccountIndexError::UntrackedBalance(AssetNameExchange::new(
1067                "bnb"
1068            )))
1069        );
1070
1071        let snapshot = UnindexedAccountEvent::new(
1072            ExchangeId::BinanceSpot,
1073            AccountEventKind::BalanceSnapshot(Snapshot(AssetBalance::new(
1074                AssetNameExchange::new("bnb"),
1075                crate::balance::Balance::new(Default::default(), Default::default()),
1076                DateTime::<Utc>::MIN_UTC,
1077            ))),
1078        );
1079        assert!(matches!(
1080            indexer.index_account_event(snapshot),
1081            Err(AccountIndexError::UntrackedBalance(_))
1082        ));
1083    }
1084
1085    #[test]
1086    fn an_unregistered_asset_on_any_other_event_is_still_a_rejection() {
1087        let indexer = binance_indexer();
1088        let cash_flow = UnindexedAccountEvent::new(
1089            ExchangeId::BinanceSpot,
1090            AccountEventKind::CashFlow(CashFlow {
1091                kind: crate::cash_flow::CashFlowKind::Rebate,
1092                asset: AssetNameExchange::new("bnb"),
1093                amount: rust_decimal::Decimal::ONE,
1094                instrument: None,
1095                time_exchange: DateTime::<Utc>::MIN_UTC,
1096                id: None,
1097            }),
1098        );
1099        assert!(matches!(
1100            indexer.index_account_event(cash_flow),
1101            Err(AccountIndexError::Rejected(IndexError::AssetIndex(_)))
1102        ));
1103
1104        // A balance on an exchange the map does not hold fails on the exchange, not the asset.
1105        let mut wrong_exchange = balance_update("bnb");
1106        wrong_exchange.exchange = ExchangeId::Kraken;
1107        assert!(matches!(
1108            indexer.index_account_event(wrong_exchange),
1109            Err(AccountIndexError::Rejected(IndexError::ExchangeIndex(_)))
1110        ));
1111    }
1112
1113    /// `StreamTerminated` carries no asset/instrument keys, so it must index by mapping only the
1114    /// exchange and passing the reason through verbatim.
1115    #[test]
1116    fn account_event_passes_stream_terminated_through_verbatim() {
1117        let indexer = binance_indexer();
1118        let reason = StreamTerminationReason::ReconnectBudgetExhausted {
1119            attempts: 3,
1120            last_error: "socket reset".to_string(),
1121        };
1122
1123        let indexed = indexer
1124            .account_event(UnindexedAccountEvent::new(
1125                ExchangeId::BinanceSpot,
1126                AccountEventKind::StreamTerminated(reason.clone()),
1127            ))
1128            .unwrap();
1129
1130        assert_eq!(indexed.exchange, indexer.map.exchange.key);
1131        assert!(
1132            matches!(indexed.kind, AccountEventKind::StreamTerminated(r) if r == reason),
1133            "expected StreamTerminated to pass through unchanged",
1134        );
1135    }
1136
1137    /// A failed re-init never fails to index: its client error is indexed as
1138    /// [`AccountEventIndexer::client_error`] indexes one, and an index error passes through.
1139    #[test]
1140    fn account_event_indexes_a_reinit_failure() {
1141        let indexer = binance_indexer();
1142        let btc = indexer
1143            .map
1144            .find_asset_index(&AssetNameExchange::new("BTC"))
1145            .unwrap();
1146        let unindexed_client = AccountStreamInitError::Client(ClientError::Api(
1147            ApiError::BalanceInsufficient(Some(AssetNameExchange::new("BTC")), "low".to_owned()),
1148        ));
1149        let indexed_client = AccountStreamInitError::Client(ClientError::Api(
1150            ApiError::BalanceInsufficient(Some(btc), "low".to_owned()),
1151        ));
1152        let index_error = IndexError::AssetIndex("ETH".to_owned());
1153
1154        for (error, expected) in [
1155            (unindexed_client, indexed_client),
1156            (
1157                AccountStreamInitError::Index(index_error.clone()),
1158                AccountStreamInitError::Index(index_error),
1159            ),
1160        ] {
1161            let indexed = indexer
1162                .account_event(UnindexedAccountEvent::new(
1163                    ExchangeId::BinanceSpot,
1164                    AccountEventKind::ReinitFailed(AccountReinitFailure::new(2, error)),
1165                ))
1166                .unwrap();
1167
1168            assert_eq!(indexed.exchange, indexer.map.exchange.key);
1169            assert_eq!(
1170                indexed.kind,
1171                AccountEventKind::ReinitFailed(AccountReinitFailure::new(2, expected))
1172            );
1173        }
1174    }
1175
1176    /// A fill recovery give-up indexes each instrument it names, and fails as a whole when one
1177    /// is not in the map rather than understating what is missing. `AllInstruments` names none.
1178    #[test]
1179    fn account_event_indexes_a_fill_recovery_give_up_by_its_instruments() {
1180        let indexer = binance_indexer();
1181        let start = DateTime::<Utc>::MIN_UTC;
1182        let end = DateTime::<Utc>::MAX_UTC;
1183        let give_up = |scope| {
1184            UnindexedAccountEvent::new(
1185                ExchangeId::BinanceSpot,
1186                AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
1187                    scope,
1188                    start,
1189                    end,
1190                    6,
1191                    FillRecoveryFailure::TimedOut { timeout_secs: 30 },
1192                )),
1193            )
1194        };
1195        let index_of = |scope| match indexer.account_event(give_up(scope)) {
1196            Ok(AccountEvent {
1197                kind: AccountEventKind::FillRecoveryGaveUp(gap),
1198                ..
1199            }) => Ok(gap),
1200            Ok(other) => panic!("expected FillRecoveryGaveUp, got {other:?}"),
1201            Err(error) => Err(error),
1202        };
1203        let Ok(btc) = indexer
1204            .map
1205            .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1206        else {
1207            panic!("BTC_USDT is mapped");
1208        };
1209
1210        assert_eq!(
1211            index_of(FillRecoveryScope::Instruments(vec![
1212                InstrumentNameExchange::new("BTC_USDT")
1213            ])),
1214            Ok(FillRecoveryGap::new(
1215                FillRecoveryScope::Instruments(vec![btc]),
1216                start,
1217                end,
1218                6,
1219                FillRecoveryFailure::TimedOut { timeout_secs: 30 },
1220            ))
1221        );
1222        assert_eq!(
1223            index_of(FillRecoveryScope::AllInstruments).map(|gap| gap.scope),
1224            Ok(FillRecoveryScope::AllInstruments)
1225        );
1226        assert!(
1227            index_of(FillRecoveryScope::Instruments(vec![
1228                InstrumentNameExchange::new("BTC_USDT"),
1229                InstrumentNameExchange::new("ETHUSDT"),
1230            ]))
1231            .is_err(),
1232            "an unmapped instrument fails the whole give-up"
1233        );
1234    }
1235
1236    /// A trade amendment indexes its instrument, and a corrected one its replacement as a trade
1237    /// is indexed. An unmapped instrument fails it.
1238    #[test]
1239    fn account_event_indexes_a_trade_amendment() {
1240        use crate::{
1241            order::id::{OrderId, StrategyId},
1242            trade::TradeId,
1243        };
1244        use rust_decimal::Decimal;
1245        use rustrade_instrument::Side;
1246
1247        let indexer = binance_indexer();
1248        let time = DateTime::<Utc>::MIN_UTC;
1249        let amended = |instrument: &str, kind| {
1250            UnindexedAccountEvent::new(
1251                ExchangeId::BinanceSpot,
1252                AccountEventKind::TradeAmended(TradeAmendment::new(
1253                    InstrumentNameExchange::new(instrument),
1254                    OrderId::new("ord-1"),
1255                    time,
1256                    Some(TradeId::new("t-1")),
1257                    kind,
1258                )),
1259            )
1260        };
1261        let index_of = |event| match indexer.account_event(event) {
1262            Ok(AccountEvent {
1263                kind: AccountEventKind::TradeAmended(amendment),
1264                ..
1265            }) => Ok(amendment),
1266            Ok(other) => panic!("expected TradeAmended, got {other:?}"),
1267            Err(error) => Err(error),
1268        };
1269        let Ok(btc) = indexer
1270            .map
1271            .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1272        else {
1273            panic!("BTC_USDT is mapped");
1274        };
1275
1276        assert_eq!(
1277            index_of(amended(
1278                "BTC_USDT",
1279                TradeAmendmentKind::Busted {
1280                    quantity: Some(Decimal::ONE)
1281                }
1282            )),
1283            Ok(TradeAmendment::new(
1284                btc,
1285                OrderId::new("ord-1"),
1286                time,
1287                Some(TradeId::new("t-1")),
1288                TradeAmendmentKind::Busted {
1289                    quantity: Some(Decimal::ONE)
1290                },
1291            ))
1292        );
1293
1294        let replacement = Trade::new(
1295            TradeId::new("t-2"),
1296            OrderId::new("ord-1"),
1297            InstrumentNameExchange::new("BTC_USDT"),
1298            StrategyId::unknown(),
1299            time,
1300            Side::Buy,
1301            Decimal::TEN,
1302            Decimal::ONE,
1303            None,
1304            AssetFees::new(AssetNameExchange::new("USDT"), Decimal::ONE, None),
1305        );
1306        let Ok(TradeAmendment {
1307            kind: TradeAmendmentKind::Corrected { replacement },
1308            ..
1309        }) = index_of(amended(
1310            "BTC_USDT",
1311            TradeAmendmentKind::Corrected { replacement },
1312        ))
1313        else {
1314            panic!("a correction indexes to a correction");
1315        };
1316        assert_eq!(replacement.instrument, btc);
1317        assert_eq!(
1318            replacement.fees.fees_quote,
1319            Some(Decimal::ONE),
1320            "indexed as a trade is, fees and all"
1321        );
1322
1323        assert!(
1324            index_of(amended(
1325                "ETHUSDT",
1326                TradeAmendmentKind::Busted { quantity: None }
1327            ))
1328            .is_err(),
1329            "an unmapped instrument fails the amendment"
1330        );
1331    }
1332
1333    /// A fill is never dropped for its fee: an unmapped fee asset, or the `"UNKNOWN"` placeholder,
1334    /// keeps the trade with its fee zeroed in quote and `fees_quote: None`, with one `warn`. A fee
1335    /// asset registered with `add_asset` indexes as a third-party fee.
1336    #[test]
1337    fn trade_keeps_a_fill_whose_fee_asset_is_unmapped() {
1338        use crate::{
1339            order::id::{OrderId, StrategyId},
1340            trade::TradeId,
1341        };
1342        use rust_decimal_macros::dec;
1343        use rustrade_instrument::{Side, index::builder::IndexedInstrumentsBuilder};
1344
1345        let fill = |fee_asset: &str, fee| {
1346            Trade::new(
1347                TradeId::new("t-1"),
1348                OrderId::new("ord-1"),
1349                InstrumentNameExchange::new("BTC_USDT"),
1350                StrategyId::unknown(),
1351                DateTime::<Utc>::MIN_UTC,
1352                Side::Buy,
1353                dec!(50000),
1354                dec!(0.01),
1355                None,
1356                AssetFees::new(AssetNameExchange::new(fee_asset), fee, None),
1357            )
1358        };
1359        let asset = |indexer: &AccountEventIndexer, name: &str| {
1360            indexer
1361                .map
1362                .find_asset_index(&AssetNameExchange::new(name))
1363                .unwrap()
1364        };
1365
1366        let indexer = binance_indexer();
1367        let usdt = asset(&indexer, "USDT");
1368        let btc = asset(&indexer, "BTC");
1369
1370        let (quote, warnings) = count_warnings(|| indexer.trade(fill("USDT", dec!(0.5))));
1371        assert_eq!(
1372            quote.unwrap().fees,
1373            AssetFees::new(usdt, dec!(0.5), Some(dec!(0.5)))
1374        );
1375        assert_eq!(warnings, 0);
1376
1377        let (base, warnings) = count_warnings(|| indexer.trade(fill("BTC", dec!(0.00001))));
1378        assert_eq!(
1379            base.unwrap().fees,
1380            AssetFees::new(btc, dec!(0.00001), Some(dec!(0.5)))
1381        );
1382        assert_eq!(warnings, 0);
1383
1384        for unmapped in ["BNB", "UNKNOWN"] {
1385            let (kept, warnings) = count_warnings(|| indexer.trade(fill(unmapped, dec!(0.001))));
1386            let kept = kept.unwrap_or_else(|error| panic!("{unmapped}: dropped: {error}"));
1387            assert_eq!(
1388                kept.quantity,
1389                dec!(0.01),
1390                "{unmapped}: the fill itself is intact"
1391            );
1392            assert_eq!(
1393                kept.fees,
1394                AssetFees::new(usdt, Decimal::ZERO, None),
1395                "{unmapped}: fee zeroed in quote, with no quote equivalent"
1396            );
1397            assert_eq!(warnings, 1, "{unmapped}: the zeroed fee is reported");
1398        }
1399
1400        let instruments = IndexedInstrumentsBuilder::new()
1401            .add_instrument(test_utils::instrument(
1402                ExchangeId::BinanceSpot,
1403                "BTC",
1404                "USDT",
1405            ))
1406            .add_asset(test_utils::exchange_asset(ExchangeId::BinanceSpot, "BNB"))
1407            .build();
1408        let with_bnb = AccountEventIndexer::new(Arc::new(
1409            generate_execution_instrument_map(&instruments, ExchangeId::BinanceSpot).unwrap(),
1410        ));
1411        let bnb = asset(&with_bnb, "BNB");
1412        let (third_party, warnings) = count_warnings(|| with_bnb.trade(fill("BNB", dec!(0.001))));
1413        assert_eq!(
1414            third_party.unwrap().fees,
1415            AssetFees::new(bnb, dec!(0.001), None),
1416            "a registered fee asset keeps its fee, with no quote equivalent"
1417        );
1418        assert_eq!(warnings, 0);
1419
1420        let mut unmapped_instrument = fill("USDT", dec!(0.5));
1421        unmapped_instrument.instrument = InstrumentNameExchange::new("ETH_USDT");
1422        assert!(
1423            indexer.trade(unmapped_instrument).is_err(),
1424            "an unmapped instrument still fails"
1425        );
1426    }
1427
1428    /// Run `index`, returning its value and the number of `WARN` events it emitted on this
1429    /// thread. A degrade is reported only by its log line, so asserting on the value alone could
1430    /// not tell a logged degrade from a silent one.
1431    fn count_warnings<T>(index: impl FnOnce() -> T) -> (T, usize) {
1432        use std::sync::atomic::{AtomicUsize, Ordering};
1433        use tracing_subscriber::layer::SubscriberExt;
1434
1435        #[derive(Default)]
1436        struct CountWarnings(Arc<AtomicUsize>);
1437
1438        impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CountWarnings {
1439            fn on_event(
1440                &self,
1441                event: &tracing::Event<'_>,
1442                _: tracing_subscriber::layer::Context<'_, S>,
1443            ) {
1444                if *event.metadata().level() == tracing::Level::WARN {
1445                    self.0.fetch_add(1, Ordering::Relaxed);
1446                }
1447            }
1448        }
1449
1450        let layer = CountWarnings::default();
1451        let count = Arc::clone(&layer.0);
1452        // Thread-local, so other tests running in parallel are not counted.
1453        let value =
1454            tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), index);
1455        (value, count.load(Ordering::Relaxed))
1456    }
1457
1458    fn btc_usdt() -> InstrumentNameExchange {
1459        binance_indexer()
1460            .map
1461            .exchange_instruments()
1462            .next()
1463            .cloned()
1464            .unwrap_or_else(|| panic!("the map holds one instrument"))
1465    }
1466
1467    #[test]
1468    fn balance_insufficient_keeps_a_held_asset_and_drops_an_unheld_one() {
1469        let indexer = binance_indexer();
1470        let btc = indexer
1471            .map
1472            .find_asset_index(&AssetNameExchange::new("BTC"))
1473            .unwrap();
1474
1475        for (asset, expected, warnings) in [
1476            (Some("BTC"), Some(btc), 0),
1477            // An instrument name, as Binance's error mapping once put there.
1478            (Some("ETH_USDT"), None, 1),
1479            (None, None, 0),
1480        ] {
1481            let (indexed, warned) = count_warnings(|| {
1482                indexer.api_error(ApiError::BalanceInsufficient(
1483                    asset.map(AssetNameExchange::new),
1484                    "low".to_string(),
1485                ))
1486            });
1487            assert_eq!(
1488                indexed,
1489                ApiError::BalanceInsufficient(expected, "low".to_string()),
1490                "{asset:?}"
1491            );
1492            assert_eq!(warned, warnings, "{asset:?}: a dropped asset is logged");
1493        }
1494    }
1495
1496    /// An invalid asset or instrument the map does not hold has no index to carry, so it becomes
1497    /// a rejection keeping its own message: `OrderRejected` for an order, `RequestRejected` for
1498    /// any other request.
1499    #[test]
1500    fn an_unresolvable_invalid_name_degrades_to_a_rejection_keeping_its_text() {
1501        let indexer = binance_indexer();
1502        for (error, text) in [
1503            (
1504                ApiError::AssetInvalid(AssetNameExchange::new("XYZ"), "bad".to_string()),
1505                "asset XYZ invalid: bad",
1506            ),
1507            (
1508                ApiError::InstrumentInvalid(
1509                    InstrumentNameExchange::new("XYZUSDT"),
1510                    "bad".to_string(),
1511                ),
1512                "instrument XYZUSDT invalid: bad",
1513            ),
1514        ] {
1515            let (indexed, warned) = count_warnings(|| indexer.api_error(error.clone()));
1516            assert_eq!(indexed, ApiError::OrderRejected(text.to_string()));
1517            assert_eq!(warned, 1, "{text}: a degrade is logged");
1518
1519            let (indexed, warned) =
1520                count_warnings(|| indexer.client_error(ClientError::Api(error)));
1521            assert_eq!(
1522                indexed,
1523                ClientError::Api(ApiError::RequestRejected(text.to_string()))
1524            );
1525            assert_eq!(warned, 1, "{text}: a degrade is logged");
1526        }
1527
1528        // A name the map holds still indexes, and nothing is logged.
1529        let (indexed, warned) = count_warnings(|| {
1530            indexer.api_error(ApiError::InstrumentInvalid(btc_usdt(), "bad".to_string()))
1531        });
1532        assert!(matches!(indexed, ApiError::InstrumentInvalid(_, _)));
1533        assert_eq!(warned, 0);
1534    }
1535
1536    /// A response whose rejection names something the map does not hold still indexes, so the
1537    /// order is settled rather than left in flight.
1538    #[test]
1539    fn a_response_whose_rejection_names_an_unresolvable_asset_still_indexes() {
1540        let indexer = binance_indexer();
1541        let rejection = || {
1542            OrderError::Rejected(ApiError::BalanceInsufficient(
1543                Some(AssetNameExchange::new("ETH_USDT")),
1544                "low".to_string(),
1545            ))
1546        };
1547        let indexed = OrderError::Rejected(ApiError::BalanceInsufficient(None, "low".to_string()));
1548
1549        assert_eq!(
1550            indexer.order_state(OrderState::inactive(rejection())),
1551            OrderState::inactive(indexed.clone())
1552        );
1553
1554        let key = OrderKey {
1555            exchange: ExchangeId::BinanceSpot,
1556            instrument: btc_usdt(),
1557            strategy: crate::order::id::StrategyId::new("test"),
1558            cid: crate::order::id::ClientOrderId::random(),
1559        };
1560        let cancel = indexer
1561            .order_response_cancel(OrderResponseCancel {
1562                key,
1563                state: Err(rejection()),
1564            })
1565            .unwrap();
1566        assert_eq!(cancel.state, Err(indexed));
1567    }
1568
1569    /// A cash flow indexes its asset and, when it names one, its instrument, and fails when
1570    /// either is not in the map.
1571    #[test]
1572    fn account_event_indexes_a_cash_flow() {
1573        use crate::cash_flow::{CashFlowId, CashFlowKind};
1574        use rust_decimal_macros::dec;
1575
1576        let indexer = binance_indexer();
1577        let time = DateTime::<Utc>::MIN_UTC;
1578        let kind = CashFlowKind::Funding {
1579            rate: Some(dec!(0.0001)),
1580            position_quantity: Some(dec!(-2)),
1581        };
1582        let flow = |asset: &str, instrument: Option<&str>| {
1583            UnindexedAccountEvent::new(
1584                ExchangeId::BinanceSpot,
1585                AccountEventKind::CashFlow(CashFlow::new(
1586                    kind.clone(),
1587                    AssetNameExchange::new(asset),
1588                    dec!(1.5),
1589                    instrument.map(InstrumentNameExchange::new),
1590                    time,
1591                    Some(CashFlowId::new("cf-1")),
1592                )),
1593            )
1594        };
1595        let index_of = |event| match indexer.account_event(event) {
1596            Ok(AccountEvent {
1597                kind: AccountEventKind::CashFlow(flow),
1598                ..
1599            }) => Ok(flow),
1600            Ok(other) => panic!("expected CashFlow, got {other:?}"),
1601            Err(error) => Err(error),
1602        };
1603        let (Ok(usdt), Ok(btc_usdt)) = (
1604            indexer
1605                .map
1606                .find_asset_index(&AssetNameExchange::new("USDT")),
1607            indexer
1608                .map
1609                .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT")),
1610        ) else {
1611            panic!("USDT and BTC_USDT are mapped");
1612        };
1613
1614        assert_eq!(
1615            index_of(flow("USDT", Some("BTC_USDT"))),
1616            Ok(CashFlow::new(
1617                kind.clone(),
1618                usdt,
1619                dec!(1.5),
1620                Some(btc_usdt),
1621                time,
1622                Some(CashFlowId::new("cf-1")),
1623            ))
1624        );
1625        assert_eq!(
1626            index_of(flow("USDT", None)).map(|flow| flow.instrument),
1627            Ok(None)
1628        );
1629        assert!(
1630            index_of(flow("ETH", None)).is_err(),
1631            "an unmapped asset fails"
1632        );
1633        assert!(
1634            index_of(flow("USDT", Some("ETHUSDT"))).is_err(),
1635            "an unmapped instrument fails"
1636        );
1637    }
1638
1639    /// A notice indexes its instrument when it names a mapped one, and is kept unattributed,
1640    /// rather than dropped, when it names one that is not mapped.
1641    #[test]
1642    fn account_event_indexes_a_notice() {
1643        use crate::notice::{AccountNotice, NoticeKind};
1644        use rust_decimal_macros::dec;
1645        use smol_str::SmolStr;
1646
1647        let indexer = binance_indexer();
1648        let time = DateTime::<Utc>::MIN_UTC;
1649        let notice = |instrument: Option<&str>| {
1650            UnindexedAccountEvent::new(
1651                ExchangeId::BinanceSpot,
1652                AccountEventKind::Notice(AccountNotice::new(
1653                    NoticeKind::MarginCall,
1654                    instrument.map(InstrumentNameExchange::new),
1655                    time,
1656                    Some(dec!(1.1)),
1657                    SmolStr::new("MARGIN_CALL"),
1658                )),
1659            )
1660        };
1661        let index_of = |event| match indexer.account_event(event) {
1662            Ok(AccountEvent {
1663                kind: AccountEventKind::Notice(notice),
1664                ..
1665            }) => notice,
1666            other => panic!("expected Notice, got {other:?}"),
1667        };
1668        let Ok(btc_usdt) = indexer
1669            .map
1670            .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1671        else {
1672            panic!("BTC_USDT is mapped");
1673        };
1674
1675        assert_eq!(
1676            index_of(notice(Some("BTC_USDT"))),
1677            AccountNotice::new(
1678                NoticeKind::MarginCall,
1679                Some(btc_usdt),
1680                time,
1681                Some(dec!(1.1)),
1682                SmolStr::new("MARGIN_CALL"),
1683            )
1684        );
1685        assert_eq!(index_of(notice(None)).instrument, None);
1686        assert_eq!(
1687            index_of(notice(Some("ETHUSDT"))).instrument,
1688            None,
1689            "an unmapped instrument is dropped from the notice, not the notice itself"
1690        );
1691    }
1692
1693    /// A margin debt update indexes its asset and instrument, and one naming an unmapped
1694    /// instrument fails rather than becoming the account's debt.
1695    #[test]
1696    fn account_event_indexes_a_margin_debt_update() {
1697        use crate::{
1698            balance::MarginDetails,
1699            margin_debt::{MarginDebtKind, MarginDebtUpdate},
1700        };
1701        use rust_decimal_macros::dec;
1702        use smol_str::SmolStr;
1703
1704        let indexer = binance_indexer();
1705        let time = DateTime::<Utc>::MIN_UTC;
1706        let debt = MarginDetails::new(dec!(0.5), dec!(0.01));
1707        let update = |asset: &str, instrument: Option<&str>| {
1708            UnindexedAccountEvent::new(
1709                ExchangeId::BinanceSpot,
1710                AccountEventKind::MarginDebt(MarginDebtUpdate::new(
1711                    AssetNameExchange::new(asset),
1712                    instrument.map(InstrumentNameExchange::new),
1713                    debt,
1714                    MarginDebtKind::Borrow,
1715                    SmolStr::new("BORROW"),
1716                    time,
1717                )),
1718            )
1719        };
1720        let index_of = |event| match indexer.account_event(event) {
1721            Ok(AccountEvent {
1722                kind: AccountEventKind::MarginDebt(update),
1723                ..
1724            }) => Ok(update),
1725            Ok(other) => panic!("expected MarginDebt, got {other:?}"),
1726            Err(error) => Err(error),
1727        };
1728        let (Ok(btc), Ok(btc_usdt)) = (
1729            indexer.map.find_asset_index(&AssetNameExchange::new("BTC")),
1730            indexer
1731                .map
1732                .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT")),
1733        ) else {
1734            panic!("BTC and BTC_USDT are mapped");
1735        };
1736
1737        assert_eq!(
1738            index_of(update("BTC", Some("BTC_USDT"))),
1739            Ok(MarginDebtUpdate::new(
1740                btc,
1741                Some(btc_usdt),
1742                debt,
1743                MarginDebtKind::Borrow,
1744                SmolStr::new("BORROW"),
1745                time,
1746            ))
1747        );
1748        assert_eq!(
1749            index_of(update("BTC", None)).map(|update| update.instrument),
1750            Ok(None)
1751        );
1752        assert!(
1753            index_of(update("ETH", None)).is_err(),
1754            "an unmapped asset fails"
1755        );
1756        assert!(
1757            index_of(update("BTC", Some("ETHUSDT"))).is_err(),
1758            "an unmapped instrument fails"
1759        );
1760    }
1761}