nautilus_common/actor/
indicators.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16#[cfg(feature = "indicators")]
17use std::cell::RefCell;
18use std::{any::Any, fmt::Debug, rc::Rc};
19
20use ahash::AHashMap;
21#[cfg(feature = "indicators")]
22use nautilus_indicators::indicator::Indicator;
23use nautilus_model::{
24    data::{Bar, BarSpecification, BarType, QuoteTick, TradeTick},
25    identifiers::InstrumentId,
26};
27
28/// Shared indicator handle used by actor and strategy registration.
29pub type SharedActorIndicator = Rc<dyn ActorIndicator>;
30
31/// Indicator callback interface used by the actor core.
32pub trait ActorIndicator: Any {
33    /// Returns a stable key used to de-duplicate indicator registrations.
34    fn key(&self) -> usize;
35
36    /// Returns this indicator as [`Any`] for adapter-specific downcasting.
37    fn as_any(&self) -> &dyn Any;
38
39    /// Checks if the indicator is initialized.
40    ///
41    /// # Errors
42    ///
43    /// Returns an error if the indicator cannot report readiness.
44    fn initialized(&self) -> anyhow::Result<bool>;
45
46    /// Handles a quote tick.
47    ///
48    /// # Errors
49    ///
50    /// Returns an error if the indicator cannot handle the quote tick.
51    fn handle_quote(&self, quote: &QuoteTick) -> anyhow::Result<()>;
52
53    /// Handles a trade tick.
54    ///
55    /// # Errors
56    ///
57    /// Returns an error if the indicator cannot handle the trade tick.
58    fn handle_trade(&self, trade: &TradeTick) -> anyhow::Result<()>;
59
60    /// Handles a bar.
61    ///
62    /// # Errors
63    ///
64    /// Returns an error if the indicator cannot handle the bar.
65    fn handle_bar(&self, bar: &Bar) -> anyhow::Result<()>;
66}
67
68#[cfg(feature = "indicators")]
69impl<T> ActorIndicator for RefCell<T>
70where
71    T: Indicator + 'static,
72{
73    fn key(&self) -> usize {
74        std::ptr::from_ref(self).cast::<()>() as usize
75    }
76
77    fn as_any(&self) -> &dyn Any {
78        self
79    }
80
81    fn initialized(&self) -> anyhow::Result<bool> {
82        Ok(self.borrow().initialized())
83    }
84
85    fn handle_quote(&self, quote: &QuoteTick) -> anyhow::Result<()> {
86        self.borrow_mut().handle_quote(quote)
87    }
88
89    fn handle_trade(&self, trade: &TradeTick) -> anyhow::Result<()> {
90        self.borrow_mut().handle_trade(trade);
91        Ok(())
92    }
93
94    fn handle_bar(&self, bar: &Bar) -> anyhow::Result<()> {
95        self.borrow_mut().handle_bar(bar);
96        Ok(())
97    }
98}
99
100/// Registry for actor and strategy indicator callbacks.
101#[derive(Clone, Default)]
102#[allow(
103    clippy::struct_field_names,
104    reason = "indicator-prefixed fields denote distinct indicator collections"
105)]
106pub struct Indicators {
107    indicators: Vec<SharedActorIndicator>,
108    indicators_for_quotes: AHashMap<InstrumentId, Vec<SharedActorIndicator>>,
109    indicators_for_trades: AHashMap<InstrumentId, Vec<SharedActorIndicator>>,
110    indicators_for_bars: AHashMap<(InstrumentId, BarSpecification), Vec<SharedActorIndicator>>,
111}
112
113impl Debug for Indicators {
114    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
115        f.debug_struct(stringify!(Indicators))
116            .field("indicators", &self.indicators.len())
117            .field("indicators_for_quotes", &self.indicators_for_quotes.len())
118            .field("indicators_for_trades", &self.indicators_for_trades.len())
119            .field("indicators_for_bars", &self.indicators_for_bars.len())
120            .finish()
121    }
122}
123
124impl Indicators {
125    /// Returns the registered indicators.
126    #[must_use]
127    pub fn registered_indicators(&self) -> Vec<SharedActorIndicator> {
128        self.indicators.clone()
129    }
130
131    /// Returns whether all registered indicators are initialized.
132    ///
133    /// # Errors
134    ///
135    /// Returns an error if a registered indicator cannot report readiness.
136    pub fn initialized(&self) -> anyhow::Result<bool> {
137        if self.indicators.is_empty() {
138            return Ok(false);
139        }
140
141        for indicator in &self.indicators {
142            if !indicator.initialized()? {
143                return Ok(false);
144            }
145        }
146
147        Ok(true)
148    }
149
150    /// Registers an indicator to receive quote ticks for an instrument.
151    pub fn register_indicator_for_quote_ticks(
152        &mut self,
153        instrument_id: InstrumentId,
154        indicator: SharedActorIndicator,
155    ) {
156        self.register_indicator(indicator.clone());
157        self.register_by_key(instrument_id, indicator, IndicatorKind::Quote);
158    }
159
160    /// Registers an indicator to receive trade ticks for an instrument.
161    pub fn register_indicator_for_trade_ticks(
162        &mut self,
163        instrument_id: InstrumentId,
164        indicator: SharedActorIndicator,
165    ) {
166        self.register_indicator(indicator.clone());
167        self.register_by_key(instrument_id, indicator, IndicatorKind::Trade);
168    }
169
170    /// Registers an indicator to receive bars for a bar type.
171    pub fn register_indicator_for_bars(
172        &mut self,
173        bar_type: BarType,
174        indicator: SharedActorIndicator,
175    ) {
176        self.register_indicator(indicator.clone());
177        self.register_bar(bar_type.id_spec_key(), indicator);
178    }
179
180    /// Handles a quote tick with registered indicators.
181    ///
182    /// # Errors
183    ///
184    /// Returns an error if a registered indicator cannot handle the quote tick.
185    pub fn handle_quote(&self, quote: &QuoteTick) -> anyhow::Result<()> {
186        if let Some(indicators) = self.indicators_for_quotes.get(&quote.instrument_id) {
187            for indicator in indicators {
188                indicator.handle_quote(quote)?;
189            }
190        }
191
192        Ok(())
193    }
194
195    /// Handles quote ticks with registered indicators.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if a registered indicator cannot handle a quote tick.
200    pub fn handle_quotes(&self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
201        for quote in quotes {
202            self.handle_quote(quote)?;
203        }
204
205        Ok(())
206    }
207
208    /// Handles a trade tick with registered indicators.
209    ///
210    /// # Errors
211    ///
212    /// Returns an error if a registered indicator cannot handle the trade tick.
213    pub fn handle_trade(&self, trade: &TradeTick) -> anyhow::Result<()> {
214        if let Some(indicators) = self.indicators_for_trades.get(&trade.instrument_id) {
215            for indicator in indicators {
216                indicator.handle_trade(trade)?;
217            }
218        }
219
220        Ok(())
221    }
222
223    /// Handles trade ticks with registered indicators.
224    ///
225    /// # Errors
226    ///
227    /// Returns an error if a registered indicator cannot handle a trade tick.
228    pub fn handle_trades(&self, trades: &[TradeTick]) -> anyhow::Result<()> {
229        for trade in trades {
230            self.handle_trade(trade)?;
231        }
232
233        Ok(())
234    }
235
236    /// Handles a bar with registered indicators.
237    ///
238    /// # Errors
239    ///
240    /// Returns an error if a registered indicator cannot handle the bar.
241    pub fn handle_bar(&self, bar: &Bar) -> anyhow::Result<()> {
242        if let Some(indicators) = self.indicators_for_bars.get(&bar.bar_type.id_spec_key()) {
243            for indicator in indicators {
244                indicator.handle_bar(bar)?;
245            }
246        }
247
248        Ok(())
249    }
250
251    /// Handles bars with registered indicators.
252    ///
253    /// # Errors
254    ///
255    /// Returns an error if a registered indicator cannot handle a bar.
256    pub fn handle_bars(&self, bars: &[Bar]) -> anyhow::Result<()> {
257        for bar in bars {
258            self.handle_bar(bar)?;
259        }
260
261        Ok(())
262    }
263
264    fn register_indicator(&mut self, indicator: SharedActorIndicator) {
265        if !contains_indicator(&self.indicators, &indicator) {
266            self.indicators.push(indicator);
267        }
268    }
269
270    fn register_by_key(
271        &mut self,
272        instrument_id: InstrumentId,
273        indicator: SharedActorIndicator,
274        kind: IndicatorKind,
275    ) {
276        let indicators = match kind {
277            IndicatorKind::Quote => self.indicators_for_quotes.entry(instrument_id).or_default(),
278            IndicatorKind::Trade => self.indicators_for_trades.entry(instrument_id).or_default(),
279        };
280
281        if !contains_indicator(indicators, &indicator) {
282            indicators.push(indicator);
283        }
284    }
285
286    fn register_bar(
287        &mut self,
288        bar_key: (InstrumentId, BarSpecification),
289        indicator: SharedActorIndicator,
290    ) {
291        let indicators = self.indicators_for_bars.entry(bar_key).or_default();
292
293        if !contains_indicator(indicators, &indicator) {
294            indicators.push(indicator);
295        }
296    }
297}
298
299#[derive(Clone, Copy)]
300enum IndicatorKind {
301    Quote,
302    Trade,
303}
304
305fn contains_indicator(
306    indicators: &[SharedActorIndicator],
307    indicator: &SharedActorIndicator,
308) -> bool {
309    let indicator_key = indicator.key();
310    indicators
311        .iter()
312        .any(|registered| registered.key() == indicator_key)
313}
314
315#[cfg(test)]
316mod tests {
317    #[cfg(feature = "indicators")]
318    use std::cell::RefCell;
319    use std::{
320        any::Any,
321        cell::Cell,
322        rc::Rc,
323        str::FromStr,
324        sync::atomic::{AtomicUsize, Ordering},
325    };
326
327    #[cfg(feature = "indicators")]
328    use nautilus_indicators::indicator::Indicator;
329    use nautilus_model::data::{Bar, BarType, QuoteTick, TradeTick};
330    use rstest::rstest;
331
332    use super::{ActorIndicator, Indicators, SharedActorIndicator};
333
334    static NEXT_KEY: AtomicUsize = AtomicUsize::new(1);
335
336    #[derive(Debug)]
337    struct TrackingIndicator {
338        key: usize,
339        initialized: Cell<bool>,
340        quotes: Cell<usize>,
341        trades: Cell<usize>,
342        bars: Cell<usize>,
343    }
344
345    impl TrackingIndicator {
346        fn new() -> Self {
347            Self {
348                key: NEXT_KEY.fetch_add(1, Ordering::Relaxed),
349                initialized: Cell::new(false),
350                quotes: Cell::new(0),
351                trades: Cell::new(0),
352                bars: Cell::new(0),
353            }
354        }
355
356        fn set_initialized(&self, initialized: bool) {
357            self.initialized.set(initialized);
358        }
359    }
360
361    impl ActorIndicator for TrackingIndicator {
362        fn key(&self) -> usize {
363            self.key
364        }
365
366        fn as_any(&self) -> &dyn Any {
367            self
368        }
369
370        fn initialized(&self) -> anyhow::Result<bool> {
371            Ok(self.initialized.get())
372        }
373
374        fn handle_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
375            self.quotes.set(self.quotes.get() + 1);
376            Ok(())
377        }
378
379        fn handle_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
380            self.trades.set(self.trades.get() + 1);
381            Ok(())
382        }
383
384        fn handle_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
385            self.bars.set(self.bars.get() + 1);
386            Ok(())
387        }
388    }
389
390    #[derive(Debug)]
391    struct ErrorIndicator {
392        key: usize,
393    }
394
395    impl ErrorIndicator {
396        fn new() -> Self {
397            Self {
398                key: NEXT_KEY.fetch_add(1, Ordering::Relaxed),
399            }
400        }
401    }
402
403    impl ActorIndicator for ErrorIndicator {
404        fn key(&self) -> usize {
405            self.key
406        }
407
408        fn as_any(&self) -> &dyn Any {
409            self
410        }
411
412        fn initialized(&self) -> anyhow::Result<bool> {
413            Ok(true)
414        }
415
416        fn handle_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
417            anyhow::bail!("indicator failed");
418        }
419
420        fn handle_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
421            anyhow::bail!("trade indicator failed");
422        }
423
424        fn handle_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
425            anyhow::bail!("bar indicator failed");
426        }
427    }
428
429    #[cfg(feature = "indicators")]
430    #[derive(Debug, Default)]
431    struct NativeIndicator {
432        initialized: bool,
433        quotes: Vec<QuoteTick>,
434        trades: Vec<TradeTick>,
435        bars: Vec<Bar>,
436    }
437
438    #[cfg(feature = "indicators")]
439    impl Indicator for NativeIndicator {
440        fn name(&self) -> String {
441            stringify!(NativeIndicator).to_string()
442        }
443
444        fn has_inputs(&self) -> bool {
445            !(self.quotes.is_empty() && self.trades.is_empty() && self.bars.is_empty())
446        }
447
448        fn initialized(&self) -> bool {
449            self.initialized
450        }
451
452        fn handle_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
453            self.quotes.push(*quote);
454            Ok(())
455        }
456
457        fn handle_trade(&mut self, trade: &TradeTick) {
458            self.trades.push(*trade);
459        }
460
461        fn handle_bar(&mut self, bar: &Bar) {
462            self.bars.push(*bar);
463        }
464
465        fn reset(&mut self) {
466            self.initialized = false;
467            self.quotes.clear();
468            self.trades.clear();
469            self.bars.clear();
470        }
471    }
472
473    #[rstest]
474    fn test_register_indicator_routes_quote_trade_and_bar_with_deduplication() {
475        let mut indicators = Indicators::default();
476        let indicator = Rc::new(TrackingIndicator::new());
477        let registered: SharedActorIndicator = indicator.clone();
478        let quote = QuoteTick::default();
479        let trade = TradeTick::default();
480        let bar = Bar::default();
481        let external_bar_type = BarType::from_str(&format!(
482            "{}-1-MINUTE-LAST-EXTERNAL",
483            bar.bar_type.instrument_id()
484        ))
485        .unwrap();
486
487        indicators.register_indicator_for_quote_ticks(quote.instrument_id, registered.clone());
488        indicators.register_indicator_for_quote_ticks(quote.instrument_id, registered.clone());
489        indicators.register_indicator_for_trade_ticks(trade.instrument_id, registered.clone());
490        indicators.register_indicator_for_trade_ticks(trade.instrument_id, registered.clone());
491        indicators.register_indicator_for_bars(external_bar_type, registered.clone());
492        indicators.register_indicator_for_bars(external_bar_type, registered);
493
494        indicators.handle_quote(&quote).unwrap();
495        indicators.handle_trade(&trade).unwrap();
496        indicators.handle_bar(&bar).unwrap();
497
498        assert_eq!(indicators.registered_indicators().len(), 1);
499        assert_eq!(indicator.quotes.get(), 1);
500        assert_eq!(indicator.trades.get(), 1);
501        assert_eq!(indicator.bars.get(), 1);
502    }
503
504    #[rstest]
505    fn test_initialized_requires_all_registered_indicators() {
506        let mut indicators = Indicators::default();
507        let first = Rc::new(TrackingIndicator::new());
508        let second = Rc::new(TrackingIndicator::new());
509        let quote = QuoteTick::default();
510
511        indicators.register_indicator_for_quote_ticks(quote.instrument_id, first.clone());
512        indicators.register_indicator_for_quote_ticks(quote.instrument_id, second.clone());
513
514        first.set_initialized(true);
515
516        assert!(!indicators.initialized().unwrap());
517
518        second.set_initialized(true);
519
520        assert!(indicators.initialized().unwrap());
521    }
522
523    #[rstest]
524    fn test_handle_quote_propagates_indicator_error() {
525        let mut indicators = Indicators::default();
526        let indicator = Rc::new(ErrorIndicator::new());
527        let quote = QuoteTick::default();
528
529        indicators.register_indicator_for_quote_ticks(quote.instrument_id, indicator);
530
531        let err = indicators.handle_quote(&quote).unwrap_err();
532
533        assert_eq!(err.to_string(), "indicator failed");
534    }
535
536    #[rstest]
537    fn test_handle_trade_and_bar_propagate_indicator_errors() {
538        let trade = TradeTick::default();
539        let bar = Bar::default();
540        let trade_indicator = Rc::new(ErrorIndicator::new());
541        let bar_indicator = Rc::new(ErrorIndicator::new());
542        let mut indicators = Indicators::default();
543
544        indicators.register_indicator_for_trade_ticks(trade.instrument_id, trade_indicator);
545        indicators.register_indicator_for_bars(bar.bar_type, bar_indicator);
546
547        let trade_error = indicators.handle_trade(&trade).unwrap_err();
548        let bar_error = indicators.handle_bar(&bar).unwrap_err();
549
550        assert_eq!(trade_error.to_string(), "trade indicator failed");
551        assert_eq!(bar_error.to_string(), "bar indicator failed");
552    }
553
554    #[rstest]
555    fn test_handle_quotes_stops_after_first_indicator_error() {
556        let quote = QuoteTick::default();
557        let error_indicator = Rc::new(ErrorIndicator::new());
558        let tracking_indicator = Rc::new(TrackingIndicator::new());
559        let mut indicators = Indicators::default();
560
561        indicators
562            .register_indicator_for_quote_ticks(quote.instrument_id, tracking_indicator.clone());
563        indicators.register_indicator_for_quote_ticks(quote.instrument_id, error_indicator);
564
565        let error = indicators.handle_quotes(&[quote, quote]).unwrap_err();
566
567        assert_eq!(error.to_string(), "indicator failed");
568        assert_eq!(tracking_indicator.quotes.get(), 1);
569    }
570
571    #[cfg(feature = "indicators")]
572    #[rstest]
573    fn test_native_indicator_bridge_forwards_readiness_and_market_data() {
574        let quote = QuoteTick::default();
575        let trade = TradeTick::default();
576        let bar = Bar::default();
577        let indicator = Rc::new(RefCell::new(NativeIndicator::default()));
578        let registered: SharedActorIndicator = indicator.clone();
579        let mut indicators = Indicators::default();
580
581        indicators.register_indicator_for_quote_ticks(quote.instrument_id, registered.clone());
582        indicators.register_indicator_for_trade_ticks(trade.instrument_id, registered.clone());
583        indicators.register_indicator_for_bars(bar.bar_type, registered);
584
585        assert!(!indicators.initialized().unwrap());
586
587        indicator.borrow_mut().initialized = true;
588        indicators.handle_quote(&quote).unwrap();
589        indicators.handle_trade(&trade).unwrap();
590        indicators.handle_bar(&bar).unwrap();
591
592        let registered = indicators.registered_indicators();
593        let indicator = indicator.borrow();
594        assert_eq!(registered.len(), 1);
595        assert!(registered[0].as_any().is::<RefCell<NativeIndicator>>());
596        assert!(indicators.initialized().unwrap());
597        assert_eq!(indicator.quotes, vec![quote]);
598        assert_eq!(indicator.trades, vec![trade]);
599        assert_eq!(indicator.bars, vec![bar]);
600    }
601}
nautilus_common/actor/indicators.rs

nautilus_common/actor/
indicators.rs
