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

nautilus_trading/algorithm/
twap.rs
