nautilus_live/execution/
emitter.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live execution event emitter for async event dispatch.
17//!
18//! This module provides [`ExecutionEventEmitter`], which combines event generation (via
19//! [`OrderEventFactory`]) with async dispatch. Adapters use the `emit_*` convenience
20//! methods to generate and send events in a single call.
21//!
22//! # Architecture
23//!
24//! ```text
25//! Adapter
26//! |-- core: ExecutionClientCore    (identity + connection state)
27//! `-- emitter: ExecutionEventEmitter   (event generation + async dispatch)
28//!     |-- factory: OrderEventFactory
29//!     `-- sender: ArcSwapOption<Sender>   (shared slot, installed at create() and start())
30//! ```
31
32use std::sync::Arc;
33
34use arc_swap::ArcSwapOption;
35use nautilus_common::{
36    factories::OrderEventFactory,
37    live::sender::EventSender,
38    messages::{ExecutionEvent, ExecutionReport},
39};
40use nautilus_core::{Params, UUID4, UnixNanos, time::AtomicTime};
41use nautilus_model::{
42    enums::{AccountType, LiquiditySide},
43    events::{
44        AccountState, OrderAcceptedBatch, OrderCancelRejected, OrderCanceledBatch, OrderEventAny,
45        OrderModifyRejected, OrderRejected, OrderSubmittedBatch,
46    },
47    identifiers::{
48        AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId, TraderId,
49        VenueOrderId,
50    },
51    orders::OrderAny,
52    reports::{FillReport, OrderStatusReport, PositionStatusReport},
53    types::{AccountBalance, Currency, MarginBalance, Money, Price, Quantity},
54};
55
56/// Event emitter for live trading - combines event generation with async dispatch.
57///
58/// This struct wraps an [`OrderEventFactory`] for event construction and an unbounded
59/// channel sender for async dispatch. It provides `emit_*` convenience methods that
60/// generate and send events in a single call.
61///
62/// The sender is installed via [`set_sender`](Self::set_sender) in the client's `start`, and in the
63/// execution client factory's `create` when the calling thread has one.
64/// Clones share the sender slot and observe later sender installations and replacements.
65#[derive(Debug, Clone)]
66pub struct ExecutionEventEmitter {
67    clock: &'static AtomicTime,
68    factory: OrderEventFactory,
69    sender: Arc<ArcSwapOption<EventSender<ExecutionEvent>>>,
70}
71
72impl ExecutionEventEmitter {
73    /// Creates a new [`ExecutionEventEmitter`] with no sender.
74    ///
75    /// Call [`set_sender`](Self::set_sender) in the client's `start`, and in the factory's `create`
76    /// when `try_get_exec_event_sender` returns `Some`.
77    #[must_use]
78    pub fn new(
79        clock: &'static AtomicTime,
80        trader_id: TraderId,
81        account_id: AccountId,
82        account_type: AccountType,
83        base_currency: Option<Currency>,
84    ) -> Self {
85        Self {
86            clock,
87            factory: OrderEventFactory::new(trader_id, account_id, account_type, base_currency),
88            sender: Arc::new(ArcSwapOption::empty()),
89        }
90    }
91
92    fn ts_init(&self) -> UnixNanos {
93        self.clock.get_time_ns()
94    }
95
96    /// Installs or replaces the sender for this emitter and all its clones.
97    ///
98    /// The slot is shared, so a clone taken before this call observes the sender it installs.
99    /// Events emitted before any install are dropped with a warning.
100    ///
101    /// Call in the client's `start`, resolved from `get_exec_event_sender`: `LiveNode` rebinds the
102    /// runner's senders on the calling thread before it starts clients, so the `start` install is
103    /// the authoritative one. Call it in the execution client factory's `create` as well when
104    /// `try_get_exec_event_sender` returns `Some`; `None` there means the calling thread has no
105    /// bound senders and is not a construction failure. See the adapter guide for hosts that drive
106    /// a client outside `LiveNode`.
107    pub fn set_sender(&mut self, sender: impl Into<EventSender<ExecutionEvent>>) {
108        self.sender.store(Some(Arc::new(sender.into())));
109    }
110
111    /// Returns true if the sender is initialized for this emitter and its clones.
112    #[must_use]
113    pub fn is_initialized(&self) -> bool {
114        self.sender.load().is_some()
115    }
116
117    /// Returns the trader ID.
118    #[must_use]
119    pub fn trader_id(&self) -> TraderId {
120        self.factory.trader_id()
121    }
122
123    /// Returns the account ID.
124    #[must_use]
125    pub fn account_id(&self) -> AccountId {
126        self.factory.account_id()
127    }
128
129    /// Sets the account ID for generated events.
130    pub fn set_account_id(&mut self, account_id: AccountId) {
131        self.factory.set_account_id(account_id);
132    }
133
134    /// Generates and emits an account state event.
135    pub fn emit_account_state(
136        &self,
137        balances: Vec<AccountBalance>,
138        margins: Vec<MarginBalance>,
139        reported: bool,
140        ts_event: UnixNanos,
141        info: Option<Params>,
142    ) {
143        if let Err(e) = self.try_emit_account_state(balances, margins, reported, ts_event, info) {
144            log::warn!("{e}");
145        }
146    }
147
148    /// Generates and emits an account state event, reporting dispatch failures.
149    ///
150    /// # Errors
151    ///
152    /// Returns an error if the sender is uninitialized or its receiver is closed.
153    pub fn try_emit_account_state(
154        &self,
155        balances: Vec<AccountBalance>,
156        margins: Vec<MarginBalance>,
157        reported: bool,
158        ts_event: UnixNanos,
159        info: Option<Params>,
160    ) -> anyhow::Result<()> {
161        let state = self.factory.generate_account_state(
162            balances,
163            margins,
164            reported,
165            ts_event,
166            self.ts_init(),
167            info,
168        );
169        self.try_send_account_state(state)
170    }
171
172    /// Generates and emits an order denied event.
173    pub fn emit_order_denied(&self, order: &OrderAny, reason: &str) {
174        let event = self
175            .factory
176            .generate_order_denied(order, reason, self.ts_init());
177        self.send_order_event(event);
178    }
179
180    /// Generates and emits an order submitted event.
181    pub fn emit_order_submitted(&self, order: &OrderAny) {
182        if let Err(e) = self.try_emit_order_submitted(order) {
183            log::warn!("{e}");
184        }
185    }
186
187    /// Generates and emits an order submitted event, reporting dispatch failures.
188    ///
189    /// # Errors
190    ///
191    /// Returns an error if the sender is uninitialized or its receiver is closed.
192    pub fn try_emit_order_submitted(&self, order: &OrderAny) -> anyhow::Result<()> {
193        let event = self.factory.generate_order_submitted(order, self.ts_init());
194        self.try_send_order_event(event)
195    }
196
197    /// Generates and emits an order rejected event.
198    pub fn emit_order_rejected(
199        &self,
200        order: &OrderAny,
201        reason: &str,
202        ts_event: UnixNanos,
203        due_post_only: bool,
204    ) {
205        let event = self.factory.generate_order_rejected(
206            order,
207            reason,
208            ts_event,
209            self.ts_init(),
210            due_post_only,
211        );
212        self.send_order_event(event);
213    }
214
215    /// Generates and emits an order accepted event.
216    pub fn emit_order_accepted(
217        &self,
218        order: &OrderAny,
219        venue_order_id: VenueOrderId,
220        ts_event: UnixNanos,
221    ) {
222        let event =
223            self.factory
224                .generate_order_accepted(order, venue_order_id, ts_event, self.ts_init());
225        self.send_order_event(event);
226    }
227
228    /// Generates and emits an order modify rejected event.
229    pub fn emit_order_modify_rejected(
230        &self,
231        order: &OrderAny,
232        venue_order_id: Option<VenueOrderId>,
233        reason: &str,
234        ts_event: UnixNanos,
235    ) {
236        let event = self.factory.generate_order_modify_rejected(
237            order,
238            venue_order_id,
239            reason,
240            ts_event,
241            self.ts_init(),
242        );
243        self.send_order_event(event);
244    }
245
246    /// Generates and emits an order cancel rejected event.
247    pub fn emit_order_cancel_rejected(
248        &self,
249        order: &OrderAny,
250        venue_order_id: Option<VenueOrderId>,
251        reason: &str,
252        ts_event: UnixNanos,
253    ) {
254        let event = self.factory.generate_order_cancel_rejected(
255            order,
256            venue_order_id,
257            reason,
258            ts_event,
259            self.ts_init(),
260        );
261        self.send_order_event(event);
262    }
263
264    /// Generates and emits an order updated event.
265    #[expect(clippy::too_many_arguments)]
266    pub fn emit_order_updated(
267        &self,
268        order: &OrderAny,
269        venue_order_id: VenueOrderId,
270        quantity: Quantity,
271        price: Option<Price>,
272        trigger_price: Option<Price>,
273        protection_price: Option<Price>,
274        ts_event: UnixNanos,
275    ) {
276        let event = self.factory.generate_order_updated(
277            order,
278            venue_order_id,
279            quantity,
280            price,
281            trigger_price,
282            protection_price,
283            ts_event,
284            self.ts_init(),
285        );
286        self.send_order_event(event);
287    }
288
289    /// Generates and emits an order canceled event.
290    pub fn emit_order_canceled(
291        &self,
292        order: &OrderAny,
293        venue_order_id: Option<VenueOrderId>,
294        ts_event: UnixNanos,
295    ) {
296        let event =
297            self.factory
298                .generate_order_canceled(order, venue_order_id, ts_event, self.ts_init());
299        self.send_order_event(event);
300    }
301
302    /// Generates and emits an order triggered event.
303    pub fn emit_order_triggered(
304        &self,
305        order: &OrderAny,
306        venue_order_id: Option<VenueOrderId>,
307        ts_event: UnixNanos,
308    ) {
309        let event =
310            self.factory
311                .generate_order_triggered(order, venue_order_id, ts_event, self.ts_init());
312        self.send_order_event(event);
313    }
314
315    /// Generates and emits an order expired event.
316    pub fn emit_order_expired(
317        &self,
318        order: &OrderAny,
319        venue_order_id: Option<VenueOrderId>,
320        ts_event: UnixNanos,
321    ) {
322        let event =
323            self.factory
324                .generate_order_expired(order, venue_order_id, ts_event, self.ts_init());
325        self.send_order_event(event);
326    }
327
328    /// Generates and emits an order filled event.
329    #[expect(clippy::too_many_arguments)]
330    pub fn emit_order_filled(
331        &self,
332        order: &OrderAny,
333        venue_order_id: VenueOrderId,
334        venue_position_id: Option<PositionId>,
335        trade_id: TradeId,
336        last_qty: Quantity,
337        last_px: Price,
338        quote_currency: Currency,
339        commission: Option<Money>,
340        liquidity_side: LiquiditySide,
341        ts_event: UnixNanos,
342    ) {
343        let event = self.factory.generate_order_filled(
344            order,
345            venue_order_id,
346            venue_position_id,
347            trade_id,
348            last_qty,
349            last_px,
350            quote_currency,
351            commission,
352            liquidity_side,
353            ts_event,
354            self.ts_init(),
355        );
356        self.send_order_event(event);
357    }
358
359    /// Constructs and emits an order rejected event from raw fields.
360    pub fn emit_order_rejected_event(
361        &self,
362        strategy_id: StrategyId,
363        instrument_id: InstrumentId,
364        client_order_id: ClientOrderId,
365        reason: &str,
366        ts_event: UnixNanos,
367        due_post_only: bool,
368    ) {
369        let event = OrderRejected::new(
370            self.factory.trader_id(),
371            strategy_id,
372            instrument_id,
373            client_order_id,
374            self.factory.account_id(),
375            reason.into(),
376            UUID4::new(),
377            ts_event,
378            self.ts_init(),
379            false,
380            due_post_only,
381        );
382        self.send_order_event(OrderEventAny::Rejected(event));
383    }
384
385    /// Constructs and emits an order modify rejected event from raw fields.
386    pub fn emit_order_modify_rejected_event(
387        &self,
388        strategy_id: StrategyId,
389        instrument_id: InstrumentId,
390        client_order_id: ClientOrderId,
391        venue_order_id: Option<VenueOrderId>,
392        reason: &str,
393        ts_event: UnixNanos,
394    ) {
395        let event = OrderModifyRejected::new(
396            self.factory.trader_id(),
397            strategy_id,
398            instrument_id,
399            client_order_id,
400            reason.into(),
401            UUID4::new(),
402            ts_event,
403            self.ts_init(),
404            false,
405            venue_order_id,
406            Some(self.factory.account_id()),
407        );
408        self.send_order_event(OrderEventAny::ModifyRejected(event));
409    }
410
411    /// Constructs and emits an order cancel rejected event from raw fields.
412    pub fn emit_order_cancel_rejected_event(
413        &self,
414        strategy_id: StrategyId,
415        instrument_id: InstrumentId,
416        client_order_id: ClientOrderId,
417        venue_order_id: Option<VenueOrderId>,
418        reason: &str,
419        ts_event: UnixNanos,
420    ) {
421        let event = OrderCancelRejected::new(
422            self.factory.trader_id(),
423            strategy_id,
424            instrument_id,
425            client_order_id,
426            reason.into(),
427            UUID4::new(),
428            ts_event,
429            self.ts_init(),
430            false,
431            venue_order_id,
432            Some(self.factory.account_id()),
433        );
434        self.send_order_event(OrderEventAny::CancelRejected(event));
435    }
436
437    /// Emits an order event.
438    pub fn send_order_event(&self, event: OrderEventAny) {
439        if let Err(e) = self.try_send_order_event(event) {
440            log::warn!("{e}");
441        }
442    }
443
444    /// Emits an order event and returns any channel error to the caller.
445    ///
446    /// # Errors
447    ///
448    /// Returns an error if the sender is uninitialized or its receiver is closed.
449    pub fn try_send_order_event(&self, event: OrderEventAny) -> anyhow::Result<()> {
450        self.try_send(ExecutionEvent::Order(event), "order event")
451    }
452
453    /// Emits a batch of order submitted events as a single channel message.
454    pub fn send_order_submitted_batch(&self, batch: OrderSubmittedBatch) {
455        if let Err(e) = self.try_send(
456            ExecutionEvent::OrderSubmittedBatch(batch),
457            "order submitted batch",
458        ) {
459            log::warn!("{e}");
460        }
461    }
462
463    /// Emits a batch of order accepted events as a single channel message.
464    pub fn send_order_accepted_batch(&self, batch: OrderAcceptedBatch) {
465        if let Err(e) = self.try_send(
466            ExecutionEvent::OrderAcceptedBatch(batch),
467            "order accepted batch",
468        ) {
469            log::warn!("{e}");
470        }
471    }
472
473    /// Emits a batch of order canceled events as a single channel message.
474    pub fn send_order_canceled_batch(&self, batch: OrderCanceledBatch) {
475        if let Err(e) = self.try_send(
476            ExecutionEvent::OrderCanceledBatch(batch),
477            "order canceled batch",
478        ) {
479            log::warn!("{e}");
480        }
481    }
482
483    /// Emits an account state event.
484    pub fn send_account_state(&self, state: AccountState) {
485        if let Err(e) = self.try_send_account_state(state) {
486            log::warn!("{e}");
487        }
488    }
489
490    /// Emits an account state event and returns any channel error to the caller.
491    ///
492    /// # Errors
493    ///
494    /// Returns an error if the sender is uninitialized or its receiver is closed.
495    pub fn try_send_account_state(&self, state: AccountState) -> anyhow::Result<()> {
496        self.try_send(ExecutionEvent::Account(state), "account state")
497    }
498
499    /// Emits an execution report.
500    pub fn send_execution_report(&self, report: ExecutionReport) {
501        if let Err(e) = self.try_send_execution_report(report) {
502            log::warn!("{e}");
503        }
504    }
505
506    /// Emits an execution report and returns any channel error to the caller.
507    ///
508    /// # Errors
509    ///
510    /// Returns an error if the sender is not initialized or the receiving channel is closed.
511    pub fn try_send_execution_report(&self, report: ExecutionReport) -> anyhow::Result<()> {
512        self.try_send(ExecutionEvent::Report(report), "execution report")
513    }
514
515    /// Emits an order status report.
516    pub fn send_order_status_report(&self, report: OrderStatusReport) {
517        self.send_execution_report(ExecutionReport::Order(Box::new(report)));
518    }
519
520    /// Emits a fill report.
521    pub fn send_fill_report(&self, report: FillReport) {
522        self.send_execution_report(ExecutionReport::Fill(Box::new(report)));
523    }
524
525    /// Emits an order status report bundled with the fills that produced it.
526    pub fn send_order_with_fills(&self, report: OrderStatusReport, fills: Vec<FillReport>) {
527        self.send_execution_report(ExecutionReport::OrderWithFills(Box::new(report), fills));
528    }
529
530    /// Emits a position status report.
531    pub fn send_position_report(&self, report: PositionStatusReport) {
532        self.send_execution_report(ExecutionReport::Position(Box::new(report)));
533    }
534
535    fn try_send(&self, event: ExecutionEvent, kind: &str) -> anyhow::Result<()> {
536        let sender = self.sender.load();
537        let sender = sender
538            .as_ref()
539            .ok_or_else(|| anyhow::anyhow!("Cannot send {kind}: sender not initialized"))?;
540        sender
541            .send(event)
542            .map_err(|e| anyhow::anyhow!("Failed to send {kind}: {e}"))
543    }
544}
545
546#[cfg(test)]
547mod tests {
548    use nautilus_core::time::get_atomic_clock_static;
549    use nautilus_model::events::order::spec::OrderSubmittedSpec;
550    use rstest::rstest;
551
552    use super::*;
553
554    fn create_emitter() -> ExecutionEventEmitter {
555        ExecutionEventEmitter::new(
556            get_atomic_clock_static(),
557            TraderId::from("TRADER-001"),
558            AccountId::from("SIM-001"),
559            AccountType::Cash,
560            None,
561        )
562    }
563
564    fn create_order_event() -> OrderEventAny {
565        OrderEventAny::Submitted(
566            OrderSubmittedSpec::builder()
567                .client_order_id(ClientOrderId::from("O-001"))
568                .build(),
569        )
570    }
571
572    #[rstest]
573    fn test_clone_before_set_sender_observes_sender() {
574        let mut emitter = create_emitter();
575        let cloned = emitter.clone();
576        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
577
578        emitter.set_sender(tx);
579
580        assert!(cloned.is_initialized());
581        cloned.send_order_event(create_order_event());
582        assert!(matches!(
583            rx.try_recv(),
584            Ok(ExecutionEvent::Order(OrderEventAny::Submitted(_)))
585        ));
586    }
587
588    #[rstest]
589    fn test_set_sender_replaces_existing_sender() {
590        let mut emitter = create_emitter();
591        let (tx_a, mut rx_a) = tokio::sync::mpsc::unbounded_channel();
592        let (tx_b, mut rx_b) = tokio::sync::mpsc::unbounded_channel();
593
594        emitter.set_sender(tx_a);
595        emitter.set_sender(tx_b);
596        emitter.send_order_event(create_order_event());
597
598        assert!(matches!(
599            rx_a.try_recv(),
600            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
601        ));
602        assert!(matches!(
603            rx_b.try_recv(),
604            Ok(ExecutionEvent::Order(OrderEventAny::Submitted(_)))
605        ));
606    }
607
608    #[rstest]
609    fn test_never_initialized_sender_drops_or_errors() {
610        let emitter = create_emitter();
611
612        emitter.send_order_event(create_order_event());
613        let error = emitter
614            .try_send_order_event(create_order_event())
615            .unwrap_err();
616
617        assert_eq!(
618            error.to_string(),
619            "Cannot send order event: sender not initialized"
620        );
621    }
622}
nautilus_live/execution/emitter.rs

nautilus_live/execution/
emitter.rs
