Skip to main content

nautilus_event_store/markers/
extractor.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//! Per-class market-data extractors for marker sidecar capture.
17
18use std::{
19    any::{Any, TypeId},
20    fmt::Debug,
21};
22
23use ahash::AHashMap;
24use nautilus_core::UnixNanos;
25use nautilus_model::{
26    data::{Bar, BookOrder, OrderBookDeltas, OrderBookDepth10, QuoteTick, TradeTick},
27    types::{Price, Quantity, fixed::FIXED_PRECISION},
28};
29
30use crate::markers::DataClass;
31
32const QUOTE_FINGERPRINT_DOMAIN: &[u8] = b"nautilus-event-store/marker/fingerprint/quote/v1";
33const TRADE_FINGERPRINT_DOMAIN: &[u8] = b"nautilus-event-store/marker/fingerprint/trade/v1";
34const BAR_FINGERPRINT_DOMAIN: &[u8] = b"nautilus-event-store/marker/fingerprint/bar/v1";
35const DEPTH10_FINGERPRINT_DOMAIN: &[u8] = b"nautilus-event-store/marker/fingerprint/depth10/v1";
36const DELTAS_FINGERPRINT_DOMAIN: &[u8] = b"nautilus-event-store/marker/fingerprint/deltas/v1";
37
38/// Extracts marker sidecar fields from a concrete market-data message.
39///
40/// The bus tap passes messages as `&dyn Any`, so implementations downcast to the type they were
41/// registered for and return `None` when the registry is miswired.
42pub trait DataMarkerExtractor: Send + Sync {
43    /// Returns the data class handled by this extractor.
44    fn data_class(&self) -> DataClass;
45
46    /// Returns the stable stream identifier for the message.
47    fn identifier(&self, msg: &dyn Any) -> Option<String>;
48
49    /// Returns `(ts_event, ts_init)` for the message.
50    fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)>;
51
52    /// Returns the class-specific canonical content fingerprint for the message.
53    fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]>;
54}
55
56/// Registry of marker extractors keyed by concrete message [`TypeId`].
57///
58/// Registration happens before the capture tap is installed. Lookups use the concrete type behind
59/// `&dyn Any`, so the hot path avoids trying every extractor.
60pub struct DataMarkerExtractorRegistry {
61    by_type: AHashMap<TypeId, Box<dyn DataMarkerExtractor>>,
62}
63
64impl DataMarkerExtractorRegistry {
65    /// Creates an empty extractor registry.
66    #[must_use]
67    pub fn new() -> Self {
68        Self {
69            by_type: AHashMap::new(),
70        }
71    }
72
73    /// Registers `ex` as the marker extractor for `T`.
74    ///
75    /// Replaces any previous extractor for `T`; callers should finish registration before sharing
76    /// the registry with the capture path.
77    pub fn register<T: 'static>(&mut self, ex: Box<dyn DataMarkerExtractor>) {
78        self.by_type.insert(TypeId::of::<T>(), ex);
79    }
80
81    /// Creates a registry containing builtin extractors for the enabled `classes`.
82    #[must_use]
83    pub fn default_registry(classes: &[DataClass]) -> Self {
84        let mut registry = Self::new();
85
86        for class in classes {
87            match class {
88                DataClass::BookDeltas => {
89                    registry.register::<OrderBookDeltas>(Box::new(OrderBookDeltasExtractor));
90                }
91                DataClass::BookDepth10 => {
92                    registry.register::<OrderBookDepth10>(Box::new(OrderBookDepth10Extractor));
93                }
94                DataClass::Quote => {
95                    registry.register::<QuoteTick>(Box::new(QuoteTickExtractor));
96                }
97                DataClass::Trade => {
98                    registry.register::<TradeTick>(Box::new(TradeTickExtractor));
99                }
100                DataClass::Bar => {
101                    registry.register::<Bar>(Box::new(BarExtractor));
102                }
103            }
104        }
105
106        registry
107    }
108
109    /// Returns the extractor registered for the concrete type behind `msg`.
110    #[must_use]
111    pub fn lookup(&self, msg: &dyn Any) -> Option<&dyn DataMarkerExtractor> {
112        self.by_type.get(&msg.type_id()).map(Box::as_ref)
113    }
114}
115
116impl Default for DataMarkerExtractorRegistry {
117    fn default() -> Self {
118        Self::new()
119    }
120}
121
122impl Debug for DataMarkerExtractorRegistry {
123    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        f.debug_struct(stringify!(DataMarkerExtractorRegistry))
125            .field("len", &self.by_type.len())
126            .finish()
127    }
128}
129
130#[derive(Debug)]
131struct QuoteTickExtractor;
132
133impl DataMarkerExtractor for QuoteTickExtractor {
134    fn data_class(&self) -> DataClass {
135        DataClass::Quote
136    }
137
138    fn identifier(&self, msg: &dyn Any) -> Option<String> {
139        msg.downcast_ref::<QuoteTick>()
140            .map(|quote| quote.instrument_id.to_string())
141    }
142
143    fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)> {
144        msg.downcast_ref::<QuoteTick>()
145            .map(|quote| (quote.ts_event, quote.ts_init))
146    }
147
148    fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]> {
149        msg.downcast_ref::<QuoteTick>().map(fingerprint_quote)
150    }
151}
152
153#[derive(Debug)]
154struct TradeTickExtractor;
155
156impl DataMarkerExtractor for TradeTickExtractor {
157    fn data_class(&self) -> DataClass {
158        DataClass::Trade
159    }
160
161    fn identifier(&self, msg: &dyn Any) -> Option<String> {
162        msg.downcast_ref::<TradeTick>()
163            .map(|trade| trade.instrument_id.to_string())
164    }
165
166    fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)> {
167        msg.downcast_ref::<TradeTick>()
168            .map(|trade| (trade.ts_event, trade.ts_init))
169    }
170
171    fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]> {
172        msg.downcast_ref::<TradeTick>().map(fingerprint_trade)
173    }
174}
175
176#[derive(Debug)]
177struct BarExtractor;
178
179impl DataMarkerExtractor for BarExtractor {
180    fn data_class(&self) -> DataClass {
181        DataClass::Bar
182    }
183
184    fn identifier(&self, msg: &dyn Any) -> Option<String> {
185        msg.downcast_ref::<Bar>()
186            .map(|bar| bar.bar_type.to_string())
187    }
188
189    fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)> {
190        msg.downcast_ref::<Bar>()
191            .map(|bar| (bar.ts_event, bar.ts_init))
192    }
193
194    fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]> {
195        msg.downcast_ref::<Bar>().map(fingerprint_bar)
196    }
197}
198
199#[derive(Debug)]
200struct OrderBookDepth10Extractor;
201
202impl DataMarkerExtractor for OrderBookDepth10Extractor {
203    fn data_class(&self) -> DataClass {
204        DataClass::BookDepth10
205    }
206
207    fn identifier(&self, msg: &dyn Any) -> Option<String> {
208        msg.downcast_ref::<OrderBookDepth10>()
209            .map(|depth| depth.instrument_id.to_string())
210    }
211
212    fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)> {
213        msg.downcast_ref::<OrderBookDepth10>()
214            .map(|depth| (depth.ts_event, depth.ts_init))
215    }
216
217    fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]> {
218        msg.downcast_ref::<OrderBookDepth10>()
219            .map(fingerprint_depth10)
220    }
221}
222
223#[derive(Debug)]
224struct OrderBookDeltasExtractor;
225
226impl DataMarkerExtractor for OrderBookDeltasExtractor {
227    fn data_class(&self) -> DataClass {
228        DataClass::BookDeltas
229    }
230
231    fn identifier(&self, msg: &dyn Any) -> Option<String> {
232        msg.downcast_ref::<OrderBookDeltas>()
233            .map(|deltas| deltas.instrument_id.to_string())
234    }
235
236    fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)> {
237        msg.downcast_ref::<OrderBookDeltas>()
238            .map(|deltas| (deltas.ts_event, deltas.ts_init))
239    }
240
241    fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]> {
242        msg.downcast_ref::<OrderBookDeltas>()
243            .map(fingerprint_deltas)
244    }
245}
246
247fn fingerprint_quote(quote: &QuoteTick) -> [u8; 32] {
248    let mut hasher = blake3::Hasher::new();
249    hasher.update(QUOTE_FINGERPRINT_DOMAIN);
250    write_price_raw(&mut hasher, quote.bid_price);
251    write_price_raw(&mut hasher, quote.ask_price);
252    write_quantity_raw(&mut hasher, quote.bid_size);
253    write_quantity_raw(&mut hasher, quote.ask_size);
254    write_unix_nanos(&mut hasher, quote.ts_event);
255    *hasher.finalize().as_bytes()
256}
257
258fn fingerprint_trade(trade: &TradeTick) -> [u8; 32] {
259    let mut hasher = blake3::Hasher::new();
260    hasher.update(TRADE_FINGERPRINT_DOMAIN);
261    write_price_raw(&mut hasher, trade.price);
262    write_quantity_raw(&mut hasher, trade.size);
263    hasher.update(&[trade.aggressor_side as u8]);
264    write_str(&mut hasher, trade.trade_id.as_str());
265    write_unix_nanos(&mut hasher, trade.ts_event);
266    *hasher.finalize().as_bytes()
267}
268
269fn fingerprint_bar(bar: &Bar) -> [u8; 32] {
270    let mut hasher = blake3::Hasher::new();
271    hasher.update(BAR_FINGERPRINT_DOMAIN);
272    write_str(&mut hasher, &bar.bar_type.to_string());
273    write_price_raw(&mut hasher, bar.open);
274    write_price_raw(&mut hasher, bar.high);
275    write_price_raw(&mut hasher, bar.low);
276    write_price_raw(&mut hasher, bar.close);
277    write_quantity_raw(&mut hasher, bar.volume);
278    write_unix_nanos(&mut hasher, bar.ts_event);
279    *hasher.finalize().as_bytes()
280}
281
282fn fingerprint_depth10(depth: &OrderBookDepth10) -> [u8; 32] {
283    let mut hasher = blake3::Hasher::new();
284    hasher.update(DEPTH10_FINGERPRINT_DOMAIN);
285    for (order, count) in depth.bids.iter().zip(depth.bid_counts) {
286        write_depth_level(&mut hasher, order, count);
287    }
288
289    for (order, count) in depth.asks.iter().zip(depth.ask_counts) {
290        write_depth_level(&mut hasher, order, count);
291    }
292    write_unix_nanos(&mut hasher, depth.ts_event);
293    *hasher.finalize().as_bytes()
294}
295
296fn fingerprint_deltas(deltas: &OrderBookDeltas) -> [u8; 32] {
297    let mut hasher = blake3::Hasher::new();
298    hasher.update(DELTAS_FINGERPRINT_DOMAIN);
299    hasher.update(&(deltas.deltas.len() as u64).to_be_bytes());
300    for delta in &deltas.deltas {
301        hasher.update(&[delta.action as u8]);
302        hasher.update(&[delta.order.side as u8]);
303        write_price_raw(&mut hasher, delta.order.price);
304        write_quantity_raw(&mut hasher, delta.order.size);
305        hasher.update(&delta.order.order_id.to_be_bytes());
306        hasher.update(&[delta.flags]);
307    }
308    write_unix_nanos(&mut hasher, deltas.ts_event);
309    *hasher.finalize().as_bytes()
310}
311
312fn write_depth_level(hasher: &mut blake3::Hasher, order: &BookOrder, count: u32) {
313    write_price_raw(hasher, order.price);
314    write_quantity_raw(hasher, order.size);
315    hasher.update(&count.to_be_bytes());
316}
317
318fn write_price_raw(hasher: &mut blake3::Hasher, price: Price) {
319    hasher.update(&[price.precision]);
320    hasher.update(&price_raw_at_precision(price).to_be_bytes());
321}
322
323fn write_quantity_raw(hasher: &mut blake3::Hasher, quantity: Quantity) {
324    hasher.update(&[quantity.precision]);
325    hasher.update(&quantity_raw_at_precision(quantity).to_be_bytes());
326}
327
328fn price_raw_at_precision(price: Price) -> i128 {
329    let scale_down = FIXED_PRECISION.saturating_sub(price.precision);
330    #[cfg(feature = "defi")]
331    let raw = price.raw;
332    #[cfg(not(feature = "defi"))]
333    let raw = i128::from(price.raw);
334
335    raw / 10_i128.pow(u32::from(scale_down))
336}
337
338fn quantity_raw_at_precision(quantity: Quantity) -> u128 {
339    let scale_down = FIXED_PRECISION.saturating_sub(quantity.precision);
340    #[cfg(feature = "defi")]
341    let raw = quantity.raw;
342    #[cfg(not(feature = "defi"))]
343    let raw = u128::from(quantity.raw);
344
345    raw / 10_u128.pow(u32::from(scale_down))
346}
347
348fn write_unix_nanos(hasher: &mut blake3::Hasher, ts: UnixNanos) {
349    hasher.update(&ts.as_u64().to_be_bytes());
350}
351
352fn write_str(hasher: &mut blake3::Hasher, value: &str) {
353    let bytes = value.as_bytes();
354    hasher.update(&(bytes.len() as u64).to_be_bytes());
355    hasher.update(bytes);
356}
357
358#[cfg(test)]
359mod tests {
360    use std::{any::Any, fmt::Write};
361
362    use nautilus_core::UnixNanos;
363    use nautilus_model::{
364        data::{
365            Bar, BarType, BookOrder, OrderBookDelta, OrderBookDeltas, OrderBookDepth10, QuoteTick,
366            TradeTick, depth::DEPTH10_LEN,
367        },
368        enums::{AggressorSide, BookAction, OrderSide},
369        identifiers::{InstrumentId, TradeId},
370        types::{Price, Quantity, price::PriceRaw, quantity::QuantityRaw},
371    };
372    use rstest::rstest;
373
374    use super::*;
375    use crate::markers::DataClass;
376
377    fn hex32(bytes: &[u8; 32]) -> String {
378        let mut out = String::with_capacity(64);
379        for byte in bytes {
380            write!(out, "{byte:02x}").expect("writing to a String is infallible");
381        }
382        out
383    }
384
385    fn quote_tick() -> QuoteTick {
386        QuoteTick::new(
387            InstrumentId::from("ETHUSDT.BINANCE"),
388            Price::from("3000.12"),
389            Price::from("3000.25"),
390            Quantity::from("1.25"),
391            Quantity::from("2.50"),
392            UnixNanos::from(1_700_000_000_000_000_100),
393            UnixNanos::from(1_700_000_000_000_000_200),
394        )
395    }
396
397    fn trade_tick() -> TradeTick {
398        TradeTick::new(
399            InstrumentId::from("ETHUSDT.BINANCE"),
400            Price::from("3000.18"),
401            Quantity::from("0.75"),
402            AggressorSide::Buyer,
403            TradeId::new("T-ABC-123"),
404            UnixNanos::from(1_700_000_000_000_000_300),
405            UnixNanos::from(1_700_000_000_000_000_400),
406        )
407    }
408
409    fn bar() -> Bar {
410        Bar::new(
411            BarType::from("ETHUSDT.BINANCE-1-MINUTE-LAST-EXTERNAL"),
412            Price::from("3000.00"),
413            Price::from("3010.50"),
414            Price::from("2995.25"),
415            Price::from("3005.75"),
416            Quantity::from("42.25"),
417            UnixNanos::from(1_700_000_000_000_000_500),
418            UnixNanos::from(1_700_000_000_000_000_600),
419        )
420    }
421
422    fn price_from_cents(cents: i64) -> Price {
423        let scale_down = FIXED_PRECISION.saturating_sub(2);
424        let scale = PriceRaw::from(10_i64.pow(u32::from(scale_down)));
425        Price::from_raw(PriceRaw::from(cents) * scale, 2)
426    }
427
428    fn quantity_from_cents(cents: u64) -> Quantity {
429        let scale_down = FIXED_PRECISION.saturating_sub(2);
430        let scale = QuantityRaw::from(10_u64.pow(u32::from(scale_down)));
431        Quantity::from_raw(QuantityRaw::from(cents) * scale, 2)
432    }
433
434    fn depth10() -> OrderBookDepth10 {
435        let instrument_id = InstrumentId::from("ETHUSDT.BINANCE");
436        let bids = std::array::from_fn(|i| {
437            let level = i64::try_from(i).expect("depth10 index fits i64");
438            let order_offset = u64::try_from(i).expect("depth10 index fits u64");
439
440            BookOrder::new(
441                OrderSide::Buy,
442                price_from_cents(300_000 - level),
443                quantity_from_cents(10_000 + order_offset),
444                1_000 + order_offset,
445            )
446        });
447        let asks = std::array::from_fn(|i| {
448            let level = i64::try_from(i).expect("depth10 index fits i64");
449            let order_offset = u64::try_from(i).expect("depth10 index fits u64");
450
451            BookOrder::new(
452                OrderSide::Sell,
453                price_from_cents(300_100 + level),
454                quantity_from_cents(20_000 + order_offset),
455                2_000 + order_offset,
456            )
457        });
458        let bid_counts =
459            std::array::from_fn(|i| 10 + u32::try_from(i).expect("depth10 index fits u32"));
460        let ask_counts =
461            std::array::from_fn(|i| 20 + u32::try_from(i).expect("depth10 index fits u32"));
462
463        OrderBookDepth10::new(
464            instrument_id,
465            bids,
466            asks,
467            bid_counts,
468            ask_counts,
469            0x20,
470            42,
471            UnixNanos::from(1_700_000_000_000_000_700),
472            UnixNanos::from(1_700_000_000_000_000_800),
473        )
474    }
475
476    fn deltas() -> OrderBookDeltas {
477        let instrument_id = InstrumentId::from("ETHUSDT.BINANCE");
478        let first = OrderBookDelta::new(
479            instrument_id,
480            BookAction::Add,
481            BookOrder::new(
482                OrderSide::Buy,
483                Price::from("3000.00"),
484                Quantity::from("1.10"),
485                10,
486            ),
487            0x01,
488            41,
489            UnixNanos::from(1_700_000_000_000_000_900),
490            UnixNanos::from(1_700_000_000_000_001_000),
491        );
492        let second = OrderBookDelta::new(
493            instrument_id,
494            BookAction::Update,
495            BookOrder::new(
496                OrderSide::Sell,
497                Price::from("3001.00"),
498                Quantity::from("2.20"),
499                11,
500            ),
501            0x20,
502            42,
503            UnixNanos::from(1_700_000_000_000_000_900),
504            UnixNanos::from(1_700_000_000_000_001_000),
505        );
506
507        OrderBookDeltas::new(instrument_id, vec![first, second])
508    }
509
510    fn extractor_for<'a>(
511        registry: &'a DataMarkerExtractorRegistry,
512        msg: &dyn Any,
513    ) -> &'a dyn DataMarkerExtractor {
514        registry
515            .lookup(msg)
516            .expect("registered data marker extractor")
517    }
518
519    #[rstest]
520    fn quote_extractor_fields_and_fingerprint() {
521        let quote = quote_tick();
522        let registry = DataMarkerExtractorRegistry::default_registry(&[DataClass::Quote]);
523        let extractor = extractor_for(&registry, &quote);
524
525        assert_eq!(extractor.data_class(), DataClass::Quote);
526        assert_eq!(
527            extractor.identifier(&quote),
528            Some("ETHUSDT.BINANCE".to_string())
529        );
530        assert_eq!(
531            extractor.timestamps(&quote),
532            Some((
533                UnixNanos::from(1_700_000_000_000_000_100),
534                UnixNanos::from(1_700_000_000_000_000_200),
535            ))
536        );
537        assert_eq!(
538            hex32(&extractor.fingerprint(&quote).expect("fingerprint")),
539            "7c6671e34f01b7b547ac8695c6d2cd19c1a37f6d2e3910d9195ed66fd4c02628"
540        );
541    }
542
543    #[rstest]
544    fn raw_writers_use_declared_precision_scale() {
545        assert_eq!(price_raw_at_precision(Price::from("3000.12")), 300_012);
546        assert_eq!(quantity_raw_at_precision(Quantity::from("1.25")), 125);
547    }
548
549    #[rstest]
550    fn trade_extractor_fields_and_fingerprint() {
551        let trade = trade_tick();
552        let registry = DataMarkerExtractorRegistry::default_registry(&[DataClass::Trade]);
553        let extractor = extractor_for(&registry, &trade);
554
555        assert_eq!(extractor.data_class(), DataClass::Trade);
556        assert_eq!(
557            extractor.identifier(&trade),
558            Some("ETHUSDT.BINANCE".to_string())
559        );
560        assert_eq!(
561            extractor.timestamps(&trade),
562            Some((
563                UnixNanos::from(1_700_000_000_000_000_300),
564                UnixNanos::from(1_700_000_000_000_000_400),
565            ))
566        );
567        assert_eq!(
568            hex32(&extractor.fingerprint(&trade).expect("fingerprint")),
569            "6b32a3187d353451a07d92a0d91051406ce4fe912010202b81616ed315f565cb"
570        );
571    }
572
573    #[rstest]
574    fn bar_extractor_fields_and_fingerprint() {
575        let bar = bar();
576        let registry = DataMarkerExtractorRegistry::default_registry(&[DataClass::Bar]);
577        let extractor = extractor_for(&registry, &bar);
578
579        assert_eq!(extractor.data_class(), DataClass::Bar);
580        assert_eq!(
581            extractor.identifier(&bar),
582            Some("ETHUSDT.BINANCE-1-MINUTE-LAST-EXTERNAL".to_string())
583        );
584        assert_eq!(
585            extractor.timestamps(&bar),
586            Some((
587                UnixNanos::from(1_700_000_000_000_000_500),
588                UnixNanos::from(1_700_000_000_000_000_600),
589            ))
590        );
591        assert_eq!(
592            hex32(&extractor.fingerprint(&bar).expect("fingerprint")),
593            "f2283ae7ed8d2e6a3874473b11935557b6bc2cbf20446419fbdff8fe51f91e84"
594        );
595    }
596
597    #[rstest]
598    fn depth10_extractor_fields_and_fingerprint() {
599        let depth = depth10();
600        let registry = DataMarkerExtractorRegistry::default_registry(&[DataClass::BookDepth10]);
601        let extractor = extractor_for(&registry, &depth);
602
603        assert_eq!(extractor.data_class(), DataClass::BookDepth10);
604        assert_eq!(
605            extractor.identifier(&depth),
606            Some("ETHUSDT.BINANCE".to_string())
607        );
608        assert_eq!(depth.bids.len(), DEPTH10_LEN);
609        assert_eq!(depth.asks.len(), DEPTH10_LEN);
610        assert_eq!(
611            extractor.timestamps(&depth),
612            Some((
613                UnixNanos::from(1_700_000_000_000_000_700),
614                UnixNanos::from(1_700_000_000_000_000_800),
615            ))
616        );
617        assert_eq!(
618            hex32(&extractor.fingerprint(&depth).expect("fingerprint")),
619            "432e1f3951c660acaefb3a44c429b050e5e8d5b6f9c3b89cc969edaad85e8e4d"
620        );
621    }
622
623    #[rstest]
624    fn deltas_extractor_fields_and_fingerprint() {
625        let deltas = deltas();
626        let registry = DataMarkerExtractorRegistry::default_registry(&[DataClass::BookDeltas]);
627        let extractor = extractor_for(&registry, &deltas);
628
629        assert_eq!(extractor.data_class(), DataClass::BookDeltas);
630        assert_eq!(
631            extractor.identifier(&deltas),
632            Some("ETHUSDT.BINANCE".to_string())
633        );
634        assert_eq!(deltas.deltas.len(), 2);
635        assert_eq!(
636            extractor.timestamps(&deltas),
637            Some((
638                UnixNanos::from(1_700_000_000_000_000_900),
639                UnixNanos::from(1_700_000_000_000_001_000),
640            ))
641        );
642        assert_eq!(
643            hex32(&extractor.fingerprint(&deltas).expect("fingerprint")),
644            "4bb7fedc4454d53e08300e1ae7a59e648747152134a921050a2c80edec6d8f9e"
645        );
646    }
647
648    #[rstest]
649    #[case::bid_price(|q: &mut QuoteTick| q.bid_price = Price::from("3000.13"))]
650    #[case::ask_price(|q: &mut QuoteTick| q.ask_price = Price::from("3000.26"))]
651    #[case::bid_size(|q: &mut QuoteTick| q.bid_size = Quantity::from("1.26"))]
652    #[case::ask_size(|q: &mut QuoteTick| q.ask_size = Quantity::from("2.51"))]
653    #[case::ts_event(|q: &mut QuoteTick| q.ts_event = UnixNanos::from(1))]
654    fn quote_fingerprint_changes_when_hashed_field_changes(#[case] mutate: fn(&mut QuoteTick)) {
655        let base = quote_tick();
656        let mut changed = base;
657        mutate(&mut changed);
658
659        assert_ne!(fingerprint_quote(&base), fingerprint_quote(&changed));
660    }
661
662    #[rstest]
663    #[case::price(|t: &mut TradeTick| t.price = Price::from("3000.19"))]
664    #[case::size(|t: &mut TradeTick| t.size = Quantity::from("0.76"))]
665    #[case::aggressor_side(|t: &mut TradeTick| t.aggressor_side = AggressorSide::Seller)]
666    #[case::trade_id(|t: &mut TradeTick| t.trade_id = TradeId::new("T-ABC-124"))]
667    #[case::ts_event(|t: &mut TradeTick| t.ts_event = UnixNanos::from(1))]
668    fn trade_fingerprint_changes_when_hashed_field_changes(#[case] mutate: fn(&mut TradeTick)) {
669        let base = trade_tick();
670        let mut changed = base;
671        mutate(&mut changed);
672
673        assert_ne!(fingerprint_trade(&base), fingerprint_trade(&changed));
674    }
675
676    #[rstest]
677    #[case::bar_type(|b: &mut Bar| b.bar_type = BarType::from("ETHUSDT.BINANCE-5-MINUTE-LAST-EXTERNAL"))]
678    #[case::open(|b: &mut Bar| b.open = Price::from("3000.01"))]
679    #[case::high(|b: &mut Bar| b.high = Price::from("3010.51"))]
680    #[case::low(|b: &mut Bar| b.low = Price::from("2995.26"))]
681    #[case::close(|b: &mut Bar| b.close = Price::from("3005.76"))]
682    #[case::volume(|b: &mut Bar| b.volume = Quantity::from("42.26"))]
683    #[case::ts_event(|b: &mut Bar| b.ts_event = UnixNanos::from(1))]
684    fn bar_fingerprint_changes_when_hashed_field_changes(#[case] mutate: fn(&mut Bar)) {
685        let base = bar();
686        let mut changed = base;
687        mutate(&mut changed);
688
689        assert_ne!(fingerprint_bar(&base), fingerprint_bar(&changed));
690    }
691
692    #[rstest]
693    #[case::bid_price(|d: &mut OrderBookDepth10| d.bids[0].price = price_from_cents(300_001))]
694    #[case::bid_size(|d: &mut OrderBookDepth10| d.bids[0].size = quantity_from_cents(10_001))]
695    #[case::bid_count(|d: &mut OrderBookDepth10| d.bid_counts[0] = 99)]
696    #[case::ask_price(|d: &mut OrderBookDepth10| d.asks[0].price = price_from_cents(300_101))]
697    #[case::ask_size(|d: &mut OrderBookDepth10| d.asks[0].size = quantity_from_cents(20_001))]
698    #[case::ask_count(|d: &mut OrderBookDepth10| d.ask_counts[0] = 99)]
699    #[case::ts_event(|d: &mut OrderBookDepth10| d.ts_event = UnixNanos::from(1))]
700    fn depth10_fingerprint_changes_when_hashed_field_changes(
701        #[case] mutate: fn(&mut OrderBookDepth10),
702    ) {
703        let base = depth10();
704        let mut changed = base;
705        mutate(&mut changed);
706
707        assert_ne!(fingerprint_depth10(&base), fingerprint_depth10(&changed));
708    }
709
710    #[rstest]
711    #[case::delta_count(|d: &mut OrderBookDeltas| {
712        let instrument_id = d.instrument_id;
713        d.deltas.push(OrderBookDelta::new(
714            instrument_id,
715            BookAction::Delete,
716            BookOrder::new(
717                OrderSide::Buy,
718                Price::from("2999.00"),
719                Quantity::from("3.30"),
720                12,
721            ),
722            0x20,
723            43,
724            UnixNanos::from(1_700_000_000_000_000_900),
725            UnixNanos::from(1_700_000_000_000_001_000),
726        ));
727    })]
728    #[case::action(|d: &mut OrderBookDeltas| d.deltas[0].action = BookAction::Delete)]
729    #[case::side(|d: &mut OrderBookDeltas| d.deltas[0].order.side = OrderSide::Sell)]
730    #[case::price(|d: &mut OrderBookDeltas| d.deltas[0].order.price = Price::from("3000.01"))]
731    #[case::size(|d: &mut OrderBookDeltas| d.deltas[0].order.size = Quantity::from("1.11"))]
732    #[case::order_id(|d: &mut OrderBookDeltas| d.deltas[0].order.order_id = 99)]
733    #[case::flags(|d: &mut OrderBookDeltas| d.deltas[0].flags = 0x02)]
734    #[case::order(|d: &mut OrderBookDeltas| d.deltas.reverse())]
735    #[case::ts_event(|d: &mut OrderBookDeltas| d.ts_event = UnixNanos::from(1))]
736    fn deltas_fingerprint_changes_when_hashed_field_changes(
737        #[case] mutate: fn(&mut OrderBookDeltas),
738    ) {
739        let base = deltas();
740        let mut changed = base.clone();
741        mutate(&mut changed);
742
743        assert_ne!(fingerprint_deltas(&base), fingerprint_deltas(&changed));
744    }
745
746    #[rstest]
747    fn fingerprint_domains_are_class_specific() {
748        let fingerprints = [
749            (DataClass::Quote, fingerprint_quote(&quote_tick())),
750            (DataClass::Trade, fingerprint_trade(&trade_tick())),
751            (DataClass::Bar, fingerprint_bar(&bar())),
752            (DataClass::BookDepth10, fingerprint_depth10(&depth10())),
753            (DataClass::BookDeltas, fingerprint_deltas(&deltas())),
754        ];
755
756        for (index, (left_class, left_fingerprint)) in fingerprints.iter().enumerate() {
757            for (right_class, right_fingerprint) in fingerprints.iter().skip(index + 1) {
758                assert_ne!(
759                    left_fingerprint, right_fingerprint,
760                    "{left_class:?} and {right_class:?} fingerprints should differ"
761                );
762            }
763        }
764    }
765
766    #[rstest]
767    fn default_registry_installs_only_enabled_builtins() {
768        let quote = quote_tick();
769        let trade = trade_tick();
770        let bar = bar();
771        let depth = depth10();
772        let deltas = deltas();
773        let registry = DataMarkerExtractorRegistry::default_registry(&[
774            DataClass::Quote,
775            DataClass::BookDepth10,
776        ]);
777
778        assert!(registry.lookup(&quote).is_some());
779        assert!(registry.lookup(&depth).is_some());
780        assert!(registry.lookup(&trade).is_none());
781        assert!(registry.lookup(&bar).is_none());
782        assert!(registry.lookup(&deltas).is_none());
783    }
784
785    #[rstest]
786    fn registry_returns_none_for_unregistered_type() {
787        let registry = DataMarkerExtractorRegistry::new();
788        let quote = quote_tick();
789        let value = 1_u8;
790
791        assert!(registry.lookup(&quote).is_none());
792        assert!(registry.lookup(&value).is_none());
793    }
794}