nautilus_data/option_chains/

aggregator.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Per-series option chain aggregator for event accumulation and snapshots.

use std::{
    cmp::Ordering,
    collections::{BTreeMap, HashMap, HashSet},
};

use nautilus_core::{DurationNanos, UnixNanos};
use nautilus_model::{
    data::{
        QuoteTick,
        option_chain::{OptionChainSlice, OptionGreeks, OptionStrikeData, StrikeRange},
    },
    enums::OptionKind,
    identifiers::{InstrumentId, OptionSeriesId},
    types::Price,
};
use rust_decimal::prelude::ToPrimitive;

use super::{
    AtmTracker,
    constants::{DEFAULT_REBALANCE_COOLDOWN, DEFAULT_REBALANCE_HYSTERESIS},
};

/// Per-series aggregator that accumulates quotes and greeks between snapshots.
///
/// Owns mutable accumulator buffers and produces immutable `OptionChainSlice`
/// snapshots on each timer tick.
#[derive(Debug)]
pub struct OptionChainAggregator {
    /// The option series identifier for this aggregator.
    series_id: OptionSeriesId,
    /// Defines which strikes to include in the active set.
    strike_range: StrikeRange,
    /// Tracks the current ATM price from market data events.
    atm_tracker: AtmTracker,
    /// All instruments for this series. Grows dynamically when the exchange
    /// lists new strikes via [`Self::add_instrument`].
    instruments: HashMap<InstrumentId, (Price, OptionKind)>,
    /// Currently active instrument IDs (subset of `instruments`).
    active_ids: HashSet<InstrumentId>,
    /// The closest ATM strike at the time of the last rebalance.
    last_atm_strike: Option<Price>,
    /// Hysteresis band for ATM rebalancing.
    hysteresis: f64,
    /// Minimum nanoseconds between rebalances.
    cooldown_ns: DurationNanos,
    /// Timestamp of the last rebalance.
    last_rebalance_ns: Option<UnixNanos>,
    /// Maximum `ts_event` seen across all quote updates.
    max_ts_event: UnixNanos,
    /// Greeks received before the corresponding quote arrived.
    pending_greeks: HashMap<InstrumentId, OptionGreeks>,
    /// Call option accumulator buffer keyed by strike price.
    call_buffer: BTreeMap<Price, OptionStrikeData>,
    /// Put option accumulator buffer keyed by strike price.
    put_buffer: BTreeMap<Price, OptionStrikeData>,
}

impl OptionChainAggregator {
    /// Creates a new aggregator for the given series.
    ///
    /// `instruments` contains ALL instruments for the series. The initial
    /// `active_ids` subset is resolved from the strike range and the current
    /// ATM price (if available). When no ATM price is set for ATM-based
    /// ranges, all instruments are active.
    pub fn new(
        series_id: OptionSeriesId,
        strike_range: StrikeRange,
        atm_tracker: AtmTracker,
        instruments: HashMap<InstrumentId, (Price, OptionKind)>,
    ) -> Self {
        let mut aggregator = Self {
            series_id,
            strike_range,
            atm_tracker,
            instruments,
            active_ids: HashSet::new(),
            last_atm_strike: None,
            hysteresis: DEFAULT_REBALANCE_HYSTERESIS,
            cooldown_ns: DEFAULT_REBALANCE_COOLDOWN,
            last_rebalance_ns: None,
            max_ts_event: UnixNanos::default(),
            pending_greeks: HashMap::new(),
            call_buffer: BTreeMap::new(),
            put_buffer: BTreeMap::new(),
        };
        // No Greeks exist at construction, so a `Delta` range resolves to its ATM fallback.
        aggregator.recompute_active_set();
        aggregator
    }

    /// Returns a mutable reference to the ATM tracker.
    pub fn atm_tracker_mut(&mut self) -> &mut AtmTracker {
        &mut self.atm_tracker
    }

    /// Returns the currently active instrument IDs.
    #[must_use]
    pub fn instrument_ids(&self) -> Vec<InstrumentId> {
        self.active_ids.iter().copied().collect()
    }

    /// Returns a reference to the active instrument ID set.
    #[must_use]
    pub fn active_ids(&self) -> &HashSet<InstrumentId> {
        &self.active_ids
    }

    /// Returns the series ID.
    #[must_use]
    pub fn series_id(&self) -> OptionSeriesId {
        self.series_id
    }

    /// Returns `true` if the given timestamp is at or past the series expiration.
    #[must_use]
    pub fn is_expired(&self, now_ns: UnixNanos) -> bool {
        now_ns >= self.series_id.expiration_ns
    }

    /// Returns a reference to the full instrument set.
    #[must_use]
    pub fn instruments(&self) -> &HashMap<InstrumentId, (Price, OptionKind)> {
        &self.instruments
    }

    /// Returns all instrument IDs in the full set.
    #[must_use]
    pub fn all_instrument_ids(&self) -> Vec<InstrumentId> {
        self.instruments.keys().copied().collect()
    }

    /// Returns `true` if the instrument catalog is empty.
    #[must_use]
    pub fn is_catalog_empty(&self) -> bool {
        self.instruments.is_empty()
    }

    /// Permanently removes an instrument from the catalog.
    ///
    /// Removes from `instruments`, `active_ids`, `pending_greeks`, and cleans
    /// buffer entries (only if no other instrument shares the same strike+kind).
    /// Returns `true` if the instrument was found and removed.
    #[must_use]
    pub fn remove_instrument(&mut self, instrument_id: &InstrumentId) -> bool {
        let Some((strike, kind)) = self.instruments.remove(instrument_id) else {
            return false;
        };

        self.active_ids.remove(instrument_id);
        self.pending_greeks.remove(instrument_id);

        // Only remove buffer entry if no sibling instrument shares the same strike+kind
        let has_sibling = self
            .instruments
            .values()
            .any(|(s, k)| *s == strike && *k == kind);

        if !has_sibling {
            let buffer = match kind {
                OptionKind::Call => &mut self.call_buffer,
                OptionKind::Put => &mut self.put_buffer,
            };
            buffer.remove(&strike);
        }

        true
    }

    /// Returns a reference to the ATM tracker.
    #[must_use]
    pub fn atm_tracker(&self) -> &AtmTracker {
        &self.atm_tracker
    }

    /// Recomputes the active instrument set from the current ATM price.
    ///
    /// Returns the new active instrument IDs. Used during bootstrap when the
    /// first ATM price arrives after deferred subscription setup.
    pub fn recompute_active_set(&mut self) -> Vec<InstrumentId> {
        let atm_price = self.atm_tracker.atm_price();
        let all_strikes = Self::sorted_strikes(&self.instruments);
        let active_strikes: HashSet<Price> = self
            .resolve_active_strikes(atm_price, &all_strikes)
            .into_iter()
            .collect();
        self.active_ids = self
            .instruments
            .iter()
            .filter(|(_, (strike, _))| active_strikes.contains(strike))
            .map(|(id, _)| *id)
            .collect();
        self.last_atm_strike =
            atm_price.and_then(|atm| Self::find_closest_strike(&all_strikes, atm));
        self.active_ids.iter().copied().collect()
    }

    /// Resolves the active strikes for the current strike range.
    ///
    /// `Delta` is resolved here from stored Greeks (see [`Self::resolve_delta`]);
    /// the price-based variants delegate to [`StrikeRange::resolve`].
    fn resolve_active_strikes(
        &self,
        atm_price: Option<Price>,
        all_strikes: &[Price],
    ) -> Vec<Price> {
        match &self.strike_range {
            StrikeRange::Delta { target, tolerance } => {
                self.resolve_delta(*target, *tolerance, atm_price, all_strikes)
            }
            _ => self.strike_range.resolve(atm_price, all_strikes),
        }
    }

