nautilus_trading/algorithm/
twap.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//! Time-Weighted Average Price (TWAP) execution algorithm.
17//!
18//! The TWAP algorithm executes orders by evenly spreading them over a specified
19//! time horizon at regular intervals. This helps reduce market impact by avoiding
20//! concentration of trade size at any given time.
21//!
22//! # Parameters
23//!
24//! Orders submitted to this algorithm must include `exec_algorithm_params` with:
25//! - `horizon_secs`: Total execution horizon in seconds.
26//! - `interval_secs`: Interval between child orders in seconds.
27//!
28//! # Example
29//!
30//! An order with `horizon_secs=60` and `interval_secs=10` will spawn 6 child
31//! orders over 60 seconds, one every 10 seconds.
32
33use std::time::Duration;
34
35use ahash::AHashMap;
36use nautilus_common::{
37    actor::{DataActor, DataActorNative},
38    timer::TimeEvent,
39};
40use nautilus_model::{
41    enums::OrderType,
42    events::OrderDeniedReason,
43    identifiers::ClientOrderId,
44    instruments::Instrument,
45    orders::{Order, OrderAny},
46    types::Quantity,
47};
48use rust_decimal::{Decimal, RoundingStrategy};
49use ustr::Ustr;
50
51use super::{
52    ExecutionAlgorithm, ExecutionAlgorithmConfig, ExecutionAlgorithmCore, ExecutionAlgorithmNative,
53};
54use crate::nautilus_execution_algorithm;
55
56/// Configuration for [`TwapAlgorithm`].
57pub type TwapAlgorithmConfig = ExecutionAlgorithmConfig;
58
59/// Time-Weighted Average Price (TWAP) execution algorithm.
60///
61/// Executes orders by evenly spreading them over a specified time horizon,
62/// at regular intervals. The algorithm receives a primary order and spawns
63/// smaller child orders that are executed at regular intervals.
64#[derive(Debug)]
65pub struct TwapAlgorithm {
66    /// The algorithm core.
67    pub core: ExecutionAlgorithmCore,
68    /// Schedules for each primary order.
69    scheduled_orders: AHashMap<ClientOrderId, TwapSchedule>,
70}
71
72impl TwapAlgorithm {
73    /// Creates a new [`TwapAlgorithm`] instance.
74    #[must_use]
75    pub fn new(config: TwapAlgorithmConfig) -> Self {
76        Self {
77            core: ExecutionAlgorithmCore::new(config),
78            scheduled_orders: AHashMap::new(),
79        }
80    }
81
82    /// Completes the execution sequence for a primary order.
83    fn complete_sequence(&mut self, primary_id: ClientOrderId) {
84        let timer_name = primary_id.as_str();
85        let core = ExecutionAlgorithmNative::exec_algorithm_core_mut(self);
86        if core.clock_mut().timer_names().contains(&timer_name) {
87            core.clock_mut().cancel_timer(timer_name);
88        }
89        core.remove_submit_params(&primary_id);
90        self.scheduled_orders.remove(&primary_id);
91        log::info!("Completed TWAP execution for {primary_id}");
92    }
93}
94
95// The clock and component lifecycle dispatch through the `DataActor` hooks,
96// so forward them to the `ExecutionAlgorithm` implementations.
97impl DataActor for TwapAlgorithm {
98    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
99        ExecutionAlgorithm::on_time_event(self, event)
100    }
101
102    fn on_stop(&mut self) -> anyhow::Result<()> {
103        ExecutionAlgorithm::on_stop(self)
104    }
105
106    fn on_resume(&mut self) -> anyhow::Result<()> {
107        ExecutionAlgorithm::on_resume(self)
108    }
109
110    fn on_reset(&mut self) -> anyhow::Result<()> {
111        ExecutionAlgorithm::on_reset(self)
112    }
113}
114
115nautilus_execution_algorithm!(TwapAlgorithm, {
116    fn on_order(&mut self, order: OrderAny) -> anyhow::Result<()> {
117        let primary_id = order.client_order_id();
118
119        if self.scheduled_orders.contains_key(&primary_id) {
120            anyhow::bail!("Order {primary_id} already being executed");
121        }
122
123        log::info!("Received order for TWAP execution: {order:?}");
124
125        // Only market orders supported
126        if order.order_type() != OrderType::Market {
127            let reason = OrderDeniedReason::UnsupportedOrderType {
128                order_type: order.order_type(),
129            }
130            .to_string();
131            return self.deny_order(&order, Ustr::from(&reason));
132        }
133
134        let instrument = {
135            let cache = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_ref();
136            cache.instrument(&order.instrument_id()).cloned()
137        };
138
139        let Some(instrument) = instrument else {
140            let reason = OrderDeniedReason::InstrumentNotFound {
141                instrument_id: order.instrument_id(),
142            }
143            .to_string();
144            return self.deny_order(&order, Ustr::from(&reason));
145        };
146
147        let Some(exec_params) = order.exec_algorithm_params() else {
148            return self.deny_order(&order, validation_failed("exec_algorithm_params not found"));
149        };
150
151        let Some(horizon_secs_str) = exec_params.get(&Ustr::from("horizon_secs")) else {
152            return self.deny_order(
153                &order,
154                validation_failed("horizon_secs not found in exec_algorithm_params"),
155            );
156        };
157
158        let horizon_secs: f64 = match horizon_secs_str.parse() {
159            Ok(value) => value,
160            Err(_) => {
161                return self.deny_order(
162                    &order,
163                    validation_failed(format!(
164                        "horizon_secs={horizon_secs_str} is not a valid number"
165                    )),
166                );
167            }
168        };
169
170        let Some(interval_secs_str) = exec_params.get(&Ustr::from("interval_secs")) else {
171            return self.deny_order(
172                &order,
173                validation_failed("interval_secs not found in exec_algorithm_params"),
174            );
175        };
176
177        let interval_secs: f64 = match interval_secs_str.parse() {
178            Ok(value) => value,
179            Err(_) => {
180                return self.deny_order(
181                    &order,
182                    validation_failed(format!(
183                        "interval_secs={interval_secs_str} is not a valid number"
184                    )),
185                );
186            }
187        };
188
189        if !horizon_secs.is_finite() || horizon_secs <= 0.0 {
190            return self.deny_order(
191                &order,
192                validation_failed(format!(
193                    "horizon_secs={horizon_secs} must be finite and positive"
194                )),
195            );
196        }
197
198        if !interval_secs.is_finite() || interval_secs <= 0.0 {
199            return self.deny_order(
200                &order,
201                validation_failed(format!(
202                    "interval_secs={interval_secs} must be finite and positive"
203                )),
204            );
205        }
206
207        if horizon_secs < interval_secs {
208            return self.deny_order(
209                &order,
210                validation_failed(format!(
211                    "horizon_secs={horizon_secs} must be greater than or equal to interval_secs={interval_secs}"
212                )),
213            );
214        }
215
216        let num_intervals = (horizon_secs / interval_secs).floor() as u64;
217        if num_intervals == 0 {
218            return self.deny_order(&order, validation_failed("num_intervals is 0"));
219        }
220
221        let total_qty = order.quantity();
222        let interval_count = Decimal::from(num_intervals);
223        let quotient = total_qty.as_decimal() / interval_count;
224        let floored = quotient.round_dp_with_strategy(
225            u32::from(instrument.size_precision()),
226            RoundingStrategy::ToZero,
227        );
228        let qty_per_interval = match instrument.try_make_qty_from_decimal(floored, None) {
229            Ok(quantity) => quantity,
230            Err(e) => {
231                return self.deny_order(
232                    &order,
233                    validation_failed(format!("invalid qty_per_interval={floored}: {e}")),
234                );
235            }
236        };
237        let remainder = total_qty.as_decimal() - floored * interval_count;
238
239        if qty_per_interval == total_qty || qty_per_interval < instrument.size_increment() {
240            log::warn!(
241                "Submitting for entire size: qty_per_interval={qty_per_interval}, order_quantity={total_qty}"
242            );
243            self.submit_order(order, None, None)?;
244            self.complete_sequence(primary_id);
245            return Ok(());
246        }
247
248        if let Some(min_qty) = instrument.min_quantity()
249            && qty_per_interval < min_qty
250        {
251            log::warn!(
252                "Submitting for entire size: qty_per_interval={qty_per_interval} < min_quantity={min_qty}"
253            );
254            self.submit_order(order, None, None)?;
255            self.complete_sequence(primary_id);
256            return Ok(());
257        }
258
259        let interval = match Duration::try_from_secs_f64(interval_secs) {
260            Ok(interval) => interval,
261            Err(e) => {
262                return self.deny_order(
263                    &order,
264                    validation_failed(format!(
265                        "interval_secs={interval_secs} is not a valid duration: {e}"
266                    )),
267                );
268            }
269        };
270
271        if interval == Duration::ZERO {
272            return self.deny_order(
273                &order,
274                validation_failed(format!(
275                    "interval_secs={interval_secs} rounds to a zero duration"
276                )),
277            );
278        }
279        let Ok(interval_ns) = u64::try_from(interval.as_nanos()) else {
280            return self.deny_order(
281                &order,
282                validation_failed(format!(
283                    "interval_secs={interval_secs} exceeds the clock nanosecond range"
284                )),
285            );
286        };
287        let timestamp_ns = ExecutionAlgorithmNative::exec_algorithm_core_mut(self)
288            .clock_mut()
289            .timestamp_ns()
290            .as_u64();
291
292        if timestamp_ns.checked_add(interval_ns).is_none() {
293            return self.deny_order(
294                &order,
295                validation_failed(format!(
296                    "interval_secs={interval_secs} exceeds the clock timestamp headroom"
297                )),
298            );
299        }
300
301        let mut scheduled_sizes: Vec<Quantity> = vec![qty_per_interval; num_intervals as usize];
302
303        if remainder > Decimal::ZERO {
304            let remainder_qty = match instrument.try_make_qty_from_decimal(remainder, None) {
305                Ok(quantity) => quantity,
306                Err(e) => {
307                    return self.deny_order(
308                        &order,
309                        validation_failed(format!("invalid qty_remainder={remainder}: {e}")),
310                    );
311                }
312            };
313            scheduled_sizes.push(remainder_qty);
314        }
315
316        let scheduled_total = scheduled_sizes
317            .iter()
318            .fold(Decimal::ZERO, |total, quantity| {
319                total + quantity.as_decimal()
320            });
321
322        if scheduled_total != total_qty.as_decimal() {
323            return self.deny_order(
324                &order,
325                validation_failed(format!(
326                    "scheduled quantity {scheduled_total} does not equal order quantity {total_qty}"
327                )),
328            );
329        }
330
331        log::info!("Order execution size schedule: {scheduled_sizes:?}");
332
333        // Add primary order to cache so on_time_event can retrieve it,
334        // it is already present when routed through the engine's submit path.
335        {
336            let cache_rc = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_rc();
337            let mut cache = cache_rc.borrow_mut();
338            if !cache.order_exists(&primary_id) {
339                cache.add_order(order.clone(), None, None, false)?;
340            }
341        }
342
343        self.scheduled_orders.insert(
344            primary_id,
345            TwapSchedule {
346                remaining_sizes: scheduled_sizes.clone(),
347                interval,
348            },
349        );
350
351        let schedule = self.scheduled_orders.get_mut(&primary_id).unwrap();
352        let first_qty = schedule.remaining_sizes.remove(0);
353        let is_single_slice = self
354            .scheduled_orders
355            .get(&primary_id)
356            .is_some_and(|schedule| schedule.remaining_sizes.is_empty());
357
358        // Single slice: submit the primary order directly
359        if is_single_slice {
360            self.submit_order(order, None, None)?;
361            self.complete_sequence(primary_id);
362            return Ok(());
363        }
364
365        // Multiple slices: spawn first child order and reduce primary
366        let tags = order.tags().map(<[Ustr]>::to_vec);
367        let time_in_force = order.time_in_force();
368        let reduce_only = order.is_reduce_only();
369        let mut order = order;
370        let spawned = self.spawn_market(
371            &mut order,
372            first_qty,
373            time_in_force,
374            reduce_only,
375            tags,
376            true,
377        );
378        self.submit_order(spawned.into(), None, None)?;
379
380        ExecutionAlgorithmNative::exec_algorithm_core_mut(self)
381            .clock_mut()
382            .set_timer(primary_id.as_str(), interval, None, None, None, None, None)?;
383
384        log::info!(
385            "Started TWAP execution for {primary_id}: horizon_secs={horizon_secs}, interval_secs={interval_secs}"
386        );
387
388        Ok(())
389    }
390
391    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
392        log::info!("Received time event: {event}");
393
394        let primary_id = ClientOrderId::new(event.name);
395
396        let primary = {
397            let cache = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_ref();
398            cache.order(&primary_id).map(|o| o.clone())
399        };
400
401        let Some(primary) = primary else {
402            log::error!("Cannot find primary order for exec_spawn_id={primary_id}");
403            self.complete_sequence(primary_id);
404            return Ok(());
405        };
406
407        if primary.is_closed() {
408            self.complete_sequence(primary_id);
409            return Ok(());
410        }
411
412        let Some(schedule) = self.scheduled_orders.get_mut(&primary_id) else {
413            log::error!("Cannot find scheduled sizes for exec_spawn_id={primary_id}");
414            return Ok(());
415        };
416
417        if schedule.remaining_sizes.is_empty() {
418            log::warn!("No more size to execute for exec_spawn_id={primary_id}");
419            return Ok(());
420        }
421
422        let quantity = schedule.remaining_sizes.remove(0);
423        let is_final_slice = schedule.remaining_sizes.is_empty();
424
425        // Final slice: submit the primary order (already reduced to remaining quantity)
426        if is_final_slice {
427            self.submit_order(primary, None, None)?;
428            self.complete_sequence(primary_id);
429            return Ok(());
430        }
431
432        // Intermediate slice: spawn child order and reduce primary
433        let tags = primary.tags().map(<[Ustr]>::to_vec);
434        let time_in_force = primary.time_in_force();
435        let reduce_only = primary.is_reduce_only();
436        let mut primary = primary;
437        let spawned = self.spawn_market(
438            &mut primary,
439            quantity,
440            time_in_force,
441            reduce_only,
442            tags,
443            true,
444        );
445        self.submit_order(spawned.into(), None, None)?;
446
447        Ok(())
448    }
449
450    fn on_stop(&mut self) -> anyhow::Result<()> {
451        ExecutionAlgorithmNative::exec_algorithm_core_mut(self)
452            .clock_mut()
453            .cancel_timers();
454        Ok(())
455    }
456
457    fn on_resume(&mut self) -> anyhow::Result<()> {
458        let primary_ids: Vec<ClientOrderId> = self.scheduled_orders.keys().copied().collect();
459
460        for primary_id in primary_ids {
461            let primary_is_open = {
462                let cache = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_ref();
463                cache.order(&primary_id).map(|primary| !primary.is_closed())
464            };
465
466            if primary_is_open.is_none() {
467                log::error!("Cannot find primary order for exec_spawn_id={primary_id}");
468            }
469            let interval = self.scheduled_orders.get(&primary_id).and_then(|schedule| {
470                (!schedule.remaining_sizes.is_empty()).then_some(schedule.interval)
471            });
472
473            let Some(interval) = interval.filter(|_| primary_is_open == Some(true)) else {
474                self.complete_sequence(primary_id);
475                continue;
476            };
477
478            ExecutionAlgorithmNative::exec_algorithm_core_mut(self)
479                .clock_mut()
480                .set_timer(primary_id.as_str(), interval, None, None, None, None, None)?;
481        }
482
483        Ok(())
484    }
485
486    fn on_reset(&mut self) -> anyhow::Result<()> {
487        self.unsubscribe_all_strategy_events();
488        ExecutionAlgorithmNative::exec_algorithm_core_mut(self).reset();
489        self.scheduled_orders.clear();
490        Ok(())
491    }
492});
493
494#[derive(Debug)]
495struct TwapSchedule {
496    remaining_sizes: Vec<Quantity>,
497    interval: Duration,
498}
499
500fn validation_failed(detail: impl Into<String>) -> Ustr {
501    let reason = OrderDeniedReason::ValidationFailed {
502        detail: detail.into(),
503    }
504    .to_string();
505    Ustr::from(&reason)
506}
507
508#[cfg(test)]
509mod tests {
510    use std::{cell::RefCell, rc::Rc};
511
512    use indexmap::IndexMap;
513    use nautilus_common::{
514        cache::Cache,
515        clock::{Clock, VirtualClock},
516        component::Component,
517        enums::ComponentTrigger,
518        messages::execution::{ModifyOrder, SubmitOrder, TradingCommand},
519        msgbus::{self, MessagingSwitchboard, TypedHandler},
520    };
521    use nautilus_core::{DurationNanos, Params, UUID4, UnixNanos};
522    use nautilus_model::{
523        enums::{OrderSide, OrderStatus, TimeInForce},
524        events::{OrderDeniedReason, OrderEventAny, order::spec::OrderCanceledSpec},
525        identifiers::{ExecAlgorithmId, InstrumentId, StrategyId, TraderId},
526        orders::{LimitOrder, MarketOrder},
527        types::Price,
528    };
529    use rstest::rstest;
530    use ustr::Ustr;
531
532    use super::*;
533
534    fn create_twap_algorithm() -> TwapAlgorithm {
535        // Use unique ID to avoid thread-local registry/msgbus conflicts in parallel tests
536        let unique_id = format!("TWAP-{}", UUID4::new());
537        let config = TwapAlgorithmConfig {
538            exec_algorithm_id: Some(ExecAlgorithmId::new(&unique_id)),
539            ..Default::default()
540        };
541        TwapAlgorithm::new(config)
542    }
543
544    fn register_algorithm_with_clock(algo: &mut TwapAlgorithm) -> Rc<RefCell<VirtualClock>> {
545        use nautilus_common::timer::TimeEventCallback;
546
547        let trader_id = TraderId::from("TRADER-001");
548        let clock = Rc::new(RefCell::new(VirtualClock::new()));
549        let cache = Rc::new(RefCell::new(Cache::default()));
550
551        // Register a no-op default handler for timer callbacks
552        clock
553            .borrow_mut()
554            .register_default_handler(TimeEventCallback::Rust(std::sync::Arc::new(|_| {})));
555
556        algo.core.register(trader_id, clock.clone(), cache).unwrap();
557
558        // Transition to Running state for tests
559        algo.transition_state(ComponentTrigger::Initialize).unwrap();
560        algo.transition_state(ComponentTrigger::Start).unwrap();
561        algo.transition_state(ComponentTrigger::StartCompleted)
562            .unwrap();
563
564        clock
565    }
566
567    fn register_algorithm(algo: &mut TwapAlgorithm) {
568        let _ = register_algorithm_with_clock(algo);
569    }
570
571    fn add_instrument_to_cache(algo: &TwapAlgorithm) {
572        use nautilus_model::instruments::{InstrumentAny, stubs::crypto_perpetual_ethusdt};
573
574        let instrument = crypto_perpetual_ethusdt();
575        let cache_rc = algo.core.cache_rc();
576        let mut cache = cache_rc.borrow_mut();
577        cache
578            .add_instrument(InstrumentAny::CryptoPerpetual(instrument))
579            .unwrap();
580    }
581
582    fn create_market_order_with_params(params: IndexMap<Ustr, Ustr>) -> OrderAny {
583        create_market_order_with_params_and_qty(params, Quantity::from("1.0"))
584    }
585
586    fn create_market_order_with_params_and_qty(
587        params: IndexMap<Ustr, Ustr>,
588        quantity: Quantity,
589    ) -> OrderAny {
590        let client_order_id = ClientOrderId::from("O-001");
591        OrderAny::Market(MarketOrder::new(
592            TraderId::from("TRADER-001"),
593            StrategyId::from("STRAT-001"),
594            InstrumentId::from("ETHUSDT-PERP.BINANCE"),
595            client_order_id,
596            OrderSide::Buy,
597            quantity,
598            TimeInForce::Gtc,
599            UUID4::new(),
600            0.into(),
601            false,
602            false,
603            None,
604            None,
605            None,
606            None,
607            Some(ExecAlgorithmId::new("TWAP")),
608            Some(params),
609            Some(client_order_id),
610            None,
611        ))
612    }
613
614    fn assert_twap_denied(algo: &mut TwapAlgorithm, order: &OrderAny, expected_reason: &str) {
615        let strategy_id = order.strategy_id();
616        {
617            let cache_rc = algo.core.cache_rc();
618            cache_rc
619                .borrow_mut()
620                .add_order(order.clone(), None, None, false)
621                .unwrap();
622        }
623        let events = Rc::new(RefCell::new(Vec::new()));
624        let handler = TypedHandler::from({
625            let events = events.clone();
626            move |event: &OrderEventAny| events.borrow_mut().push(event.clone())
627        });
628        let topic = format!("events.order.{strategy_id}");
629        msgbus::subscribe_order_events(topic.clone().into(), handler.clone(), None);
630
631        algo.on_order(order.clone()).unwrap();
632        algo.on_order(order.clone()).unwrap();
633
634        msgbus::unsubscribe_order_events(topic.into(), &handler);
635        let cached_order = algo.cache().order(&order.client_order_id()).unwrap();
636        let events = events.borrow();
637
638        assert_eq!(cached_order.status(), OrderStatus::Denied);
639        assert_eq!(events.len(), 1);
640        assert!(matches!(
641            &events[0],
642            OrderEventAny::Denied(event)
643                if event.reason == expected_reason
644                    && event.strategy_id == strategy_id
645                    && event.client_order_id == order.client_order_id()
646        ));
647        assert!(algo.scheduled_orders.is_empty());
648        assert!(algo.clock().timer_names().is_empty());
649    }
650
651    #[rstest]
652    fn test_twap_creation() {
653        let algo = create_twap_algorithm();
654        assert!(algo.id().inner().starts_with("TWAP"));
655        assert!(algo.scheduled_orders.is_empty());
656    }
657
658    #[rstest]
659    fn test_twap_registration() {
660        let mut algo = create_twap_algorithm();
661        register_algorithm(&mut algo);
662
663        assert_eq!(algo.trader_id(), Some(TraderId::from("TRADER-001")));
664    }
665
666    #[rstest]
667    fn test_twap_reset_clears_scheduled_sizes() {
668        let mut algo = create_twap_algorithm();
669        algo.scheduled_orders.insert(
670            ClientOrderId::new("O-001"),
671            TwapSchedule {
672                remaining_sizes: vec![Quantity::from("1.0")],
673                interval: Duration::from_secs(1),
674            },
675        );
676        algo.scheduled_orders.insert(
677            ClientOrderId::new("O-002"),
678            TwapSchedule {
679                remaining_sizes: vec![Quantity::from("2.0")],
680                interval: Duration::from_secs(2),
681            },
682        );
683
684        assert!(!algo.scheduled_orders.is_empty());
685
686        // Dispatch through the DataActor entry point the component lifecycle uses
687        DataActor::on_reset(&mut algo).unwrap();
688
689        assert!(algo.scheduled_orders.is_empty());
690    }
691
692    #[rstest]
693    fn test_twap_rejects_non_market_orders() {
694        let mut algo = create_twap_algorithm();
695        register_algorithm(&mut algo);
696
697        let order = OrderAny::Limit(LimitOrder::new(
698            TraderId::from("TRADER-001"),
699            StrategyId::from("STRAT-001"),
700            InstrumentId::from("BTC/USDT.BINANCE"),
701            ClientOrderId::from("O-001"),
702            OrderSide::Buy,
703            Quantity::from("1.0"),
704            Price::from("50000.0"),
705            TimeInForce::Gtc,
706            None,  // expire_time
707            false, // post_only
708            false, // reduce_only
709            false, // quote_quantity
710            None,  // display_qty
711            None,  // emulation_trigger
712            None,  // trigger_instrument_id
713            None,  // contingency_type
714            None,  // order_list_id
715            None,  // linked_order_ids
716            None,  // parent_order_id
717            None,  // exec_algorithm_id
718            None,  // exec_algorithm_params
719            None,  // exec_spawn_id
720            None,  // tags
721            UUID4::new(),
722            0.into(),
723        ));
724
725        let reason = OrderDeniedReason::UnsupportedOrderType {
726            order_type: OrderType::Limit,
727        }
728        .to_string();
729        assert_twap_denied(&mut algo, &order, &reason);
730    }
731
732    #[rstest]
733    fn test_twap_denies_missing_instrument() {
734        let mut algo = create_twap_algorithm();
735        register_algorithm(&mut algo);
736
737        let mut params = IndexMap::new();
738        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
739        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
740        let order = create_market_order_with_params(params);
741        let reason = OrderDeniedReason::InstrumentNotFound {
742            instrument_id: order.instrument_id(),
743        }
744        .to_string();
745
746        assert_twap_denied(&mut algo, &order, &reason);
747    }
748
749    #[rstest]
750    fn test_twap_rejects_missing_params() {
751        let mut algo = create_twap_algorithm();
752        register_algorithm(&mut algo);
753
754        add_instrument_to_cache(&algo);
755
756        let order = OrderAny::Market(MarketOrder::new(
757            TraderId::from("TRADER-001"),
758            StrategyId::from("STRAT-001"),
759            InstrumentId::from("ETHUSDT-PERP.BINANCE"),
760            ClientOrderId::from("O-001"),
761            OrderSide::Buy,
762            Quantity::from("1.0"),
763            TimeInForce::Gtc,
764            UUID4::new(),
765            0.into(),
766            false,
767            false,
768            None,
769            None,
770            None,
771            None,
772            None,
773            None, // No exec_algorithm_params
774            None,
775            None,
776        ));
777
778        assert_twap_denied(
779            &mut algo,
780            &order,
781            "VALIDATION_FAILED: exec_algorithm_params not found",
782        );
783    }
784
785    #[rstest]
786    #[case(
787        None,
788        Some("10"),
789        "VALIDATION_FAILED: horizon_secs not found in exec_algorithm_params"
790    )]
791    #[case(
792        Some("60"),
793        None,
794        "VALIDATION_FAILED: interval_secs not found in exec_algorithm_params"
795    )]
796    #[case(
797        Some("not-a-number"),
798        Some("10"),
799        "VALIDATION_FAILED: horizon_secs=not-a-number is not a valid number"
800    )]
801    #[case(
802        Some("60"),
803        Some("not-a-number"),
804        "VALIDATION_FAILED: interval_secs=not-a-number is not a valid number"
805    )]
806    fn test_twap_denies_missing_or_malformed_schedule_parameter(
807        #[case] horizon_secs: Option<&str>,
808        #[case] interval_secs: Option<&str>,
809        #[case] expected_reason: &str,
810    ) {
811        let mut algo = create_twap_algorithm();
812        register_algorithm(&mut algo);
813        add_instrument_to_cache(&algo);
814
815        let mut params = IndexMap::new();
816
817        if let Some(horizon_secs) = horizon_secs {
818            params.insert(Ustr::from("horizon_secs"), Ustr::from(horizon_secs));
819        }
820
821        if let Some(interval_secs) = interval_secs {
822            params.insert(Ustr::from("interval_secs"), Ustr::from(interval_secs));
823        }
824
825        assert_twap_denied(
826            &mut algo,
827            &create_market_order_with_params(params),
828            expected_reason,
829        );
830    }
831
832    #[rstest]
833    fn test_twap_rejects_horizon_less_than_interval() {
834        let mut algo = create_twap_algorithm();
835        register_algorithm(&mut algo);
836
837        add_instrument_to_cache(&algo);
838
839        let mut params = IndexMap::new();
840        params.insert(Ustr::from("horizon_secs"), Ustr::from("30"));
841        params.insert(Ustr::from("interval_secs"), Ustr::from("60"));
842
843        let order = create_market_order_with_params(params);
844        assert_twap_denied(
845            &mut algo,
846            &order,
847            "VALIDATION_FAILED: horizon_secs=30 must be greater than or equal to interval_secs=60",
848        );
849    }
850
851    #[rstest]
852    fn test_twap_rejects_duplicate_order() {
853        let mut algo = create_twap_algorithm();
854        register_algorithm(&mut algo);
855
856        add_instrument_to_cache(&algo);
857
858        let mut params = IndexMap::new();
859        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
860        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
861
862        let order1 = create_market_order_with_params(params.clone());
863        let order2 = create_market_order_with_params(params);
864
865        algo.on_order(order1).unwrap();
866        let result = algo.on_order(order2);
867
868        assert!(result.is_err());
869        assert!(
870            result
871                .unwrap_err()
872                .to_string()
873                .contains("already being executed")
874        );
875    }
876
877    #[rstest]
878    fn test_twap_calculates_size_schedule_evenly() {
879        let mut algo = create_twap_algorithm();
880        register_algorithm(&mut algo);
881
882        add_instrument_to_cache(&algo);
883
884        // 1.2 qty over 60s with 20s intervals = 3 intervals of 0.4 each (divides evenly)
885        let mut params = IndexMap::new();
886        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
887        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
888
889        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
890        let primary_id = order.client_order_id();
891
892        algo.on_order(order).unwrap();
893
894        // First slice spawned immediately, remaining 2 slices scheduled (no remainder)
895        let remaining = &algo
896            .scheduled_orders
897            .get(&primary_id)
898            .unwrap()
899            .remaining_sizes;
900        assert_eq!(remaining.len(), 2);
901
902        for qty in remaining {
903            assert_eq!(*qty, Quantity::from("0.4"));
904        }
905    }
906
907    #[rstest]
908    fn test_twap_reduces_cached_primary_after_first_child_spawn() {
909        let mut algo = create_twap_algorithm();
910        register_algorithm(&mut algo);
911
912        add_instrument_to_cache(&algo);
913
914        let mut params = IndexMap::new();
915        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
916        params.insert(Ustr::from("interval_secs"), Ustr::from("30"));
917
918        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
919        let primary_id = order.client_order_id();
920
921        algo.on_order(order).unwrap();
922
923        let cache = algo.cache();
924        let primary = cache.order(&primary_id).unwrap();
925        let spawned = cache.order(&ClientOrderId::from("O-001-E1")).unwrap();
926
927        assert_eq!(primary.quantity(), Quantity::from("0.6"));
928        assert_eq!(spawned.quantity(), Quantity::from("0.6"));
929        assert_eq!(spawned.exec_spawn_id(), Some(primary_id));
930    }
931
932    #[rstest]
933    fn test_twap_refused_modify_preserves_remaining_quantity_schedule() {
934        let mut algo = create_twap_algorithm();
935        register_algorithm(&mut algo);
936        add_instrument_to_cache(&algo);
937
938        let mut params = IndexMap::new();
939        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
940        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
941        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
942        let primary_id = order.client_order_id();
943        algo.on_order(order).unwrap();
944
945        let command = ModifyOrder::new(
946            TraderId::from("TRADER-001"),
947            None,
948            StrategyId::from("STRAT-001"),
949            InstrumentId::from("ETHUSDT-PERP.BINANCE"),
950            primary_id,
951            None,
952            Some(Quantity::from("0.4")),
953            None,
954            None,
955            UUID4::new(),
956            0.into(),
957            None,
958            None,
959        );
960
961        algo.handle_modify_order(command).unwrap();
962
963        let primary_quantity = algo.cache().order(&primary_id).unwrap().quantity();
964        let scheduled_quantity = algo.scheduled_orders[&primary_id]
965            .remaining_sizes
966            .iter()
967            .fold(Decimal::ZERO, |total, quantity| {
968                total + quantity.as_decimal()
969            });
970        assert_eq!(primary_quantity.as_decimal(), scheduled_quantity);
971        assert_eq!(primary_quantity, Quantity::from("0.8"));
972    }
973
974    #[rstest]
975    fn test_twap_on_order_accepts_already_cached_primary() {
976        let mut algo = create_twap_algorithm();
977        register_algorithm(&mut algo);
978
979        add_instrument_to_cache(&algo);
980
981        let mut params = IndexMap::new();
982        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
983        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
984
985        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
986        let primary_id = order.client_order_id();
987
988        // The engine submit path caches the primary before routing to the algorithm
989        {
990            let cache_rc = algo.core.cache_rc();
991            let mut cache = cache_rc.borrow_mut();
992            cache.add_order(order.clone(), None, None, false).unwrap();
993        }
994
995        algo.on_order(order).unwrap();
996
997        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 2);
998    }
999
1000    #[rstest]
1001    fn test_twap_calculates_size_schedule_with_remainder() {
1002        let mut algo = create_twap_algorithm();
1003        register_algorithm(&mut algo);
1004
1005        add_instrument_to_cache(&algo);
1006        let instrument = nautilus_model::instruments::stubs::crypto_perpetual_ethusdt();
1007
1008        // 1.0 qty over 60s with 20s intervals = 3 intervals
1009        let mut params = IndexMap::new();
1010        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1011        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1012
1013        let order = create_market_order_with_params(params);
1014        let primary_id = order.client_order_id();
1015
1016        algo.on_order(order).unwrap();
1017
1018        // First slice spawned, 3 remaining (2 regular + 1 remainder)
1019        let remaining = &algo
1020            .scheduled_orders
1021            .get(&primary_id)
1022            .unwrap()
1023            .remaining_sizes;
1024        assert_eq!(remaining.len(), 3);
1025        assert_eq!(
1026            remaining,
1027            &[
1028                Quantity::from("0.333"),
1029                Quantity::from("0.333"),
1030                Quantity::from("0.001"),
1031            ]
1032        );
1033
1034        for quantity in remaining {
1035            assert_eq!(quantity.precision, instrument.size_precision());
1036            assert_eq!(quantity.raw() % instrument.size_increment().raw(), 0);
1037        }
1038
1039        let first = algo
1040            .cache()
1041            .order(&ClientOrderId::from("O-001-E1"))
1042            .unwrap()
1043            .quantity();
1044        let total = remaining
1045            .iter()
1046            .fold(first.as_decimal(), |sum, qty| sum + qty.as_decimal());
1047        assert_eq!(total, Quantity::from("1.0").as_decimal());
1048    }
1049
1050    #[rstest]
1051    fn test_twap_children_use_instrument_size_precision() {
1052        let mut algo = create_twap_algorithm();
1053        register_algorithm(&mut algo);
1054
1055        add_instrument_to_cache(&algo);
1056        let instrument = nautilus_model::instruments::stubs::crypto_perpetual_ethusdt();
1057
1058        let mut params = IndexMap::new();
1059        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1060        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1061
1062        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.000000000"));
1063        let primary_id = order.client_order_id();
1064
1065        algo.on_order(order).unwrap();
1066
1067        let spawned = algo
1068            .cache()
1069            .order(&ClientOrderId::from("O-001-E1"))
1070            .unwrap()
1071            .quantity();
1072        assert_eq!(spawned.precision, instrument.size_precision());
1073        assert!(
1074            algo.scheduled_orders[&primary_id]
1075                .remaining_sizes
1076                .iter()
1077                .all(|quantity| quantity.precision == instrument.size_precision())
1078        );
1079    }
1080
1081    #[rstest]
1082    fn test_twap_on_time_event_spawns_next_slice() {
1083        let mut algo = create_twap_algorithm();
1084        register_algorithm(&mut algo);
1085
1086        add_instrument_to_cache(&algo);
1087
1088        // Use qty that divides evenly: 1.2 / 3 = 0.4 each
1089        let mut params = IndexMap::new();
1090        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1091        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1092
1093        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1094        let primary_id = order.client_order_id();
1095
1096        let submit_params = Params::new(
1097            [(
1098                "routing_profile".to_string(),
1099                serde_json::Value::String("intermediate-slice".to_string()),
1100            )]
1101            .into(),
1102        );
1103        algo.core
1104            .remember_submit_params(primary_id, Some(submit_params.clone()));
1105
1106        algo.on_order(order).unwrap();
1107
1108        // Verify 2 slices remain after first spawn (no remainder)
1109        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 2);
1110
1111        // Simulate timer firing
1112        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
1113        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
1114
1115        // One slice consumed
1116        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 1);
1117        assert_eq!(algo.core.submit_params(&primary_id), Some(submit_params));
1118    }
1119
1120    #[rstest]
1121    fn test_twap_data_actor_dispatch_spawns_next_slice() {
1122        let mut algo = create_twap_algorithm();
1123        register_algorithm(&mut algo);
1124
1125        add_instrument_to_cache(&algo);
1126
1127        let mut params = IndexMap::new();
1128        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1129        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1130
1131        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1132        let primary_id = order.client_order_id();
1133
1134        algo.on_order(order).unwrap();
1135        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 2);
1136
1137        // Dispatch through the DataActor entry point the clock callback uses
1138        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
1139        algo.handle_time_event(&event);
1140
1141        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 1);
1142    }
1143
1144    #[rstest]
1145    fn test_twap_on_time_event_completes_on_final_slice() {
1146        let mut algo = create_twap_algorithm();
1147        register_algorithm(&mut algo);
1148
1149        add_instrument_to_cache(&algo);
1150
1151        // 2 intervals: first spawned immediately, one in scheduled_sizes
1152        let mut params = IndexMap::new();
1153        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1154        params.insert(Ustr::from("interval_secs"), Ustr::from("30"));
1155
1156        let order = create_market_order_with_params(params);
1157        let primary_id = order.client_order_id();
1158
1159        let submit_params = Params::new(
1160            [(
1161                "routing_profile".to_string(),
1162                serde_json::Value::String("final-slice".to_string()),
1163            )]
1164            .into(),
1165        );
1166        algo.core
1167            .remember_submit_params(primary_id, Some(submit_params.clone()));
1168
1169        algo.on_order(order).unwrap();
1170        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 1);
1171        assert_eq!(
1172            algo.core.submit_params(&primary_id),
1173            Some(submit_params.clone())
1174        );
1175
1176        let received = Rc::new(RefCell::new(None::<SubmitOrder>));
1177        let handler = msgbus::TypedIntoHandler::from({
1178            let captured = received.clone();
1179            move |cmd: TradingCommand| {
1180                if let TradingCommand::SubmitOrder(cmd) = cmd {
1181                    *captured.borrow_mut() = Some(cmd);
1182                }
1183            }
1184        });
1185        msgbus::register_trading_command_endpoint(
1186            MessagingSwitchboard::risk_engine_queue_execute(),
1187            handler,
1188        );
1189
1190        // Simulate timer firing for final slice
1191        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
1192        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
1193
1194        // Sequence completed, scheduled_sizes removed
1195        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1196        assert_eq!(
1197            received
1198                .borrow()
1199                .as_ref()
1200                .and_then(|cmd| cmd.params.clone()),
1201            Some(submit_params),
1202        );
1203        assert_eq!(algo.core.submit_params(&primary_id), None);
1204    }
1205
1206    #[rstest]
1207    fn test_twap_on_time_event_completes_when_primary_closed() {
1208        let mut algo = create_twap_algorithm();
1209        register_algorithm(&mut algo);
1210
1211        add_instrument_to_cache(&algo);
1212
1213        let mut params = IndexMap::new();
1214        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1215        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1216
1217        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1218        let primary_id = order.client_order_id();
1219
1220        let submit_params = Params::new(
1221            [(
1222                "routing_profile".to_string(),
1223                serde_json::Value::String("closed-primary".to_string()),
1224            )]
1225            .into(),
1226        );
1227        algo.core
1228            .remember_submit_params(primary_id, Some(submit_params));
1229
1230        algo.on_order(order).unwrap();
1231        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 2);
1232
1233        // Mark primary order as closed (canceled)
1234        {
1235            let cache_rc = algo.core.cache_rc();
1236            let mut cache = cache_rc.borrow_mut();
1237            let primary = cache.order(&primary_id).map(|o| o.clone()).unwrap();
1238
1239            let canceled = OrderCanceledSpec::builder()
1240                .trader_id(primary.trader_id())
1241                .strategy_id(primary.strategy_id())
1242                .instrument_id(primary.instrument_id())
1243                .client_order_id(primary.client_order_id())
1244                .build();
1245            cache
1246                .update_order(&OrderEventAny::Canceled(canceled))
1247                .unwrap();
1248        }
1249
1250        // Timer fires but primary is closed
1251        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
1252        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
1253
1254        // Sequence should complete early since primary is closed
1255        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1256        assert_eq!(algo.core.submit_params(&primary_id), None);
1257    }
1258
1259    #[rstest]
1260    fn test_twap_on_time_event_completes_when_primary_missing() {
1261        let mut algo = create_twap_algorithm();
1262        register_algorithm(&mut algo);
1263        add_instrument_to_cache(&algo);
1264
1265        let mut params = IndexMap::new();
1266        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1267        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1268
1269        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1270        let primary_id = order.client_order_id();
1271        algo.on_order(order).unwrap();
1272
1273        {
1274            let cache_rc = algo.core.cache_rc();
1275            let mut cache = cache_rc.borrow_mut();
1276            let primary = cache.order(&primary_id).map(|order| order.clone()).unwrap();
1277            let canceled = OrderCanceledSpec::builder()
1278                .trader_id(primary.trader_id())
1279                .strategy_id(primary.strategy_id())
1280                .instrument_id(primary.instrument_id())
1281                .client_order_id(primary.client_order_id())
1282                .build();
1283            cache
1284                .update_order(&OrderEventAny::Canceled(canceled))
1285                .unwrap();
1286            cache.purge_order(primary_id);
1287        }
1288        assert!(algo.cache().order(&primary_id).is_none());
1289
1290        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
1291        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
1292
1293        // A vanished primary is terminal: the schedule must not outlive it and block the ID
1294        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1295        assert!(algo.clock().timer_names().is_empty());
1296    }
1297
1298    #[rstest]
1299    fn test_twap_on_stop_cancels_timers() {
1300        let mut algo = create_twap_algorithm();
1301        register_algorithm(&mut algo);
1302
1303        add_instrument_to_cache(&algo);
1304
1305        let mut params = IndexMap::new();
1306        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1307        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1308
1309        let order = create_market_order_with_params(params);
1310        let primary_id = order.client_order_id();
1311
1312        algo.on_order(order).unwrap();
1313
1314        // Verify timer is set
1315        assert!(
1316            algo.clock()
1317                .timer_names()
1318                .iter()
1319                .any(|name| name.as_str() == primary_id.as_str())
1320        );
1321
1322        // Stop through the DataActor entry point the component lifecycle uses
1323        DataActor::on_stop(&mut algo).unwrap();
1324
1325        // Timer should be canceled
1326        assert!(algo.clock().timer_names().is_empty());
1327        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 3);
1328    }
1329
1330    #[rstest]
1331    fn test_twap_on_resume_rearms_timer_without_submitting() {
1332        let mut algo = create_twap_algorithm();
1333        register_algorithm(&mut algo);
1334        add_instrument_to_cache(&algo);
1335
1336        let mut params = IndexMap::new();
1337        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1338        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1339
1340        let order = create_market_order_with_params(params);
1341        let primary_id = order.client_order_id();
1342        algo.on_order(order).unwrap();
1343        let order_count = algo
1344            .cache()
1345            .orders_total_count(None, None, None, None, None);
1346
1347        Component::stop(&mut algo).unwrap();
1348
1349        // Observe the command bus directly: resubmitting the already-cached primary would
1350        // leave the order count unchanged, so the count alone cannot prove nothing was sent.
1351        let received = Rc::new(RefCell::new(None::<SubmitOrder>));
1352        let handler = msgbus::TypedIntoHandler::from({
1353            let captured = received.clone();
1354            move |cmd: TradingCommand| {
1355                if let TradingCommand::SubmitOrder(cmd) = cmd {
1356                    *captured.borrow_mut() = Some(cmd);
1357                }
1358            }
1359        });
1360        msgbus::register_trading_command_endpoint(
1361            MessagingSwitchboard::risk_engine_queue_execute(),
1362            handler,
1363        );
1364
1365        let resume_time = algo.clock().timestamp_ns();
1366        Component::resume(&mut algo).unwrap();
1367
1368        assert!(received.borrow().is_none());
1369        assert_eq!(algo.clock().timer_count(), 1);
1370        assert_eq!(
1371            algo.clock().next_time_ns(primary_id.as_str()),
1372            Some(resume_time + DurationNanos::from_secs(20))
1373        );
1374        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 3);
1375        assert_eq!(
1376            algo.cache()
1377                .orders_total_count(None, None, None, None, None),
1378            order_count
1379        );
1380    }
1381
1382    #[rstest]
1383    fn test_twap_on_resume_executes_remaining_slices() {
1384        let mut algo = create_twap_algorithm();
1385        let clock = register_algorithm_with_clock(&mut algo);
1386        add_instrument_to_cache(&algo);
1387
1388        let mut params = IndexMap::new();
1389        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1390        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1391
1392        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1393        let primary_id = order.client_order_id();
1394        algo.on_order(order).unwrap();
1395
1396        Component::stop(&mut algo).unwrap();
1397        Component::resume(&mut algo).unwrap();
1398        assert_eq!(algo.clock().timer_count(), 1);
1399
1400        let first_events = clock.borrow_mut().advance_time(20_000_000_000.into(), true);
1401        assert_eq!(first_events.len(), 1);
1402        algo.handle_time_event(&first_events[0]);
1403        assert_eq!(algo.scheduled_orders[&primary_id].remaining_sizes.len(), 1);
1404
1405        let final_events = clock.borrow_mut().advance_time(40_000_000_000.into(), true);
1406        assert_eq!(final_events.len(), 1);
1407        algo.handle_time_event(&final_events[0]);
1408        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1409    }
1410
1411    #[rstest]
1412    fn test_twap_on_resume_completes_closed_primary() {
1413        let mut algo = create_twap_algorithm();
1414        register_algorithm(&mut algo);
1415        add_instrument_to_cache(&algo);
1416
1417        let mut params = IndexMap::new();
1418        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1419        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1420
1421        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1422        let primary_id = order.client_order_id();
1423        algo.on_order(order).unwrap();
1424        Component::stop(&mut algo).unwrap();
1425
1426        {
1427            let cache_rc = algo.core.cache_rc();
1428            let mut cache = cache_rc.borrow_mut();
1429            let primary = cache.order(&primary_id).map(|order| order.clone()).unwrap();
1430            let canceled = OrderCanceledSpec::builder()
1431                .trader_id(primary.trader_id())
1432                .strategy_id(primary.strategy_id())
1433                .instrument_id(primary.instrument_id())
1434                .client_order_id(primary.client_order_id())
1435                .build();
1436            cache
1437                .update_order(&OrderEventAny::Canceled(canceled))
1438                .unwrap();
1439        }
1440
1441        Component::resume(&mut algo).unwrap();
1442
1443        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1444        assert_eq!(algo.clock().timer_count(), 0);
1445    }
1446
1447    #[rstest]
1448    fn test_twap_on_resume_completes_missing_primary() {
1449        let mut algo = create_twap_algorithm();
1450        register_algorithm(&mut algo);
1451        add_instrument_to_cache(&algo);
1452
1453        let mut params = IndexMap::new();
1454        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1455        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1456
1457        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1458        let primary_id = order.client_order_id();
1459        algo.on_order(order).unwrap();
1460        Component::stop(&mut algo).unwrap();
1461
1462        {
1463            let cache_rc = algo.core.cache_rc();
1464            let mut cache = cache_rc.borrow_mut();
1465            let primary = cache.order(&primary_id).map(|order| order.clone()).unwrap();
1466            let canceled = OrderCanceledSpec::builder()
1467                .trader_id(primary.trader_id())
1468                .strategy_id(primary.strategy_id())
1469                .instrument_id(primary.instrument_id())
1470                .client_order_id(primary.client_order_id())
1471                .build();
1472            cache
1473                .update_order(&OrderEventAny::Canceled(canceled))
1474                .unwrap();
1475            cache.purge_order(primary_id);
1476        }
1477
1478        assert!(algo.cache().order(&primary_id).is_none());
1479        Component::resume(&mut algo).unwrap();
1480
1481        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1482        assert_eq!(algo.clock().timer_count(), 0);
1483    }
1484
1485    #[rstest]
1486    fn test_twap_fractional_interval_secs() {
1487        let mut algo = create_twap_algorithm();
1488        register_algorithm(&mut algo);
1489
1490        add_instrument_to_cache(&algo);
1491
1492        // Use fractional interval like Python tests: 3 second horizon, 0.5 second interval
1493        let mut params = IndexMap::new();
1494        params.insert(Ustr::from("horizon_secs"), Ustr::from("3"));
1495        params.insert(Ustr::from("interval_secs"), Ustr::from("0.5"));
1496
1497        let order = create_market_order_with_params(params);
1498        let primary_id = order.client_order_id();
1499
1500        // Should not error - fractional seconds should parse correctly
1501        algo.on_order(order).unwrap();
1502
1503        // 3 / 0.5 = 6 intervals, first spawned immediately, 5 remaining (plus possible remainder)
1504        let remaining = &algo
1505            .scheduled_orders
1506            .get(&primary_id)
1507            .unwrap()
1508            .remaining_sizes;
1509        assert!(remaining.len() >= 5);
1510    }
1511
1512    #[rstest]
1513    fn test_twap_submits_entire_size_when_qty_per_interval_below_size_increment() {
1514        use nautilus_model::instruments::{InstrumentAny, stubs::equity_aapl};
1515
1516        let mut algo = create_twap_algorithm();
1517        register_algorithm(&mut algo);
1518
1519        // Use equity with size_increment of 1 (whole shares only)
1520        let instrument = equity_aapl();
1521        let instrument_id = instrument.id();
1522        {
1523            let cache_rc = algo.core.cache_rc();
1524            let mut cache = cache_rc.borrow_mut();
1525            cache
1526                .add_instrument(InstrumentAny::Equity(instrument))
1527                .unwrap();
1528        }
1529
1530        // 2 shares over 60s with 10s intervals = 6 intervals
1531        // 2 / 6 = 0.333... which is less than size_increment of 1
1532        let mut params = IndexMap::new();
1533        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1534        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
1535
1536        let client_order_id = ClientOrderId::from("O-002");
1537        let order = OrderAny::Market(MarketOrder::new(
1538            TraderId::from("TRADER-001"),
1539            StrategyId::from("STRAT-001"),
1540            instrument_id,
1541            client_order_id,
1542            OrderSide::Buy,
1543            Quantity::from("2"),
1544            TimeInForce::Gtc,
1545            UUID4::new(),
1546            0.into(),
1547            false,
1548            false,
1549            None,
1550            None,
1551            None,
1552            None,
1553            Some(ExecAlgorithmId::new("TWAP")),
1554            Some(params),
1555            Some(client_order_id),
1556            None,
1557        ));
1558
1559        let primary_id = order.client_order_id();
1560
1561        let submit_params = Params::new(
1562            [(
1563                "routing_profile".to_string(),
1564                serde_json::Value::String("whole-size".to_string()),
1565            )]
1566            .into(),
1567        );
1568        algo.core
1569            .remember_submit_params(primary_id, Some(submit_params));
1570        algo.on_order(order).unwrap();
1571
1572        // Should submit entire size directly (no scheduling)
1573        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1574        assert_eq!(algo.core.submit_params(&primary_id), None);
1575    }
1576
1577    #[rstest]
1578    fn test_twap_submits_entire_size_when_qty_per_interval_below_min_quantity() {
1579        use nautilus_model::instruments::{InstrumentAny, stubs::crypto_perpetual_ethusdt};
1580
1581        let mut algo = create_twap_algorithm();
1582        register_algorithm(&mut algo);
1583
1584        let mut instrument = crypto_perpetual_ethusdt();
1585        instrument.min_quantity = Some(Quantity::from("0.5"));
1586        {
1587            let cache_rc = algo.core.cache_rc();
1588            let mut cache = cache_rc.borrow_mut();
1589            cache
1590                .add_instrument(InstrumentAny::CryptoPerpetual(instrument))
1591                .unwrap();
1592        }
1593
1594        let mut params = IndexMap::new();
1595        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1596        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
1597
1598        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
1599        let primary_id = order.client_order_id();
1600
1601        let submit_params = Params::new(
1602            [(
1603                "routing_profile".to_string(),
1604                serde_json::Value::String("minimum-quantity".to_string()),
1605            )]
1606            .into(),
1607        );
1608        algo.core
1609            .remember_submit_params(primary_id, Some(submit_params));
1610
1611        algo.on_order(order).unwrap();
1612
1613        assert!(algo.scheduled_orders.get(&primary_id).is_none());
1614        assert_eq!(algo.core.submit_params(&primary_id), None);
1615    }
1616
1617    #[rstest]
1618    fn test_twap_rejects_negative_interval_secs() {
1619        let mut algo = create_twap_algorithm();
1620        register_algorithm(&mut algo);
1621
1622        add_instrument_to_cache(&algo);
1623
1624        let mut params = IndexMap::new();
1625        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1626        params.insert(Ustr::from("interval_secs"), Ustr::from("-0.5"));
1627
1628        let order = create_market_order_with_params(params);
1629        assert_twap_denied(
1630            &mut algo,
1631            &order,
1632            "VALIDATION_FAILED: interval_secs=-0.5 must be finite and positive",
1633        );
1634    }
1635
1636    #[rstest]
1637    fn test_twap_rejects_negative_horizon_secs() {
1638        let mut algo = create_twap_algorithm();
1639        register_algorithm(&mut algo);
1640
1641        add_instrument_to_cache(&algo);
1642
1643        let mut params = IndexMap::new();
1644        params.insert(Ustr::from("horizon_secs"), Ustr::from("-10"));
1645        params.insert(Ustr::from("interval_secs"), Ustr::from("1"));
1646
1647        let order = create_market_order_with_params(params);
1648        assert_twap_denied(
1649            &mut algo,
1650            &order,
1651            "VALIDATION_FAILED: horizon_secs=-10 must be finite and positive",
1652        );
1653    }
1654
1655    #[rstest]
1656    fn test_twap_rejects_zero_interval_secs() {
1657        let mut algo = create_twap_algorithm();
1658        register_algorithm(&mut algo);
1659
1660        add_instrument_to_cache(&algo);
1661
1662        let mut params = IndexMap::new();
1663        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1664        params.insert(Ustr::from("interval_secs"), Ustr::from("0"));
1665
1666        let order = create_market_order_with_params(params);
1667        assert_twap_denied(
1668            &mut algo,
1669            &order,
1670            "VALIDATION_FAILED: interval_secs=0 must be finite and positive",
1671        );
1672    }
1673
1674    #[rstest]
1675    fn test_twap_rejects_huge_finite_interval_before_submission() {
1676        let mut algo = create_twap_algorithm();
1677        register_algorithm(&mut algo);
1678
1679        add_instrument_to_cache(&algo);
1680
1681        let received = Rc::new(RefCell::new(None::<SubmitOrder>));
1682        let handler = msgbus::TypedIntoHandler::from({
1683            let captured = received.clone();
1684            move |cmd: TradingCommand| {
1685                if let TradingCommand::SubmitOrder(cmd) = cmd {
1686                    *captured.borrow_mut() = Some(cmd);
1687                }
1688            }
1689        });
1690        msgbus::register_trading_command_endpoint(
1691            MessagingSwitchboard::risk_engine_queue_execute(),
1692            handler,
1693        );
1694
1695        let mut params = IndexMap::new();
1696        params.insert(Ustr::from("horizon_secs"), Ustr::from("2e20"));
1697        params.insert(Ustr::from("interval_secs"), Ustr::from("1e20"));
1698
1699        let duration_error = Duration::try_from_secs_f64(1e20).unwrap_err();
1700        let reason = OrderDeniedReason::ValidationFailed {
1701            detail: format!(
1702                "interval_secs=100000000000000000000 is not a valid duration: {duration_error}"
1703            ),
1704        }
1705        .to_string();
1706        assert_twap_denied(&mut algo, &create_market_order_with_params(params), &reason);
1707
1708        assert!(received.borrow().is_none());
1709    }
1710
1711    #[rstest]
1712    fn test_twap_rejects_subnanosecond_interval_before_submission() {
1713        let mut algo = create_twap_algorithm();
1714        register_algorithm(&mut algo);
1715
1716        add_instrument_to_cache(&algo);
1717
1718        let received = Rc::new(RefCell::new(None::<SubmitOrder>));
1719        let handler = msgbus::TypedIntoHandler::from({
1720            let captured = received.clone();
1721            move |cmd: TradingCommand| {
1722                if let TradingCommand::SubmitOrder(cmd) = cmd {
1723                    *captured.borrow_mut() = Some(cmd);
1724                }
1725            }
1726        });
1727        msgbus::register_trading_command_endpoint(
1728            MessagingSwitchboard::risk_engine_queue_execute(),
1729            handler,
1730        );
1731
1732        let mut params = IndexMap::new();
1733        params.insert(Ustr::from("horizon_secs"), Ustr::from("2e-10"));
1734        params.insert(Ustr::from("interval_secs"), Ustr::from("1e-10"));
1735
1736        assert_twap_denied(
1737            &mut algo,
1738            &create_market_order_with_params(params),
1739            "VALIDATION_FAILED: interval_secs=0.0000000001 rounds to a zero duration",
1740        );
1741
1742        assert!(received.borrow().is_none());
1743    }
1744
1745    #[rstest]
1746    fn test_twap_rejects_interval_exceeding_timestamp_headroom_before_submission() {
1747        let mut algo = create_twap_algorithm();
1748        register_algorithm(&mut algo);
1749
1750        add_instrument_to_cache(&algo);
1751        DataActorNative::clock_mut(&mut algo)
1752            .as_any_mut()
1753            .downcast_mut::<VirtualClock>()
1754            .unwrap()
1755            .set_time(UnixNanos::new(u64::MAX - 500_000_000));
1756
1757        let received = Rc::new(RefCell::new(None::<SubmitOrder>));
1758        let handler = msgbus::TypedIntoHandler::from({
1759            let captured = received.clone();
1760            move |cmd: TradingCommand| {
1761                if let TradingCommand::SubmitOrder(cmd) = cmd {
1762                    *captured.borrow_mut() = Some(cmd);
1763                }
1764            }
1765        });
1766        msgbus::register_trading_command_endpoint(
1767            MessagingSwitchboard::risk_engine_queue_execute(),
1768            handler,
1769        );
1770
1771        let mut params = IndexMap::new();
1772        params.insert(Ustr::from("horizon_secs"), Ustr::from("2"));
1773        params.insert(Ustr::from("interval_secs"), Ustr::from("1"));
1774
1775        assert_twap_denied(
1776            &mut algo,
1777            &create_market_order_with_params(params),
1778            "VALIDATION_FAILED: interval_secs=1 exceeds the clock timestamp headroom",
1779        );
1780
1781        assert!(received.borrow().is_none());
1782    }
1783
1784    #[rstest]
1785    fn test_twap_rejects_nan_interval_secs() {
1786        let mut algo = create_twap_algorithm();
1787        register_algorithm(&mut algo);
1788
1789        add_instrument_to_cache(&algo);
1790
1791        let mut params = IndexMap::new();
1792        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1793        params.insert(Ustr::from("interval_secs"), Ustr::from("NaN"));
1794
1795        let order = create_market_order_with_params(params);
1796        assert_twap_denied(
1797            &mut algo,
1798            &order,
1799            "VALIDATION_FAILED: interval_secs=NaN must be finite and positive",
1800        );
1801    }
1802
1803    #[rstest]
1804    fn test_twap_rejects_infinity_horizon_secs() {
1805        let mut algo = create_twap_algorithm();
1806        register_algorithm(&mut algo);
1807
1808        add_instrument_to_cache(&algo);
1809
1810        let mut params = IndexMap::new();
1811        params.insert(Ustr::from("horizon_secs"), Ustr::from("inf"));
1812        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
1813
1814        let order = create_market_order_with_params(params);
1815        assert_twap_denied(
1816            &mut algo,
1817            &order,
1818            "VALIDATION_FAILED: horizon_secs=inf must be finite and positive",
1819        );
1820    }
1821}
nautilus_trading/algorithm/twap.rs

nautilus_trading/algorithm/
twap.rs
