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

nautilus_data/engine/
bar.rs