    /// Resolves strikes whose buffered or pending Greeks have an absolute delta
    /// within `tolerance` of `target`.
    ///
    /// A strike qualifies when either its call or put delta magnitude matches
    /// (calls have positive delta, puts negative; both are compared by absolute
    /// value), so a typical target selects an OTM strike on each side of ATM.
    /// Strikes with only pending Greeks (received before their first quote) are
    /// eligible. Before the resolver changes from a fallback set to a selected
    /// set, every current fallback leg must have Greeks. This avoids unsubscribing
    /// legs whose Greeks have not arrived yet, including when the fallback window
    /// shifts with ATM. When no Greeks fall in the band, this falls back to the
    /// ATM-relative window from [`StrikeRange::resolve`].
    fn resolve_delta(
        &self,
        target: f64,
        tolerance: f64,
        atm_price: Option<Price>,
        all_strikes: &[Price],
    ) -> Vec<Price> {
        let selected: Vec<Price> = self
            .deltas_by_strike()
            .into_iter()
            .filter(|(_, deltas)| {
                deltas
                    .iter()
                    .any(|delta| Self::delta_within_band(*delta, target, tolerance))
            })
            .map(|(strike, _)| strike)
            .collect();

        let fallback_strikes = self.strike_range.resolve(atm_price, all_strikes);

        if selected.is_empty() {
            return fallback_strikes;
        }

        let selected_ids = self.instrument_ids_for_strikes(&selected);
        let fallback_ids = self.instrument_ids_for_strikes(&fallback_strikes);

        if self.active_ids != selected_ids && !self.delta_window_ready(&fallback_ids) {
            return fallback_strikes;
        }

        selected
    }

    fn instrument_ids_for_strikes(&self, strikes: &[Price]) -> HashSet<InstrumentId> {
        let strike_set: HashSet<Price> = strikes.iter().copied().collect();
        self.instruments
            .iter()
            .filter(|(_, (strike, _))| strike_set.contains(strike))
            .map(|(id, _)| *id)
            .collect()
    }

    fn delta_window_ready(&self, instrument_ids: &HashSet<InstrumentId>) -> bool {
        !instrument_ids.is_empty()
            && instrument_ids
                .iter()
                .all(|id| self.instrument_has_greeks(id))
    }

    fn instrument_has_greeks(&self, instrument_id: &InstrumentId) -> bool {
        if self.pending_greeks.contains_key(instrument_id) {
            return true;
        }

        let Some((strike, kind)) = self.instruments.get(instrument_id) else {
            return false;
        };
        let buffer = match kind {
            OptionKind::Call => &self.call_buffer,
            OptionKind::Put => &self.put_buffer,
        };

        buffer
            .get(strike)
            .and_then(|data| data.greeks.as_ref())
            .is_some()
    }

    /// Collects every reported delta per strike, from buffered Greeks and from
    /// Greeks still pending their first quote.
    fn deltas_by_strike(&self) -> BTreeMap<Price, Vec<f64>> {
        let mut deltas_by_strike: BTreeMap<Price, Vec<f64>> = BTreeMap::new();

        for (strike, data) in self.call_buffer.iter().chain(self.put_buffer.iter()) {
            if let Some(greeks) = data.greeks.as_ref() {
                deltas_by_strike
                    .entry(*strike)
                    .or_default()
                    .push(greeks.delta);
            }
        }

        for (id, greeks) in &self.pending_greeks {
            if let Some((strike, _)) = self.instruments.get(id) {
                deltas_by_strike
                    .entry(*strike)
                    .or_default()
                    .push(greeks.delta);
            }
        }

        deltas_by_strike
    }

    /// Returns `true` when `delta`'s magnitude is within `tolerance` of `target`.
    ///
    /// Compares by absolute value so a put (negative delta) matches the same
    /// target as the equivalent call.
    fn delta_within_band(delta: f64, target: f64, tolerance: f64) -> bool {
        (delta.abs() - target).abs() <= tolerance
    }

    /// Adds a newly discovered instrument to the series.
    ///
    /// Returns `true` if the instrument was newly inserted. Returns `false`
    /// if it was already known (no-op). When the new instrument's strike
    /// falls within the current active range, it is immediately added to
    /// `active_ids`.
    #[must_use]
    pub fn add_instrument(
        &mut self,
        instrument_id: InstrumentId,
        strike: Price,
        kind: OptionKind,
    ) -> bool {
        if self.instruments.contains_key(&instrument_id) {
            return false;
        }

        self.instruments.insert(instrument_id, (strike, kind));

        // Determine if the new strike is in the current active range
        let all_strikes = Self::sorted_strikes(&self.instruments);
        let atm_price = self.atm_tracker.atm_price();
        let active_strikes: HashSet<Price> = self
            .resolve_active_strikes(atm_price, &all_strikes)
            .into_iter()
            .collect();

        if active_strikes.contains(&strike) {
            self.active_ids.insert(instrument_id);
        }

        true
    }

    /// Returns sorted, deduplicated strikes from the given instruments.
    fn sorted_strikes(instruments: &HashMap<InstrumentId, (Price, OptionKind)>) -> Vec<Price> {
        let mut strikes: Vec<Price> = instruments.values().map(|(s, _)| *s).collect();
        strikes.sort();
        strikes.dedup();
        strikes
    }

    /// Finds the strike in `all_strikes` closest to `atm`.
    fn find_closest_strike(all_strikes: &[Price], atm: Price) -> Option<Price> {
        all_strikes
            .iter()
            .min_by_key(|strike| strike.raw().abs_diff(atm.raw()))
            .copied()
    }

    /// Handles an incoming quote tick by updating the accumulator buffers.
    pub fn update_quote(&mut self, quote: &QuoteTick) {
        if self.is_expired(quote.ts_event) {
            log::warn!(
                "Dropping quote for {}, series {} expired at {}",
                quote.instrument_id,
                self.series_id,
                self.series_id.expiration_ns,
            );
            return;
        }

        if !self.active_ids.contains(&quote.instrument_id) {
            return;
        }

        if let Some(&(strike, kind)) = self.instruments.get(&quote.instrument_id) {
            // Track max ts_event across all quotes
            if quote.ts_event > self.max_ts_event {
                self.max_ts_event = quote.ts_event;
            }

            let buffer = match kind {
                OptionKind::Call => &mut self.call_buffer,
                OptionKind::Put => &mut self.put_buffer,
            };

            match buffer.get_mut(&strike) {
                Some(data) => data.quote = *quote,
                None => {
                    // Check for pending greeks that arrived before this first quote
                    let greeks = self.pending_greeks.remove(&quote.instrument_id);
                    buffer.insert(
                        strike,
                        OptionStrikeData {
                            quote: *quote,
                            greeks,
                        },
                    );
                }
            }
        }
    }

