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    identifiers::ClientOrderId,
43    instruments::Instrument,
44    orders::{Order, OrderAny},
45    types::{Quantity, quantity::QuantityRaw},
46};
47use ustr::Ustr;
48
49use super::{
50    ExecutionAlgorithm, ExecutionAlgorithmConfig, ExecutionAlgorithmCore, ExecutionAlgorithmNative,
51};
52use crate::nautilus_execution_algorithm;
53
54/// Configuration for [`TwapAlgorithm`].
55pub type TwapAlgorithmConfig = ExecutionAlgorithmConfig;
56
57/// Time-Weighted Average Price (TWAP) execution algorithm.
58///
59/// Executes orders by evenly spreading them over a specified time horizon,
60/// at regular intervals. The algorithm receives a primary order and spawns
61/// smaller child orders that are executed at regular intervals.
62#[derive(Debug)]
63pub struct TwapAlgorithm {
64    /// The algorithm core.
65    pub core: ExecutionAlgorithmCore,
66    /// Scheduled sizes for each primary order.
67    scheduled_sizes: AHashMap<ClientOrderId, Vec<Quantity>>,
68}
69
70impl TwapAlgorithm {
71    /// Creates a new [`TwapAlgorithm`] instance.
72    #[must_use]
73    pub fn new(config: TwapAlgorithmConfig) -> Self {
74        Self {
75            core: ExecutionAlgorithmCore::new(config),
76            scheduled_sizes: AHashMap::new(),
77        }
78    }
79
80    /// Completes the execution sequence for a primary order.
81    fn complete_sequence(&mut self, primary_id: ClientOrderId) {
82        let timer_name = primary_id.as_str();
83        let core = ExecutionAlgorithmNative::exec_algorithm_core_mut(self);
84        if core.clock_mut().timer_names().contains(&timer_name) {
85            core.clock_mut().cancel_timer(timer_name);
86        }
87        self.scheduled_sizes.remove(&primary_id);
88        log::info!("Completed TWAP execution for {primary_id}");
89    }
90}
91
92// The clock and component lifecycle dispatch through the `DataActor` hooks,
93// so forward them to the `ExecutionAlgorithm` implementations.
94impl DataActor for TwapAlgorithm {
95    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
96        ExecutionAlgorithm::on_time_event(self, event)
97    }
98
99    fn on_stop(&mut self) -> anyhow::Result<()> {
100        ExecutionAlgorithm::on_stop(self)
101    }
102
103    fn on_reset(&mut self) -> anyhow::Result<()> {
104        ExecutionAlgorithm::on_reset(self)
105    }
106}
107
108nautilus_execution_algorithm!(TwapAlgorithm, {
109    fn on_order(&mut self, order: OrderAny) -> anyhow::Result<()> {
110        let primary_id = order.client_order_id();
111
112        if self.scheduled_sizes.contains_key(&primary_id) {
113            anyhow::bail!("Order {primary_id} already being executed");
114        }
115
116        log::info!("Received order for TWAP execution: {order:?}");
117
118        // Only market orders supported
119        if order.order_type() != OrderType::Market {
120            log::error!(
121                "Cannot execute order: only implemented for market orders, order_type={:?}",
122                order.order_type()
123            );
124            return Ok(());
125        }
126
127        let instrument = {
128            let cache = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_ref();
129            cache.instrument(&order.instrument_id()).cloned()
130        };
131
132        let Some(instrument) = instrument else {
133            log::error!(
134                "Cannot execute order: instrument {} not found",
135                order.instrument_id()
136            );
137            return Ok(());
138        };
139
140        let Some(exec_params) = order.exec_algorithm_params() else {
141            log::error!(
142                "Cannot execute order: exec_algorithm_params not found for primary order {primary_id}"
143            );
144            return Ok(());
145        };
146
147        let Some(horizon_secs_str) = exec_params.get(&Ustr::from("horizon_secs")) else {
148            log::error!("Cannot execute order: horizon_secs not found in exec_algorithm_params");
149            return Ok(());
150        };
151
152        let horizon_secs: f64 = horizon_secs_str.parse().map_err(|e| {
153            log::error!("Cannot parse horizon_secs: {e}");
154            anyhow::anyhow!("Invalid horizon_secs")
155        })?;
156
157        let Some(interval_secs_str) = exec_params.get(&Ustr::from("interval_secs")) else {
158            log::error!("Cannot execute order: interval_secs not found in exec_algorithm_params");
159            return Ok(());
160        };
161
162        let interval_secs: f64 = interval_secs_str.parse().map_err(|e| {
163            log::error!("Cannot parse interval_secs: {e}");
164            anyhow::anyhow!("Invalid interval_secs")
165        })?;
166
167        if !horizon_secs.is_finite() || horizon_secs <= 0.0 {
168            log::error!(
169                "Cannot execute order: horizon_secs={horizon_secs} must be finite and positive"
170            );
171            return Ok(());
172        }
173
174        if !interval_secs.is_finite() || interval_secs <= 0.0 {
175            log::error!(
176                "Cannot execute order: interval_secs={interval_secs} must be finite and positive"
177            );
178            return Ok(());
179        }
180
181        if horizon_secs < interval_secs {
182            log::error!(
183                "Cannot execute order: horizon_secs={horizon_secs} was less than interval_secs={interval_secs}"
184            );
185            return Ok(());
186        }
187
188        let num_intervals = (horizon_secs / interval_secs).floor() as u64;
189        if num_intervals == 0 {
190            log::error!("Cannot execute order: num_intervals is 0");
191            return Ok(());
192        }
193
194        let total_qty = order.quantity();
195        let total_raw = total_qty.raw;
196        let precision = total_qty.precision;
197
198        let qty_per_interval_raw = total_raw / (num_intervals as QuantityRaw);
199        let qty_per_interval = Quantity::from_raw(qty_per_interval_raw, precision);
200
201        if qty_per_interval == total_qty || qty_per_interval < instrument.size_increment() {
202            log::warn!(
203                "Submitting for entire size: qty_per_interval={qty_per_interval}, order_quantity={total_qty}"
204            );
205            self.submit_order(order, None, None)?;
206            return Ok(());
207        }
208
209        if let Some(min_qty) = instrument.min_quantity()
210            && qty_per_interval < min_qty
211        {
212            log::warn!(
213                "Submitting for entire size: qty_per_interval={qty_per_interval} < min_quantity={min_qty}"
214            );
215            self.submit_order(order, None, None)?;
216            return Ok(());
217        }
218
219        let mut scheduled_sizes: Vec<Quantity> = vec![qty_per_interval; num_intervals as usize];
220
221        // Remainder goes in the last slice
222        let scheduled_total = qty_per_interval_raw * (num_intervals as QuantityRaw);
223        let remainder_raw = total_raw - scheduled_total;
224        if remainder_raw > 0 {
225            let remainder = Quantity::from_raw(remainder_raw, total_qty.precision);
226            scheduled_sizes.push(remainder);
227        }
228
229        log::info!("Order execution size schedule: {scheduled_sizes:?}");
230
231        // Add primary order to cache so on_time_event can retrieve it,
232        // it is already present when routed through the engine's submit path.
233        {
234            let cache_rc = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_rc();
235            let mut cache = cache_rc.borrow_mut();
236            if !cache.order_exists(&primary_id) {
237                cache.add_order(order.clone(), None, None, false)?;
238            }
239        }
240
241        self.scheduled_sizes
242            .insert(primary_id, scheduled_sizes.clone());
243
244        let first_qty = self.scheduled_sizes.get_mut(&primary_id).unwrap().remove(0);
245        let is_single_slice = self
246            .scheduled_sizes
247            .get(&primary_id)
248            .is_some_and(|s| s.is_empty());
249
250        // Single slice: submit the primary order directly
251        if is_single_slice {
252            self.submit_order(order, None, None)?;
253            self.complete_sequence(primary_id);
254            return Ok(());
255        }
256
257        // Multiple slices: spawn first child order and reduce primary
258        let tags = order.tags().map(|t| t.to_vec());
259        let time_in_force = order.time_in_force();
260        let reduce_only = order.is_reduce_only();
261        let mut order = order;
262        let spawned = self.spawn_market(
263            &mut order,
264            first_qty,
265            time_in_force,
266            reduce_only,
267            tags,
268            true,
269        );
270        self.submit_order(spawned.into(), None, None)?;
271
272        ExecutionAlgorithmNative::exec_algorithm_core_mut(self)
273            .clock_mut()
274            .set_timer(
275                primary_id.as_str(),
276                Duration::from_secs_f64(interval_secs),
277                None,
278                None,
279                None,
280                None,
281                None,
282            )?;
283
284        log::info!(
285            "Started TWAP execution for {primary_id}: horizon_secs={horizon_secs}, interval_secs={interval_secs}"
286        );
287
288        Ok(())
289    }
290
291    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
292        log::info!("Received time event: {event:?}");
293
294        let primary_id = ClientOrderId::new(event.name.as_str());
295
296        let primary = {
297            let cache = ExecutionAlgorithmNative::exec_algorithm_core(self).cache_ref();
298            cache.order(&primary_id).map(|o| o.clone())
299        };
300
301        let Some(primary) = primary else {
302            log::error!("Cannot find primary order for exec_spawn_id={primary_id}");
303            return Ok(());
304        };
305
306        if primary.is_closed() {
307            self.complete_sequence(primary_id);
308            return Ok(());
309        }
310
311        let Some(scheduled_sizes) = self.scheduled_sizes.get_mut(&primary_id) else {
312            log::error!("Cannot find scheduled sizes for exec_spawn_id={primary_id}");
313            return Ok(());
314        };
315
316        if scheduled_sizes.is_empty() {
317            log::warn!("No more size to execute for exec_spawn_id={primary_id}");
318            return Ok(());
319        }
320
321        let quantity = scheduled_sizes.remove(0);
322        let is_final_slice = scheduled_sizes.is_empty();
323
324        // Final slice: submit the primary order (already reduced to remaining quantity)
325        if is_final_slice {
326            self.submit_order(primary, None, None)?;
327            self.complete_sequence(primary_id);
328            return Ok(());
329        }
330
331        // Intermediate slice: spawn child order and reduce primary
332        let tags = primary.tags().map(|t| t.to_vec());
333        let time_in_force = primary.time_in_force();
334        let reduce_only = primary.is_reduce_only();
335        let mut primary = primary;
336        let spawned = self.spawn_market(
337            &mut primary,
338            quantity,
339            time_in_force,
340            reduce_only,
341            tags,
342            true,
343        );
344        self.submit_order(spawned.into(), None, None)?;
345
346        Ok(())
347    }
348
349    fn on_stop(&mut self) -> anyhow::Result<()> {
350        ExecutionAlgorithmNative::exec_algorithm_core_mut(self)
351            .clock_mut()
352            .cancel_timers();
353        Ok(())
354    }
355
356    fn on_reset(&mut self) -> anyhow::Result<()> {
357        self.unsubscribe_all_strategy_events();
358        ExecutionAlgorithmNative::exec_algorithm_core_mut(self).reset();
359        self.scheduled_sizes.clear();
360        Ok(())
361    }
362});
363
364#[cfg(test)]
365mod tests {
366    use std::{cell::RefCell, rc::Rc};
367
368    use indexmap::IndexMap;
369    use nautilus_common::{
370        cache::Cache,
371        clock::{Clock, TestClock},
372        component::Component,
373        enums::ComponentTrigger,
374    };
375    use nautilus_core::UUID4;
376    use nautilus_model::{
377        enums::{OrderSide, TimeInForce},
378        events::{OrderEventAny, order::spec::OrderCanceledSpec},
379        identifiers::{ExecAlgorithmId, InstrumentId, StrategyId, TraderId},
380        orders::{LimitOrder, MarketOrder},
381        types::Price,
382    };
383    use rstest::rstest;
384    use ustr::Ustr;
385
386    use super::*;
387
388    fn create_twap_algorithm() -> TwapAlgorithm {
389        // Use unique ID to avoid thread-local registry/msgbus conflicts in parallel tests
390        let unique_id = format!("TWAP-{}", UUID4::new());
391        let config = TwapAlgorithmConfig {
392            exec_algorithm_id: Some(ExecAlgorithmId::new(&unique_id)),
393            ..Default::default()
394        };
395        TwapAlgorithm::new(config)
396    }
397
398    fn register_algorithm(algo: &mut TwapAlgorithm) {
399        use nautilus_common::timer::TimeEventCallback;
400
401        let trader_id = TraderId::from("TRADER-001");
402        let clock = Rc::new(RefCell::new(TestClock::new()));
403        let cache = Rc::new(RefCell::new(Cache::default()));
404
405        // Register a no-op default handler for timer callbacks
406        clock
407            .borrow_mut()
408            .register_default_handler(TimeEventCallback::Rust(std::sync::Arc::new(|_| {})));
409
410        algo.core.register(trader_id, clock, cache).unwrap();
411
412        // Transition to Running state for tests
413        algo.transition_state(ComponentTrigger::Initialize).unwrap();
414        algo.transition_state(ComponentTrigger::Start).unwrap();
415        algo.transition_state(ComponentTrigger::StartCompleted)
416            .unwrap();
417    }
418
419    fn add_instrument_to_cache(algo: &TwapAlgorithm) {
420        use nautilus_model::instruments::{InstrumentAny, stubs::crypto_perpetual_ethusdt};
421
422        let instrument = crypto_perpetual_ethusdt();
423        let cache_rc = algo.core.cache_rc();
424        let mut cache = cache_rc.borrow_mut();
425        cache
426            .add_instrument(InstrumentAny::CryptoPerpetual(instrument))
427            .unwrap();
428    }
429
430    fn create_market_order_with_params(params: IndexMap<Ustr, Ustr>) -> OrderAny {
431        create_market_order_with_params_and_qty(params, Quantity::from("1.0"))
432    }
433
434    fn create_market_order_with_params_and_qty(
435        params: IndexMap<Ustr, Ustr>,
436        quantity: Quantity,
437    ) -> OrderAny {
438        OrderAny::Market(MarketOrder::new(
439            TraderId::from("TRADER-001"),
440            StrategyId::from("STRAT-001"),
441            InstrumentId::from("ETHUSDT-PERP.BINANCE"),
442            ClientOrderId::from("O-001"),
443            OrderSide::Buy,
444            quantity,
445            TimeInForce::Gtc,
446            UUID4::new(),
447            0.into(),
448            false,
449            false,
450            None,
451            None,
452            None,
453            None,
454            Some(ExecAlgorithmId::new("TWAP")),
455            Some(params),
456            None,
457            None,
458        ))
459    }
460
461    #[rstest]
462    fn test_twap_creation() {
463        let algo = create_twap_algorithm();
464        assert!(algo.id().inner().starts_with("TWAP"));
465        assert!(algo.scheduled_sizes.is_empty());
466    }
467
468    #[rstest]
469    fn test_twap_registration() {
470        let mut algo = create_twap_algorithm();
471        register_algorithm(&mut algo);
472
473        assert_eq!(algo.trader_id(), Some(TraderId::from("TRADER-001")));
474    }
475
476    #[rstest]
477    fn test_twap_reset_clears_scheduled_sizes() {
478        let mut algo = create_twap_algorithm();
479        let primary_id = ClientOrderId::new("O-001");
480
481        algo.scheduled_sizes
482            .insert(primary_id, vec![Quantity::from("1.0")]);
483
484        assert!(!algo.scheduled_sizes.is_empty());
485
486        // Dispatch through the DataActor entry point the component lifecycle uses
487        DataActor::on_reset(&mut algo).unwrap();
488
489        assert!(algo.scheduled_sizes.is_empty());
490    }
491
492    #[rstest]
493    fn test_twap_rejects_non_market_orders() {
494        let mut algo = create_twap_algorithm();
495        register_algorithm(&mut algo);
496
497        let order = OrderAny::Limit(LimitOrder::new(
498            TraderId::from("TRADER-001"),
499            StrategyId::from("STRAT-001"),
500            InstrumentId::from("BTC/USDT.BINANCE"),
501            ClientOrderId::from("O-001"),
502            OrderSide::Buy,
503            Quantity::from("1.0"),
504            Price::from("50000.0"),
505            TimeInForce::Gtc,
506            None,  // expire_time
507            false, // post_only
508            false, // reduce_only
509            false, // quote_quantity
510            None,  // display_qty
511            None,  // emulation_trigger
512            None,  // trigger_instrument_id
513            None,  // contingency_type
514            None,  // order_list_id
515            None,  // linked_order_ids
516            None,  // parent_order_id
517            None,  // exec_algorithm_id
518            None,  // exec_algorithm_params
519            None,  // exec_spawn_id
520            None,  // tags
521            UUID4::new(),
522            0.into(),
523        ));
524
525        // Should not error, just log and return
526        let result = algo.on_order(order);
527        assert!(result.is_ok());
528    }
529
530    #[rstest]
531    fn test_twap_rejects_missing_params() {
532        let mut algo = create_twap_algorithm();
533        register_algorithm(&mut algo);
534
535        let order = OrderAny::Market(MarketOrder::new(
536            TraderId::from("TRADER-001"),
537            StrategyId::from("STRAT-001"),
538            InstrumentId::from("BTC/USDT.BINANCE"),
539            ClientOrderId::from("O-001"),
540            OrderSide::Buy,
541            Quantity::from("1.0"),
542            TimeInForce::Gtc,
543            UUID4::new(),
544            0.into(),
545            false,
546            false,
547            None,
548            None,
549            None,
550            None,
551            None,
552            None, // No exec_algorithm_params
553            None,
554            None,
555        ));
556
557        // Should not error, just log and return
558        let result = algo.on_order(order);
559        assert!(result.is_ok());
560    }
561
562    #[rstest]
563    fn test_twap_rejects_horizon_less_than_interval() {
564        let mut algo = create_twap_algorithm();
565        register_algorithm(&mut algo);
566
567        add_instrument_to_cache(&algo);
568
569        let mut params = IndexMap::new();
570        params.insert(Ustr::from("horizon_secs"), Ustr::from("30"));
571        params.insert(Ustr::from("interval_secs"), Ustr::from("60"));
572
573        let order = create_market_order_with_params(params);
574        let result = algo.on_order(order);
575
576        assert!(result.is_ok());
577        assert!(algo.scheduled_sizes.is_empty());
578    }
579
580    #[rstest]
581    fn test_twap_rejects_duplicate_order() {
582        let mut algo = create_twap_algorithm();
583        register_algorithm(&mut algo);
584
585        add_instrument_to_cache(&algo);
586
587        let mut params = IndexMap::new();
588        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
589        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
590
591        let order1 = create_market_order_with_params(params.clone());
592        let order2 = create_market_order_with_params(params);
593
594        algo.on_order(order1).unwrap();
595        let result = algo.on_order(order2);
596
597        assert!(result.is_err());
598        assert!(
599            result
600                .unwrap_err()
601                .to_string()
602                .contains("already being executed")
603        );
604    }
605
606    #[rstest]
607    fn test_twap_calculates_size_schedule_evenly() {
608        let mut algo = create_twap_algorithm();
609        register_algorithm(&mut algo);
610
611        add_instrument_to_cache(&algo);
612
613        // 1.2 qty over 60s with 20s intervals = 3 intervals of 0.4 each (divides evenly)
614        let mut params = IndexMap::new();
615        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
616        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
617
618        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
619        let primary_id = order.client_order_id();
620
621        algo.on_order(order).unwrap();
622
623        // First slice spawned immediately, remaining 2 slices scheduled (no remainder)
624        let remaining = algo.scheduled_sizes.get(&primary_id).unwrap();
625        assert_eq!(remaining.len(), 2);
626
627        for qty in remaining {
628            assert_eq!(*qty, Quantity::from("0.4"));
629        }
630    }
631
632    #[rstest]
633    fn test_twap_reduces_cached_primary_after_first_child_spawn() {
634        let mut algo = create_twap_algorithm();
635        register_algorithm(&mut algo);
636
637        add_instrument_to_cache(&algo);
638
639        let mut params = IndexMap::new();
640        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
641        params.insert(Ustr::from("interval_secs"), Ustr::from("30"));
642
643        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
644        let primary_id = order.client_order_id();
645
646        algo.on_order(order).unwrap();
647
648        let cache = algo.cache();
649        let primary = cache.order(&primary_id).unwrap();
650        let spawned = cache.order(&ClientOrderId::from("O-001-E1")).unwrap();
651
652        assert_eq!(primary.quantity(), Quantity::from("0.6"));
653        assert_eq!(spawned.quantity(), Quantity::from("0.6"));
654        assert_eq!(spawned.exec_spawn_id(), Some(primary_id));
655    }
656
657    #[rstest]
658    fn test_twap_on_order_accepts_already_cached_primary() {
659        let mut algo = create_twap_algorithm();
660        register_algorithm(&mut algo);
661
662        add_instrument_to_cache(&algo);
663
664        let mut params = IndexMap::new();
665        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
666        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
667
668        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
669        let primary_id = order.client_order_id();
670
671        // The engine submit path caches the primary before routing to the algorithm
672        {
673            let cache_rc = algo.core.cache_rc();
674            let mut cache = cache_rc.borrow_mut();
675            cache.add_order(order.clone(), None, None, false).unwrap();
676        }
677
678        algo.on_order(order).unwrap();
679
680        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 2);
681    }
682
683    #[rstest]
684    fn test_twap_calculates_size_schedule_with_remainder() {
685        let mut algo = create_twap_algorithm();
686        register_algorithm(&mut algo);
687
688        add_instrument_to_cache(&algo);
689
690        // 1.0 qty over 60s with 20s intervals = 3 intervals
691        // Raw is scaled to FIXED_PRECISION: 9 (standard) or 16 (high-precision)
692        let mut params = IndexMap::new();
693        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
694        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
695
696        let order = create_market_order_with_params(params);
697        let primary_id = order.client_order_id();
698
699        algo.on_order(order).unwrap();
700
701        // First slice spawned, 3 remaining (2 regular + 1 remainder)
702        let remaining = algo.scheduled_sizes.get(&primary_id).unwrap();
703        assert_eq!(remaining.len(), 3);
704
705        // Expected raw values depend on FIXED_PRECISION
706        // Standard (9):  1_000_000_000 / 3 = 333_333_333, remainder = 1
707        // High (16): 10_000_000_000_000_000 / 3 = 3_333_333_333_333_333, remainder = 1
708        #[cfg(feature = "high-precision")]
709        {
710            assert_eq!(remaining[0].raw, 3_333_333_333_333_333);
711            assert_eq!(remaining[1].raw, 3_333_333_333_333_333);
712            assert_eq!(remaining[2].raw, 1);
713        }
714        #[cfg(not(feature = "high-precision"))]
715        {
716            assert_eq!(remaining[0].raw, 333_333_333);
717            assert_eq!(remaining[1].raw, 333_333_333);
718            assert_eq!(remaining[2].raw, 1);
719        }
720    }
721
722    #[rstest]
723    fn test_twap_on_time_event_spawns_next_slice() {
724        let mut algo = create_twap_algorithm();
725        register_algorithm(&mut algo);
726
727        add_instrument_to_cache(&algo);
728
729        // Use qty that divides evenly: 1.2 / 3 = 0.4 each
730        let mut params = IndexMap::new();
731        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
732        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
733
734        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
735        let primary_id = order.client_order_id();
736
737        algo.on_order(order).unwrap();
738
739        // Verify 2 slices remain after first spawn (no remainder)
740        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 2);
741
742        // Simulate timer firing
743        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
744        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
745
746        // One slice consumed
747        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 1);
748    }
749
750    #[rstest]
751    fn test_twap_data_actor_dispatch_spawns_next_slice() {
752        let mut algo = create_twap_algorithm();
753        register_algorithm(&mut algo);
754
755        add_instrument_to_cache(&algo);
756
757        let mut params = IndexMap::new();
758        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
759        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
760
761        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
762        let primary_id = order.client_order_id();
763
764        algo.on_order(order).unwrap();
765        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 2);
766
767        // Dispatch through the DataActor entry point the clock callback uses
768        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
769        algo.handle_time_event(&event);
770
771        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 1);
772    }
773
774    #[rstest]
775    fn test_twap_on_time_event_completes_on_final_slice() {
776        let mut algo = create_twap_algorithm();
777        register_algorithm(&mut algo);
778
779        add_instrument_to_cache(&algo);
780
781        // 2 intervals: first spawned immediately, one in scheduled_sizes
782        let mut params = IndexMap::new();
783        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
784        params.insert(Ustr::from("interval_secs"), Ustr::from("30"));
785
786        let order = create_market_order_with_params(params);
787        let primary_id = order.client_order_id();
788
789        algo.on_order(order).unwrap();
790        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 1);
791
792        // Simulate timer firing for final slice
793        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
794        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
795
796        // Sequence completed, scheduled_sizes removed
797        assert!(algo.scheduled_sizes.get(&primary_id).is_none());
798    }
799
800    #[rstest]
801    fn test_twap_on_time_event_completes_when_primary_closed() {
802        let mut algo = create_twap_algorithm();
803        register_algorithm(&mut algo);
804
805        add_instrument_to_cache(&algo);
806
807        let mut params = IndexMap::new();
808        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
809        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
810
811        let order = create_market_order_with_params_and_qty(params, Quantity::from("1.2"));
812        let primary_id = order.client_order_id();
813
814        algo.on_order(order).unwrap();
815        assert_eq!(algo.scheduled_sizes.get(&primary_id).unwrap().len(), 2);
816
817        // Mark primary order as closed (canceled)
818        {
819            let cache_rc = algo.core.cache_rc();
820            let mut cache = cache_rc.borrow_mut();
821            let primary = cache.order(&primary_id).map(|o| o.clone()).unwrap();
822
823            let canceled = OrderCanceledSpec::builder()
824                .trader_id(primary.trader_id())
825                .strategy_id(primary.strategy_id())
826                .instrument_id(primary.instrument_id())
827                .client_order_id(primary.client_order_id())
828                .build();
829            cache
830                .update_order(&OrderEventAny::Canceled(canceled))
831                .unwrap();
832        }
833
834        // Timer fires but primary is closed
835        let event = TimeEvent::new(primary_id.inner(), UUID4::new(), 0.into(), 0.into());
836        ExecutionAlgorithm::on_time_event(&mut algo, &event).unwrap();
837
838        // Sequence should complete early since primary is closed
839        assert!(algo.scheduled_sizes.get(&primary_id).is_none());
840    }
841
842    #[rstest]
843    fn test_twap_on_stop_cancels_timers() {
844        let mut algo = create_twap_algorithm();
845        register_algorithm(&mut algo);
846
847        add_instrument_to_cache(&algo);
848
849        let mut params = IndexMap::new();
850        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
851        params.insert(Ustr::from("interval_secs"), Ustr::from("20"));
852
853        let order = create_market_order_with_params(params);
854        let primary_id = order.client_order_id();
855
856        algo.on_order(order).unwrap();
857
858        // Verify timer is set
859        assert!(
860            algo.clock()
861                .timer_names()
862                .iter()
863                .any(|name| name.as_str() == primary_id.as_str())
864        );
865
866        // Stop through the DataActor entry point the component lifecycle uses
867        DataActor::on_stop(&mut algo).unwrap();
868
869        // Timer should be canceled
870        assert!(algo.clock().timer_names().is_empty());
871    }
872
873    #[rstest]
874    fn test_twap_fractional_interval_secs() {
875        let mut algo = create_twap_algorithm();
876        register_algorithm(&mut algo);
877
878        add_instrument_to_cache(&algo);
879
880        // Use fractional interval like Python tests: 3 second horizon, 0.5 second interval
881        let mut params = IndexMap::new();
882        params.insert(Ustr::from("horizon_secs"), Ustr::from("3"));
883        params.insert(Ustr::from("interval_secs"), Ustr::from("0.5"));
884
885        let order = create_market_order_with_params(params);
886        let primary_id = order.client_order_id();
887
888        // Should not error - fractional seconds should parse correctly
889        algo.on_order(order).unwrap();
890
891        // 3 / 0.5 = 6 intervals, first spawned immediately, 5 remaining (plus possible remainder)
892        let remaining = algo.scheduled_sizes.get(&primary_id).unwrap();
893        assert!(remaining.len() >= 5);
894    }
895
896    #[rstest]
897    fn test_twap_submits_entire_size_when_qty_per_interval_below_size_increment() {
898        use nautilus_model::instruments::{InstrumentAny, stubs::equity_aapl};
899
900        let mut algo = create_twap_algorithm();
901        register_algorithm(&mut algo);
902
903        // Use equity with size_increment of 1 (whole shares only)
904        let instrument = equity_aapl();
905        let instrument_id = instrument.id();
906        {
907            let cache_rc = algo.core.cache_rc();
908            let mut cache = cache_rc.borrow_mut();
909            cache
910                .add_instrument(InstrumentAny::Equity(instrument))
911                .unwrap();
912        }
913
914        // 2 shares over 60s with 10s intervals = 6 intervals
915        // 2 / 6 = 0.333... which is less than size_increment of 1
916        let mut params = IndexMap::new();
917        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
918        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
919
920        let order = OrderAny::Market(MarketOrder::new(
921            TraderId::from("TRADER-001"),
922            StrategyId::from("STRAT-001"),
923            instrument_id,
924            ClientOrderId::from("O-002"),
925            OrderSide::Buy,
926            Quantity::from("2"),
927            TimeInForce::Gtc,
928            UUID4::new(),
929            0.into(),
930            false,
931            false,
932            None,
933            None,
934            None,
935            None,
936            Some(ExecAlgorithmId::new("TWAP")),
937            Some(params),
938            None,
939            None,
940        ));
941
942        let primary_id = order.client_order_id();
943        algo.on_order(order).unwrap();
944
945        // Should submit entire size directly (no scheduling)
946        assert!(algo.scheduled_sizes.get(&primary_id).is_none());
947    }
948
949    #[rstest]
950    fn test_twap_rejects_negative_interval_secs() {
951        let mut algo = create_twap_algorithm();
952        register_algorithm(&mut algo);
953
954        add_instrument_to_cache(&algo);
955
956        let mut params = IndexMap::new();
957        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
958        params.insert(Ustr::from("interval_secs"), Ustr::from("-0.5"));
959
960        let order = create_market_order_with_params(params);
961
962        // Should not error but should reject the order (no scheduling)
963        let result = algo.on_order(order);
964        assert!(result.is_ok());
965        assert!(algo.scheduled_sizes.is_empty());
966    }
967
968    #[rstest]
969    fn test_twap_rejects_negative_horizon_secs() {
970        let mut algo = create_twap_algorithm();
971        register_algorithm(&mut algo);
972
973        add_instrument_to_cache(&algo);
974
975        let mut params = IndexMap::new();
976        params.insert(Ustr::from("horizon_secs"), Ustr::from("-10"));
977        params.insert(Ustr::from("interval_secs"), Ustr::from("1"));
978
979        let order = create_market_order_with_params(params);
980
981        // Should not error but should reject the order (no scheduling)
982        let result = algo.on_order(order);
983        assert!(result.is_ok());
984        assert!(algo.scheduled_sizes.is_empty());
985    }
986
987    #[rstest]
988    fn test_twap_rejects_zero_interval_secs() {
989        let mut algo = create_twap_algorithm();
990        register_algorithm(&mut algo);
991
992        add_instrument_to_cache(&algo);
993
994        let mut params = IndexMap::new();
995        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
996        params.insert(Ustr::from("interval_secs"), Ustr::from("0"));
997
998        let order = create_market_order_with_params(params);
999
1000        // Should not error but should reject the order (no scheduling)
1001        let result = algo.on_order(order);
1002        assert!(result.is_ok());
1003        assert!(algo.scheduled_sizes.is_empty());
1004    }
1005
1006    #[rstest]
1007    fn test_twap_rejects_nan_interval_secs() {
1008        let mut algo = create_twap_algorithm();
1009        register_algorithm(&mut algo);
1010
1011        add_instrument_to_cache(&algo);
1012
1013        let mut params = IndexMap::new();
1014        params.insert(Ustr::from("horizon_secs"), Ustr::from("60"));
1015        params.insert(Ustr::from("interval_secs"), Ustr::from("NaN"));
1016
1017        let order = create_market_order_with_params(params);
1018
1019        let result = algo.on_order(order);
1020        assert!(result.is_ok());
1021        assert!(algo.scheduled_sizes.is_empty());
1022    }
1023
1024    #[rstest]
1025    fn test_twap_rejects_infinity_horizon_secs() {
1026        let mut algo = create_twap_algorithm();
1027        register_algorithm(&mut algo);
1028
1029        add_instrument_to_cache(&algo);
1030
1031        let mut params = IndexMap::new();
1032        params.insert(Ustr::from("horizon_secs"), Ustr::from("inf"));
1033        params.insert(Ustr::from("interval_secs"), Ustr::from("10"));
1034
1035        let order = create_market_order_with_params(params);
1036
1037        let result = algo.on_order(order);
1038        assert!(result.is_ok());
1039        assert!(algo.scheduled_sizes.is_empty());
1040    }
1041}
nautilus_trading/algorithm/twap.rs

nautilus_trading/algorithm/
twap.rs
