nautilus_risk/engine/
mod.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 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//! Risk management engine implementation.
17
18pub mod config;
19
20#[cfg(test)]
21mod tests;
22
23use std::{cell::RefCell, collections::HashMap, fmt::Debug, rc::Rc};
24
25use config::RiskEngineConfig;
26use nautilus_common::{
27    cache::Cache,
28    clock::Clock,
29    logging::{CMD, EVT, RECV},
30    messages::execution::{ModifyOrder, SubmitOrder, SubmitOrderList, TradingCommand},
31    msgbus,
32    throttler::Throttler,
33};
34use nautilus_core::UUID4;
35use nautilus_model::{
36    accounts::{Account, AccountAny},
37    enums::{InstrumentClass, OrderSide, OrderStatus, TimeInForce, TradingState},
38    events::{OrderDenied, OrderEventAny, OrderModifyRejected},
39    identifiers::InstrumentId,
40    instruments::{Instrument, InstrumentAny},
41    orders::{Order, OrderAny, OrderList},
42    types::{Currency, Money, Price, Quantity},
43};
44use nautilus_portfolio::Portfolio;
45use rust_decimal::{Decimal, prelude::ToPrimitive};
46use ustr::Ustr;
47
48type SubmitOrderFn = Box<dyn Fn(SubmitOrder)>;
49type ModifyOrderFn = Box<dyn Fn(ModifyOrder)>;
50
51/// Central risk management engine that validates and controls trading operations.
52///
53/// The `RiskEngine` provides comprehensive pre-trade risk checks including order validation,
54/// balance verification, position sizing limits, and trading state management. It acts as
55/// a gateway between strategy orders and execution, ensuring all trades comply with
56/// defined risk parameters and regulatory constraints.
57#[allow(dead_code)]
58pub struct RiskEngine {
59    clock: Rc<RefCell<dyn Clock>>,
60    cache: Rc<RefCell<Cache>>,
61    portfolio: Portfolio,
62    pub throttled_submit_order: Throttler<SubmitOrder, SubmitOrderFn>,
63    pub throttled_modify_order: Throttler<ModifyOrder, ModifyOrderFn>,
64    max_notional_per_order: HashMap<InstrumentId, Decimal>,
65    trading_state: TradingState,
66    config: RiskEngineConfig,
67}
68
69impl Debug for RiskEngine {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        f.debug_struct(stringify!(RiskEngine)).finish()
72    }
73}
74
75impl RiskEngine {
76    /// Creates a new [`RiskEngine`] instance.
77    pub fn new(
78        config: RiskEngineConfig,
79        portfolio: Portfolio,
80        clock: Rc<RefCell<dyn Clock>>,
81        cache: Rc<RefCell<Cache>>,
82    ) -> Self {
83        let throttled_submit_order =
84            Self::create_submit_order_throttler(&config, clock.clone(), cache.clone());
85
86        let throttled_modify_order =
87            Self::create_modify_order_throttler(&config, clock.clone(), cache.clone());
88
89        Self {
90            clock,
91            cache,
92            portfolio,
93            throttled_submit_order,
94            throttled_modify_order,
95            max_notional_per_order: HashMap::new(),
96            trading_state: TradingState::Active,
97            config,
98        }
99    }
100
101    fn create_submit_order_throttler(
102        config: &RiskEngineConfig,
103        clock: Rc<RefCell<dyn Clock>>,
104        cache: Rc<RefCell<Cache>>,
105    ) -> Throttler<SubmitOrder, SubmitOrderFn> {
106        let success_handler = {
107            Box::new(move |submit_order: SubmitOrder| {
108                msgbus::send_any(
109                    "ExecEngine.execute".into(),
110                    &TradingCommand::SubmitOrder(submit_order),
111                );
112            }) as Box<dyn Fn(SubmitOrder)>
113        };
114
115        let failure_handler = {
116            let cache = cache;
117            let clock = clock.clone();
118            Box::new(move |submit_order: SubmitOrder| {
119                let reason = "REJECTED BY THROTTLER";
120                log::warn!(
121                    "SubmitOrder for {} DENIED: {}",
122                    submit_order.client_order_id,
123                    reason
124                );
125
126                Self::handle_submit_order_cache(&cache, &submit_order);
127
128                let denied = Self::create_order_denied(&submit_order, reason, &clock);
129
130                msgbus::send_any("ExecEngine.process".into(), &denied);
131            }) as Box<dyn Fn(SubmitOrder)>
132        };
133
134        Throttler::new(
135            config.max_order_submit.limit,
136            config.max_order_submit.interval_ns,
137            clock,
138            "ORDER_SUBMIT_THROTTLER".to_string(),
139            success_handler,
140            Some(failure_handler),
141            Ustr::from(&UUID4::new().to_string()),
142        )
143    }
144
145    fn create_modify_order_throttler(
146        config: &RiskEngineConfig,
147        clock: Rc<RefCell<dyn Clock>>,
148        cache: Rc<RefCell<Cache>>,
149    ) -> Throttler<ModifyOrder, ModifyOrderFn> {
150        let success_handler = {
151            Box::new(move |order: ModifyOrder| {
152                msgbus::send_any(
153                    "ExecEngine.execute".into(),
154                    &TradingCommand::ModifyOrder(order),
155                );
156            }) as Box<dyn Fn(ModifyOrder)>
157        };
158
159        let failure_handler = {
160            let cache = cache;
161            let clock = clock.clone();
162            Box::new(move |order: ModifyOrder| {
163                let reason = "Exceeded MAX_ORDER_MODIFY_RATE";
164                log::warn!(
165                    "SubmitOrder for {} DENIED: {}",
166                    order.client_order_id,
167                    reason
168                );
169
170                let order = match Self::get_existing_order(&cache, &order) {
171                    Some(order) => order,
172                    None => return,
173                };
174
175                let rejected = Self::create_modify_rejected(&order, reason, &clock);
176
177                msgbus::send_any("ExecEngine.process".into(), &rejected);
178            }) as Box<dyn Fn(ModifyOrder)>
179        };
180
181        Throttler::new(
182            config.max_order_modify.limit,
183            config.max_order_modify.interval_ns,
184            clock,
185            "ORDER_MODIFY_THROTTLER".to_string(),
186            success_handler,
187            Some(failure_handler),
188            Ustr::from(&UUID4::new().to_string()),
189        )
190    }
191
192    fn handle_submit_order_cache(cache: &Rc<RefCell<Cache>>, submit_order: &SubmitOrder) {
193        let mut cache = cache.borrow_mut();
194        if !cache.order_exists(&submit_order.client_order_id) {
195            cache
196                .add_order(submit_order.order.clone(), None, None, false)
197                .map_err(|e| {
198                    log::error!("Cannot add order to cache: {e}");
199                })
200                .unwrap();
201        }
202    }
203
204    fn get_existing_order(cache: &Rc<RefCell<Cache>>, order: &ModifyOrder) -> Option<OrderAny> {
205        let cache = cache.borrow();
206        if let Some(order) = cache.order(&order.client_order_id) {
207            Some(order.clone())
208        } else {
209            log::error!(
210                "Order with command.client_order_id: {} not found",
211                order.client_order_id
212            );
213            None
214        }
215    }
216
217    fn create_order_denied(
218        submit_order: &SubmitOrder,
219        reason: &str,
220        clock: &Rc<RefCell<dyn Clock>>,
221    ) -> OrderEventAny {
222        let timestamp = clock.borrow().timestamp_ns();
223        OrderEventAny::Denied(OrderDenied::new(
224            submit_order.trader_id,
225            submit_order.strategy_id,
226            submit_order.instrument_id,
227            submit_order.client_order_id,
228            reason.into(),
229            UUID4::new(),
230            timestamp,
231            timestamp,
232        ))
233    }
234
235    fn create_modify_rejected(
236        order: &OrderAny,
237        reason: &str,
238        clock: &Rc<RefCell<dyn Clock>>,
239    ) -> OrderEventAny {
240        let timestamp = clock.borrow().timestamp_ns();
241        OrderEventAny::ModifyRejected(OrderModifyRejected::new(
242            order.trader_id(),
243            order.strategy_id(),
244            order.instrument_id(),
245            order.client_order_id(),
246            reason.into(),
247            UUID4::new(),
248            timestamp,
249            timestamp,
250            false,
251            order.venue_order_id(),
252            None,
253        ))
254    }
255
256    // -- COMMANDS --------------------------------------------------------------------------------
257
258    /// Executes a trading command through the risk management pipeline.
259    pub fn execute(&mut self, command: TradingCommand) {
260        // This will extend to other commands such as `RiskCommand`
261        self.handle_command(command);
262    }
263
264    /// Processes an order event for risk monitoring and state updates.
265    pub fn process(&mut self, event: OrderEventAny) {
266        // This will extend to other events such as `RiskEvent`
267        self.handle_event(event);
268    }
269
270    /// Sets the trading state for risk control enforcement.
271    pub fn set_trading_state(&mut self, state: TradingState) {
272        if state == self.trading_state {
273            log::warn!("No change to trading state: already set to {state:?}");
274            return;
275        }
276
277        self.trading_state = state;
278
279        let _ts_now = self.clock.borrow().timestamp_ns();
280
281        // TODO: Create a new Event "TradingStateChanged" in OrderEventAny enum.
282        // let event = OrderEventAny::TradingStateChanged(TradingStateChanged::new(..,self.trading_state,..));
283
284        msgbus::publish("events.risk".into(), &"message"); // TODO: Send the new Event here
285
286        log::info!("Trading state set to {state:?}");
287    }
288
289    /// Sets the maximum notional value per order for the specified instrument.
290    pub fn set_max_notional_per_order(&mut self, instrument_id: InstrumentId, new_value: Decimal) {
291        self.max_notional_per_order.insert(instrument_id, new_value);
292
293        let new_value_str = new_value.to_string();
294        log::info!("Set MAX_NOTIONAL_PER_ORDER: {instrument_id} {new_value_str}");
295    }
296
297    // -- COMMAND HANDLERS ------------------------------------------------------------------------
298
299    // Renamed from `execute_command`
300    fn handle_command(&mut self, command: TradingCommand) {
301        if self.config.debug {
302            log::debug!("{CMD}{RECV} {command:?}");
303        }
304
305        match command {
306            TradingCommand::SubmitOrder(submit_order) => self.handle_submit_order(submit_order),
307            TradingCommand::SubmitOrderList(submit_order_list) => {
308                self.handle_submit_order_list(submit_order_list);
309            }
310            TradingCommand::ModifyOrder(modify_order) => self.handle_modify_order(modify_order),
311            _ => {
312                log::error!("Cannot handle command: {command}");
313            }
314        }
315    }
316
317    fn handle_submit_order(&self, command: SubmitOrder) {
318        if self.config.bypass {
319            self.send_to_execution(TradingCommand::SubmitOrder(command));
320            return;
321        }
322
323        let order = &command.order;
324        if let Some(position_id) = command.position_id
325            && order.is_reduce_only()
326        {
327            let position_exists = {
328                let cache = self.cache.borrow();
329                cache
330                    .position(&position_id)
331                    .map(|pos| (pos.side, pos.quantity))
332            };
333
334            if let Some((pos_side, pos_quantity)) = position_exists {
335                if !order.would_reduce_only(pos_side, pos_quantity) {
336                    self.deny_command(
337                        TradingCommand::SubmitOrder(command),
338                        &format!("Reduce only order would increase position {position_id}"),
339                    );
340                    return; // Denied
341                }
342            } else {
343                self.deny_command(
344                    TradingCommand::SubmitOrder(command),
345                    &format!("Position {position_id} not found for reduce-only order"),
346                );
347                return;
348            }
349        }
350
351        let instrument_exists = {
352            let cache = self.cache.borrow();
353            cache.instrument(&order.instrument_id()).cloned()
354        };
355
356        let instrument = if let Some(instrument) = instrument_exists {
357            instrument
358        } else {
359            self.deny_command(
360                TradingCommand::SubmitOrder(command.clone()),
361                &format!("Instrument for {} not found", command.instrument_id),
362            );
363            return; // Denied
364        };
365
366        ////////////////////////////////////////////////////////////////////////////////
367        // PRE-TRADE ORDER(S) CHECKS
368        ////////////////////////////////////////////////////////////////////////////////
369        if !self.check_order(instrument.clone(), order.clone()) {
370            return; // Denied
371        }
372
373        if !self.check_orders_risk(instrument.clone(), Vec::from([order.clone()])) {
374            return; // Denied
375        }
376
377        self.execution_gateway(instrument, TradingCommand::SubmitOrder(command.clone()));
378    }
379
380    fn handle_submit_order_list(&self, command: SubmitOrderList) {
381        if self.config.bypass {
382            self.send_to_execution(TradingCommand::SubmitOrderList(command));
383            return;
384        }
385
386        let instrument_exists = {
387            let cache = self.cache.borrow();
388            cache.instrument(&command.instrument_id).cloned()
389        };
390
391        let instrument = if let Some(instrument) = instrument_exists {
392            instrument
393        } else {
394            self.deny_command(
395                TradingCommand::SubmitOrderList(command.clone()),
396                &format!("no instrument found for {}", command.instrument_id),
397            );
398            return; // Denied
399        };
400
401        ////////////////////////////////////////////////////////////////////////////////
402        // PRE-TRADE ORDER(S) CHECKS
403        ////////////////////////////////////////////////////////////////////////////////
404        for order in command.order_list.orders.clone() {
405            if !self.check_order(instrument.clone(), order) {
406                return; // Denied
407            }
408        }
409
410        if !self.check_orders_risk(instrument.clone(), command.order_list.clone().orders) {
411            self.deny_order_list(
412                command.order_list.clone(),
413                &format!("OrderList {} DENIED", command.order_list.id),
414            );
415            return; // Denied
416        }
417
418        self.execution_gateway(instrument, TradingCommand::SubmitOrderList(command));
419    }
420
421    fn handle_modify_order(&self, command: ModifyOrder) {
422        ////////////////////////////////////////////////////////////////////////////////
423        // VALIDATE COMMAND
424        ////////////////////////////////////////////////////////////////////////////////
425        let order_exists = {
426            let cache = self.cache.borrow();
427            cache.order(&command.client_order_id).cloned()
428        };
429
430        let order = if let Some(order) = order_exists {
431            order
432        } else {
433            log::error!(
434                "ModifyOrder DENIED: Order with command.client_order_id: {} not found",
435                command.client_order_id
436            );
437            return;
438        };
439
440        if order.is_closed() {
441            self.reject_modify_order(
442                order,
443                &format!(
444                    "Order with command.client_order_id: {} already closed",
445                    command.client_order_id
446                ),
447            );
448            return;
449        } else if order.status() == OrderStatus::PendingCancel {
450            self.reject_modify_order(
451                order,
452                &format!(
453                    "Order with command.client_order_id: {} is already pending cancel",
454                    command.client_order_id
455                ),
456            );
457            return;
458        }
459
460        // Get instrument for orders
461        let maybe_instrument = {
462            let cache = self.cache.borrow();
463            cache.instrument(&command.instrument_id).cloned()
464        };
465
466        let instrument = if let Some(instrument) = maybe_instrument {
467            instrument
468        } else {
469            self.reject_modify_order(
470                order,
471                &format!("no instrument found for {}", command.instrument_id),
472            );
473            return; // Denied
474        };
475
476        // Check Price
477        let mut risk_msg = self.check_price(&instrument, command.price);
478        if let Some(risk_msg) = risk_msg {
479            self.reject_modify_order(order, &risk_msg);
480            return; // Denied
481        }
482
483        // Check Trigger
484        risk_msg = self.check_price(&instrument, command.trigger_price);
485        if let Some(risk_msg) = risk_msg {
486            self.reject_modify_order(order, &risk_msg);
487            return; // Denied
488        }
489
490        // Check Quantity
491        risk_msg = self.check_quantity(&instrument, command.quantity);
492        if let Some(risk_msg) = risk_msg {
493            self.reject_modify_order(order, &risk_msg);
494            return; // Denied
495        }
496
497        // Check TradingState
498        match self.trading_state {
499            TradingState::Halted => {
500                self.reject_modify_order(order, "TradingState is HALTED: Cannot modify order");
501            }
502            TradingState::Reducing => {
503                if let Some(quantity) = command.quantity
504                    && quantity > order.quantity()
505                    && ((order.is_buy() && self.portfolio.is_net_long(&instrument.id()))
506                        || (order.is_sell() && self.portfolio.is_net_short(&instrument.id())))
507                {
508                    self.reject_modify_order(
509                        order,
510                        &format!(
511                            "TradingState is REDUCING and update will increase exposure {}",
512                            instrument.id()
513                        ),
514                    );
515                }
516            }
517            _ => {}
518        }
519
520        // TODO: Fix message bus usage
521        // self.throttled_modify_order.send(command);
522    }
523
524    // -- PRE-TRADE CHECKS ------------------------------------------------------------------------
525
526    fn check_order(&self, instrument: InstrumentAny, order: OrderAny) -> bool {
527        ////////////////////////////////////////////////////////////////////////////////
528        // VALIDATION CHECKS
529        ////////////////////////////////////////////////////////////////////////////////
530        if order.time_in_force() == TimeInForce::Gtd {
531            // SAFETY: GTD guarantees an expire time
532            let expire_time = order.expire_time().unwrap();
533            if expire_time <= self.clock.borrow().timestamp_ns() {
534                self.deny_order(
535                    order,
536                    &format!("GTD {} already past", expire_time.to_rfc3339()),
537                );
538                return false; // Denied
539            }
540        }
541
542        if !self.check_order_price(instrument.clone(), order.clone())
543            || !self.check_order_quantity(instrument, order)
544        {
545            return false; // Denied
546        }
547
548        true
549    }
550
551    fn check_order_price(&self, instrument: InstrumentAny, order: OrderAny) -> bool {
552        ////////////////////////////////////////////////////////////////////////////////
553        // CHECK PRICE
554        ////////////////////////////////////////////////////////////////////////////////
555        if order.price().is_some() {
556            let risk_msg = self.check_price(&instrument, order.price());
557            if let Some(risk_msg) = risk_msg {
558                self.deny_order(order, &risk_msg);
559                return false; // Denied
560            }
561        }
562
563        ////////////////////////////////////////////////////////////////////////////////
564        // CHECK TRIGGER
565        ////////////////////////////////////////////////////////////////////////////////
566        if order.trigger_price().is_some() {
567            let risk_msg = self.check_price(&instrument, order.trigger_price());
568            if let Some(risk_msg) = risk_msg {
569                self.deny_order(order, &risk_msg);
570                return false; // Denied
571            }
572        }
573
574        true
575    }
576
577    fn check_order_quantity(&self, instrument: InstrumentAny, order: OrderAny) -> bool {
578        let risk_msg = self.check_quantity(&instrument, Some(order.quantity()));
579        if let Some(risk_msg) = risk_msg {
580            self.deny_order(order, &risk_msg);
581            return false; // Denied
582        }
583
584        true
585    }
586
587    fn check_orders_risk(&self, instrument: InstrumentAny, orders: Vec<OrderAny>) -> bool {
588        ////////////////////////////////////////////////////////////////////////////////
589        // CHECK TRIGGER
590        ////////////////////////////////////////////////////////////////////////////////
591        let mut last_px: Option<Price> = None;
592        let mut max_notional: Option<Money> = None;
593
594        // Determine max notional
595        let max_notional_setting = self.max_notional_per_order.get(&instrument.id());
596        if let Some(max_notional_setting_val) = max_notional_setting.copied() {
597            max_notional = Some(Money::new(
598                max_notional_setting_val
599                    .to_f64()
600                    .expect("Invalid decimal conversion"),
601                instrument.quote_currency(),
602            ));
603        }
604
605        // Get account for risk checks
606        let account_exists = {
607            let cache = self.cache.borrow();
608            cache.account_for_venue(&instrument.id().venue).cloned()
609        };
610
611        let account = if let Some(account) = account_exists {
612            account
613        } else {
614            log::debug!("Cannot find account for venue {}", instrument.id().venue);
615            return true; // TODO: Temporary early return until handling routing/multiple venues
616        };
617        let cash_account = match account {
618            AccountAny::Cash(cash_account) => cash_account,
619            AccountAny::Margin(_) => return true, // TODO: Determine risk controls for margin
620        };
621        let free = cash_account.balance_free(Some(instrument.quote_currency()));
622        if self.config.debug {
623            log::debug!("Free cash: {free:?}");
624        }
625
626        let mut cum_notional_buy: Option<Money> = None;
627        let mut cum_notional_sell: Option<Money> = None;
628        let mut base_currency: Option<Currency> = None;
629        for order in &orders {
630            // Determine last price based on order type
631            last_px = match order {
632                OrderAny::Market(_) | OrderAny::MarketToLimit(_) => {
633                    if last_px.is_none() {
634                        let cache = self.cache.borrow();
635                        if let Some(last_quote) = cache.quote(&instrument.id()) {
636                            match order.order_side() {
637                                OrderSide::Buy => Some(last_quote.ask_price),
638                                OrderSide::Sell => Some(last_quote.bid_price),
639                                _ => panic!("Invalid order side"),
640                            }
641                        } else {
642                            let cache = self.cache.borrow();
643                            let last_trade = cache.trade(&instrument.id());
644
645                            if let Some(last_trade) = last_trade {
646                                Some(last_trade.price)
647                            } else {
648                                log::warn!(
649                                    "Cannot check MARKET order risk: no prices for {}",
650                                    instrument.id()
651                                );
652                                continue;
653                            }
654                        }
655                    } else {
656                        last_px
657                    }
658                }
659                OrderAny::StopMarket(_) | OrderAny::MarketIfTouched(_) => order.trigger_price(),
660                OrderAny::TrailingStopMarket(_) | OrderAny::TrailingStopLimit(_) => {
661                    if let Some(trigger_price) = order.trigger_price() {
662                        Some(trigger_price)
663                    } else {
664                        log::warn!(
665                            "Cannot check {} order risk: no trigger price was set", // TODO: Use last_trade += offset
666                            order.order_type()
667                        );
668                        continue;
669                    }
670                }
671                _ => order.price(),
672            };
673
674            let last_px = if let Some(px) = last_px {
675                px
676            } else {
677                log::error!("Cannot check order risk: no price available");
678                continue;
679            };
680
681            let notional =
682                instrument.calculate_notional_value(order.quantity(), last_px, Some(true));
683
684            if self.config.debug {
685                log::debug!("Notional: {notional:?}");
686            }
687
688            // Check MAX notional per order limit
689            if let Some(max_notional_value) = max_notional
690                && notional > max_notional_value
691            {
692                self.deny_order(
693                        order.clone(),
694                        &format!(
695                            "NOTIONAL_EXCEEDS_MAX_PER_ORDER: max_notional={max_notional_value:?}, notional={notional:?}"
696                        ),
697                    );
698                return false; // Denied
699            }
700
701            // Check MIN notional instrument limit
702            if let Some(min_notional) = instrument.min_notional()
703                && notional.currency == min_notional.currency
704                && notional < min_notional
705            {
706                self.deny_order(
707                        order.clone(),
708                        &format!(
709                            "NOTIONAL_LESS_THAN_MIN_FOR_INSTRUMENT: min_notional={min_notional:?}, notional={notional:?}"
710                        ),
711                    );
712                return false; // Denied
713            }
714
715            // // Check MAX notional instrument limit
716            if let Some(max_notional) = instrument.max_notional()
717                && notional.currency == max_notional.currency
718                && notional > max_notional
719            {
720                self.deny_order(
721                        order.clone(),
722                        &format!(
723                            "NOTIONAL_GREATER_THAN_MAX_FOR_INSTRUMENT: max_notional={max_notional:?}, notional={notional:?}"
724                        ),
725                    );
726                return false; // Denied
727            }
728
729            // Calculate OrderBalanceImpact (valid for CashAccount only)
730            let notional = instrument.calculate_notional_value(order.quantity(), last_px, None);
731            let order_balance_impact = match order.order_side() {
732                OrderSide::Buy => Money::from_raw(-notional.raw, notional.currency),
733                OrderSide::Sell => Money::from_raw(notional.raw, notional.currency),
734                OrderSide::NoOrderSide => {
735                    panic!("invalid `OrderSide`, was {}", order.order_side());
736                }
737            };
738
739            if self.config.debug {
740                log::debug!("Balance impact: {order_balance_impact}");
741            }
742
743            if let Some(free_val) = free
744                && (free_val.as_decimal() + order_balance_impact.as_decimal()) < Decimal::ZERO
745            {
746                self.deny_order(
747                    order.clone(),
748                    &format!(
749                        "NOTIONAL_EXCEEDS_FREE_BALANCE: free={free_val:?}, notional={notional:?}"
750                    ),
751                );
752                return false;
753            }
754
755            if base_currency.is_none() {
756                base_currency = instrument.base_currency();
757            }
758            if order.is_buy() {
759                match cum_notional_buy.as_mut() {
760                    Some(cum_notional_buy_val) => {
761                        cum_notional_buy_val.raw += -order_balance_impact.raw;
762                    }
763                    None => {
764                        cum_notional_buy = Some(Money::from_raw(
765                            -order_balance_impact.raw,
766                            order_balance_impact.currency,
767                        ));
768                    }
769                }
770
771                if self.config.debug {
772                    log::debug!("Cumulative notional BUY: {cum_notional_buy:?}");
773                }
774
775                if let (Some(free), Some(cum_notional_buy)) = (free, cum_notional_buy)
776                    && cum_notional_buy > free
777                {
778                    self.deny_order(order.clone(), &format!("CUM_NOTIONAL_EXCEEDS_FREE_BALANCE: free={free}, cum_notional={cum_notional_buy}"));
779                    return false; // Denied
780                }
781            } else if order.is_sell() {
782                if cash_account.base_currency.is_some() {
783                    match cum_notional_sell.as_mut() {
784                        Some(cum_notional_buy_val) => {
785                            cum_notional_buy_val.raw += order_balance_impact.raw;
786                        }
787                        None => {
788                            cum_notional_sell = Some(Money::from_raw(
789                                order_balance_impact.raw,
790                                order_balance_impact.currency,
791                            ));
792                        }
793                    }
794                    if self.config.debug {
795                        log::debug!("Cumulative notional SELL: {cum_notional_sell:?}");
796                    }
797
798                    if let (Some(free), Some(cum_notional_sell)) = (free, cum_notional_sell)
799                        && cum_notional_sell > free
800                    {
801                        self.deny_order(order.clone(), &format!("CUM_NOTIONAL_EXCEEDS_FREE_BALANCE: free={free}, cum_notional={cum_notional_sell}"));
802                        return false; // Denied
803                    }
804                }
805                // Account is already of type Cash, so no check
806                else if let Some(base_currency) = base_currency {
807                    let cash_value = Money::from_raw(
808                        order
809                            .quantity()
810                            .raw
811                            .try_into()
812                            .map_err(|e| log::error!("Unable to convert Quantity to f64: {e}"))
813                            .unwrap(),
814                        base_currency,
815                    );
816
817                    if self.config.debug {
818                        log::debug!("Cash value: {cash_value:?}");
819                        log::debug!(
820                            "Total: {:?}",
821                            cash_account.balance_total(Some(base_currency))
822                        );
823                        log::debug!(
824                            "Locked: {:?}",
825                            cash_account.balance_locked(Some(base_currency))
826                        );
827                        log::debug!("Free: {:?}", cash_account.balance_free(Some(base_currency)));
828                    }
829
830                    match cum_notional_sell {
831                        Some(mut cum_notional_sell) => {
832                            cum_notional_sell.raw += cash_value.raw;
833                        }
834                        None => cum_notional_sell = Some(cash_value),
835                    }
836
837                    if self.config.debug {
838                        log::debug!("Cumulative notional SELL: {cum_notional_sell:?}");
839                    }
840                    if let (Some(free), Some(cum_notional_sell)) = (free, cum_notional_sell)
841                        && cum_notional_sell.raw > free.raw
842                    {
843                        self.deny_order(order.clone(), &format!("CUM_NOTIONAL_EXCEEDS_FREE_BALANCE: free={free}, cum_notional={cum_notional_sell}"));
844                        return false; // Denied
845                    }
846                }
847            }
848        }
849
850        // Finally
851        true // Passed
852    }
853
854    fn check_price(&self, instrument: &InstrumentAny, price: Option<Price>) -> Option<String> {
855        let price_val = price?;
856
857        if price_val.precision > instrument.price_precision() {
858            return Some(format!(
859                "price {} invalid (precision {} > {})",
860                price_val,
861                price_val.precision,
862                instrument.price_precision()
863            ));
864        }
865
866        if instrument.instrument_class() != InstrumentClass::Option && price_val.raw <= 0 {
867            return Some(format!("price {price_val} invalid (<= 0)"));
868        }
869
870        None
871    }
872
873    fn check_quantity(
874        &self,
875        instrument: &InstrumentAny,
876        quantity: Option<Quantity>,
877    ) -> Option<String> {
878        let quantity_val = quantity?;
879
880        // Check precision
881        if quantity_val.precision > instrument.size_precision() {
882            return Some(format!(
883                "quantity {} invalid (precision {} > {})",
884                quantity_val,
885                quantity_val.precision,
886                instrument.size_precision()
887            ));
888        }
889
890        // Check maximum quantity
891        if let Some(max_quantity) = instrument.max_quantity()
892            && quantity_val > max_quantity
893        {
894            return Some(format!(
895                "quantity {quantity_val} invalid (> maximum trade size of {max_quantity})"
896            ));
897        }
898
899        // // Check minimum quantity
900        if let Some(min_quantity) = instrument.min_quantity()
901            && quantity_val < min_quantity
902        {
903            return Some(format!(
904                "quantity {quantity_val} invalid (< minimum trade size of {min_quantity})"
905            ));
906        }
907
908        None
909    }
910
911    // -- DENIALS ---------------------------------------------------------------------------------
912
913    fn deny_command(&self, command: TradingCommand, reason: &str) {
914        match command {
915            TradingCommand::SubmitOrder(submit_order) => {
916                self.deny_order(submit_order.order, reason);
917            }
918            TradingCommand::SubmitOrderList(submit_order_list) => {
919                self.deny_order_list(submit_order_list.order_list, reason);
920            }
921            _ => {
922                panic!("Cannot deny command {command}");
923            }
924        }
925    }
926
927    fn deny_order(&self, order: OrderAny, reason: &str) {
928        log::warn!(
929            "SubmitOrder for {} DENIED: {}",
930            order.client_order_id(),
931            reason
932        );
933
934        if order.status() != OrderStatus::Initialized {
935            return;
936        }
937
938        let mut cache = self.cache.borrow_mut();
939        if !cache.order_exists(&order.client_order_id()) {
940            cache
941                .add_order(order.clone(), None, None, false)
942                .map_err(|e| {
943                    log::error!("Cannot add order to cache: {e}");
944                })
945                .unwrap();
946        }
947
948        let denied = OrderEventAny::Denied(OrderDenied::new(
949            order.trader_id(),
950            order.strategy_id(),
951            order.instrument_id(),
952            order.client_order_id(),
953            reason.into(),
954            UUID4::new(),
955            self.clock.borrow().timestamp_ns(),
956            self.clock.borrow().timestamp_ns(),
957        ));
958
959        msgbus::send_any("ExecEngine.process".into(), &denied);
960    }
961
962    fn deny_order_list(&self, order_list: OrderList, reason: &str) {
963        for order in order_list.orders {
964            if !order.is_closed() {
965                self.deny_order(order, reason);
966            }
967        }
968    }
969
970    fn reject_modify_order(&self, order: OrderAny, reason: &str) {
971        let ts_event = self.clock.borrow().timestamp_ns();
972        let denied = OrderEventAny::ModifyRejected(OrderModifyRejected::new(
973            order.trader_id(),
974            order.strategy_id(),
975            order.instrument_id(),
976            order.client_order_id(),
977            reason.into(),
978            UUID4::new(),
979            ts_event,
980            ts_event,
981            false,
982            order.venue_order_id(),
983            order.account_id(),
984        ));
985
986        msgbus::send_any("ExecEngine.process".into(), &denied);
987    }
988
989    // -- EGRESS ----------------------------------------------------------------------------------
990
991    fn execution_gateway(&self, instrument: InstrumentAny, command: TradingCommand) {
992        match self.trading_state {
993            TradingState::Halted => match command {
994                TradingCommand::SubmitOrder(submit_order) => {
995                    self.deny_order(submit_order.order, "TradingState::HALTED");
996                }
997                TradingCommand::SubmitOrderList(submit_order_list) => {
998                    self.deny_order_list(submit_order_list.order_list, "TradingState::HALTED");
999                }
1000                _ => {}
1001            },
1002            TradingState::Reducing => match command {
1003                TradingCommand::SubmitOrder(submit_order) => {
1004                    let order = submit_order.order;
1005                    if order.is_buy() && self.portfolio.is_net_long(&instrument.id()) {
1006                        self.deny_order(
1007                            order,
1008                            &format!(
1009                                "BUY when TradingState::REDUCING and LONG {}",
1010                                instrument.id()
1011                            ),
1012                        );
1013                    } else if order.is_sell() && self.portfolio.is_net_short(&instrument.id()) {
1014                        self.deny_order(
1015                            order,
1016                            &format!(
1017                                "SELL when TradingState::REDUCING and SHORT {}",
1018                                instrument.id()
1019                            ),
1020                        );
1021                    }
1022                }
1023                TradingCommand::SubmitOrderList(submit_order_list) => {
1024                    let order_list = submit_order_list.order_list;
1025                    for order in &order_list.orders {
1026                        if order.is_buy() && self.portfolio.is_net_long(&instrument.id()) {
1027                            self.deny_order_list(
1028                                order_list,
1029                                &format!(
1030                                    "BUY when TradingState::REDUCING and LONG {}",
1031                                    instrument.id()
1032                                ),
1033                            );
1034                            return;
1035                        } else if order.is_sell() && self.portfolio.is_net_short(&instrument.id()) {
1036                            self.deny_order_list(
1037                                order_list,
1038                                &format!(
1039                                    "SELL when TradingState::REDUCING and SHORT {}",
1040                                    instrument.id()
1041                                ),
1042                            );
1043                            return;
1044                        }
1045                    }
1046                }
1047                _ => {}
1048            },
1049            TradingState::Active => match command {
1050                TradingCommand::SubmitOrder(_submit_order) => {
1051                    // TODO: Fix message bus usage
1052                    // self.throttled_submit_order.send(submit_order);
1053                }
1054                TradingCommand::SubmitOrderList(_submit_order_list) => {
1055                    todo!("NOT IMPLEMENTED");
1056                }
1057                _ => {}
1058            },
1059        }
1060    }
1061
1062    fn send_to_execution(&self, command: TradingCommand) {
1063        msgbus::send_any("ExecEngine.execute".into(), &command);
1064    }
1065
1066    fn handle_event(&mut self, event: OrderEventAny) {
1067        // We intend to extend the risk engine to be able to handle additional events.
1068        // For now we just log.
1069        if self.config.debug {
1070            log::debug!("{RECV}{EVT} {event:?}");
1071        }
1072    }
1073}