nautilus_data/option_chains/
aggregator.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-series option chain aggregator for event accumulation and snapshots.
17
18use std::collections::{BTreeMap, HashMap, HashSet};
19
20use nautilus_core::UnixNanos;
21use nautilus_model::{
22    data::{
23        QuoteTick,
24        option_chain::{OptionChainSlice, OptionGreeks, OptionStrikeData, StrikeRange},
25    },
26    enums::OptionKind,
27    identifiers::{InstrumentId, OptionSeriesId},
28    types::Price,
29};
30
31use super::{
32    AtmTracker,
33    constants::{DEFAULT_REBALANCE_COOLDOWN_NS, DEFAULT_REBALANCE_HYSTERESIS},
34};
35
36/// Per-series aggregator that accumulates quotes and greeks between snapshots.
37///
38/// Owns mutable accumulator buffers and produces immutable `OptionChainSlice`
39/// snapshots on each timer tick.
40#[derive(Debug)]
41pub struct OptionChainAggregator {
42    /// The option series identifier for this aggregator.
43    series_id: OptionSeriesId,
44    /// Defines which strikes to include in the active set.
45    strike_range: StrikeRange,
46    /// Tracks the current ATM price from market data events.
47    atm_tracker: AtmTracker,
48    /// All instruments for this series. Grows dynamically when the exchange
49    /// lists new strikes via [`Self::add_instrument`].
50    instruments: HashMap<InstrumentId, (Price, OptionKind)>,
51    /// Currently active instrument IDs (subset of `instruments`).
52    active_ids: HashSet<InstrumentId>,
53    /// The closest ATM strike at the time of the last rebalance.
54    last_atm_strike: Option<Price>,
55    /// Hysteresis band for ATM rebalancing.
56    hysteresis: f64,
57    /// Minimum nanoseconds between rebalances.
58    cooldown_ns: u64,
59    /// Timestamp of the last rebalance.
60    last_rebalance_ns: Option<UnixNanos>,
61    /// Maximum `ts_event` seen across all quote updates.
62    max_ts_event: UnixNanos,
63    /// Greeks received before the corresponding quote arrived.
64    pending_greeks: HashMap<InstrumentId, OptionGreeks>,
65    /// Call option accumulator buffer keyed by strike price.
66    call_buffer: BTreeMap<Price, OptionStrikeData>,
67    /// Put option accumulator buffer keyed by strike price.
68    put_buffer: BTreeMap<Price, OptionStrikeData>,
69}
70
71impl OptionChainAggregator {
72    /// Creates a new aggregator for the given series.
73    ///
74    /// `instruments` contains ALL instruments for the series. The initial
75    /// `active_ids` subset is resolved from the strike range and the current
76    /// ATM price (if available). When no ATM price is set for ATM-based
77    /// ranges, all instruments are active.
78    pub fn new(
79        series_id: OptionSeriesId,
80        strike_range: StrikeRange,
81        atm_tracker: AtmTracker,
82        instruments: HashMap<InstrumentId, (Price, OptionKind)>,
83    ) -> Self {
84        let mut aggregator = Self {
85            series_id,
86            strike_range,
87            atm_tracker,
88            instruments,
89            active_ids: HashSet::new(),
90            last_atm_strike: None,
91            hysteresis: DEFAULT_REBALANCE_HYSTERESIS,
92            cooldown_ns: DEFAULT_REBALANCE_COOLDOWN_NS,
93            last_rebalance_ns: None,
94            max_ts_event: UnixNanos::default(),
95            pending_greeks: HashMap::new(),
96            call_buffer: BTreeMap::new(),
97            put_buffer: BTreeMap::new(),
98        };
99        // No Greeks exist at construction, so a `Delta` range resolves to its ATM fallback.
100        aggregator.recompute_active_set();
101        aggregator
102    }
103
104    /// Returns a mutable reference to the ATM tracker.
105    pub fn atm_tracker_mut(&mut self) -> &mut AtmTracker {
106        &mut self.atm_tracker
107    }
108
109    /// Returns the currently active instrument IDs.
110    #[must_use]
111    pub fn instrument_ids(&self) -> Vec<InstrumentId> {
112        self.active_ids.iter().copied().collect()
113    }
114
115    /// Returns a reference to the active instrument ID set.
116    #[must_use]
117    pub fn active_ids(&self) -> &HashSet<InstrumentId> {
118        &self.active_ids
119    }
120
121    /// Returns the series ID.
122    #[must_use]
123    pub fn series_id(&self) -> OptionSeriesId {
124        self.series_id
125    }
126
127    /// Returns `true` if the given timestamp is at or past the series expiration.
128    #[must_use]
129    pub fn is_expired(&self, now_ns: UnixNanos) -> bool {
130        now_ns >= self.series_id.expiration_ns
131    }
132
133    /// Returns a reference to the full instrument set.
134    #[must_use]
135    pub fn instruments(&self) -> &HashMap<InstrumentId, (Price, OptionKind)> {
136        &self.instruments
137    }
138
139    /// Returns all instrument IDs in the full set.
140    #[must_use]
141    pub fn all_instrument_ids(&self) -> Vec<InstrumentId> {
142        self.instruments.keys().copied().collect()
143    }
144
145    /// Returns `true` if the instrument catalog is empty.
146    #[must_use]
147    pub fn is_catalog_empty(&self) -> bool {
148        self.instruments.is_empty()
149    }
150
151    /// Permanently removes an instrument from the catalog.
152    ///
153    /// Removes from `instruments`, `active_ids`, `pending_greeks`, and cleans
154    /// buffer entries (only if no other instrument shares the same strike+kind).
155    /// Returns `true` if the instrument was found and removed.
156    #[must_use]
157    pub fn remove_instrument(&mut self, instrument_id: &InstrumentId) -> bool {
158        let Some((strike, kind)) = self.instruments.remove(instrument_id) else {
159            return false;
160        };
161
162        self.active_ids.remove(instrument_id);
163        self.pending_greeks.remove(instrument_id);
164
165        // Only remove buffer entry if no sibling instrument shares the same strike+kind
166        let has_sibling = self
167            .instruments
168            .values()
169            .any(|(s, k)| *s == strike && *k == kind);
170
171        if !has_sibling {
172            let buffer = match kind {
173                OptionKind::Call => &mut self.call_buffer,
174                OptionKind::Put => &mut self.put_buffer,
175            };
176            buffer.remove(&strike);
177        }
178
179        true
180    }
181
182    /// Returns a reference to the ATM tracker.
183    #[must_use]
184    pub fn atm_tracker(&self) -> &AtmTracker {
185        &self.atm_tracker
186    }
187
188    /// Recomputes the active instrument set from the current ATM price.
189    ///
190    /// Returns the new active instrument IDs. Used during bootstrap when the
191    /// first ATM price arrives after deferred subscription setup.
192    pub fn recompute_active_set(&mut self) -> Vec<InstrumentId> {
193        let atm_price = self.atm_tracker.atm_price();
194        let all_strikes = Self::sorted_strikes(&self.instruments);
195        let active_strikes: HashSet<Price> = self
196            .resolve_active_strikes(atm_price, &all_strikes)
197            .into_iter()
198            .collect();
199        self.active_ids = self
200            .instruments
201            .iter()
202            .filter(|(_, (strike, _))| active_strikes.contains(strike))
203            .map(|(id, _)| *id)
204            .collect();
205        self.last_atm_strike =
206            atm_price.and_then(|atm| Self::find_closest_strike(&all_strikes, atm));
207        self.active_ids.iter().copied().collect()
208    }
209
210    /// Resolves the active strikes for the current strike range.
211    ///
212    /// `Delta` is resolved here from stored Greeks (see [`Self::resolve_delta`]);
213    /// the price-based variants delegate to [`StrikeRange::resolve`].
214    fn resolve_active_strikes(
215        &self,
216        atm_price: Option<Price>,
217        all_strikes: &[Price],
218    ) -> Vec<Price> {
219        match &self.strike_range {
220            StrikeRange::Delta { target, tolerance } => {
221                self.resolve_delta(*target, *tolerance, atm_price, all_strikes)
222            }
223            _ => self.strike_range.resolve(atm_price, all_strikes),
224        }
225    }
226
227    /// Resolves strikes whose buffered or pending Greeks have an absolute delta
228    /// within `tolerance` of `target`.
229    ///
230    /// A strike qualifies when either its call or put delta magnitude matches
231    /// (calls have positive delta, puts negative; both are compared by absolute
232    /// value), so a typical target selects an OTM strike on each side of ATM.
233    /// Strikes with only pending Greeks (received before their first quote) are
234    /// eligible. Before the resolver changes from a fallback set to a selected
235    /// set, every current fallback leg must have Greeks. This avoids unsubscribing
236    /// legs whose Greeks have not arrived yet, including when the fallback window
237    /// shifts with ATM. When no Greeks fall in the band, this falls back to the
238    /// ATM-relative window from [`StrikeRange::resolve`].
239    fn resolve_delta(
240        &self,
241        target: f64,
242        tolerance: f64,
243        atm_price: Option<Price>,
244        all_strikes: &[Price],
245    ) -> Vec<Price> {
246        let selected: Vec<Price> = self
247            .deltas_by_strike()
248            .into_iter()
249            .filter(|(_, deltas)| {
250                deltas
251                    .iter()
252                    .any(|delta| Self::delta_within_band(*delta, target, tolerance))
253            })
254            .map(|(strike, _)| strike)
255            .collect();
256
257        let fallback_strikes = self.strike_range.resolve(atm_price, all_strikes);
258
259        if selected.is_empty() {
260            return fallback_strikes;
261        }
262
263        let selected_ids = self.instrument_ids_for_strikes(&selected);
264        let fallback_ids = self.instrument_ids_for_strikes(&fallback_strikes);
265
266        if self.active_ids != selected_ids && !self.delta_window_ready(&fallback_ids) {
267            return fallback_strikes;
268        }
269
270        selected
271    }
272
273    fn instrument_ids_for_strikes(&self, strikes: &[Price]) -> HashSet<InstrumentId> {
274        let strike_set: HashSet<Price> = strikes.iter().copied().collect();
275        self.instruments
276            .iter()
277            .filter(|(_, (strike, _))| strike_set.contains(strike))
278            .map(|(id, _)| *id)
279            .collect()
280    }
281
282    fn delta_window_ready(&self, instrument_ids: &HashSet<InstrumentId>) -> bool {
283        !instrument_ids.is_empty()
284            && instrument_ids
285                .iter()
286                .all(|id| self.instrument_has_greeks(id))
287    }
288
289    fn instrument_has_greeks(&self, instrument_id: &InstrumentId) -> bool {
290        if self.pending_greeks.contains_key(instrument_id) {
291            return true;
292        }
293
294        let Some((strike, kind)) = self.instruments.get(instrument_id) else {
295            return false;
296        };
297        let buffer = match kind {
298            OptionKind::Call => &self.call_buffer,
299            OptionKind::Put => &self.put_buffer,
300        };
301
302        buffer
303            .get(strike)
304            .and_then(|data| data.greeks.as_ref())
305            .is_some()
306    }
307
308    /// Collects every reported delta per strike, from buffered Greeks and from
309    /// Greeks still pending their first quote.
310    fn deltas_by_strike(&self) -> BTreeMap<Price, Vec<f64>> {
311        let mut deltas_by_strike: BTreeMap<Price, Vec<f64>> = BTreeMap::new();
312
313        for (strike, data) in self.call_buffer.iter().chain(self.put_buffer.iter()) {
314            if let Some(greeks) = data.greeks.as_ref() {
315                deltas_by_strike
316                    .entry(*strike)
317                    .or_default()
318                    .push(greeks.delta);
319            }
320        }
321
322        for (id, greeks) in &self.pending_greeks {
323            if let Some((strike, _)) = self.instruments.get(id) {
324                deltas_by_strike
325                    .entry(*strike)
326                    .or_default()
327                    .push(greeks.delta);
328            }
329        }
330
331        deltas_by_strike
332    }
333
334    /// Returns `true` when `delta`'s magnitude is within `tolerance` of `target`.
335    ///
336    /// Compares by absolute value so a put (negative delta) matches the same
337    /// target as the equivalent call.
338    fn delta_within_band(delta: f64, target: f64, tolerance: f64) -> bool {
339        (delta.abs() - target).abs() <= tolerance
340    }
341
342    /// Adds a newly discovered instrument to the series.
343    ///
344    /// Returns `true` if the instrument was newly inserted. Returns `false`
345    /// if it was already known (no-op). When the new instrument's strike
346    /// falls within the current active range, it is immediately added to
347    /// `active_ids`.
348    #[must_use]
349    pub fn add_instrument(
350        &mut self,
351        instrument_id: InstrumentId,
352        strike: Price,
353        kind: OptionKind,
354    ) -> bool {
355        if self.instruments.contains_key(&instrument_id) {
356            return false;
357        }
358
359        self.instruments.insert(instrument_id, (strike, kind));
360
361        // Determine if the new strike is in the current active range
362        let all_strikes = Self::sorted_strikes(&self.instruments);
363        let atm_price = self.atm_tracker.atm_price();
364        let active_strikes: HashSet<Price> = self
365            .resolve_active_strikes(atm_price, &all_strikes)
366            .into_iter()
367            .collect();
368
369        if active_strikes.contains(&strike) {
370            self.active_ids.insert(instrument_id);
371        }
372
373        true
374    }
375
376    /// Returns sorted, deduplicated strikes from the given instruments.
377    fn sorted_strikes(instruments: &HashMap<InstrumentId, (Price, OptionKind)>) -> Vec<Price> {
378        let mut strikes: Vec<Price> = instruments.values().map(|(s, _)| *s).collect();
379        strikes.sort();
380        strikes.dedup();
381        strikes
382    }
383
384    /// Finds the strike in `all_strikes` closest to `atm`.
385    fn find_closest_strike(all_strikes: &[Price], atm: Price) -> Option<Price> {
386        all_strikes
387            .iter()
388            .min_by(|a, b| {
389                let da = (a.as_f64() - atm.as_f64()).abs();
390                let db = (b.as_f64() - atm.as_f64()).abs();
391                da.partial_cmp(&db).unwrap_or(std::cmp::Ordering::Equal)
392            })
393            .copied()
394    }
395
396    /// Handles an incoming quote tick by updating the accumulator buffers.
397    pub fn update_quote(&mut self, quote: &QuoteTick) {
398        if self.is_expired(quote.ts_event) {
399            log::warn!(
400                "Dropping quote for {}, series {} expired at {}",
401                quote.instrument_id,
402                self.series_id,
403                self.series_id.expiration_ns,
404            );
405            return;
406        }
407
408        if !self.active_ids.contains(&quote.instrument_id) {
409            return;
410        }
411
412        if let Some(&(strike, kind)) = self.instruments.get(&quote.instrument_id) {
413            // Track max ts_event across all quotes
414            if quote.ts_event > self.max_ts_event {
415                self.max_ts_event = quote.ts_event;
416            }
417
418            let buffer = match kind {
419                OptionKind::Call => &mut self.call_buffer,
420                OptionKind::Put => &mut self.put_buffer,
421            };
422
423            match buffer.get_mut(&strike) {
424                Some(data) => data.quote = *quote,
425                None => {
426                    // Check for pending greeks that arrived before this first quote
427                    let greeks = self.pending_greeks.remove(&quote.instrument_id);
428                    buffer.insert(
429                        strike,
430                        OptionStrikeData {
431                            quote: *quote,
432                            greeks,
433                        },
434                    );
435                }
436            }
437        }
438    }
439
440    /// Handles incoming greeks by updating the accumulator buffers.
441    ///
442    /// If no quote has arrived yet for this instrument (no buffer entry),
443    /// the greeks are stored in `pending_greeks` and will be attached when
444    /// the first quote arrives.
445    pub fn update_greeks(&mut self, greeks: &OptionGreeks) {
446        if self.is_expired(greeks.ts_event) {
447            log::warn!(
448                "Dropping greeks for {}, series {} expired at {}",
449                greeks.instrument_id,
450                self.series_id,
451                self.series_id.expiration_ns,
452            );
453            return;
454        }
455
456        if !self.active_ids.contains(&greeks.instrument_id) {
457            return;
458        }
459
460        if let Some(&(strike, kind)) = self.instruments.get(&greeks.instrument_id) {
461            let buffer = match kind {
462                OptionKind::Call => &mut self.call_buffer,
463                OptionKind::Put => &mut self.put_buffer,
464            };
465
466            match buffer.get_mut(&strike) {
467                Some(data) => data.greeks = Some(*greeks),
468                None => {
469                    // No quote yet: park the greeks for later
470                    self.pending_greeks.insert(greeks.instrument_id, *greeks);
471                }
472            }
473        }
474    }
475
476    /// Creates a point-in-time snapshot from accumulated buffers, applying strike filtering.
477    ///
478    /// Buffers are preserved (keep-latest semantics) so instruments that didn't
479    /// quote since the last tick are still included in subsequent snapshots.
480    ///
481    /// # Panics
482    ///
483    /// Panics if strike prices cannot be compared (NaN values).
484    pub fn snapshot(&self, ts_init: UnixNanos) -> OptionChainSlice {
485        let atm_price = self.atm_tracker.atm_price();
486
487        // Use catalog strikes for ATM strike (most accurate closest-strike lookup)
488        let catalog_strikes = Self::sorted_strikes(&self.instruments);
489        let atm_strike = atm_price.and_then(|atm| Self::find_closest_strike(&catalog_strikes, atm));
490
491        // Filter buffers using active set strikes directly. The active set is already
492        // the result of strike range resolution from the last rebalance. Re-resolving
493        // here would shift the window during hysteresis/cooldown, dropping buffered data.
494        let active_strikes: HashSet<Price> = self
495            .active_ids
496            .iter()
497            .filter_map(|id| self.instruments.get(id).map(|(s, _)| *s))
498            .collect();
499
500        // Build filtered snapshot (clone from buffers)
501        let mut calls = BTreeMap::new();
502
503        for (strike, data) in &self.call_buffer {
504            if active_strikes.contains(strike) {
505                calls.insert(*strike, data.clone());
506            }
507        }
508        let mut puts = BTreeMap::new();
509
510        for (strike, data) in &self.put_buffer {
511            if active_strikes.contains(strike) {
512                puts.insert(*strike, data.clone());
513            }
514        }
515
516        // Use the max observed ts_event from quotes, falling back to ts_init
517        let ts_event = if self.max_ts_event == UnixNanos::default() {
518            ts_init
519        } else {
520            self.max_ts_event
521        };
522
523        OptionChainSlice {
524            series_id: self.series_id,
525            atm_strike,
526            calls,
527            puts,
528            ts_event,
529            ts_init,
530        }
531    }
532
533    /// Returns `true` if both buffers are empty.
534    #[must_use]
535    pub fn is_buffer_empty(&self) -> bool {
536        self.call_buffer.is_empty() && self.put_buffer.is_empty()
537    }
538
539    /// Checks whether the instrument set should be rebalanced around the current ATM.
540    ///
541    /// Returns `None` when no rebalancing is needed (fixed ranges, no ATM price,
542    /// ATM strike unchanged, hysteresis not exceeded, or cooldown not elapsed).
543    /// Returns `Some(RebalanceAction)` with instrument add/remove lists when the
544    /// closest ATM strike shifts past the hysteresis threshold.
545    ///
546    /// `Delta` ranges resolve from Greeks rather than an ATM window, so their
547    /// active set can change while the closest ATM strike is unchanged. They skip
548    /// the ATM-shift and hysteresis gates and rebalance on any resolved-set change,
549    /// with the cooldown still applied to throttle churn.
550    #[must_use]
551    pub fn check_rebalance(&self, now_ns: UnixNanos) -> Option<RebalanceAction> {
552        // Fixed ranges never rebalance
553        if matches!(self.strike_range, StrikeRange::Fixed(_)) {
554            return None;
555        }
556
557        let atm_price = self.atm_tracker.atm_price()?;
558        let all_strikes = Self::sorted_strikes(&self.instruments);
559        let current_atm_strike = Self::find_closest_strike(&all_strikes, atm_price)?;
560
561        let is_delta = matches!(self.strike_range, StrikeRange::Delta { .. });
562
563        if !is_delta {
564            // No change: no rebalance
565            if self.last_atm_strike == Some(current_atm_strike) {
566                return None;
567            }
568
569            // Hysteresis check: price must cross hysteresis fraction of the gap to next strike
570            if let Some(last_strike) = self.last_atm_strike
571                && self.hysteresis > 0.0
572            {
573                let last_f = last_strike.as_f64();
574                let atm_f = atm_price.as_f64();
575                let direction = atm_f - last_f;
576
577                // Find the next strike in the direction of price movement
578                let next_strike = if direction > 0.0 {
579                    all_strikes.iter().find(|s| s.as_f64() > last_f)
580                } else {
581                    all_strikes.iter().rev().find(|s| s.as_f64() < last_f)
582                };
583
584                if let Some(next) = next_strike {
585                    let gap = (next.as_f64() - last_f).abs();
586                    let threshold = last_f + direction.signum() * self.hysteresis * gap;
587                    // Check if price has not crossed the threshold
588                    if direction > 0.0 && atm_f < threshold {
589                        return None;
590                    }
591
592                    if direction < 0.0 && atm_f > threshold {
593                        return None;
594                    }
595                }
596            }
597        }
598
599        // Cooldown check
600        if self.cooldown_ns > 0
601            && let Some(last_ts) = self.last_rebalance_ns
602            && now_ns.as_u64().saturating_sub(last_ts.as_u64()) < self.cooldown_ns
603        {
604            return None;
605        }
606
607        // Compute new active set
608        let new_active_strikes: HashSet<Price> = self
609            .resolve_active_strikes(Some(atm_price), &all_strikes)
610            .into_iter()
611            .collect();
612        let new_active: HashSet<InstrumentId> = self
613            .instruments
614            .iter()
615            .filter(|(_, (s, _))| new_active_strikes.contains(s))
616            .map(|(id, _)| *id)
617            .collect();
618
619        let add: Vec<InstrumentId> = new_active.difference(&self.active_ids).copied().collect();
620        let remove: Vec<InstrumentId> = self.active_ids.difference(&new_active).copied().collect();
621
622        // Suppress no-op delta rebalances so the cooldown timestamp is not reset on
623        // every snapshot while the resolved set is stable.
624        if is_delta && add.is_empty() && remove.is_empty() {
625            return None;
626        }
627
628        Some(RebalanceAction { add, remove })
629    }
630
631    /// Applies a rebalance action: updates the active ID set, cleans stale buffers,
632    /// and records the rebalance timestamp.
633    pub fn apply_rebalance(&mut self, action: &RebalanceAction, now_ns: UnixNanos) {
634        for id in &action.add {
635            self.active_ids.insert(*id);
636        }
637
638        for id in &action.remove {
639            self.active_ids.remove(id);
640        }
641
642        // Clean buffers for strikes no longer in active set
643        let active_strikes: HashSet<Price> = self
644            .active_ids
645            .iter()
646            .filter_map(|id| self.instruments.get(id))
647            .map(|(s, _)| *s)
648            .collect();
649        self.call_buffer
650            .retain(|strike, _| active_strikes.contains(strike));
651        self.put_buffer
652            .retain(|strike, _| active_strikes.contains(strike));
653        self.pending_greeks
654            .retain(|id, _| self.active_ids.contains(id));
655
656        // Update last_atm_strike and record rebalance timestamp
657        if let Some(atm) = self.atm_tracker.atm_price() {
658            let all_strikes = Self::sorted_strikes(&self.instruments);
659            self.last_atm_strike = Self::find_closest_strike(&all_strikes, atm);
660        }
661        self.last_rebalance_ns = Some(now_ns);
662    }
663}
664
665/// Describes instruments to add and remove during an ATM rebalance.
666#[derive(Clone, Debug, PartialEq, Eq)]
667pub struct RebalanceAction {
668    /// Instruments to subscribe to (newly in range).
669    pub add: Vec<InstrumentId>,
670    /// Instruments to unsubscribe from (no longer in range).
671    pub remove: Vec<InstrumentId>,
672}
673
674#[cfg(test)]
675impl OptionChainAggregator {
676    fn call_buffer_len(&self) -> usize {
677        self.call_buffer.len()
678    }
679
680    fn put_buffer_len(&self) -> usize {
681        self.put_buffer.len()
682    }
683
684    fn get_call_greeks_from_buffer(&self, strike: &Price) -> Option<&OptionGreeks> {
685        self.call_buffer.get(strike).and_then(|d| d.greeks.as_ref())
686    }
687
688    pub(crate) fn last_atm_strike(&self) -> Option<Price> {
689        self.last_atm_strike
690    }
691
692    fn set_hysteresis(&mut self, h: f64) {
693        self.hysteresis = h;
694    }
695
696    fn set_cooldown_ns(&mut self, ns: u64) {
697        self.cooldown_ns = ns;
698    }
699
700    fn pending_greeks_count(&self) -> usize {
701        self.pending_greeks.len()
702    }
703}
704
705#[cfg(test)]
706mod tests {
707    use nautilus_model::{data::greeks::OptionGreekValues, identifiers::Venue, types::Quantity};
708    use rstest::*;
709
710    use super::*;
711
712    fn make_series_id() -> OptionSeriesId {
713        OptionSeriesId::new(
714            Venue::new("DERIBIT"),
715            ustr::Ustr::from("BTC"),
716            ustr::Ustr::from("BTC"),
717            UnixNanos::from(1_700_000_000_000_000_000u64),
718        )
719    }
720
721    fn make_quote(instrument_id: InstrumentId, bid: &str, ask: &str) -> QuoteTick {
722        QuoteTick::new(
723            instrument_id,
724            Price::from(bid),
725            Price::from(ask),
726            Quantity::from("1.0"),
727            Quantity::from("1.0"),
728            UnixNanos::from(1u64),
729            UnixNanos::from(1u64),
730        )
731    }
732
733    fn now() -> UnixNanos {
734        // A base timestamp for tests (far enough from zero to avoid edge cases)
735        UnixNanos::from(1_000_000_000_000_000_000u64)
736    }
737
738    /// Sets ATM price on an aggregator via a synthetic `OptionGreeks` with the given forward price.
739    fn set_atm_via_greeks(agg: &mut OptionChainAggregator, price: f64) {
740        let greeks = OptionGreeks {
741            instrument_id: InstrumentId::from("BTC-20240101-50000-C.DERIBIT"),
742            underlying_price: Some(price),
743            ..Default::default()
744        };
745        agg.atm_tracker_mut().update_from_option_greeks(&greeks);
746    }
747
748    fn make_aggregator() -> (OptionChainAggregator, InstrumentId, InstrumentId) {
749        let call_id = InstrumentId::from("BTC-20240101-50000-C.DERIBIT");
750        let put_id = InstrumentId::from("BTC-20240101-50000-P.DERIBIT");
751        let strike = Price::from("50000");
752
753        let mut instrument_map = HashMap::new();
754        instrument_map.insert(call_id, (strike, OptionKind::Call));
755        instrument_map.insert(put_id, (strike, OptionKind::Put));
756
757        let tracker = AtmTracker::new();
758        let agg = OptionChainAggregator::new(
759            make_series_id(),
760            StrikeRange::Fixed(vec![strike]),
761            tracker,
762            instrument_map,
763        );
764
765        (agg, call_id, put_id)
766    }
767
768    #[rstest]
769    fn test_aggregator_instrument_ids() {
770        let (agg, call_id, put_id) = make_aggregator();
771        let ids = agg.instrument_ids();
772        assert_eq!(ids.len(), 2);
773        assert!(ids.contains(&call_id));
774        assert!(ids.contains(&put_id));
775    }
776
777    #[rstest]
778    fn test_aggregator_update_quote() {
779        let (mut agg, call_id, _) = make_aggregator();
780        let quote = make_quote(call_id, "100.00", "101.00");
781
782        agg.update_quote(&quote);
783
784        assert_eq!(agg.call_buffer_len(), 1);
785        assert_eq!(agg.put_buffer_len(), 0);
786    }
787
788    #[rstest]
789    fn test_aggregator_update_greeks() {
790        let (mut agg, call_id, _) = make_aggregator();
791        let quote = make_quote(call_id, "100.00", "101.00");
792        agg.update_quote(&quote);
793
794        let greeks = OptionGreeks {
795            instrument_id: call_id,
796            greeks: OptionGreekValues {
797                delta: 0.55,
798                ..Default::default()
799            },
800            ..Default::default()
801        };
802        agg.update_greeks(&greeks);
803
804        let strike = Price::from("50000");
805        let data = agg.get_call_greeks_from_buffer(&strike);
806        assert!(data.is_some());
807        assert_eq!(data.unwrap().delta, 0.55);
808    }
809
810    #[rstest]
811    fn test_aggregator_snapshot_preserves_state() {
812        let (mut agg, call_id, _) = make_aggregator();
813        let quote = make_quote(call_id, "100.00", "101.00");
814        agg.update_quote(&quote);
815
816        let slice = agg.snapshot(UnixNanos::from(100u64));
817        assert_eq!(slice.call_count(), 1);
818        assert_eq!(slice.ts_init, UnixNanos::from(100u64));
819
820        // Buffers should still contain data (keep-latest semantics)
821        assert!(!agg.is_buffer_empty());
822
823        // Second snapshot should return the same data
824        let slice2 = agg.snapshot(UnixNanos::from(200u64));
825        assert_eq!(slice2.call_count(), 1);
826        assert_eq!(slice2.ts_init, UnixNanos::from(200u64));
827    }
828
829    #[rstest]
830    fn test_aggregator_ignores_unknown_instrument() {
831        let (mut agg, _, _) = make_aggregator();
832        let unknown_id = InstrumentId::from("ETH-20240101-3000-C.DERIBIT");
833        let quote = make_quote(unknown_id, "100.00", "101.00");
834
835        agg.update_quote(&quote);
836
837        assert!(agg.is_buffer_empty());
838    }
839
840    #[rstest]
841    fn test_check_rebalance_returns_none() {
842        let (agg, _, _) = make_aggregator();
843        assert!(agg.check_rebalance(now()).is_none());
844    }
845
846    // -- Rebalance tests --
847
848    /// Builds instruments with 5 strike prices (45000..55000 step 2500) and `AtmRelative` +-1.
849    /// Hysteresis and cooldown are disabled so existing rebalance tests pass unchanged.
850    fn make_multi_strike_aggregator() -> OptionChainAggregator {
851        let strikes = [45000, 47500, 50000, 52500, 55000];
852        let mut instruments = HashMap::new();
853
854        for s in &strikes {
855            let strike = Price::from(&s.to_string());
856            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
857            let put_id = InstrumentId::from(&format!("BTC-20240101-{s}-P.DERIBIT"));
858            instruments.insert(call_id, (strike, OptionKind::Call));
859            instruments.insert(put_id, (strike, OptionKind::Put));
860        }
861
862        let tracker = AtmTracker::new();
863        let mut agg = OptionChainAggregator::new(
864            make_series_id(),
865            StrikeRange::AtmRelative {
866                strikes_above: 1,
867                strikes_below: 1,
868            },
869            tracker,
870            instruments,
871        );
872        // Disable guards so existing tests exercise pure rebalance logic
873        agg.set_hysteresis(0.0);
874        agg.set_cooldown_ns(0);
875        agg
876    }
877
878    #[rstest]
879    fn test_check_rebalance_fixed_always_none() {
880        // Fixed range + ATM price set: still returns None
881        let (mut agg, _, _) = make_aggregator();
882        set_atm_via_greeks(&mut agg, 50000.0);
883        assert!(agg.check_rebalance(now()).is_none());
884    }
885
886    #[rstest]
887    fn test_check_rebalance_no_atm_returns_none() {
888        let agg = make_multi_strike_aggregator();
889        // No ATM price set: None
890        assert!(agg.check_rebalance(now()).is_none());
891    }
892
893    #[rstest]
894    fn test_check_rebalance_atm_unchanged_returns_none() {
895        let mut agg = make_multi_strike_aggregator();
896        // Set ATM to 50000 and apply initial rebalance
897        set_atm_via_greeks(&mut agg, 50000.0);
898        // First check detects ATM shift from None to 50000
899        let action = agg.check_rebalance(now()).unwrap();
900        agg.apply_rebalance(&action, now());
901
902        // ATM moves slightly but stays closest to 50000
903        set_atm_via_greeks(&mut agg, 50200.0);
904        assert!(agg.check_rebalance(now()).is_none());
905    }
906
907    #[rstest]
908    fn test_check_rebalance_detects_atm_shift() {
909        let mut agg = make_multi_strike_aggregator();
910        // Set ATM near 50000
911        set_atm_via_greeks(&mut agg, 50000.0);
912        let action = agg.check_rebalance(now()).unwrap();
913        agg.apply_rebalance(&action, now());
914        // Active: 47500, 50000, 52500 (ATM=50000, +-1 strike)
915        assert_eq!(agg.instrument_ids().len(), 6); // 3 strikes * 2
916
917        // Now shift ATM to 55000
918        set_atm_via_greeks(&mut agg, 55000.0);
919        let action2 = agg.check_rebalance(now()).unwrap();
920        // Should have instruments to add (55000) and remove (47500)
921        assert!(!action2.add.is_empty() || !action2.remove.is_empty());
922    }
923
924    #[rstest]
925    fn test_apply_rebalance_updates_instrument_map() {
926        let mut agg = make_multi_strike_aggregator();
927        // Set ATM near 50000
928        set_atm_via_greeks(&mut agg, 50000.0);
929        let action = agg.check_rebalance(now()).unwrap();
930        agg.apply_rebalance(&action, now());
931
932        // Active should be 3 strikes (47500, 50000, 52500)
933        let active_ids = agg.instrument_ids();
934        assert_eq!(active_ids.len(), 6); // 3 strikes * 2 (call + put)
935
936        // Now shift to 55000
937        set_atm_via_greeks(&mut agg, 55000.0);
938        let action2 = agg.check_rebalance(now()).unwrap();
939        agg.apply_rebalance(&action2, now());
940
941        // Active should now be (52500, 55000): 2 strikes at the top end
942        let active_ids2 = agg.instrument_ids();
943        assert_eq!(active_ids2.len(), 4); // 2 strikes * 2
944    }
945
946    #[rstest]
947    fn test_apply_rebalance_cleans_buffers() {
948        let mut agg = make_multi_strike_aggregator();
949        // Set ATM near 50000
950        set_atm_via_greeks(&mut agg, 50000.0);
951        let action = agg.check_rebalance(now()).unwrap();
952        agg.apply_rebalance(&action, now());
953
954        // Feed quotes for the 47500 call
955        let call_47500 = InstrumentId::from("BTC-20240101-47500-C.DERIBIT");
956        let quote = make_quote(call_47500, "100.00", "101.00");
957        agg.update_quote(&quote);
958        assert_eq!(agg.call_buffer_len(), 1);
959
960        // Now shift ATM up so 47500 is out of range
961        set_atm_via_greeks(&mut agg, 55000.0);
962        let action2 = agg.check_rebalance(now()).unwrap();
963        agg.apply_rebalance(&action2, now());
964
965        // Buffer for 47500 should be cleaned
966        assert_eq!(agg.call_buffer_len(), 0);
967    }
968
969    #[rstest]
970    fn test_initial_active_set_empty_when_no_atm() {
971        let agg = make_multi_strike_aggregator();
972        // AtmRelative with no ATM price: empty active set (deferred)
973        assert_eq!(agg.instrument_ids().len(), 0);
974        assert_eq!(agg.all_instrument_ids().len(), 10);
975    }
976
977    #[rstest]
978    fn test_catalog_vs_active_separation() {
979        let mut agg = make_multi_strike_aggregator();
980        // Set ATM near 50000 to narrow active set
981        set_atm_via_greeks(&mut agg, 50000.0);
982        let action = agg.check_rebalance(now()).unwrap();
983        agg.apply_rebalance(&action, now());
984
985        // Catalog should still have all 10 instruments
986        assert_eq!(agg.instruments().len(), 10);
987        // Active should be a subset
988        assert_eq!(agg.instrument_ids().len(), 6);
989    }
990
991    // -- add_instrument tests --
992
993    #[rstest]
994    fn test_add_instrument_already_known() {
995        let (mut agg, call_id, _) = make_aggregator();
996        let strike = Price::from("50000");
997        let count_before = agg.instruments().len();
998
999        let result = agg.add_instrument(call_id, strike, OptionKind::Call);
1000
1001        assert!(!result);
1002        assert_eq!(agg.instruments().len(), count_before);
1003    }
1004
1005    #[rstest]
1006    fn test_add_instrument_new_in_active_range() {
1007        let (mut agg, _, _) = make_aggregator();
1008        // Fixed range includes strike 50000; adding another instrument at same strike
1009        let new_id = InstrumentId::from("BTC-20240101-50000-C2.DERIBIT");
1010        let strike = Price::from("50000");
1011
1012        let result = agg.add_instrument(new_id, strike, OptionKind::Call);
1013
1014        assert!(result);
1015        assert_eq!(agg.instruments().len(), 3);
1016        assert!(agg.active_ids().contains(&new_id));
1017    }
1018
1019    #[rstest]
1020    fn test_add_instrument_new_out_of_range() {
1021        let (mut agg, _, _) = make_aggregator();
1022        // Fixed range only includes 50000; adding instrument at 60000
1023        let new_id = InstrumentId::from("BTC-20240101-60000-C.DERIBIT");
1024        let strike = Price::from("60000");
1025
1026        let result = agg.add_instrument(new_id, strike, OptionKind::Call);
1027
1028        assert!(result);
1029        assert_eq!(agg.instruments().len(), 3);
1030        assert!(!agg.active_ids().contains(&new_id));
1031    }
1032
1033    #[rstest]
1034    fn test_add_instrument_available_for_rebalance() {
1035        let mut agg = make_multi_strike_aggregator();
1036        // Set ATM near 50000 and apply initial rebalance
1037        set_atm_via_greeks(&mut agg, 50000.0);
1038        let action = agg.check_rebalance(now()).unwrap();
1039        agg.apply_rebalance(&action, now());
1040        // Active: 47500, 50000, 52500 (6 instruments)
1041        assert_eq!(agg.instrument_ids().len(), 6);
1042
1043        // Add a new instrument at strike 57500 (out of current range)
1044        let new_id = InstrumentId::from("BTC-20240101-57500-C.DERIBIT");
1045        let strike = Price::from("57500");
1046        let result = agg.add_instrument(new_id, strike, OptionKind::Call);
1047        assert!(result);
1048        assert!(!agg.active_ids().contains(&new_id));
1049
1050        // Shift ATM to 57500: rebalance should pick up the new instrument
1051        set_atm_via_greeks(&mut agg, 57500.0);
1052        let action2 = agg.check_rebalance(now()).unwrap();
1053        agg.apply_rebalance(&action2, now());
1054
1055        assert!(agg.active_ids().contains(&new_id));
1056    }
1057
1058    // -- Hysteresis tests --
1059
1060    #[rstest]
1061    fn test_hysteresis_blocks_small_movement() {
1062        let strikes = [47500, 50000, 52500];
1063        let mut instruments = HashMap::new();
1064
1065        for s in &strikes {
1066            let strike = Price::from(&s.to_string());
1067            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
1068            instruments.insert(call_id, (strike, OptionKind::Call));
1069        }
1070        let tracker = AtmTracker::new();
1071        let mut agg = OptionChainAggregator::new(
1072            make_series_id(),
1073            StrikeRange::AtmRelative {
1074                strikes_above: 1,
1075                strikes_below: 1,
1076            },
1077            tracker,
1078            instruments,
1079        );
1080        agg.set_hysteresis(0.6);
1081        agg.set_cooldown_ns(0);
1082
1083        // Set ATM to 50000
1084        set_atm_via_greeks(&mut agg, 50000.0);
1085        let action = agg.check_rebalance(now()).unwrap();
1086        agg.apply_rebalance(&action, now());
1087        assert_eq!(agg.last_atm_strike(), Some(Price::from("50000")));
1088
1089        // Move ATM slightly toward 52500: gap=2500, threshold=50000+0.6*2500=51500
1090        // 51000 does NOT cross 51500
1091        set_atm_via_greeks(&mut agg, 51000.0);
1092        assert!(agg.check_rebalance(now()).is_none());
1093    }
1094
1095    #[rstest]
1096    fn test_hysteresis_allows_large_movement() {
1097        let strikes = [47500, 50000, 52500];
1098        let mut instruments = HashMap::new();
1099
1100        for s in &strikes {
1101            let strike = Price::from(&s.to_string());
1102            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
1103            instruments.insert(call_id, (strike, OptionKind::Call));
1104        }
1105        let tracker = AtmTracker::new();
1106        let mut agg = OptionChainAggregator::new(
1107            make_series_id(),
1108            StrikeRange::AtmRelative {
1109                strikes_above: 1,
1110                strikes_below: 1,
1111            },
1112            tracker,
1113            instruments,
1114        );
1115        agg.set_hysteresis(0.6);
1116        agg.set_cooldown_ns(0);
1117
1118        // Set ATM to 50000
1119        set_atm_via_greeks(&mut agg, 50000.0);
1120        let action = agg.check_rebalance(now()).unwrap();
1121        agg.apply_rebalance(&action, now());
1122
1123        // Move ATM well past threshold: 52000 > 51500
1124        set_atm_via_greeks(&mut agg, 52000.0);
1125        assert!(agg.check_rebalance(now()).is_some());
1126    }
1127
1128    #[rstest]
1129    fn test_zero_hysteresis_disables_guard() {
1130        let mut agg = make_multi_strike_aggregator();
1131        agg.set_hysteresis(0.0);
1132        agg.set_cooldown_ns(0);
1133
1134        set_atm_via_greeks(&mut agg, 50000.0);
1135        let action = agg.check_rebalance(now()).unwrap();
1136        agg.apply_rebalance(&action, now());
1137
1138        // Any shift past the strike boundary triggers rebalance
1139        set_atm_via_greeks(&mut agg, 52500.0);
1140        assert!(agg.check_rebalance(now()).is_some());
1141    }
1142
1143    // -- Cooldown tests --
1144
1145    #[rstest]
1146    fn test_cooldown_blocks_rapid_rebalance() {
1147        let mut agg = make_multi_strike_aggregator();
1148        agg.set_hysteresis(0.0);
1149        agg.set_cooldown_ns(5_000_000_000); // 5s
1150
1151        set_atm_via_greeks(&mut agg, 50000.0);
1152        let t0 = now();
1153        let action = agg.check_rebalance(t0).unwrap();
1154        agg.apply_rebalance(&action, t0);
1155
1156        // Shift ATM immediately: cooldown blocks
1157        set_atm_via_greeks(&mut agg, 55000.0);
1158        let t1 = UnixNanos::from(t0.as_u64() + 1_000_000_000); // 1s later
1159        assert!(agg.check_rebalance(t1).is_none());
1160    }
1161
1162    #[rstest]
1163    fn test_cooldown_allows_after_elapsed() {
1164        let mut agg = make_multi_strike_aggregator();
1165        agg.set_hysteresis(0.0);
1166        agg.set_cooldown_ns(5_000_000_000); // 5s
1167
1168        set_atm_via_greeks(&mut agg, 50000.0);
1169        let t0 = now();
1170        let action = agg.check_rebalance(t0).unwrap();
1171        agg.apply_rebalance(&action, t0);
1172
1173        // Shift ATM after cooldown elapses
1174        set_atm_via_greeks(&mut agg, 55000.0);
1175        let t1 = UnixNanos::from(t0.as_u64() + 6_000_000_000); // 6s later
1176        assert!(agg.check_rebalance(t1).is_some());
1177    }
1178
1179    #[rstest]
1180    fn test_zero_cooldown_disables_guard() {
1181        let mut agg = make_multi_strike_aggregator();
1182        agg.set_hysteresis(0.0);
1183        agg.set_cooldown_ns(0);
1184
1185        set_atm_via_greeks(&mut agg, 50000.0);
1186        let t0 = now();
1187        let action = agg.check_rebalance(t0).unwrap();
1188        agg.apply_rebalance(&action, t0);
1189
1190        // Shift ATM immediately: no cooldown block
1191        set_atm_via_greeks(&mut agg, 55000.0);
1192        assert!(agg.check_rebalance(t0).is_some());
1193    }
1194
1195    // -- Pending greeks tests --
1196
1197    #[rstest]
1198    fn test_pending_greeks_consumed_on_first_quote() {
1199        let (mut agg, call_id, _) = make_aggregator();
1200
1201        // Send greeks before any quote
1202        let greeks = OptionGreeks {
1203            instrument_id: call_id,
1204            greeks: OptionGreekValues {
1205                delta: 0.55,
1206                ..Default::default()
1207            },
1208            ..Default::default()
1209        };
1210        agg.update_greeks(&greeks);
1211        assert_eq!(agg.pending_greeks_count(), 1);
1212
1213        // Now send the first quote: pending greeks should be consumed
1214        let quote = make_quote(call_id, "100.00", "101.00");
1215        agg.update_quote(&quote);
1216        assert_eq!(agg.pending_greeks_count(), 0);
1217
1218        // Verify greeks were attached
1219        let strike = Price::from("50000");
1220        let data = agg.get_call_greeks_from_buffer(&strike);
1221        assert!(data.is_some());
1222        assert_eq!(data.unwrap().delta, 0.55);
1223    }
1224
1225    // -- ts_event tracking tests --
1226
1227    #[rstest]
1228    fn test_snapshot_ts_event_reflects_max_quote_timestamp() {
1229        let (mut agg, call_id, put_id) = make_aggregator();
1230
1231        let quote1 = QuoteTick::new(
1232            call_id,
1233            Price::from("100.00"),
1234            Price::from("101.00"),
1235            Quantity::from("1.0"),
1236            Quantity::from("1.0"),
1237            UnixNanos::from(500u64), // ts_event
1238            UnixNanos::from(500u64),
1239        );
1240        agg.update_quote(&quote1);
1241
1242        let quote2 = QuoteTick::new(
1243            put_id,
1244            Price::from("50.00"),
1245            Price::from("51.00"),
1246            Quantity::from("1.0"),
1247            Quantity::from("1.0"),
1248            UnixNanos::from(800u64), // ts_event: later
1249            UnixNanos::from(800u64),
1250        );
1251        agg.update_quote(&quote2);
1252
1253        let slice = agg.snapshot(UnixNanos::from(1000u64));
1254        assert_eq!(slice.ts_event, UnixNanos::from(800u64));
1255        assert_eq!(slice.ts_init, UnixNanos::from(1000u64));
1256    }
1257
1258    #[rstest]
1259    fn test_snapshot_ts_event_fallback_when_no_quotes() {
1260        let (agg, _, _) = make_aggregator();
1261        let slice = agg.snapshot(UnixNanos::from(1000u64));
1262        // No quotes: ts_event falls back to ts_init
1263        assert_eq!(slice.ts_event, UnixNanos::from(1000u64));
1264    }
1265
1266    #[rstest]
1267    fn test_snapshot_retains_buffered_data_during_hysteresis_window() {
1268        // Setup: 3 strikes at 47500/50000/52500, AtmRelative +-1, hysteresis enabled
1269        let strikes = [47500, 50000, 52500];
1270        let mut instruments = HashMap::new();
1271
1272        for s in &strikes {
1273            let strike = Price::from(&s.to_string());
1274            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
1275            instruments.insert(call_id, (strike, OptionKind::Call));
1276        }
1277        let tracker = AtmTracker::new();
1278        let mut agg = OptionChainAggregator::new(
1279            make_series_id(),
1280            StrikeRange::AtmRelative {
1281                strikes_above: 1,
1282                strikes_below: 1,
1283            },
1284            tracker,
1285            instruments,
1286        );
1287        agg.set_hysteresis(0.6);
1288        agg.set_cooldown_ns(0);
1289
1290        // Set ATM to 50000, rebalance -> active: {47500, 50000, 52500}
1291        set_atm_via_greeks(&mut agg, 50000.0);
1292        let action = agg.check_rebalance(now()).unwrap();
1293        agg.apply_rebalance(&action, now());
1294        assert_eq!(agg.instrument_ids().len(), 3);
1295
1296        // Buffer quotes for all active strikes
1297        let q1 = make_quote(
1298            InstrumentId::from("BTC-20240101-47500-C.DERIBIT"),
1299            "3000.00",
1300            "3100.00",
1301        );
1302        let q2 = make_quote(
1303            InstrumentId::from("BTC-20240101-50000-C.DERIBIT"),
1304            "1500.00",
1305            "1600.00",
1306        );
1307        let q3 = make_quote(
1308            InstrumentId::from("BTC-20240101-52500-C.DERIBIT"),
1309            "500.00",
1310            "600.00",
1311        );
1312        agg.update_quote(&q1);
1313        agg.update_quote(&q2);
1314        agg.update_quote(&q3);
1315        assert_eq!(agg.call_buffer_len(), 3);
1316
1317        // Move ATM slightly toward 52500 but within hysteresis band (no rebalance)
1318        set_atm_via_greeks(&mut agg, 51000.0);
1319        assert!(agg.check_rebalance(now()).is_none());
1320
1321        // Snapshot must still include all 3 buffered strikes
1322        let slice = agg.snapshot(UnixNanos::from(100u64));
1323        assert_eq!(slice.call_count(), 3);
1324    }
1325
1326    #[rstest]
1327    fn test_remove_instrument_from_catalog() {
1328        let (mut agg, call_id, put_id) = make_aggregator();
1329        assert_eq!(agg.instruments().len(), 2);
1330
1331        let removed = agg.remove_instrument(&call_id);
1332        assert!(removed);
1333        assert_eq!(agg.instruments().len(), 1);
1334        assert!(!agg.active_ids().contains(&call_id));
1335        assert!(agg.instruments().contains_key(&put_id));
1336    }
1337
1338    #[rstest]
1339    fn test_remove_instrument_cleans_buffer() {
1340        let (mut agg, call_id, _) = make_aggregator();
1341        let quote = make_quote(call_id, "100.00", "101.00");
1342        agg.update_quote(&quote);
1343        assert_eq!(agg.call_buffer_len(), 1);
1344
1345        let _ = agg.remove_instrument(&call_id);
1346        // No sibling call at same strike, buffer entry should be removed
1347        assert_eq!(agg.call_buffer_len(), 0);
1348    }
1349
1350    #[rstest]
1351    fn test_remove_instrument_preserves_sibling_buffer() {
1352        let (mut agg, call_id, _) = make_aggregator();
1353        // Add a second call at the same strike
1354        let sibling_id = InstrumentId::from("BTC-20240101-50000-C2.DERIBIT");
1355        let strike = Price::from("50000");
1356        let _ = agg.add_instrument(sibling_id, strike, OptionKind::Call);
1357
1358        let quote = make_quote(call_id, "100.00", "101.00");
1359        agg.update_quote(&quote);
1360        assert_eq!(agg.call_buffer_len(), 1);
1361
1362        // Remove original: sibling still shares the strike+kind
1363        let _ = agg.remove_instrument(&call_id);
1364        assert_eq!(agg.call_buffer_len(), 1); // buffer preserved
1365        assert!(agg.instruments().contains_key(&sibling_id));
1366    }
1367
1368    #[rstest]
1369    fn test_remove_instrument_unknown_noop() {
1370        let (mut agg, _, _) = make_aggregator();
1371        let unknown = InstrumentId::from("ETH-20240101-3000-C.DERIBIT");
1372        assert!(!agg.remove_instrument(&unknown));
1373        assert_eq!(agg.instruments().len(), 2);
1374    }
1375
1376    #[rstest]
1377    fn test_remove_instrument_cleans_pending_greeks() {
1378        let (mut agg, call_id, _) = make_aggregator();
1379        let greeks = OptionGreeks {
1380            instrument_id: call_id,
1381            greeks: OptionGreekValues {
1382                delta: 0.55,
1383                ..Default::default()
1384            },
1385            ..Default::default()
1386        };
1387        agg.update_greeks(&greeks);
1388        assert_eq!(agg.pending_greeks_count(), 1);
1389
1390        let _ = agg.remove_instrument(&call_id);
1391        assert_eq!(agg.pending_greeks_count(), 0);
1392    }
1393
1394    #[rstest]
1395    fn test_is_catalog_empty_after_full_removal() {
1396        let (mut agg, call_id, put_id) = make_aggregator();
1397        assert!(!agg.is_catalog_empty());
1398
1399        let _ = agg.remove_instrument(&call_id);
1400        assert!(!agg.is_catalog_empty());
1401
1402        let _ = agg.remove_instrument(&put_id);
1403        assert!(agg.is_catalog_empty());
1404    }
1405
1406    // -- Expiry guard tests --
1407
1408    #[rstest]
1409    fn test_expired_quote_is_dropped() {
1410        let (mut agg, call_id, _) = make_aggregator();
1411        // Series expires at 1_700_000_000_000_000_000; send quote AT that timestamp
1412        let expired_quote = QuoteTick::new(
1413            call_id,
1414            Price::from("100.00"),
1415            Price::from("101.00"),
1416            Quantity::from("1.0"),
1417            Quantity::from("1.0"),
1418            UnixNanos::from(1_700_000_000_000_000_000u64),
1419            UnixNanos::from(1_700_000_000_000_000_000u64),
1420        );
1421        agg.update_quote(&expired_quote);
1422        assert!(agg.is_buffer_empty());
1423    }
1424
1425    #[rstest]
1426    fn test_expired_greeks_are_dropped() {
1427        let (mut agg, call_id, _) = make_aggregator();
1428        // First add a valid quote so greeks would normally land in the buffer
1429        let quote = make_quote(call_id, "100.00", "101.00");
1430        agg.update_quote(&quote);
1431        assert_eq!(agg.call_buffer_len(), 1);
1432
1433        // Send greeks at expiry timestamp: should be dropped
1434        let greeks = OptionGreeks {
1435            instrument_id: call_id,
1436            ts_event: UnixNanos::from(1_700_000_000_000_000_000u64),
1437            greeks: OptionGreekValues {
1438                delta: 0.55,
1439                ..Default::default()
1440            },
1441            ..Default::default()
1442        };
1443        agg.update_greeks(&greeks);
1444
1445        let strike = Price::from("50000");
1446        assert!(agg.get_call_greeks_from_buffer(&strike).is_none());
1447    }
1448
1449    // -- Delta range tests --
1450
1451    /// Builds a `Delta`-range aggregator over `strikes` (call + put per strike),
1452    /// with hysteresis and cooldown disabled so rebalance decisions reflect pure
1453    /// delta resolution.
1454    fn make_delta_aggregator(
1455        strikes: &[i64],
1456        target: f64,
1457        tolerance: f64,
1458    ) -> OptionChainAggregator {
1459        let mut instruments = HashMap::new();
1460
1461        for s in strikes {
1462            let strike = Price::from(&s.to_string());
1463            instruments.insert(option_id(*s, OptionKind::Call), (strike, OptionKind::Call));
1464            instruments.insert(option_id(*s, OptionKind::Put), (strike, OptionKind::Put));
1465        }
1466        let tracker = AtmTracker::new();
1467        let mut agg = OptionChainAggregator::new(
1468            make_series_id(),
1469            StrikeRange::Delta { target, tolerance },
1470            tracker,
1471            instruments,
1472        );
1473        agg.set_hysteresis(0.0);
1474        agg.set_cooldown_ns(0);
1475        agg
1476    }
1477
1478    fn option_id(strike: i64, kind: OptionKind) -> InstrumentId {
1479        let suffix = match kind {
1480            OptionKind::Call => "C",
1481            OptionKind::Put => "P",
1482        };
1483        InstrumentId::from(&format!("BTC-20240101-{strike}-{suffix}.DERIBIT"))
1484    }
1485
1486    /// Feeds a quote then greeks (with the given `delta`) for one option leg.
1487    fn feed_quote_and_greeks(
1488        agg: &mut OptionChainAggregator,
1489        strike: i64,
1490        kind: OptionKind,
1491        delta: f64,
1492    ) {
1493        let id = option_id(strike, kind);
1494        agg.update_quote(&make_quote(id, "100.00", "101.00"));
1495        agg.update_greeks(&OptionGreeks {
1496            instrument_id: id,
1497            greeks: OptionGreekValues {
1498                delta,
1499                ..Default::default()
1500            },
1501            ..Default::default()
1502        });
1503    }
1504
1505    #[rstest]
1506    #[case(0.30, 0.30, 0.03, true)] // exact target
1507    #[case(-0.30, 0.30, 0.03, true)] // negative delta, magnitude matches
1508    #[case(0.28, 0.30, 0.03, true)] // inside band, below target
1509    #[case(0.32, 0.30, 0.03, true)] // inside band, above target
1510    #[case(0.20, 0.30, 0.03, false)] // below band
1511    #[case(0.40, 0.30, 0.03, false)] // above band
1512    fn test_delta_within_band(
1513        #[case] delta: f64,
1514        #[case] target: f64,
1515        #[case] tolerance: f64,
1516        #[case] expected: bool,
1517    ) {
1518        assert_eq!(
1519            OptionChainAggregator::delta_within_band(delta, target, tolerance),
1520            expected
1521        );
1522    }
1523
1524    #[rstest]
1525    fn test_delta_target_hit() {
1526        let strikes = [40000, 45000, 50000, 55000, 60000];
1527        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1528        // Bootstrap the ATM-relative fallback so the window is active and greeks can land.
1529        set_atm_via_greeks(&mut agg, 50000.0);
1530        agg.recompute_active_set();
1531        assert_eq!(agg.instrument_ids().len(), 10); // all 5 strikes (fallback)
1532
1533        // Only the 55000 call sits at the 0.30 target.
1534        feed_quote_and_greeks(&mut agg, 40000, OptionKind::Call, 0.95);
1535        feed_quote_and_greeks(&mut agg, 40000, OptionKind::Put, -0.95);
1536        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.80);
1537        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.80);
1538        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.55);
1539        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.55);
1540        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
1541        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.12);
1542        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Call, 0.12);
1543        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Put, -0.10);
1544
1545        let active = agg.recompute_active_set();
1546        assert_eq!(active.len(), 2); // 55000 call + put
1547        assert!(active.contains(&option_id(55000, OptionKind::Call)));
1548        assert!(active.contains(&option_id(55000, OptionKind::Put)));
1549    }
1550
1551    #[rstest]
1552    fn test_delta_tolerance_band() {
1553        let strikes = [45000, 50000, 55000, 60000];
1554        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.05);
1555        set_atm_via_greeks(&mut agg, 50000.0);
1556        agg.recompute_active_set();
1557
1558        // Band is [0.25, 0.35]: 0.50 and 0.10 are outside, 0.32 and 0.30 inside.
1559        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.50);
1560        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.50);
1561        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.32);
1562        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.50);
1563        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
1564        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.50);
1565        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Call, 0.10);
1566        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Put, -0.10);
1567
1568        let active = agg.recompute_active_set();
1569        assert_eq!(active.len(), 4); // 50000 + 55000, both legs each
1570        assert!(active.contains(&option_id(50000, OptionKind::Call)));
1571        assert!(active.contains(&option_id(55000, OptionKind::Call)));
1572        assert!(!active.contains(&option_id(45000, OptionKind::Call)));
1573        assert!(!active.contains(&option_id(60000, OptionKind::Call)));
1574    }
1575
1576    #[rstest]
1577    fn test_delta_matches_put_by_magnitude() {
1578        let strikes = [45000, 50000, 55000];
1579        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1580        set_atm_via_greeks(&mut agg, 50000.0);
1581        agg.recompute_active_set();
1582
1583        // Only the 45000 put matches the target, isolating put-side matching.
1584        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
1585        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.30);
1586        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.55);
1587        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.55);
1588        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.55);
1589        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.55);
1590
1591        let active = agg.recompute_active_set();
1592        // |-0.30| == target, so the 45000 strike (both legs) is selected.
1593        assert_eq!(active.len(), 2);
1594        assert!(active.contains(&option_id(45000, OptionKind::Put)));
1595        assert!(active.contains(&option_id(45000, OptionKind::Call)));
1596    }
1597
1598    #[rstest]
1599    fn test_delta_no_greeks_falls_back_to_atm_window() {
1600        // 13 strikes so the ATM-relative fallback window is a proper subset.
1601        let strikes: Vec<i64> = (0..13).map(|i| 40000 + i * 1000).collect();
1602        let mut agg = make_delta_aggregator(&strikes, 0.25, 0.05);
1603        set_atm_via_greeks(&mut agg, 46000.0); // centered
1604
1605        let active = agg.recompute_active_set();
1606
1607        // No greeks -> a bounded ATM-relative window: neither empty nor the full chain.
1608        // The exact window width is asserted in the model-level resolve test.
1609        assert!(active.len() > 2);
1610        assert!(active.len() < strikes.len() * 2);
1611        assert!(active.contains(&option_id(46000, OptionKind::Call))); // ATM included
1612        assert!(!active.contains(&option_id(40000, OptionKind::Call))); // extreme excluded
1613        assert!(!active.contains(&option_id(52000, OptionKind::Call))); // extreme excluded
1614    }
1615
1616    #[rstest]
1617    fn test_delta_pending_only_greeks_eligible() {
1618        let strikes = [45000, 50000, 55000];
1619        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1620        set_atm_via_greeks(&mut agg, 50000.0);
1621        agg.recompute_active_set();
1622
1623        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
1624        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.55);
1625        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.55);
1626        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.55);
1627        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.55);
1628
1629        // Greeks arrive before any quote, so they land in pending_greeks.
1630        agg.update_greeks(&OptionGreeks {
1631            instrument_id: option_id(55000, OptionKind::Call),
1632            greeks: OptionGreekValues {
1633                delta: 0.30,
1634                ..Default::default()
1635            },
1636            ..Default::default()
1637        });
1638        assert_eq!(agg.pending_greeks_count(), 1);
1639
1640        let active = agg.recompute_active_set();
1641        // Pending-only greeks are eligible for delta resolution.
1642        assert_eq!(active.len(), 2);
1643        assert!(active.contains(&option_id(55000, OptionKind::Call)));
1644    }
1645
1646    #[rstest]
1647    fn test_delta_waits_for_fallback_window_greeks_before_narrowing() {
1648        let strikes = [45000, 50000, 55000];
1649        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1650        set_atm_via_greeks(&mut agg, 50000.0);
1651        agg.recompute_active_set();
1652        assert_eq!(agg.instrument_ids().len(), 6); // fallback: all 3 strikes
1653
1654        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
1655
1656        assert!(agg.check_rebalance(now()).is_none());
1657        assert_eq!(agg.instrument_ids().len(), 6);
1658    }
1659
1660    #[rstest]
1661    fn test_delta_waits_when_fallback_window_shifts_during_warmup() {
1662        let strikes: Vec<i64> = (0..13).map(|i| 40000 + i * 1000).collect();
1663        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1664        set_atm_via_greeks(&mut agg, 46000.0);
1665        agg.recompute_active_set();
1666        assert!(
1667            agg.active_ids()
1668                .contains(&option_id(42000, OptionKind::Call))
1669        );
1670        assert!(
1671            agg.active_ids()
1672                .contains(&option_id(51000, OptionKind::Put))
1673        );
1674
1675        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.30);
1676        set_atm_via_greeks(&mut agg, 47000.0);
1677
1678        let action = agg
1679            .check_rebalance(now())
1680            .expect("fallback window shift should rebalance active legs");
1681
1682        assert!(action.add.contains(&option_id(52000, OptionKind::Call)));
1683        assert!(!action.remove.contains(&option_id(42000, OptionKind::Call)));
1684        assert!(!action.remove.contains(&option_id(51000, OptionKind::Put)));
1685    }
1686
1687    #[rstest]
1688    fn test_delta_rebalances_on_greeks_with_atm_unchanged() {
1689        let strikes = [45000, 50000, 55000];
1690        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1691        set_atm_via_greeks(&mut agg, 50000.0);
1692        agg.recompute_active_set();
1693        assert_eq!(agg.last_atm_strike(), Some(Price::from("50000")));
1694        assert_eq!(agg.instrument_ids().len(), 6); // fallback: all 3 strikes
1695
1696        // Greeks arrive; only 55000 matches. The closest ATM strike is unchanged.
1697        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
1698        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.55);
1699        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.45);
1700        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.45);
1701        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
1702        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.12);
1703
1704        let action = agg
1705            .check_rebalance(now())
1706            .expect("delta range should rebalance when greeks narrow the set");
1707        assert!(action.add.is_empty());
1708        assert!(!action.remove.is_empty());
1709
1710        agg.apply_rebalance(&action, now());
1711        assert_eq!(agg.instrument_ids().len(), 2); // narrowed to the 55000 legs
1712        assert!(
1713            agg.active_ids()
1714                .contains(&option_id(55000, OptionKind::Call))
1715        );
1716    }
1717
1718    #[rstest]
1719    fn test_delta_no_op_rebalance_returns_none() {
1720        let strikes = [45000, 50000, 55000];
1721        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
1722        set_atm_via_greeks(&mut agg, 50000.0);
1723        agg.recompute_active_set();
1724        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
1725        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.55);
1726        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.45);
1727        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.45);
1728        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
1729        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.12);
1730
1731        // First rebalance narrows to the 55000 legs.
1732        let action = agg.check_rebalance(now()).unwrap();
1733        agg.apply_rebalance(&action, now());
1734        assert_eq!(agg.instrument_ids().len(), 2);
1735
1736        // Greeks unchanged -> stable set -> no-op suppressed (cooldown disabled).
1737        assert!(agg.check_rebalance(now()).is_none());
1738    }
1739}
nautilus_data/option_chains/aggregator.rs

nautilus_data/option_chains/
aggregator.rs
