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: Option<Sender>   (set in start())
30//! ```
31
32use nautilus_common::{
33    factories::OrderEventFactory,
34    messages::{ExecutionEvent, ExecutionReport},
35};
36use nautilus_core::{UUID4, UnixNanos, time::AtomicTime};
37use nautilus_model::{
38    enums::{AccountType, LiquiditySide},
39    events::{
40        AccountState, OrderAcceptedBatch, OrderCancelRejected, OrderCanceledBatch, OrderEventAny,
41        OrderModifyRejected, OrderRejected, OrderSubmittedBatch,
42    },
43    identifiers::{
44        AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId, TraderId,
45        VenueOrderId,
46    },
47    orders::OrderAny,
48    reports::{FillReport, OrderStatusReport, PositionStatusReport},
49    types::{AccountBalance, Currency, MarginBalance, Money, Price, Quantity},
50};
51
52/// Event emitter for live trading - combines event generation with async dispatch.
53///
54/// This struct wraps an [`OrderEventFactory`] for event construction and an unbounded
55/// channel sender for async dispatch. It provides `emit_*` convenience methods that
56/// generate and send events in a single call.
57///
58/// The sender is set during the adapter's `start()` phase via [`set_sender`](Self::set_sender).
59#[derive(Debug, Clone)]
60pub struct ExecutionEventEmitter {
61    clock: &'static AtomicTime,
62    factory: OrderEventFactory,
63    sender: Option<tokio::sync::mpsc::UnboundedSender<ExecutionEvent>>,
64}
65
66impl ExecutionEventEmitter {
67    /// Creates a new [`ExecutionEventEmitter`] with no sender.
68    ///
69    /// Call [`set_sender`](Self::set_sender) in the adapter's `start()` method.
70    #[must_use]
71    pub fn new(
72        clock: &'static AtomicTime,
73        trader_id: TraderId,
74        account_id: AccountId,
75        account_type: AccountType,
76        base_currency: Option<Currency>,
77    ) -> Self {
78        Self {
79            clock,
80            factory: OrderEventFactory::new(trader_id, account_id, account_type, base_currency),
81            sender: None,
82        }
83    }
84
85    fn ts_init(&self) -> UnixNanos {
86        self.clock.get_time_ns()
87    }
88
89    /// Sets the sender. Call in adapter's `start()`.
90    pub fn set_sender(&mut self, sender: tokio::sync::mpsc::UnboundedSender<ExecutionEvent>) {
91        self.sender = Some(sender);
92    }
93
94    /// Returns true if the sender is initialized.
95    #[must_use]
96    pub fn is_initialized(&self) -> bool {
97        self.sender.is_some()
98    }
99
100    /// Returns the trader ID.
101    #[must_use]
102    pub fn trader_id(&self) -> TraderId {
103        self.factory.trader_id()
104    }
105
106    /// Returns the account ID.
107    #[must_use]
108    pub fn account_id(&self) -> AccountId {
109        self.factory.account_id()
110    }
111
112    /// Sets the account ID for generated events.
113    pub fn set_account_id(&mut self, account_id: AccountId) {
114        self.factory.set_account_id(account_id);
115    }
116
117    /// Generates and emits an account state event.
118    pub fn emit_account_state(
119        &self,
120        balances: Vec<AccountBalance>,
121        margins: Vec<MarginBalance>,
122        reported: bool,
123        ts_event: UnixNanos,
124    ) {
125        let state = self.factory.generate_account_state(
126            balances,
127            margins,
128            reported,
129            ts_event,
130            self.ts_init(),
131        );
132        self.send_account_state(state);
133    }
134
135    /// Generates and emits an order denied event.
136    pub fn emit_order_denied(&self, order: &OrderAny, reason: &str) {
137        let event = self
138            .factory
139            .generate_order_denied(order, reason, self.ts_init());
140        self.send_order_event(event);
141    }
142
143    /// Generates and emits an order submitted event.
144    pub fn emit_order_submitted(&self, order: &OrderAny) {
145        let event = self.factory.generate_order_submitted(order, self.ts_init());
146        self.send_order_event(event);
147    }
148
149    /// Generates and emits an order rejected event.
150    pub fn emit_order_rejected(
151        &self,
152        order: &OrderAny,
153        reason: &str,
154        ts_event: UnixNanos,
155        due_post_only: bool,
156    ) {
157        let event = self.factory.generate_order_rejected(
158            order,
159            reason,
160            ts_event,
161            self.ts_init(),
162            due_post_only,
163        );
164        self.send_order_event(event);
165    }
166
167    /// Generates and emits an order accepted event.
168    pub fn emit_order_accepted(
169        &self,
170        order: &OrderAny,
171        venue_order_id: VenueOrderId,
172        ts_event: UnixNanos,
173    ) {
174        let event =
175            self.factory
176                .generate_order_accepted(order, venue_order_id, ts_event, self.ts_init());
177        self.send_order_event(event);
178    }
179
180    /// Generates and emits an order modify rejected event.
181    pub fn emit_order_modify_rejected(
182        &self,
183        order: &OrderAny,
184        venue_order_id: Option<VenueOrderId>,
185        reason: &str,
186        ts_event: UnixNanos,
187    ) {
188        let event = self.factory.generate_order_modify_rejected(
189            order,
190            venue_order_id,
191            reason,
192            ts_event,
193            self.ts_init(),
194        );
195        self.send_order_event(event);
196    }
197
198    /// Generates and emits an order cancel rejected event.
199    pub fn emit_order_cancel_rejected(
200        &self,
201        order: &OrderAny,
202        venue_order_id: Option<VenueOrderId>,
203        reason: &str,
204        ts_event: UnixNanos,
205    ) {
206        let event = self.factory.generate_order_cancel_rejected(
207            order,
208            venue_order_id,
209            reason,
210            ts_event,
211            self.ts_init(),
212        );
213        self.send_order_event(event);
214    }
215
216    /// Generates and emits an order updated event.
217    #[expect(clippy::too_many_arguments)]
218    pub fn emit_order_updated(
219        &self,
220        order: &OrderAny,
221        venue_order_id: VenueOrderId,
222        quantity: Quantity,
223        price: Option<Price>,
224        trigger_price: Option<Price>,
225        protection_price: Option<Price>,
226        ts_event: UnixNanos,
227    ) {
228        let event = self.factory.generate_order_updated(
229            order,
230            venue_order_id,
231            quantity,
232            price,
233            trigger_price,
234            protection_price,
235            ts_event,
236            self.ts_init(),
237        );
238        self.send_order_event(event);
239    }
240
241    /// Generates and emits an order canceled event.
242    pub fn emit_order_canceled(
243        &self,
244        order: &OrderAny,
245        venue_order_id: Option<VenueOrderId>,
246        ts_event: UnixNanos,
247    ) {
248        let event =
249            self.factory
250                .generate_order_canceled(order, venue_order_id, ts_event, self.ts_init());
251        self.send_order_event(event);
252    }
253
254    /// Generates and emits an order triggered event.
255    pub fn emit_order_triggered(
256        &self,
257        order: &OrderAny,
258        venue_order_id: Option<VenueOrderId>,
259        ts_event: UnixNanos,
260    ) {
261        let event =
262            self.factory
263                .generate_order_triggered(order, venue_order_id, ts_event, self.ts_init());
264        self.send_order_event(event);
265    }
266
267    /// Generates and emits an order expired event.
268    pub fn emit_order_expired(
269        &self,
270        order: &OrderAny,
271        venue_order_id: Option<VenueOrderId>,
272        ts_event: UnixNanos,
273    ) {
274        let event =
275            self.factory
276                .generate_order_expired(order, venue_order_id, ts_event, self.ts_init());
277        self.send_order_event(event);
278    }
279
280    /// Generates and emits an order filled event.
281    #[expect(clippy::too_many_arguments)]
282    pub fn emit_order_filled(
283        &self,
284        order: &OrderAny,
285        venue_order_id: VenueOrderId,
286        venue_position_id: Option<PositionId>,
287        trade_id: TradeId,
288        last_qty: Quantity,
289        last_px: Price,
290        quote_currency: Currency,
291        commission: Option<Money>,
292        liquidity_side: LiquiditySide,
293        ts_event: UnixNanos,
294    ) {
295        let event = self.factory.generate_order_filled(
296            order,
297            venue_order_id,
298            venue_position_id,
299            trade_id,
300            last_qty,
301            last_px,
302            quote_currency,
303            commission,
304            liquidity_side,
305            ts_event,
306            self.ts_init(),
307        );
308        self.send_order_event(event);
309    }
310
311    /// Constructs and emits an order rejected event from raw fields.
312    pub fn emit_order_rejected_event(
313        &self,
314        strategy_id: StrategyId,
315        instrument_id: InstrumentId,
316        client_order_id: ClientOrderId,
317        reason: &str,
318        ts_event: UnixNanos,
319        due_post_only: bool,
320    ) {
321        let event = OrderRejected::new(
322            self.factory.trader_id(),
323            strategy_id,
324            instrument_id,
325            client_order_id,
326            self.factory.account_id(),
327            reason.into(),
328            UUID4::new(),
329            ts_event,
330            self.ts_init(),
331            false,
332            due_post_only,
333        );
334        self.send_order_event(OrderEventAny::Rejected(event));
335    }
336
337    /// Constructs and emits an order modify rejected event from raw fields.
338    pub fn emit_order_modify_rejected_event(
339        &self,
340        strategy_id: StrategyId,
341        instrument_id: InstrumentId,
342        client_order_id: ClientOrderId,
343        venue_order_id: Option<VenueOrderId>,
344        reason: &str,
345        ts_event: UnixNanos,
346    ) {
347        let event = OrderModifyRejected::new(
348            self.factory.trader_id(),
349            strategy_id,
350            instrument_id,
351            client_order_id,
352            reason.into(),
353            UUID4::new(),
354            ts_event,
355            self.ts_init(),
356            false,
357            venue_order_id,
358            Some(self.factory.account_id()),
359        );
360        self.send_order_event(OrderEventAny::ModifyRejected(event));
361    }
362
363    /// Constructs and emits an order cancel rejected event from raw fields.
364    pub fn emit_order_cancel_rejected_event(
365        &self,
366        strategy_id: StrategyId,
367        instrument_id: InstrumentId,
368        client_order_id: ClientOrderId,
369        venue_order_id: Option<VenueOrderId>,
370        reason: &str,
371        ts_event: UnixNanos,
372    ) {
373        let event = OrderCancelRejected::new(
374            self.factory.trader_id(),
375            strategy_id,
376            instrument_id,
377            client_order_id,
378            reason.into(),
379            UUID4::new(),
380            ts_event,
381            self.ts_init(),
382            false,
383            venue_order_id,
384            Some(self.factory.account_id()),
385        );
386        self.send_order_event(OrderEventAny::CancelRejected(event));
387    }
388
389    /// Emits an order event.
390    pub fn send_order_event(&self, event: OrderEventAny) {
391        if let Some(sender) = &self.sender {
392            if let Err(e) = sender.send(ExecutionEvent::Order(event)) {
393                log::warn!("Failed to send order event: {e}");
394            }
395        } else {
396            log::warn!("Cannot send order event: sender not initialized");
397        }
398    }
399
400    /// Emits a batch of order submitted events as a single channel message.
401    pub fn send_order_submitted_batch(&self, batch: OrderSubmittedBatch) {
402        if let Some(sender) = &self.sender {
403            if let Err(e) = sender.send(ExecutionEvent::OrderSubmittedBatch(batch)) {
404                log::warn!("Failed to send order submitted batch: {e}");
405            }
406        } else {
407            log::warn!("Cannot send order submitted batch: sender not initialized");
408        }
409    }
410
411    /// Emits a batch of order accepted events as a single channel message.
412    pub fn send_order_accepted_batch(&self, batch: OrderAcceptedBatch) {
413        if let Some(sender) = &self.sender {
414            if let Err(e) = sender.send(ExecutionEvent::OrderAcceptedBatch(batch)) {
415                log::warn!("Failed to send order accepted batch: {e}");
416            }
417        } else {
418            log::warn!("Cannot send order accepted batch: sender not initialized");
419        }
420    }
421
422    /// Emits a batch of order canceled events as a single channel message.
423    pub fn send_order_canceled_batch(&self, batch: OrderCanceledBatch) {
424        if let Some(sender) = &self.sender {
425            if let Err(e) = sender.send(ExecutionEvent::OrderCanceledBatch(batch)) {
426                log::warn!("Failed to send order canceled batch: {e}");
427            }
428        } else {
429            log::warn!("Cannot send order canceled batch: sender not initialized");
430        }
431    }
432
433    /// Emits an account state event.
434    pub fn send_account_state(&self, state: AccountState) {
435        if let Some(sender) = &self.sender {
436            if let Err(e) = sender.send(ExecutionEvent::Account(state)) {
437                log::warn!("Failed to send account state: {e}");
438            }
439        } else {
440            log::warn!("Cannot send account state: sender not initialized");
441        }
442    }
443
444    /// Emits an execution report.
445    pub fn send_execution_report(&self, report: ExecutionReport) {
446        if let Some(sender) = &self.sender {
447            if let Err(e) = sender.send(ExecutionEvent::Report(report)) {
448                log::warn!("Failed to send execution report: {e}");
449            }
450        } else {
451            log::warn!("Cannot send execution report: sender not initialized");
452        }
453    }
454
455    /// Emits an order status report.
456    pub fn send_order_status_report(&self, report: OrderStatusReport) {
457        self.send_execution_report(ExecutionReport::Order(Box::new(report)));
458    }
459
460    /// Emits a fill report.
461    pub fn send_fill_report(&self, report: FillReport) {
462        self.send_execution_report(ExecutionReport::Fill(Box::new(report)));
463    }
464
465    /// Emits an order status report bundled with the fills that produced it.
466    pub fn send_order_with_fills(&self, report: OrderStatusReport, fills: Vec<FillReport>) {
467        self.send_execution_report(ExecutionReport::OrderWithFills(Box::new(report), fills));
468    }
469
470    /// Emits a position status report.
471    pub fn send_position_report(&self, report: PositionStatusReport) {
472        self.send_execution_report(ExecutionReport::Position(Box::new(report)));
473    }
474}
nautilus_live/execution/emitter.rs

nautilus_live/execution/
emitter.rs