    /// Handles incoming greeks by updating the accumulator buffers.
    ///
    /// If no quote has arrived yet for this instrument (no buffer entry),
    /// the greeks are stored in `pending_greeks` and will be attached when
    /// the first quote arrives.
    pub fn update_greeks(&mut self, greeks: &OptionGreeks) {
        if self.is_expired(greeks.ts_event) {
            log::warn!(
                "Dropping greeks for {}, series {} expired at {}",
                greeks.instrument_id,
                self.series_id,
                self.series_id.expiration_ns,
            );
            return;
        }

        if !self.active_ids.contains(&greeks.instrument_id) {
            return;
        }

        if let Some(&(strike, kind)) = self.instruments.get(&greeks.instrument_id) {
            let buffer = match kind {
                OptionKind::Call => &mut self.call_buffer,
                OptionKind::Put => &mut self.put_buffer,
            };

            match buffer.get_mut(&strike) {
                Some(data) => data.greeks = Some(*greeks),
                None => {
                    // No quote yet: park the greeks for later
                    self.pending_greeks.insert(greeks.instrument_id, *greeks);
                }
            }
        }
    }

    /// Creates a point-in-time snapshot from accumulated buffers, applying strike filtering.
    ///
    /// Buffers are preserved (keep-latest semantics) so instruments that didn't
    /// quote since the last tick are still included in subsequent snapshots.
    ///
    /// # Panics
    ///
    /// Panics if strike prices cannot be compared (NaN values).
    pub fn snapshot(&self, ts_init: UnixNanos) -> OptionChainSlice {
        let atm_price = self.atm_tracker.atm_price();

        // Use catalog strikes for ATM strike (most accurate closest-strike lookup)
        let catalog_strikes = Self::sorted_strikes(&self.instruments);
        let atm_strike = atm_price.and_then(|atm| Self::find_closest_strike(&catalog_strikes, atm));

        // Filter buffers using active set strikes directly. The active set is already
        // the result of strike range resolution from the last rebalance. Re-resolving
        // here would shift the window during hysteresis/cooldown, dropping buffered data.
        let active_strikes: HashSet<Price> = self
            .active_ids
            .iter()
            .filter_map(|id| self.instruments.get(id).map(|(s, _)| *s))
            .collect();

        // Build filtered snapshot (clone from buffers)
        let mut calls = BTreeMap::new();

        for (strike, data) in &self.call_buffer {
            if active_strikes.contains(strike) {
                calls.insert(*strike, data.clone());
            }
        }
        let mut puts = BTreeMap::new();

        for (strike, data) in &self.put_buffer {
            if active_strikes.contains(strike) {
                puts.insert(*strike, data.clone());
            }
        }

        // Use the max observed ts_event from quotes, falling back to ts_init
        let ts_event = if self.max_ts_event == UnixNanos::default() {
            ts_init
        } else {
            self.max_ts_event
        };

        OptionChainSlice {
            series_id: self.series_id,
            atm_strike,
            calls,
            puts,
            ts_event,
            ts_init,
        }
    }

    /// Returns `true` if both buffers are empty.
    #[must_use]
    pub fn is_buffer_empty(&self) -> bool {
        self.call_buffer.is_empty() && self.put_buffer.is_empty()
    }

    /// Checks whether the instrument set should be rebalanced around the current ATM.
    ///
    /// Returns `None` when no rebalancing is needed (fixed ranges, no ATM price,
    /// ATM strike unchanged, hysteresis not exceeded, or cooldown not elapsed).
    /// Returns `Some(RebalanceAction)` with instrument add/remove lists when the
    /// closest ATM strike shifts past the hysteresis threshold.
    ///
    /// `Delta` ranges resolve from Greeks rather than an ATM window, so their
    /// active set can change while the closest ATM strike is unchanged. They skip
    /// the ATM-shift and hysteresis gates and rebalance on any resolved-set change,
    /// with the cooldown still applied to throttle churn.
    #[must_use]
    pub fn check_rebalance(&self, now_ns: UnixNanos) -> Option<RebalanceAction> {
        // Fixed ranges never rebalance
        if matches!(self.strike_range, StrikeRange::Fixed(_)) {
            return None;
        }

        let atm_price = self.atm_tracker.atm_price()?;
        let all_strikes = Self::sorted_strikes(&self.instruments);
        let current_atm_strike = Self::find_closest_strike(&all_strikes, atm_price)?;

        let is_delta = matches!(self.strike_range, StrikeRange::Delta { .. });

        if !is_delta {
            // No change: no rebalance
            if self.last_atm_strike == Some(current_atm_strike) {
                return None;
            }

            // Hysteresis check: price must cross hysteresis fraction of the gap to next strike
            if let Some(last_strike) = self.last_atm_strike
                && self.hysteresis > 0.0
            {
                // Find the next strike in the direction of price movement
                let next_strike = match atm_price.cmp(&last_strike) {
                    Ordering::Greater => all_strikes.iter().find(|s| **s > last_strike),
                    Ordering::Less => all_strikes.iter().rev().find(|s| **s < last_strike),
                    Ordering::Equal => None,
                };

                if let Some(next) = next_strike {
                    let progress = (atm_price.as_decimal() - last_strike.as_decimal()).abs();
                    let gap = (next.as_decimal() - last_strike.as_decimal()).abs();
                    let progress_ratio = (progress / gap).to_f64().unwrap_or(f64::MAX);
                    if progress_ratio < self.hysteresis {
                        return None;
                    }
                }
            }
        }

        // Cooldown check
        if !self.cooldown_ns.is_zero()
            && let Some(last_ts) = self.last_rebalance_ns
            && now_ns.saturating_duration_since(last_ts) < self.cooldown_ns
        {
            return None;
        }

        // Compute new active set
        let new_active_strikes: HashSet<Price> = self
            .resolve_active_strikes(Some(atm_price), &all_strikes)
            .into_iter()
            .collect();
        let new_active: HashSet<InstrumentId> = self
            .instruments
            .iter()
            .filter(|(_, (s, _))| new_active_strikes.contains(s))
            .map(|(id, _)| *id)
            .collect();

        let add: Vec<InstrumentId> = new_active.difference(&self.active_ids).copied().collect();
        let remove: Vec<InstrumentId> = self.active_ids.difference(&new_active).copied().collect();

        // Suppress no-op delta rebalances so the cooldown timestamp is not reset on
        // every snapshot while the resolved set is stable.
        if is_delta && add.is_empty() && remove.is_empty() {
            return None;
        }

        Some(RebalanceAction { add, remove })
    }

    /// Applies a rebalance action: updates the active ID set, cleans stale buffers,
    /// and records the rebalance timestamp.
    pub fn apply_rebalance(&mut self, action: &RebalanceAction, now_ns: UnixNanos) {
        for id in &action.add {
            self.active_ids.insert(*id);
        }

        for id in &action.remove {
            self.active_ids.remove(id);
        }

        // Clean buffers for strikes no longer in active set
        let active_strikes: HashSet<Price> = self
            .active_ids
            .iter()
            .filter_map(|id| self.instruments.get(id))
            .map(|(s, _)| *s)
            .collect();
        self.call_buffer
            .retain(|strike, _| active_strikes.contains(strike));
        self.put_buffer
            .retain(|strike, _| active_strikes.contains(strike));
        self.pending_greeks
            .retain(|id, _| self.active_ids.contains(id));

        // Update last_atm_strike and record rebalance timestamp
        if let Some(atm) = self.atm_tracker.atm_price() {
            let all_strikes = Self::sorted_strikes(&self.instruments);
            self.last_atm_strike = Self::find_closest_strike(&all_strikes, atm);
        }
        self.last_rebalance_ns = Some(now_ns);
    }
}

/// Describes instruments to add and remove during an ATM rebalance.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RebalanceAction {
    /// Instruments to subscribe to (newly in range).
    pub add: Vec<InstrumentId>,
    /// Instruments to unsubscribe from (no longer in range).
    pub remove: Vec<InstrumentId>,
}

