nautilus_data/engine/
bar.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
16use std::fmt::Debug;
17
18use anyhow::Context;
19use jiff::tz::TimeZone;
20use nautilus_common::{
21    messages::data::{SubscribeBars, SubscribeCommand},
22    msgbus::{self, MStr, Topic, TypedHandler},
23};
24use nautilus_core::{UUID4, datetime::get_timezone};
25use nautilus_model::data::{Bar, BarType, QuoteTick, TradeTick};
26
27use super::{
28    AggregationSource, BAR_AGGREGATOR_PRIORITY, BarAggregation, BarAggregator, BarBarHandler,
29    BarQuoteHandler, BarTradeHandler, DataCommand, DataEngine, DataResponse, Instrument,
30    InstrumentAny, PriceType, Rc, RefCell, RenkoBarAggregator, RequestBarAggregation,
31    RequestCommand, SubscribeQuotes, SubscribeTrades, TickBarAggregator,
32    TickImbalanceBarAggregator, TickRunsBarAggregator, TimeBarAggregator, UnsubscribeBars,
33    UnsubscribeCommand, UnsubscribeQuotes, UnsubscribeTrades, ValueBarAggregator,
34    ValueImbalanceBarAggregator, ValueRunsBarAggregator, VolumeBarAggregator,
35    VolumeImbalanceBarAggregator, VolumeRunsBarAggregator, log_error_on_cache_insert,
36    process_engine_bar, request_bar_aggregation_from_params, request_params,
37    requests::time_zone_from_params, switchboard,
38};
39
40impl DataEngine {
41    pub(super) fn prepare_request_bar_aggregators_from_state(
42        &mut self,
43        request_id: UUID4,
44        state: &RequestBarAggregation,
45    ) -> anyhow::Result<()> {
46        if !self.can_start_request_bar_aggregators(request_id, state) {
47            anyhow::bail!(
48                "Cannot request aggregated bars: one of the aggregators in `bar_types` is already running"
49            );
50        }
51
52        self.request_bar_aggregations
53            .insert(request_id, state.clone());
54
55        if let Err(e) = self.init_request_bar_aggregators(request_id, state) {
56            self.cleanup_request_bar_aggregators(&request_id);
57            return Err(e);
58        }
59
60        Ok(())
61    }
62
63    pub(super) fn prepare_request_bar_aggregators(
64        &mut self,
65        req: &RequestCommand,
66    ) -> anyhow::Result<()> {
67        let request_id = *req.request_id();
68
69        let Some(state) = request_bar_aggregation_from_params(request_params(req))? else {
70            return Ok(());
71        };
72
73        self.prepare_request_bar_aggregators_from_state(request_id, &state)
74    }
75
76    fn can_start_request_bar_aggregators(
77        &self,
78        request_id: UUID4,
79        state: &RequestBarAggregation,
80    ) -> bool {
81        let aggregator_request_id = state.aggregator_request_id(request_id);
82        state.bar_types.iter().all(|bar_type| {
83            let key = bar_aggregator_key(*bar_type, aggregator_request_id);
84            self.bar_aggregators
85                .get(&key)
86                .is_none_or(|aggregator| !aggregator.borrow().is_running())
87        })
88    }
89
90    fn init_request_bar_aggregators(
91        &mut self,
92        request_id: UUID4,
93        state: &RequestBarAggregation,
94    ) -> anyhow::Result<()> {
95        let aggregator_request_id = state.aggregator_request_id(request_id);
96
97        for bar_type in &state.bar_types {
98            self.create_bar_aggregator_for_key(
99                *bar_type,
100                aggregator_request_id,
101                state.skip_first_non_full_bar,
102                state.time_zone.clone(),
103            )?;
104            self.setup_bar_aggregator(*bar_type, true, aggregator_request_id)?;
105
106            let key = bar_aggregator_key(*bar_type, aggregator_request_id);
107            if let Some(aggregator) = self.bar_aggregators.get(&key) {
108                if state.disable_build_with_no_updates {
109                    aggregator.borrow_mut().set_build_with_no_updates(false);
110                }
111
112                aggregator.borrow_mut().set_is_running(true);
113            }
114        }
115
116        self.set_request_bar_aggregator_chain_handlers(request_id, state);
117
118        Ok(())
119    }
120
121    fn set_request_bar_aggregator_chain_handlers(
122        &self,
123        request_id: UUID4,
124        state: &RequestBarAggregation,
125    ) {
126        let aggregator_request_id = state.aggregator_request_id(request_id);
127
128        for bar_type in &state.bar_types {
129            let key = bar_aggregator_key(*bar_type, aggregator_request_id);
130
131            let Some(aggregator) = self.bar_aggregators.get(&key).cloned() else {
132                continue;
133            };
134
135            let downstream: Vec<_> = state
136                .bar_types
137                .iter()
138                .filter(|candidate| {
139                    candidate.is_composite()
140                        && candidate.composite().standard() == bar_type.standard()
141                })
142                .filter_map(|candidate| {
143                    let key = bar_aggregator_key(*candidate, aggregator_request_id);
144                    self.bar_aggregators.get(&key).cloned()
145                })
146                .collect();
147
148            let cache = Rc::clone(&self.cache);
149
150            let handler: Box<dyn FnMut(Bar)> = Box::new(move |bar: Bar| {
151                // Request-generated bars are delivered only through the cache.
152                if let Err(e) = cache.as_ref().borrow_mut().add_bar_historical(bar) {
153                    log_error_on_cache_insert(&e);
154                }
155
156                for aggregator in &downstream {
157                    aggregator.borrow_mut().handle_bar(bar);
158                }
159            });
160
161            aggregator.borrow_mut().set_historical_mode(true, handler);
162        }
163    }
164
165    pub(super) fn cleanup_request_bar_aggregators(&mut self, request_id: &UUID4) -> bool {
166        let Some(state) = self.request_bar_aggregations.remove(request_id) else {
167            return false;
168        };
169
170        let aggregator_request_id = state.aggregator_request_id(*request_id);
171
172        for bar_type in state.bar_types {
173            let key = bar_aggregator_key(bar_type, aggregator_request_id);
174            let has_live_handlers =
175                state.update_subscriptions && self.bar_aggregator_handlers.contains_key(&key);
176
177            let keep_running = if has_live_handlers {
178                match self.setup_bar_aggregator(bar_type, false, aggregator_request_id) {
179                    Ok(()) => true,
180                    Err(e) => {
181                        log::error!(
182                            "Error starting live request bar aggregator for {bar_type}: {e}"
183                        );
184                        false
185                    }
186                }
187            } else {
188                false
189            };
190
191            if let Some(aggregator) = self.bar_aggregators.get(&key) {
192                aggregator.borrow_mut().set_is_running(keep_running);
193            }
194
195            if !state.update_subscriptions
196                && let Err(e) = self.stop_bar_aggregator(bar_type, aggregator_request_id)
197            {
198                log::error!("Error stopping request bar aggregator for {bar_type}: {e}");
199            }
200        }
201
202        true
203    }
204
205    pub(super) fn process_request_bar_aggregation_response(
206        &mut self,
207        resp: &DataResponse,
208        completed: bool,
209    ) {
210        let correlation_id = *resp.correlation_id();
211
212        let Some(state) = self.request_bar_aggregations.get(&correlation_id).cloned() else {
213            return;
214        };
215
216        match resp {
217            DataResponse::Quotes(r) => {
218                for quote in r.data.iter() {
219                    self.update_request_bar_aggregators_from_quote(&state, correlation_id, *quote);
220                }
221            }
222            DataResponse::Trades(r) => {
223                for trade in r.data.iter() {
224                    self.update_request_bar_aggregators_from_trade(&state, correlation_id, *trade);
225                }
226            }
227            DataResponse::Bars(r) => {
228                for bar in r.data.iter() {
229                    self.update_request_bar_aggregators_from_bar(&state, correlation_id, *bar);
230                }
231            }
232            _ => {}
233        }
234
235        if completed {
236            self.cleanup_request_bar_aggregators(&correlation_id);
237        }
238    }
239
240    pub(super) fn update_request_bar_aggregators_from_quote(
241        &self,
242        state: &RequestBarAggregation,
243        request_id: UUID4,
244        quote: QuoteTick,
245    ) {
246        let aggregator_request_id = state.aggregator_request_id(request_id);
247
248        for bar_type in &state.bar_types {
249            if bar_type.is_composite()
250                || bar_type.instrument_id() != quote.instrument_id
251                || bar_type.spec().price_type == PriceType::Last
252            {
253                continue;
254            }
255
256            self.update_request_bar_aggregator(*bar_type, aggregator_request_id, |aggregator| {
257                aggregator.handle_quote(quote);
258            });
259        }
260    }
261
262    pub(super) fn update_request_bar_aggregators_from_trade(
263        &self,
264        state: &RequestBarAggregation,
265        request_id: UUID4,
266        trade: TradeTick,
267    ) {
268        let aggregator_request_id = state.aggregator_request_id(request_id);
269
270        for bar_type in &state.bar_types {
271            if bar_type.is_composite()
272                || bar_type.instrument_id() != trade.instrument_id
273                || bar_type.spec().price_type != PriceType::Last
274            {
275                continue;
276            }
277
278            self.update_request_bar_aggregator(*bar_type, aggregator_request_id, |aggregator| {
279                aggregator.handle_trade(trade);
280            });
281        }
282    }
283
284    pub(super) fn update_request_bar_aggregators_from_bar(
285        &self,
286        state: &RequestBarAggregation,
287        request_id: UUID4,
288        bar: Bar,
289    ) {
290        let aggregator_request_id = state.aggregator_request_id(request_id);
291
292        for bar_type in &state.bar_types {
293            if !bar_type.is_composite()
294                || bar_type.composite().standard() != bar.bar_type.standard()
295            {
296                continue;
297            }
298
299            self.update_request_bar_aggregator(*bar_type, aggregator_request_id, |aggregator| {
300                aggregator.handle_bar(bar);
301            });
302        }
303    }
304
305    pub(super) fn update_request_bar_aggregator<F>(
306        &self,
307        bar_type: BarType,
308        request_id: Option<UUID4>,
309        update: F,
310    ) where
311        F: FnOnce(&mut dyn BarAggregator),
312    {
313        let key = bar_aggregator_key(bar_type, request_id);
314
315        let Some(aggregator) = self.bar_aggregators.get(&key) else {
316            log::error!("Cannot update request bar aggregator: no aggregator found for {bar_type}");
317            return;
318        };
319
320        update(aggregator.borrow_mut().as_mut());
321    }
322
323    pub(super) fn subscribe_bars(&mut self, cmd: &SubscribeBars) -> anyhow::Result<()> {
324        let bar_type = cmd
325            .single_bar_type()
326            .expect("native engine receives one bar type");
327
328        match bar_type.aggregation_source() {
329            AggregationSource::Internal => self.start_bar_aggregation(cmd)?,
330            AggregationSource::External => {
331                if bar_type.instrument_id().is_synthetic() {
332                    anyhow::bail!(
333                        "Cannot subscribe for externally aggregated synthetic instrument bar data"
334                    );
335                }
336            }
337        }
338
339        Ok(())
340    }
341
342    pub(super) fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) {
343        let bar_type = cmd
344            .single_bar_type()
345            .expect("native engine receives one bar type");
346
347        // Don't remove aggregator if other exact-topic subscribers still exist
348        let topic = switchboard::get_bars_topic(bar_type.standard());
349        if msgbus::exact_subscriber_count_bars(topic) > 0 {
350            return;
351        }
352
353        let retained = self
354            .subscriptions_bar_aggregation
355            .get(&bar_type.standard())
356            .map(|subscription| subscription.command.clone());
357
358        let command = retained.map_or_else(
359            || cmd.clone(),
360            |subscribe| {
361                UnsubscribeBars::new(
362                    subscribe.bar_types,
363                    subscribe.client_id,
364                    subscribe.venue,
365                    cmd.command_id,
366                    cmd.ts_init,
367                    Some(subscribe.command_id),
368                    subscribe.params,
369                )
370            },
371        );
372
373        if self
374            .bar_aggregators
375            .contains_key(&bar_aggregator_key(bar_type, None))
376        {
377            match self.stop_bar_aggregator(bar_type, None) {
378                Ok(()) => {
379                    self.subscriptions_bar_aggregation
380                        .remove(&bar_type.standard());
381                    self.unsubscribe_bar_aggregator(&command);
382                }
383                Err(e) => log::error!("Error stopping bar aggregator for {bar_type}: {e}"),
384            }
385        }
386
387        // After stopping a composite, release its source through `unsubscribe_bars`, which frees
388        // the client feed recorded in the source's retained command.
389        if command
390            .single_bar_type()
391            .expect("retained aggregation has one bar type")
392            .is_composite()
393        {
394            let source_type = command
395                .single_bar_type()
396                .expect("retained aggregation has one bar type")
397                .composite();
398
399            if self
400                .bar_aggregators
401                .contains_key(&bar_aggregator_key(source_type, None))
402            {
403                self.unsubscribe_bars(&UnsubscribeBars::new(
404                    source_type,
405                    command.client_id,
406                    command.venue,
407                    UUID4::new(),
408                    command.ts_init,
409                    Some(command.command_id),
410                    command.params.clone(),
411                ));
412            }
413        }
414    }
415
416    fn create_bar_aggregator(
417        &self,
418        instrument: &InstrumentAny,
419        bar_type: BarType,
420        skip_first_non_full_bar: Option<bool>,
421        time_zone: TimeZone,
422    ) -> Box<dyn BarAggregator> {
423        let cache = Rc::clone(&self.cache);
424        let validate_sequence = self.config.validate_data_sequence;
425
426        let handler = move |bar: Bar| {
427            process_engine_bar(&cache, validate_sequence, true, bar);
428        };
429
430        let clock = Rc::clone(&self.clock);
431        let config = self.config.clone();
432
433        let price_precision = instrument.price_precision();
434        let size_precision = instrument.size_precision();
435
436        if bar_type.spec().is_time_aggregated() {
437            let time_bars_origin_offset = config
438                .time_bars_origin_offset
439                .get(&bar_type.spec().aggregation)
440                .map(|duration| jiff::SignedDuration::try_from(*duration).unwrap_or_default());
441
442            Box::new(TimeBarAggregator::new(
443                bar_type,
444                price_precision,
445                size_precision,
446                clock,
447                handler,
448                config.time_bars_build_with_no_updates,
449                config.time_bars_timestamp_on_close,
450                config.time_bars_interval_type,
451                time_bars_origin_offset,
452                time_zone,
453                config.time_bars_build_delay,
454                skip_first_non_full_bar.unwrap_or(config.time_bars_skip_first_non_full_bar),
455            ))
456        } else {
457            match bar_type.spec().aggregation {
458                BarAggregation::Tick => Box::new(TickBarAggregator::new(
459                    bar_type,
460                    price_precision,
461                    size_precision,
462                    handler,
463                )) as Box<dyn BarAggregator>,
464                BarAggregation::TickImbalance => Box::new(TickImbalanceBarAggregator::new(
465                    bar_type,
466                    price_precision,
467                    size_precision,
468                    handler,
469                )) as Box<dyn BarAggregator>,
470                BarAggregation::TickRuns => Box::new(TickRunsBarAggregator::new(
471                    bar_type,
472                    price_precision,
473                    size_precision,
474                    handler,
475                )) as Box<dyn BarAggregator>,
476                BarAggregation::Volume => Box::new(VolumeBarAggregator::new(
477                    bar_type,
478                    price_precision,
479                    size_precision,
480                    handler,
481                )) as Box<dyn BarAggregator>,
482                BarAggregation::VolumeImbalance => Box::new(VolumeImbalanceBarAggregator::new(
483                    bar_type,
484                    price_precision,
485                    size_precision,
486                    handler,
487                )) as Box<dyn BarAggregator>,
488                BarAggregation::VolumeRuns => Box::new(VolumeRunsBarAggregator::new(
489                    bar_type,
490                    price_precision,
491                    size_precision,
492                    handler,
493                )) as Box<dyn BarAggregator>,
494                BarAggregation::Value => Box::new(ValueBarAggregator::new(
495                    bar_type,
496                    price_precision,
497                    size_precision,
498                    handler,
499                )) as Box<dyn BarAggregator>,
500                BarAggregation::ValueImbalance => Box::new(ValueImbalanceBarAggregator::new(
501                    bar_type,
502                    price_precision,
503                    size_precision,
504                    handler,
505                )) as Box<dyn BarAggregator>,
506                BarAggregation::ValueRuns => Box::new(ValueRunsBarAggregator::new(
507                    bar_type,
508                    price_precision,
509                    size_precision,
510                    handler,
511                )) as Box<dyn BarAggregator>,
512                BarAggregation::Renko => Box::new(RenkoBarAggregator::new(
513                    bar_type,
514                    price_precision,
515                    size_precision,
516                    instrument.price_increment(),
517                    handler,
518                )) as Box<dyn BarAggregator>,
519                other => unreachable!(
520                    "Unsupported internal bar aggregation dispatch for {other:?}; update `create_bar_aggregator`"
521                ),
522            }
523        }
524    }
525
526    pub(super) fn create_bar_aggregator_for_key(
527        &mut self,
528        bar_type: BarType,
529        request_id: Option<UUID4>,
530        skip_first_non_full_bar: Option<bool>,
531        time_zone: Option<TimeZone>,
532    ) -> anyhow::Result<()> {
533        let key = bar_aggregator_key(bar_type, request_id);
534
535        let time_zone = match time_zone {
536            Some(zone) => zone,
537            None => get_timezone(self.config.time_bars_time_zone.as_deref().unwrap_or("UTC"))
538                .context("invalid `time_bars_time_zone` configuration")?,
539        };
540
541        if let Some(aggregator) = self.bar_aggregators.get(&key) {
542            let aggregator = aggregator.borrow();
543            if let Some(aggregator) = aggregator.as_any().downcast_ref::<TimeBarAggregator>() {
544                anyhow::ensure!(
545                    aggregator.time_zone() == &time_zone,
546                    "Cannot reuse bar aggregator {bar_type} with a different time zone"
547                );
548            }
549
550            return Ok(());
551        }
552
553        let instrument = {
554            let cache = self.cache.borrow();
555            cache
556                .instrument(&bar_type.instrument_id())
557                .ok_or_else(|| {
558                    anyhow::anyhow!(
559                        "Cannot start bar aggregation: no instrument found for {}",
560                        bar_type.instrument_id(),
561                    )
562                })?
563                .clone()
564        };
565
566        let aggregator =
567            self.create_bar_aggregator(&instrument, bar_type, skip_first_non_full_bar, time_zone);
568        debug_assert_eq!(
569            aggregator.bar_type(),
570            key.0,
571            "aggregator bar type must match its standardized key"
572        );
573        self.bar_aggregators
574            .insert(key, Rc::new(RefCell::new(aggregator)));
575
576        Ok(())
577    }
578
579    fn start_bar_aggregation(&mut self, cmd: &SubscribeBars) -> anyhow::Result<()> {
580        let bar_type = cmd
581            .single_bar_type()
582            .expect("native engine receives one bar type");
583        let time_zone = time_zone_from_params(cmd.params.as_ref())?;
584        let skip_first_non_full_bar = cmd
585            .params
586            .as_ref()
587            .and_then(|params| params.get_bool("skip_first_non_full_bar"));
588        self.create_bar_aggregator_for_key(
589            bar_type,
590            None,
591            skip_first_non_full_bar,
592            time_zone.clone(),
593        )?;
594        let key = bar_aggregator_key(bar_type, None);
595
596        if self
597            .bar_aggregators
598            .get(&key)
599            .is_some_and(|aggregator| aggregator.borrow().is_running())
600            && self.bar_aggregator_handlers.contains_key(&key)
601        {
602            if let Some(source_command) = self
603                .subscriptions_bar_aggregation
604                .get(&bar_type.standard())
605                .and_then(|subscription| subscription.source.clone())
606            {
607                self.execute(DataCommand::Subscribe(source_command));
608            }
609
610            log::debug!("Reusing running bar aggregator for {bar_type}");
611            return Ok(());
612        }
613
614        self.start_bar_aggregator(bar_type, None, skip_first_non_full_bar, time_zone)?;
615        let source = self.subscribe_bar_aggregator(cmd);
616        self.subscriptions_bar_aggregation.insert(
617            bar_type.standard(),
618            BarAggregationSubscription {
619                command: cmd.clone(),
620                source,
621            },
622        );
623
624        Ok(())
625    }
626
627    fn start_bar_aggregator(
628        &mut self,
629        bar_type: BarType,
630        request_id: Option<UUID4>,
631        skip_first_non_full_bar: Option<bool>,
632        time_zone: Option<TimeZone>,
633    ) -> anyhow::Result<()> {
634        let key = bar_aggregator_key(bar_type, request_id);
635        let bar_type_std = bar_type.standard();
636
637        self.create_bar_aggregator_for_key(
638            bar_type,
639            request_id,
640            skip_first_non_full_bar,
641            time_zone,
642        )?;
643        let aggregator = self
644            .bar_aggregators
645            .get(&key)
646            .map(Rc::clone)
647            .ok_or_else(|| anyhow::anyhow!("Cannot start bar aggregation for {bar_type}"))?;
648        let defer_subscription_activation = request_id.is_none()
649            && aggregator.borrow().is_running()
650            && !self.bar_aggregator_handlers.contains_key(&key);
651
652        if !self.bar_aggregator_handlers.contains_key(&key) {
653            // Subscribe to underlying data topics
654            let mut subscriptions = Vec::new();
655
656            if bar_type.is_composite() {
657                let topic = switchboard::get_bars_topic(bar_type.composite());
658                let handler = TypedHandler::new(BarBarHandler::new(&aggregator, bar_type_std));
659                msgbus::subscribe_bars(topic.into(), handler.clone(), None);
660                subscriptions.push(BarAggregatorSubscription::Bar { topic, handler });
661            } else if bar_type.spec().price_type == PriceType::Last {
662                let topic = switchboard::get_trades_topic(bar_type.instrument_id());
663                let handler = TypedHandler::new(BarTradeHandler::new(&aggregator, bar_type_std));
664                msgbus::subscribe_trades(
665                    topic.into(),
666                    handler.clone(),
667                    Some(BAR_AGGREGATOR_PRIORITY),
668                );
669                subscriptions.push(BarAggregatorSubscription::Trade { topic, handler });
670            } else {
671                // Warn if imbalance/runs aggregation is wired to quotes (needs aggressor_side from trades)
672                if matches!(
673                    bar_type.spec().aggregation,
674                    BarAggregation::TickImbalance
675                        | BarAggregation::VolumeImbalance
676                        | BarAggregation::ValueImbalance
677                        | BarAggregation::TickRuns
678                        | BarAggregation::VolumeRuns
679                        | BarAggregation::ValueRuns
680                ) {
681                    log::warn!(
682                        "Bar type {bar_type} uses imbalance/runs aggregation which requires trade \
683                         data with `aggressor_side`, but `price_type` is not LAST so it will receive \
684                         quote data: bars will not emit correctly",
685                    );
686                }
687
688                let topic = switchboard::get_quotes_topic(bar_type.instrument_id());
689                let handler = TypedHandler::new(BarQuoteHandler::new(&aggregator, bar_type_std));
690                msgbus::subscribe_quotes(
691                    topic.into(),
692                    handler.clone(),
693                    Some(BAR_AGGREGATOR_PRIORITY),
694                );
695                subscriptions.push(BarAggregatorSubscription::Quote { topic, handler });
696            }
697
698            self.bar_aggregator_handlers.insert(key, subscriptions);
699        }
700
701        if defer_subscription_activation {
702            return Ok(());
703        }
704
705        self.setup_bar_aggregator(bar_type, false, request_id)?;
706
707        aggregator.borrow_mut().set_is_running(true);
708
709        Ok(())
710    }
711
712    fn subscribe_bar_aggregator(&mut self, cmd: &SubscribeBars) -> Option<SubscribeCommand> {
713        let subscribe = self.bar_aggregator_source_command(cmd)?;
714        self.execute(DataCommand::Subscribe(subscribe.clone()));
715        Some(subscribe)
716    }
717
718    fn bar_aggregator_source_command(&self, cmd: &SubscribeBars) -> Option<SubscribeCommand> {
719        let bar_type = cmd
720            .single_bar_type()
721            .expect("native engine receives one bar type");
722        let key = bar_aggregator_key(bar_type, None);
723        if !self.bar_aggregators.contains_key(&key) {
724            log::error!("Cannot subscribe bar aggregator: no aggregator found for {bar_type}");
725            return None;
726        }
727
728        if bar_type.is_composite() {
729            let composite_bar_type = bar_type.composite();
730            if composite_bar_type.is_externally_aggregated() {
731                let subscribe = SubscribeBars::new(
732                    composite_bar_type,
733                    cmd.client_id,
734                    cmd.venue,
735                    UUID4::new(),
736                    cmd.ts_init,
737                    Some(cmd.command_id),
738                    cmd.params.clone(),
739                );
740                return Some(SubscribeCommand::Bars(subscribe));
741            }
742        } else if bar_type.spec().price_type == PriceType::Last {
743            let subscribe = SubscribeTrades::new(
744                bar_type.instrument_id(),
745                cmd.client_id,
746                cmd.venue,
747                UUID4::new(),
748                cmd.ts_init,
749                Some(cmd.command_id),
750                cmd.params.clone(),
751            );
752            return Some(SubscribeCommand::Trades(subscribe));
753        } else {
754            let subscribe = SubscribeQuotes::new(
755                bar_type.instrument_id(),
756                cmd.client_id,
757                cmd.venue,
758                UUID4::new(),
759                cmd.ts_init,
760                Some(cmd.command_id),
761                cmd.params.clone(),
762            );
763            return Some(SubscribeCommand::Quotes(subscribe));
764        }
765
766        None
767    }
768
769    /// Sets up a bar aggregator.
770    ///
771    /// This method handles historical mode, message bus subscriptions, and time bar aggregator setup.
772    pub(super) fn setup_bar_aggregator(
773        &self,
774        bar_type: BarType,
775        historical: bool,
776        request_id: Option<UUID4>,
777    ) -> anyhow::Result<()> {
778        let key = bar_aggregator_key(bar_type, request_id);
779
780        let aggregator = self.bar_aggregators.get(&key).ok_or_else(|| {
781            anyhow::anyhow!("Cannot setup bar aggregator: no aggregator found for {bar_type}")
782        })?;
783
784        // Set historical mode and handler
785        let cache = Rc::clone(&self.cache);
786        let validate_sequence = self.config.validate_data_sequence;
787        let publish = !historical;
788
789        let handler: Box<dyn FnMut(Bar)> = Box::new(move |bar: Bar| {
790            process_engine_bar(&cache, validate_sequence, publish, bar);
791        });
792
793        aggregator
794            .borrow_mut()
795            .set_historical_mode(historical, handler);
796
797        // For TimeBarAggregator, set clock and start timer
798        if bar_type.spec().is_time_aggregated() {
799            use nautilus_common::clock::VirtualClock;
800
801            if historical {
802                // Each aggregator gets its own independent clock
803                let test_clock = Rc::new(RefCell::new(VirtualClock::new()));
804                aggregator.borrow_mut().set_clock(test_clock);
805                // Set weak reference for historical mode (start_timer called later from preprocess_historical_events)
806                // Store weak reference so start_timer can use it when called later
807                let aggregator_weak = Rc::downgrade(aggregator);
808                aggregator.borrow_mut().set_aggregator_weak(aggregator_weak);
809            } else {
810                aggregator.borrow_mut().set_clock(Rc::clone(&self.clock));
811                aggregator
812                    .borrow_mut()
813                    .start_timer(Some(Rc::clone(aggregator)));
814            }
815        }
816
817        Ok(())
818    }
819
820    fn unsubscribe_bar_aggregator(&mut self, cmd: &UnsubscribeBars) {
821        let bar_type = cmd
822            .single_bar_type()
823            .expect("native engine receives one bar type");
824
825        if bar_type.is_composite() {
826            let composite_bar_type = bar_type.composite();
827            if composite_bar_type.is_externally_aggregated() {
828                let unsubscribe = UnsubscribeBars::new(
829                    composite_bar_type,
830                    cmd.client_id,
831                    cmd.venue,
832                    UUID4::new(),
833                    cmd.ts_init,
834                    Some(cmd.command_id),
835                    cmd.params.clone(),
836                );
837                self.execute(DataCommand::Unsubscribe(UnsubscribeCommand::Bars(
838                    unsubscribe,
839                )));
840            }
841        } else if bar_type.spec().price_type == PriceType::Last {
842            let unsubscribe = UnsubscribeTrades::new(
843                bar_type.instrument_id(),
844                cmd.client_id,
845                cmd.venue,
846                UUID4::new(),
847                cmd.ts_init,
848                Some(cmd.command_id),
849                cmd.params.clone(),
850            );
851            self.execute(DataCommand::Unsubscribe(UnsubscribeCommand::Trades(
852                unsubscribe,
853            )));
854        } else {
855            let unsubscribe = UnsubscribeQuotes::new(
856                bar_type.instrument_id(),
857                cmd.client_id,
858                cmd.venue,
859                UUID4::new(),
860                cmd.ts_init,
861                Some(cmd.command_id),
862                cmd.params.clone(),
863            );
864            self.execute(DataCommand::Unsubscribe(UnsubscribeCommand::Quotes(
865                unsubscribe,
866            )));
867        }
868    }
869
870    pub(super) fn stop_bar_aggregator(
871        &mut self,
872        bar_type: BarType,
873        request_id: Option<UUID4>,
874    ) -> anyhow::Result<()> {
875        let key = bar_aggregator_key(bar_type, request_id);
876
877        let aggregator = self.bar_aggregators.shift_remove(&key).ok_or_else(|| {
878            anyhow::anyhow!("Cannot stop bar aggregator: no aggregator to stop for {bar_type}")
879        })?;
880
881        aggregator.borrow_mut().stop();
882
883        // Unsubscribe any registered message handlers
884        if let Some(subs) = self.bar_aggregator_handlers.remove(&key) {
885            for sub in subs {
886                match sub {
887                    BarAggregatorSubscription::Bar { topic, handler } => {
888                        msgbus::unsubscribe_bars(topic.into(), &handler);
889                    }
890                    BarAggregatorSubscription::Trade { topic, handler } => {
891                        msgbus::unsubscribe_trades(topic.into(), &handler);
892                    }
893                    BarAggregatorSubscription::Quote { topic, handler } => {
894                        msgbus::unsubscribe_quotes(topic.into(), &handler);
895                    }
896                }
897            }
898        }
899
900        Ok(())
901    }
902}
903
904#[derive(Clone, Debug)]
905pub(super) struct BarAggregationSubscription {
906    pub(super) command: SubscribeBars,
907    pub(super) source: Option<SubscribeCommand>,
908}
909
910/// Identifies a bar aggregator instance.
911///
912/// Live subscriptions key on `(bar_type.standard(), None)`. Request-scoped
913/// aggregators carrying a `request_id` key on `(bar_type.standard(), Some(id))`
914/// so they can run alongside a live aggregator on the same bar type.
915pub(crate) type BarAggregatorKey = (BarType, Option<UUID4>);
916
917#[inline]
918pub(crate) fn bar_aggregator_key(bar_type: BarType, request_id: Option<UUID4>) -> BarAggregatorKey {
919    (bar_type.standard(), request_id)
920}
921
922/// Typed subscription for bar aggregator handlers.
923///
924/// Stores the topic and handler for each data type so we can properly
925/// unsubscribe from the typed routers.
926#[derive(Clone)]
927pub enum BarAggregatorSubscription {
928    Bar {
929        topic: MStr<Topic>,
930        handler: TypedHandler<Bar>,
931    },
932    Trade {
933        topic: MStr<Topic>,
934        handler: TypedHandler<TradeTick>,
935    },
936    Quote {
937        topic: MStr<Topic>,
938        handler: TypedHandler<QuoteTick>,
939    },
940}
941
942impl Debug for BarAggregatorSubscription {
943    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
944        match self {
945            Self::Bar { topic, handler } => f
946                .debug_struct(stringify!(Bar))
947                .field("topic", topic)
948                .field("handler_id", &handler.id())
949                .finish(),
950            Self::Trade { topic, handler } => f
951                .debug_struct(stringify!(Trade))
952                .field("topic", topic)
953                .field("handler_id", &handler.id())
954                .finish(),
955            Self::Quote { topic, handler } => f
956                .debug_struct(stringify!(Quote))
957                .field("topic", topic)
958                .field("handler_id", &handler.id())
959                .finish(),
960        }
961    }
962}
963
964#[cfg(test)]
965mod tests {
966    use std::{cell::RefCell, rc::Rc};
967
968    use indexmap::IndexMap;
969    use jiff::Timestamp;
970    use nautilus_common::{
971        cache::Cache,
972        clock::{Clock, VirtualClock},
973        messages::data::SubscribeBars,
974        msgbus::MessageBus,
975    };
976    use nautilus_core::{UUID4, UnixNanos, datetime::get_timezone};
977    use nautilus_model::{
978        data::{Bar, BarType},
979        identifiers::TraderId,
980        instruments::{InstrumentAny, stubs::audusd_sim},
981        types::{Price, Quantity},
982    };
983    use rstest::rstest;
984
985    use super::{bar_aggregator_key, request_bar_aggregation_from_params, time_zone_from_params};
986    use crate::engine::{DataEngine, config::DataEngineConfig};
987
988    #[rstest]
989    #[case::utc_default(None, None, "2026-11-02T00:00:00Z")]
990    #[case::config(Some("America/New_York"), None, "2026-11-02T05:00:00Z")]
991    #[case::utc_override(Some("America/New_York"), Some("UTC"), "2026-11-02T00:00:00Z")]
992    fn test_calendar_zone_precedence_and_reuse(
993        #[case] config_zone: Option<&str>,
994        #[case] override_zone: Option<&str>,
995        #[case] expected: &str,
996    ) {
997        let _msgbus = MessageBus::new(TraderId::new("TESTER-001"), UUID4::new(), None, None)
998            .register_message_bus();
999        let clock = Rc::new(RefCell::new(VirtualClock::new()));
1000        clock.borrow_mut().set_time(UnixNanos::from(
1001            "2026-11-01T12:00:00Z".parse::<Timestamp>().unwrap(),
1002        ));
1003        let cache = Rc::new(RefCell::new(Cache::default()));
1004        cache
1005            .borrow_mut()
1006            .add_instrument(InstrumentAny::CurrencyPair(audusd_sim()))
1007            .unwrap();
1008        let config = DataEngineConfig::builder()
1009            .maybe_time_bars_time_zone(config_zone.map(str::to_owned))
1010            .time_bars_skip_first_non_full_bar(true)
1011            .build();
1012        let mut engine = DataEngine::new(
1013            Rc::clone(&clock) as Rc<RefCell<dyn Clock>>,
1014            Rc::clone(&cache),
1015            Some(config),
1016        );
1017        let bar_type = BarType::from("AUD/USD.SIM-1-DAY-LAST-INTERNAL");
1018        let mut params = IndexMap::from([(
1019            "skip_first_non_full_bar".to_owned(),
1020            serde_json::json!(false),
1021        )]);
1022
1023        if let Some(zone) = override_zone {
1024            params.insert("time_zone".to_owned(), serde_json::json!(zone));
1025        }
1026
1027        let cmd = SubscribeBars::new(
1028            bar_type,
1029            None,
1030            Some(bar_type.instrument_id().venue),
1031            UUID4::new(),
1032            UnixNanos::default(),
1033            None,
1034            Some(nautilus_core::Params::new(params)),
1035        );
1036        engine.subscribe_bars(&cmd).unwrap();
1037        engine.subscribe_bars(&cmd).unwrap();
1038        let mut params = IndexMap::clone(cmd.params.as_ref().unwrap());
1039        params.insert("time_zone".to_owned(), serde_json::json!("Asia/Tokyo"));
1040
1041        let cmd = SubscribeBars {
1042            params: Some(nautilus_core::Params::new(params)),
1043            ..cmd
1044        };
1045
1046        let e = engine.subscribe_bars(&cmd).unwrap_err();
1047
1048        // Request-local instances can use another zone without replacing the live instance
1049        engine
1050            .create_bar_aggregator_for_key(
1051                bar_type,
1052                Some(UUID4::new()),
1053                None,
1054                Some(get_timezone("Asia/Tokyo").unwrap()),
1055            )
1056            .unwrap();
1057        assert_eq!(
1058            e.to_string(),
1059            format!("Cannot reuse bar aggregator {bar_type} with a different time zone")
1060        );
1061        assert_eq!(
1062            clock.borrow().next_time_ns(&format!("TIME_BAR_{bar_type}")),
1063            Some(UnixNanos::from(expected.parse::<Timestamp>().unwrap()))
1064        );
1065        let now = clock.borrow().timestamp_ns();
1066        engine
1067            .bar_aggregators
1068            .get(&bar_aggregator_key(bar_type, None))
1069            .unwrap()
1070            .borrow_mut()
1071            .update(Price::from("0.65000"), Quantity::from(2), now);
1072        let close = UnixNanos::from(expected.parse::<Timestamp>().unwrap());
1073        let events = clock.borrow_mut().advance_time(close, true);
1074        let handlers = clock.borrow().match_handlers(events);
1075        for handler in handlers {
1076            handler.callback.call(handler.event);
1077        }
1078
1079        assert_eq!(
1080            cache.borrow().bar(&bar_type).copied(),
1081            Some(Bar::new(
1082                bar_type,
1083                Price::from("0.65000"),
1084                Price::from("0.65000"),
1085                Price::from("0.65000"),
1086                Price::from("0.65000"),
1087                Quantity::from(2),
1088                close,
1089                close
1090            ))
1091        );
1092    }
1093
1094    #[rstest]
1095    #[case::wrong_type(serde_json::json!(17), "`time_zone` parameter must be a string")]
1096    #[case::unknown_zone(serde_json::json!("Not/A_Zone"), "invalid `time_zone` parameter \"Not/A_Zone\"")]
1097    fn test_calendar_zone_parameter_rejects_invalid_value(
1098        #[case] value: serde_json::Value,
1099        #[case] expected: &str,
1100    ) {
1101        let params = nautilus_core::Params::new([("time_zone".to_owned(), value)].into());
1102        assert_eq!(
1103            time_zone_from_params(Some(&params))
1104                .unwrap_err()
1105                .to_string(),
1106            expected
1107        );
1108    }
1109
1110    #[rstest]
1111    fn test_calendar_request_zone_checks_shared_instance() {
1112        let clock = Rc::new(RefCell::new(VirtualClock::new()));
1113        let cache = Rc::new(RefCell::new(Cache::default()));
1114        cache
1115            .borrow_mut()
1116            .add_instrument(InstrumentAny::CurrencyPair(audusd_sim()))
1117            .unwrap();
1118        let mut engine = DataEngine::new(clock, cache, None);
1119        let bar_type = BarType::from("AUD/USD.SIM-1-MONTH-LAST-INTERNAL");
1120        engine
1121            .create_bar_aggregator_for_key(bar_type, None, None, None)
1122            .unwrap();
1123        let params: nautilus_core::Params = serde_json::from_value(serde_json::json!({
1124            "bar_types": [bar_type.to_string()], "update_subscriptions": true,
1125            "time_zone": "America/New_York"
1126        }))
1127        .unwrap();
1128        let state = request_bar_aggregation_from_params(Some(&params))
1129            .unwrap()
1130            .unwrap();
1131        let id = UUID4::new();
1132        let error = engine
1133            .prepare_request_bar_aggregators_from_state(id, &state)
1134            .unwrap_err();
1135        let mut params = IndexMap::clone(&params);
1136        params.insert("update_subscriptions".to_owned(), serde_json::json!(false));
1137        let params = nautilus_core::Params::new(params);
1138        let state = request_bar_aggregation_from_params(Some(&params))
1139            .unwrap()
1140            .unwrap();
1141        engine
1142            .prepare_request_bar_aggregators_from_state(id, &state)
1143            .unwrap();
1144        let instance = engine
1145            .bar_aggregators
1146            .get(&bar_aggregator_key(bar_type, Some(id)))
1147            .unwrap()
1148            .borrow();
1149        let request_zone = instance
1150            .as_any()
1151            .downcast_ref::<crate::aggregation::TimeBarAggregator>()
1152            .unwrap()
1153            .time_zone()
1154            .clone();
1155        let live = engine
1156            .bar_aggregators
1157            .get(&bar_aggregator_key(bar_type, None))
1158            .unwrap()
1159            .borrow();
1160        let live_zone = live
1161            .as_any()
1162            .downcast_ref::<crate::aggregation::TimeBarAggregator>()
1163            .unwrap()
1164            .time_zone()
1165            .clone();
1166        assert_eq!(
1167            error.to_string(),
1168            format!("Cannot reuse bar aggregator {bar_type} with a different time zone")
1169        );
1170        assert_eq!(request_zone, get_timezone("America/New_York").unwrap());
1171        assert_eq!(live_zone, jiff::tz::TimeZone::UTC);
1172        assert_eq!(engine.bar_aggregators.len(), 2);
1173    }
1174}
nautilus_data/engine/bar.rs

nautilus_data/engine/
bar.rs
