Skip to main content

rustrade_execution/
indexer.rs

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