#[cfg(test)]
impl OptionChainAggregator {
    fn call_buffer_len(&self) -> usize {
        self.call_buffer.len()
    }

    fn put_buffer_len(&self) -> usize {
        self.put_buffer.len()
    }

    fn get_call_greeks_from_buffer(&self, strike: &Price) -> Option<&OptionGreeks> {
        self.call_buffer.get(strike).and_then(|d| d.greeks.as_ref())
    }

    pub(crate) fn last_atm_strike(&self) -> Option<Price> {
        self.last_atm_strike
    }

    fn set_hysteresis(&mut self, h: f64) {
        self.hysteresis = h;
    }

    fn set_cooldown_ns(&mut self, ns: DurationNanos) {
        self.cooldown_ns = ns;
    }

    fn pending_greeks_count(&self) -> usize {
        self.pending_greeks.len()
    }
}

#[cfg(test)]
mod tests {
    use nautilus_model::{data::greeks::OptionGreekValues, identifiers::Venue, types::Quantity};
    use rstest::*;

    use super::*;

    fn make_series_id() -> OptionSeriesId {
        OptionSeriesId::new(
            Venue::new("DERIBIT"),
            ustr::Ustr::from("BTC"),
            ustr::Ustr::from("BTC"),
            UnixNanos::from(1_700_000_000_000_000_000u64),
        )
    }

    fn make_quote(instrument_id: InstrumentId, bid: &str, ask: &str) -> QuoteTick {
        QuoteTick::new(
            instrument_id,
            Price::from(bid),
            Price::from(ask),
            Quantity::from("1.0"),
            Quantity::from("1.0"),
            UnixNanos::from(1u64),
            UnixNanos::from(1u64),
        )
    }

    fn now() -> UnixNanos {
        // A base timestamp for tests (far enough from zero to avoid edge cases)
        UnixNanos::from(1_000_000_000_000_000_000u64)
    }

    /// Sets ATM price on an aggregator via a synthetic `OptionGreeks` with the given forward price.
    fn set_atm_via_greeks(agg: &mut OptionChainAggregator, price: f64) {
        let greeks = OptionGreeks {
            instrument_id: InstrumentId::from("BTC-20240101-50000-C.DERIBIT"),
            underlying_price: Some(price),
            ..Default::default()
        };
        agg.atm_tracker_mut().update_from_option_greeks(&greeks);
    }

    fn make_aggregator() -> (OptionChainAggregator, InstrumentId, InstrumentId) {
        let call_id = InstrumentId::from("BTC-20240101-50000-C.DERIBIT");
        let put_id = InstrumentId::from("BTC-20240101-50000-P.DERIBIT");
        let strike = Price::from("50000");

        let mut instrument_map = HashMap::new();
        instrument_map.insert(call_id, (strike, OptionKind::Call));
        instrument_map.insert(put_id, (strike, OptionKind::Put));

        let tracker = AtmTracker::new();
        let agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::Fixed(vec![strike]),
            tracker,
            instrument_map,
        );

        (agg, call_id, put_id)
    }

    #[rstest]
    fn test_find_closest_strike_preserves_subprecision_atm() {
        let mut atm = Price::from("100.75");
        atm.precision = 0;
        let strikes = [Price::from("100"), Price::from("101")];
        assert_eq!(
            OptionChainAggregator::find_closest_strike(&strikes, atm),
            Some(strikes[1])
        );
    }

    #[rstest]
    fn test_find_closest_strike_prefers_exact_high_value_match() {
        let collapsed = Price::from("9007199253.999000000");
        let atm = Price::from("9007199253.999000001");
        let strikes = [collapsed, atm];
        assert_eq!(collapsed.as_f64(), atm.as_f64());

        let result = OptionChainAggregator::find_closest_strike(&strikes, atm);

        assert_eq!(result, Some(atm));
    }

    #[rstest]
    fn test_aggregator_instrument_ids() {
        let (agg, call_id, put_id) = make_aggregator();
        let ids = agg.instrument_ids();
        assert_eq!(ids.len(), 2);
        assert!(ids.contains(&call_id));
        assert!(ids.contains(&put_id));
    }

    #[rstest]
    fn test_aggregator_update_quote() {
        let (mut agg, call_id, _) = make_aggregator();
        let quote = make_quote(call_id, "100.00", "101.00");

        agg.update_quote(&quote);

        assert_eq!(agg.call_buffer_len(), 1);
        assert_eq!(agg.put_buffer_len(), 0);
    }

    #[rstest]
    fn test_aggregator_update_greeks() {
        let (mut agg, call_id, _) = make_aggregator();
        let quote = make_quote(call_id, "100.00", "101.00");
        agg.update_quote(&quote);

        let greeks = OptionGreeks {
            instrument_id: call_id,
            greeks: OptionGreekValues {
                delta: 0.55,
                ..Default::default()
            },
            ..Default::default()
        };
        agg.update_greeks(&greeks);

        let strike = Price::from("50000");
        let data = agg.get_call_greeks_from_buffer(&strike);
        assert!(data.is_some());
        assert_eq!(data.unwrap().delta, 0.55);
    }

    #[rstest]
    fn test_aggregator_snapshot_preserves_state() {
        let (mut agg, call_id, _) = make_aggregator();
        let quote = make_quote(call_id, "100.00", "101.00");
        agg.update_quote(&quote);

        let slice = agg.snapshot(UnixNanos::from(100u64));
        assert_eq!(slice.call_count(), 1);
        assert_eq!(slice.ts_init, UnixNanos::from(100u64));

        // Buffers should still contain data (keep-latest semantics)
        assert!(!agg.is_buffer_empty());

        // Second snapshot should return the same data
        let slice2 = agg.snapshot(UnixNanos::from(200u64));
        assert_eq!(slice2.call_count(), 1);
        assert_eq!(slice2.ts_init, UnixNanos::from(200u64));
    }

    #[rstest]
    fn test_aggregator_ignores_unknown_instrument() {
        let (mut agg, _, _) = make_aggregator();
        let unknown_id = InstrumentId::from("ETH-20240101-3000-C.DERIBIT");
        let quote = make_quote(unknown_id, "100.00", "101.00");

        agg.update_quote(&quote);

        assert!(agg.is_buffer_empty());
    }

    #[rstest]
    fn test_check_rebalance_returns_none() {
        let (agg, _, _) = make_aggregator();
        assert!(agg.check_rebalance(now()).is_none());
    }

    // -- Rebalance tests --

    /// Builds instruments with 5 strike prices (45000..55000 step 2500) and `AtmRelative` +-1.
    /// Hysteresis and cooldown are disabled so existing rebalance tests pass unchanged.
    fn make_multi_strike_aggregator() -> OptionChainAggregator {
        let strikes = [45000, 47500, 50000, 52500, 55000];
        let mut instruments = HashMap::new();

        for s in &strikes {
            let strike = Price::from(&s.to_string());
            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
            let put_id = InstrumentId::from(&format!("BTC-20240101-{s}-P.DERIBIT"));
            instruments.insert(call_id, (strike, OptionKind::Call));
            instruments.insert(put_id, (strike, OptionKind::Put));
        }

        let tracker = AtmTracker::new();
        let mut agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::AtmRelative {
                strikes_above: 1,
                strikes_below: 1,
            },
            tracker,
            instruments,
        );
        // Disable guards so existing tests exercise pure rebalance logic
        agg.set_hysteresis(0.0);
        agg.set_cooldown_ns(DurationNanos::default());
        agg
    }

    #[rstest]
    fn test_check_rebalance_fixed_always_none() {
        // Fixed range + ATM price set: still returns None
        let (mut agg, _, _) = make_aggregator();
        set_atm_via_greeks(&mut agg, 50000.0);
        assert!(agg.check_rebalance(now()).is_none());
    }

    #[rstest]
    fn test_check_rebalance_no_atm_returns_none() {
        let agg = make_multi_strike_aggregator();
        // No ATM price set: None
        assert!(agg.check_rebalance(now()).is_none());
    }

    #[rstest]
    fn test_check_rebalance_atm_unchanged_returns_none() {
        let mut agg = make_multi_strike_aggregator();
        // Set ATM to 50000 and apply initial rebalance
        set_atm_via_greeks(&mut agg, 50000.0);
        // First check detects ATM shift from None to 50000
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());

        // ATM moves slightly but stays closest to 50000
        set_atm_via_greeks(&mut agg, 50200.0);
        assert!(agg.check_rebalance(now()).is_none());
    }

    #[rstest]
    fn test_check_rebalance_detects_atm_shift() {
        let mut agg = make_multi_strike_aggregator();
        // Set ATM near 50000
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());
        // Active: 47500, 50000, 52500 (ATM=50000, +-1 strike)
        assert_eq!(agg.instrument_ids().len(), 6); // 3 strikes * 2

        // Now shift ATM to 55000
        set_atm_via_greeks(&mut agg, 55000.0);
        let action2 = agg.check_rebalance(now()).unwrap();
        // Should have instruments to add (55000) and remove (47500)
        assert!(!action2.add.is_empty() || !action2.remove.is_empty());
    }

    #[rstest]
    fn test_apply_rebalance_updates_instrument_map() {
        let mut agg = make_multi_strike_aggregator();
        // Set ATM near 50000
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());

        // Active should be 3 strikes (47500, 50000, 52500)
        let active_ids = agg.instrument_ids();
        assert_eq!(active_ids.len(), 6); // 3 strikes * 2 (call + put)

        // Now shift to 55000
        set_atm_via_greeks(&mut agg, 55000.0);
        let action2 = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action2, now());

        // Active should now be (52500, 55000): 2 strikes at the top end
        let active_ids2 = agg.instrument_ids();
        assert_eq!(active_ids2.len(), 4); // 2 strikes * 2
    }

    #[rstest]
    fn test_apply_rebalance_cleans_buffers() {
        let mut agg = make_multi_strike_aggregator();
        // Set ATM near 50000
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());

        // Feed quotes for the 47500 call
        let call_47500 = InstrumentId::from("BTC-20240101-47500-C.DERIBIT");
        let quote = make_quote(call_47500, "100.00", "101.00");
        agg.update_quote(&quote);
        assert_eq!(agg.call_buffer_len(), 1);

        // Now shift ATM up so 47500 is out of range
        set_atm_via_greeks(&mut agg, 55000.0);
        let action2 = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action2, now());

        // Buffer for 47500 should be cleaned
        assert_eq!(agg.call_buffer_len(), 0);
    }

    #[rstest]
    fn test_initial_active_set_empty_when_no_atm() {
        let agg = make_multi_strike_aggregator();
        // AtmRelative with no ATM price: empty active set (deferred)
        assert_eq!(agg.instrument_ids().len(), 0);
        assert_eq!(agg.all_instrument_ids().len(), 10);
    }

    #[rstest]
    fn test_catalog_vs_active_separation() {
        let mut agg = make_multi_strike_aggregator();
        // Set ATM near 50000 to narrow active set
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());

        // Catalog should still have all 10 instruments
        assert_eq!(agg.instruments().len(), 10);
        // Active should be a subset
        assert_eq!(agg.instrument_ids().len(), 6);
    }

    // -- add_instrument tests --

    #[rstest]
    fn test_add_instrument_already_known() {
        let (mut agg, call_id, _) = make_aggregator();
        let strike = Price::from("50000");
        let count_before = agg.instruments().len();

        let result = agg.add_instrument(call_id, strike, OptionKind::Call);

        assert!(!result);
        assert_eq!(agg.instruments().len(), count_before);
    }

    #[rstest]
    fn test_add_instrument_new_in_active_range() {
        let (mut agg, _, _) = make_aggregator();
        // Fixed range includes strike 50000; adding another instrument at same strike
        let new_id = InstrumentId::from("BTC-20240101-50000-C2.DERIBIT");
        let strike = Price::from("50000");

        let result = agg.add_instrument(new_id, strike, OptionKind::Call);

        assert!(result);
        assert_eq!(agg.instruments().len(), 3);
        assert!(agg.active_ids().contains(&new_id));
    }

    #[rstest]
    fn test_add_instrument_new_out_of_range() {
        let (mut agg, _, _) = make_aggregator();
        // Fixed range only includes 50000; adding instrument at 60000
        let new_id = InstrumentId::from("BTC-20240101-60000-C.DERIBIT");
        let strike = Price::from("60000");

        let result = agg.add_instrument(new_id, strike, OptionKind::Call);

        assert!(result);
        assert_eq!(agg.instruments().len(), 3);
        assert!(!agg.active_ids().contains(&new_id));
    }

    #[rstest]
    fn test_add_instrument_available_for_rebalance() {
        let mut agg = make_multi_strike_aggregator();
        // Set ATM near 50000 and apply initial rebalance
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());
        // Active: 47500, 50000, 52500 (6 instruments)
        assert_eq!(agg.instrument_ids().len(), 6);

        // Add a new instrument at strike 57500 (out of current range)
        let new_id = InstrumentId::from("BTC-20240101-57500-C.DERIBIT");
        let strike = Price::from("57500");
        let result = agg.add_instrument(new_id, strike, OptionKind::Call);
        assert!(result);
        assert!(!agg.active_ids().contains(&new_id));

        // Shift ATM to 57500: rebalance should pick up the new instrument
        set_atm_via_greeks(&mut agg, 57500.0);
        let action2 = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action2, now());

        assert!(agg.active_ids().contains(&new_id));
    }

    // -- Hysteresis tests --

    #[rstest]
    fn test_hysteresis_blocks_small_movement() {
        let strikes = [47500, 50000, 52500];
        let mut instruments = HashMap::new();

        for s in &strikes {
            let strike = Price::from(&s.to_string());
            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
            instruments.insert(call_id, (strike, OptionKind::Call));
        }
        let tracker = AtmTracker::new();
        let mut agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::AtmRelative {
                strikes_above: 1,
                strikes_below: 1,
            },
            tracker,
            instruments,
        );
        agg.set_hysteresis(0.6);
        agg.set_cooldown_ns(DurationNanos::default());

        // Set ATM to 50000
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());
        assert_eq!(agg.last_atm_strike(), Some(Price::from("50000")));

        // Move ATM slightly toward 52500: gap=2500, threshold=50000+0.6*2500=51500
        // 51000 does NOT cross 51500
        set_atm_via_greeks(&mut agg, 51000.0);
        assert!(agg.check_rebalance(now()).is_none());
    }

    #[rstest]
    fn test_hysteresis_allows_large_movement() {
        let strikes = [47500, 50000, 52500];
        let mut instruments = HashMap::new();

        for s in &strikes {
            let strike = Price::from(&s.to_string());
            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
            instruments.insert(call_id, (strike, OptionKind::Call));
        }
        let tracker = AtmTracker::new();
        let mut agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::AtmRelative {
                strikes_above: 1,
                strikes_below: 1,
            },
            tracker,
            instruments,
        );
        agg.set_hysteresis(0.6);
        agg.set_cooldown_ns(DurationNanos::default());

        // Set ATM to 50000
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());

        // Move ATM well past threshold: 52000 > 51500
        set_atm_via_greeks(&mut agg, 52000.0);
        assert!(agg.check_rebalance(now()).is_some());
    }

    #[rstest]
    fn test_hysteresis_exact_strike_gap_boundary() {
        let lower = Price::from("9007199253.999000000");
        let upper = Price::from("9007199253.999002800");
        let blocked = Price::from("9007199253.999001673");
        let allowed = Price::from("9007199253.999001680");
        let mut instruments = HashMap::new();
        instruments.insert(
            InstrumentId::from("BTC-LOW-C.DERIBIT"),
            (lower, OptionKind::Call),
        );
        instruments.insert(
            InstrumentId::from("BTC-HIGH-C.DERIBIT"),
            (upper, OptionKind::Call),
        );
        let mut tracker = AtmTracker::new();
        tracker.set_initial_price(lower);
        let mut agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::AtmRelative {
                strikes_above: 0,
                strikes_below: 0,
            },
            tracker,
            instruments,
        );
        agg.set_hysteresis(0.6);
        agg.set_cooldown_ns(DurationNanos::default());

        agg.atm_tracker_mut().set_initial_price(blocked);
        assert!(agg.check_rebalance(now()).is_none());

        agg.atm_tracker_mut().set_initial_price(allowed);
        assert!(agg.check_rebalance(now()).is_some());
    }

    #[rstest]
    fn test_zero_hysteresis_disables_guard() {
        let mut agg = make_multi_strike_aggregator();
        agg.set_hysteresis(0.0);
        agg.set_cooldown_ns(DurationNanos::default());

        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());

        // Any shift past the strike boundary triggers rebalance
        set_atm_via_greeks(&mut agg, 52500.0);
        assert!(agg.check_rebalance(now()).is_some());
    }

    // -- Cooldown tests --

    #[rstest]
    fn test_cooldown_blocks_rapid_rebalance() {
        let mut agg = make_multi_strike_aggregator();
        agg.set_hysteresis(0.0);
        agg.set_cooldown_ns(DurationNanos::from_secs(5));

        set_atm_via_greeks(&mut agg, 50000.0);
        let t0 = now();
        let action = agg.check_rebalance(t0).unwrap();
        agg.apply_rebalance(&action, t0);

        // Shift ATM immediately: cooldown blocks
        set_atm_via_greeks(&mut agg, 55000.0);
        let t1 = UnixNanos::from(t0.as_u64() + 1_000_000_000); // 1s later
        assert!(agg.check_rebalance(t1).is_none());
    }

    #[rstest]
    fn test_cooldown_allows_after_elapsed() {
        let mut agg = make_multi_strike_aggregator();
        agg.set_hysteresis(0.0);
        agg.set_cooldown_ns(DurationNanos::from_secs(5));

        set_atm_via_greeks(&mut agg, 50000.0);
        let t0 = now();
        let action = agg.check_rebalance(t0).unwrap();
        agg.apply_rebalance(&action, t0);

        // Shift ATM after cooldown elapses
        set_atm_via_greeks(&mut agg, 55000.0);
        let t1 = UnixNanos::from(t0.as_u64() + 6_000_000_000); // 6s later
        assert!(agg.check_rebalance(t1).is_some());
    }

    #[rstest]
    fn test_zero_cooldown_disables_guard() {
        let mut agg = make_multi_strike_aggregator();
        agg.set_hysteresis(0.0);
        agg.set_cooldown_ns(DurationNanos::default());

        set_atm_via_greeks(&mut agg, 50000.0);
        let t0 = now();
        let action = agg.check_rebalance(t0).unwrap();
        agg.apply_rebalance(&action, t0);

        // Shift ATM immediately: no cooldown block
        set_atm_via_greeks(&mut agg, 55000.0);
        assert!(agg.check_rebalance(t0).is_some());
    }

    // -- Pending greeks tests --

    #[rstest]
    fn test_pending_greeks_consumed_on_first_quote() {
        let (mut agg, call_id, _) = make_aggregator();

        // Send greeks before any quote
        let greeks = OptionGreeks {
            instrument_id: call_id,
            greeks: OptionGreekValues {
                delta: 0.55,
                ..Default::default()
            },
            ..Default::default()
        };
        agg.update_greeks(&greeks);
        assert_eq!(agg.pending_greeks_count(), 1);

        // Now send the first quote: pending greeks should be consumed
        let quote = make_quote(call_id, "100.00", "101.00");
        agg.update_quote(&quote);
        assert_eq!(agg.pending_greeks_count(), 0);

        // Verify greeks were attached
        let strike = Price::from("50000");
        let data = agg.get_call_greeks_from_buffer(&strike);
        assert!(data.is_some());
        assert_eq!(data.unwrap().delta, 0.55);
    }

    // -- ts_event tracking tests --

    #[rstest]
    fn test_snapshot_ts_event_reflects_max_quote_timestamp() {
        let (mut agg, call_id, put_id) = make_aggregator();

        let quote1 = QuoteTick::new(
            call_id,
            Price::from("100.00"),
            Price::from("101.00"),
            Quantity::from("1.0"),
            Quantity::from("1.0"),
            UnixNanos::from(500u64), // ts_event
            UnixNanos::from(500u64),
        );
        agg.update_quote(&quote1);

        let quote2 = QuoteTick::new(
            put_id,
            Price::from("50.00"),
            Price::from("51.00"),
            Quantity::from("1.0"),
            Quantity::from("1.0"),
            UnixNanos::from(800u64), // ts_event: later
            UnixNanos::from(800u64),
        );
        agg.update_quote(&quote2);

        let slice = agg.snapshot(UnixNanos::from(1000u64));
        assert_eq!(slice.ts_event, UnixNanos::from(800u64));
        assert_eq!(slice.ts_init, UnixNanos::from(1000u64));
    }

    #[rstest]
    fn test_snapshot_ts_event_fallback_when_no_quotes() {
        let (agg, _, _) = make_aggregator();
        let slice = agg.snapshot(UnixNanos::from(1000u64));
        // No quotes: ts_event falls back to ts_init
        assert_eq!(slice.ts_event, UnixNanos::from(1000u64));
    }

    #[rstest]
    fn test_snapshot_retains_buffered_data_during_hysteresis_window() {
        // Setup: 3 strikes at 47500/50000/52500, AtmRelative +-1, hysteresis enabled
        let strikes = [47500, 50000, 52500];
        let mut instruments = HashMap::new();

        for s in &strikes {
            let strike = Price::from(&s.to_string());
            let call_id = InstrumentId::from(&format!("BTC-20240101-{s}-C.DERIBIT"));
            instruments.insert(call_id, (strike, OptionKind::Call));
        }
        let tracker = AtmTracker::new();
        let mut agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::AtmRelative {
                strikes_above: 1,
                strikes_below: 1,
            },
            tracker,
            instruments,
        );
        agg.set_hysteresis(0.6);
        agg.set_cooldown_ns(DurationNanos::default());

        // Set ATM to 50000, rebalance -> active: {47500, 50000, 52500}
        set_atm_via_greeks(&mut agg, 50000.0);
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());
        assert_eq!(agg.instrument_ids().len(), 3);

        // Buffer quotes for all active strikes
        let q1 = make_quote(
            InstrumentId::from("BTC-20240101-47500-C.DERIBIT"),
            "3000.00",
            "3100.00",
        );
        let q2 = make_quote(
            InstrumentId::from("BTC-20240101-50000-C.DERIBIT"),
            "1500.00",
            "1600.00",
        );
        let q3 = make_quote(
            InstrumentId::from("BTC-20240101-52500-C.DERIBIT"),
            "500.00",
            "600.00",
        );
        agg.update_quote(&q1);
        agg.update_quote(&q2);
        agg.update_quote(&q3);
        assert_eq!(agg.call_buffer_len(), 3);

        // Move ATM slightly toward 52500 but within hysteresis band (no rebalance)
        set_atm_via_greeks(&mut agg, 51000.0);
        assert!(agg.check_rebalance(now()).is_none());

        // Snapshot must still include all 3 buffered strikes
        let slice = agg.snapshot(UnixNanos::from(100u64));
        assert_eq!(slice.call_count(), 3);
    }

    #[rstest]
    fn test_remove_instrument_from_catalog() {
        let (mut agg, call_id, put_id) = make_aggregator();
        assert_eq!(agg.instruments().len(), 2);

        let removed = agg.remove_instrument(&call_id);
        assert!(removed);
        assert_eq!(agg.instruments().len(), 1);
        assert!(!agg.active_ids().contains(&call_id));
        assert!(agg.instruments().contains_key(&put_id));
    }

    #[rstest]
    fn test_remove_instrument_cleans_buffer() {
        let (mut agg, call_id, _) = make_aggregator();
        let quote = make_quote(call_id, "100.00", "101.00");
        agg.update_quote(&quote);
        assert_eq!(agg.call_buffer_len(), 1);

        let _ = agg.remove_instrument(&call_id);
        // No sibling call at same strike, buffer entry should be removed
        assert_eq!(agg.call_buffer_len(), 0);
    }

    #[rstest]
    fn test_remove_instrument_preserves_sibling_buffer() {
        let (mut agg, call_id, _) = make_aggregator();
        // Add a second call at the same strike
        let sibling_id = InstrumentId::from("BTC-20240101-50000-C2.DERIBIT");
        let strike = Price::from("50000");
        let _ = agg.add_instrument(sibling_id, strike, OptionKind::Call);

        let quote = make_quote(call_id, "100.00", "101.00");
        agg.update_quote(&quote);
        assert_eq!(agg.call_buffer_len(), 1);

        // Remove original: sibling still shares the strike+kind
        let _ = agg.remove_instrument(&call_id);
        assert_eq!(agg.call_buffer_len(), 1); // buffer preserved
        assert!(agg.instruments().contains_key(&sibling_id));
    }

    #[rstest]
    fn test_remove_instrument_unknown_noop() {
        let (mut agg, _, _) = make_aggregator();
        let unknown = InstrumentId::from("ETH-20240101-3000-C.DERIBIT");
        assert!(!agg.remove_instrument(&unknown));
        assert_eq!(agg.instruments().len(), 2);
    }

    #[rstest]
    fn test_remove_instrument_cleans_pending_greeks() {
        let (mut agg, call_id, _) = make_aggregator();
        let greeks = OptionGreeks {
            instrument_id: call_id,
            greeks: OptionGreekValues {
                delta: 0.55,
                ..Default::default()
            },
            ..Default::default()
        };
        agg.update_greeks(&greeks);
        assert_eq!(agg.pending_greeks_count(), 1);

        let _ = agg.remove_instrument(&call_id);
        assert_eq!(agg.pending_greeks_count(), 0);
    }

    #[rstest]
    fn test_is_catalog_empty_after_full_removal() {
        let (mut agg, call_id, put_id) = make_aggregator();
        assert!(!agg.is_catalog_empty());

        let _ = agg.remove_instrument(&call_id);
        assert!(!agg.is_catalog_empty());

        let _ = agg.remove_instrument(&put_id);
        assert!(agg.is_catalog_empty());
    }

    // -- Expiry guard tests --

    #[rstest]
    fn test_expired_quote_is_dropped() {
        let (mut agg, call_id, _) = make_aggregator();
        // Series expires at 1_700_000_000_000_000_000; send quote AT that timestamp
        let expired_quote = QuoteTick::new(
            call_id,
            Price::from("100.00"),
            Price::from("101.00"),
            Quantity::from("1.0"),
            Quantity::from("1.0"),
            UnixNanos::from(1_700_000_000_000_000_000u64),
            UnixNanos::from(1_700_000_000_000_000_000u64),
        );
        agg.update_quote(&expired_quote);
        assert!(agg.is_buffer_empty());
    }

    #[rstest]
    fn test_expired_greeks_are_dropped() {
        let (mut agg, call_id, _) = make_aggregator();
        // First add a valid quote so greeks would normally land in the buffer
        let quote = make_quote(call_id, "100.00", "101.00");
        agg.update_quote(&quote);
        assert_eq!(agg.call_buffer_len(), 1);

        // Send greeks at expiry timestamp: should be dropped
        let greeks = OptionGreeks {
            instrument_id: call_id,
            ts_event: UnixNanos::from(1_700_000_000_000_000_000u64),
            greeks: OptionGreekValues {
                delta: 0.55,
                ..Default::default()
            },
            ..Default::default()
        };
        agg.update_greeks(&greeks);

        let strike = Price::from("50000");
        assert!(agg.get_call_greeks_from_buffer(&strike).is_none());
    }

    // -- Delta range tests --

    /// Builds a `Delta`-range aggregator over `strikes` (call + put per strike),
    /// with hysteresis and cooldown disabled so rebalance decisions reflect pure
    /// delta resolution.
    fn make_delta_aggregator(
        strikes: &[i64],
        target: f64,
        tolerance: f64,
    ) -> OptionChainAggregator {
        let mut instruments = HashMap::new();

        for s in strikes {
            let strike = Price::from(&s.to_string());
            instruments.insert(option_id(*s, OptionKind::Call), (strike, OptionKind::Call));
            instruments.insert(option_id(*s, OptionKind::Put), (strike, OptionKind::Put));
        }
        let tracker = AtmTracker::new();
        let mut agg = OptionChainAggregator::new(
            make_series_id(),
            StrikeRange::Delta { target, tolerance },
            tracker,
            instruments,
        );
        agg.set_hysteresis(0.0);
        agg.set_cooldown_ns(DurationNanos::default());
        agg
    }

    fn option_id(strike: i64, kind: OptionKind) -> InstrumentId {
        let suffix = match kind {
            OptionKind::Call => "C",
            OptionKind::Put => "P",
        };
        InstrumentId::from(&format!("BTC-20240101-{strike}-{suffix}.DERIBIT"))
    }

    /// Feeds a quote then greeks (with the given `delta`) for one option leg.
    fn feed_quote_and_greeks(
        agg: &mut OptionChainAggregator,
        strike: i64,
        kind: OptionKind,
        delta: f64,
    ) {
        let id = option_id(strike, kind);
        agg.update_quote(&make_quote(id, "100.00", "101.00"));
        agg.update_greeks(&OptionGreeks {
            instrument_id: id,
            greeks: OptionGreekValues {
                delta,
                ..Default::default()
            },
            ..Default::default()
        });
    }

    #[rstest]
    #[case(0.30, 0.30, 0.03, true)] // exact target
    #[case(-0.30, 0.30, 0.03, true)] // negative delta, magnitude matches
    #[case(0.28, 0.30, 0.03, true)] // inside band, below target
    #[case(0.32, 0.30, 0.03, true)] // inside band, above target
    #[case(0.20, 0.30, 0.03, false)] // below band
    #[case(0.40, 0.30, 0.03, false)] // above band
    fn test_delta_within_band(
        #[case] delta: f64,
        #[case] target: f64,
        #[case] tolerance: f64,
        #[case] expected: bool,
    ) {
        assert_eq!(
            OptionChainAggregator::delta_within_band(delta, target, tolerance),
            expected
        );
    }

    #[rstest]
    fn test_delta_target_hit() {
        let strikes = [40000, 45000, 50000, 55000, 60000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        // Bootstrap the ATM-relative fallback so the window is active and greeks can land.
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();
        assert_eq!(agg.instrument_ids().len(), 10); // all 5 strikes (fallback)

        // Only the 55000 call sits at the 0.30 target.
        feed_quote_and_greeks(&mut agg, 40000, OptionKind::Call, 0.95);
        feed_quote_and_greeks(&mut agg, 40000, OptionKind::Put, -0.95);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.80);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.80);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.55);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.12);
        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Call, 0.12);
        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Put, -0.10);

        let active = agg.recompute_active_set();
        assert_eq!(active.len(), 2); // 55000 call + put
        assert!(active.contains(&option_id(55000, OptionKind::Call)));
        assert!(active.contains(&option_id(55000, OptionKind::Put)));
    }

    #[rstest]
    fn test_delta_tolerance_band() {
        let strikes = [45000, 50000, 55000, 60000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.05);
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();

        // Band is [0.25, 0.35]: 0.50 and 0.10 are outside, 0.32 and 0.30 inside.
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.50);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.50);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.32);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.50);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.50);
        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Call, 0.10);
        feed_quote_and_greeks(&mut agg, 60000, OptionKind::Put, -0.10);

        let active = agg.recompute_active_set();
        assert_eq!(active.len(), 4); // 50000 + 55000, both legs each
        assert!(active.contains(&option_id(50000, OptionKind::Call)));
        assert!(active.contains(&option_id(55000, OptionKind::Call)));
        assert!(!active.contains(&option_id(45000, OptionKind::Call)));
        assert!(!active.contains(&option_id(60000, OptionKind::Call)));
    }

    #[rstest]
    fn test_delta_matches_put_by_magnitude() {
        let strikes = [45000, 50000, 55000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();

        // Only the 45000 put matches the target, isolating put-side matching.
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.30);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.55);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.55);

        let active = agg.recompute_active_set();
        // |-0.30| == target, so the 45000 strike (both legs) is selected.
        assert_eq!(active.len(), 2);
        assert!(active.contains(&option_id(45000, OptionKind::Put)));
        assert!(active.contains(&option_id(45000, OptionKind::Call)));
    }

    #[rstest]
    fn test_delta_no_greeks_falls_back_to_atm_window() {
        // 13 strikes so the ATM-relative fallback window is a proper subset.
        let strikes: Vec<i64> = (0..13).map(|i| 40000 + i * 1000).collect();
        let mut agg = make_delta_aggregator(&strikes, 0.25, 0.05);
        set_atm_via_greeks(&mut agg, 46000.0); // centered

        let active = agg.recompute_active_set();

        // No greeks -> a bounded ATM-relative window: neither empty nor the full chain.
        // The exact window width is asserted in the model-level resolve test.
        assert!(active.len() > 2);
        assert!(active.len() < strikes.len() * 2);
        assert!(active.contains(&option_id(46000, OptionKind::Call))); // ATM included
        assert!(!active.contains(&option_id(40000, OptionKind::Call))); // extreme excluded
        assert!(!active.contains(&option_id(52000, OptionKind::Call))); // extreme excluded
    }

    #[rstest]
    fn test_delta_pending_only_greeks_eligible() {
        let strikes = [45000, 50000, 55000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();

        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.55);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.55);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.55);

        // Greeks arrive before any quote, so they land in pending_greeks.
        agg.update_greeks(&OptionGreeks {
            instrument_id: option_id(55000, OptionKind::Call),
            greeks: OptionGreekValues {
                delta: 0.30,
                ..Default::default()
            },
            ..Default::default()
        });
        assert_eq!(agg.pending_greeks_count(), 1);

        let active = agg.recompute_active_set();
        // Pending-only greeks are eligible for delta resolution.
        assert_eq!(active.len(), 2);
        assert!(active.contains(&option_id(55000, OptionKind::Call)));
    }

    #[rstest]
    fn test_delta_waits_for_fallback_window_greeks_before_narrowing() {
        let strikes = [45000, 50000, 55000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();
        assert_eq!(agg.instrument_ids().len(), 6); // fallback: all 3 strikes

        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);

        assert!(agg.check_rebalance(now()).is_none());
        assert_eq!(agg.instrument_ids().len(), 6);
    }

    #[rstest]
    fn test_delta_waits_when_fallback_window_shifts_during_warmup() {
        let strikes: Vec<i64> = (0..13).map(|i| 40000 + i * 1000).collect();
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        set_atm_via_greeks(&mut agg, 46000.0);
        agg.recompute_active_set();
        assert!(
            agg.active_ids()
                .contains(&option_id(42000, OptionKind::Call))
        );
        assert!(
            agg.active_ids()
                .contains(&option_id(51000, OptionKind::Put))
        );

        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.30);
        set_atm_via_greeks(&mut agg, 47000.0);

        let action = agg
            .check_rebalance(now())
            .expect("fallback window shift should rebalance active legs");

        assert!(action.add.contains(&option_id(52000, OptionKind::Call)));
        assert!(!action.remove.contains(&option_id(42000, OptionKind::Call)));
        assert!(!action.remove.contains(&option_id(51000, OptionKind::Put)));
    }

    #[rstest]
    fn test_delta_rebalances_on_greeks_with_atm_unchanged() {
        let strikes = [45000, 50000, 55000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();
        assert_eq!(agg.last_atm_strike(), Some(Price::from("50000")));
        assert_eq!(agg.instrument_ids().len(), 6); // fallback: all 3 strikes

        // Greeks arrive; only 55000 matches. The closest ATM strike is unchanged.
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.55);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.45);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.45);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.12);

        let action = agg
            .check_rebalance(now())
            .expect("delta range should rebalance when greeks narrow the set");
        assert!(action.add.is_empty());
        assert!(!action.remove.is_empty());

        agg.apply_rebalance(&action, now());
        assert_eq!(agg.instrument_ids().len(), 2); // narrowed to the 55000 legs
        assert!(
            agg.active_ids()
                .contains(&option_id(55000, OptionKind::Call))
        );
    }

    #[rstest]
    fn test_delta_no_op_rebalance_returns_none() {
        let strikes = [45000, 50000, 55000];
        let mut agg = make_delta_aggregator(&strikes, 0.30, 0.03);
        set_atm_via_greeks(&mut agg, 50000.0);
        agg.recompute_active_set();
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Call, 0.55);
        feed_quote_and_greeks(&mut agg, 45000, OptionKind::Put, -0.55);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Call, 0.45);
        feed_quote_and_greeks(&mut agg, 50000, OptionKind::Put, -0.45);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Call, 0.30);
        feed_quote_and_greeks(&mut agg, 55000, OptionKind::Put, -0.12);

        // First rebalance narrows to the 55000 legs.
        let action = agg.check_rebalance(now()).unwrap();
        agg.apply_rebalance(&action, now());
        assert_eq!(agg.instrument_ids().len(), 2);

        // Greeks unchanged -> stable set -> no-op suppressed (cooldown disabled).
        assert!(agg.check_rebalance(now()).is_none());
    }
}