hypercall/catalog_manager/

manager.rs

//! CatalogManager - Main reconciliation loop.
//!
//! Runs at a configurable interval and reconciles desired markets/instruments
//! against actual database state. Uses advisory locks for distributed safety.

use anyhow::{Context, Result};
use hypercall_db::CatalogReader;
use hypercall_db_diesel::DieselDb;
use metrics::{counter, gauge};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{interval, Instant};
use tracing::{debug, error, info, warn};

use crate::price_oracle::hyperliquid_oracle::HyperliquidMarkPriceOracle;
use crate::rsm::unified_engine::MarketRequest;
use crate::shared::traits::MarkPriceOracle;
use catalog_manager::{
    format_symbol, generate_expiry_schedule_at_date, generate_strike_set_at_time,
    plan_extension_at_time, timestamp_to_code, CatalogConfig, ExpiryInfo, ExtensionRequest,
};
use hypercall_types::MarketActionMessage;

/// `watch`-channel payload published to the engine whenever the catalog
/// manager rewrites any `instruments.trading_mode` row. Keyed on
/// underlying — the engine calls
/// `UnifiedEngine::apply_underlying_trading_mode_update` with this map
/// to refresh its in-memory `instrument_trading_modes` cache.
pub type TradingModeUpdate = std::collections::HashMap<String, hypercall_types::TradingModes>;

/// CatalogManager handles continuous market/instrument reconciliation
pub struct CatalogManager {
    /// Async Diesel database connection pool
    diesel_db: Arc<DieselDb>,
    /// Catalog configuration
    config: CatalogConfig,
    /// Channel to send market creation requests (uses MarketRequest with response channel)
    market_sender: mpsc::Sender<MarketRequest>,
    /// Mark price oracles for spot prices
    mark_price_oracles: HashMap<String, Arc<HyperliquidMarkPriceOracle>>,
    /// Reconciliation interval
    interval_secs: u64,
    /// Live notify channel to the engine main loop. Every time
    /// `sync_trading_modes_to_db` rewrites any row we publish the full
    /// current `underlying -> TradingModes` map here. `watch` semantics
    /// mean the engine only observes the latest value, which is what
    /// we want -- we never care about superseded updates.
    trading_mode_notify: Option<tokio::sync::watch::Sender<TradingModeUpdate>>,
}

impl CatalogManager {
    /// Create a new CatalogManager
    pub fn new(
        diesel_db: Arc<DieselDb>,
        config: CatalogConfig,
        market_sender: mpsc::Sender<MarketRequest>,
        mark_price_oracles: HashMap<String, Arc<HyperliquidMarkPriceOracle>>,
        interval_secs: u64,
    ) -> Self {
        Self {
            diesel_db,
            config,
            market_sender,
            mark_price_oracles,
            interval_secs,
            trading_mode_notify: None,
        }
    }

    /// Attach a `tokio::sync::watch::Sender` that this catalog manager
    /// will publish `underlying → trading_mode` updates on whenever a
    /// periodic sync rewrites any row. Optional — call sites that don't
    /// need live notifications can omit this and the catalog manager
    /// behaves exactly as before (warn-only drift).
    pub fn with_trading_mode_notify(
        mut self,
        sender: tokio::sync::watch::Sender<TradingModeUpdate>,
    ) -> Self {
        self.trading_mode_notify = Some(sender);
        self
    }

    /// Execute one reconciliation tick
    async fn reconcile_tick(&self) -> Result<()> {
        let start = Instant::now();
        counter!("catalog_manager_ticks_total").increment(1);

        // Use transaction-scoped advisory lock: acquire on a dedicated connection,
        // hold it during reconciliation, then let it auto-release when dropped.
        // We use try_advisory_lock (session-level) on a connection we keep alive
        // during the entire reconciliation to prevent other instances from running.
        //
        // IMPORTANT: This connection is held for the duration of reconciliation.
        // All reconciliation queries use separate pool connections (DieselDb methods
        // each checkout their own). Ensure the pool has at least 2 async connections.
        let mut lock_conn = self
            .diesel_db
            .get_conn()
            .await
            .context("Failed to acquire DB connection for advisory lock")?;

        let acquired = self
            .diesel_db
            .try_acquire_advisory_lock_on_conn(
                &mut lock_conn,
                super::CATALOG_MANAGER_ADVISORY_LOCK_ID,
            )
            .await?;
        if !acquired {
            debug!("Another CatalogManager instance holds the lock, skipping this tick");
            return Ok(());
        }

        let result = self.do_reconciliation().await;

        if let Err(e) = self
            .diesel_db
            .release_advisory_lock_on_conn(&mut lock_conn, super::CATALOG_MANAGER_ADVISORY_LOCK_ID)
            .await
        {
            warn!("Failed to release advisory lock: {}", e);
        }

        let elapsed = start.elapsed();
        info!("CatalogManager tick completed in {:?}", elapsed);

        // Only update success timestamp on successful reconciliation
        if result.is_ok() {
            gauge!("catalog_manager_last_success_timestamp_seconds")
                .set(chrono::Utc::now().timestamp() as f64);
        }

        result
    }

    /// Perform the actual reconciliation work
    async fn do_reconciliation(&self) -> Result<()> {
        let expiry_times = self.config.expiry_times()?;
        let today = chrono::Utc::now().date_naive();

        // Process each underlying with its own expiry schedule: the dates
        // are usually shared policy, with per-underlying overrides for bespoke
        // listings like SPCX.
        for (underlying, underlying_config) in &self.config.underlyings {
            info!("Processing underlying: {}", underlying);

            let schedule_config = underlying_config
                .schedule
                .as_ref()
                .unwrap_or(&self.config.expiry.schedule);
            let schedule = generate_expiry_schedule_at_date(
                today,
                schedule_config.daily_count,
                schedule_config.weekly_count,
                schedule_config.monthly_count,
                schedule_config.weekdays_only,
                expiry_times.for_underlying(underlying),
            )?;

            info!(
                "Target expiry schedule for {}: {} expiries",
                underlying,
                schedule.expiries.len()
            );

            // Get spot price from oracle
            let spot_price = match self.get_spot_price(underlying).await {
                Some(price) => price,
                None => {
                    warn!("No spot price for {}, skipping", underlying);
                    continue;
                }
            };

            info!("{} spot price: {}", underlying, spot_price);

            let max_expiry_code = underlying_config.max_expiry_code;

            for expiry_info in &schedule.expiries {
                if let Some(cap) = max_expiry_code {
                    if expiry_info.code > cap {
                        continue;
                    }
                }
                if let Err(e) = self
                    .process_expiry(underlying, expiry_info, spot_price)
                    .await
                {
                    error!(
                        "Failed to process {}-{}: {:?}",
                        underlying, expiry_info.code, e
                    );
                    counter!("catalog_manager_errors_total").increment(1);
                }
            }
        }

        Ok(())
    }

    /// Process a single expiry for an underlying
    async fn process_expiry(
        &self,
        underlying: &str,
        expiry_info: &ExpiryInfo,
        spot_price: f64,
    ) -> Result<()> {
        let expiry_timestamp = expiry_info.timestamp;
        // DB stores expiry as YYYYMMDD code, not Unix timestamp
        let expiry_code = expiry_info.code as i64;

        // Check if market exists (using YYYYMMDD code for DB query)
        let market_exists = self
            .diesel_db
            .market_exists(underlying, expiry_code)
            .await?;

        // Get or create catalog state (using YYYYMMDD code for DB query)
        let catalog_state = self
            .diesel_db
            .get_catalog_listing_state(underlying, expiry_code)
            .await?;

        let (ref_price, strike_reference_timestamp_secs) = if let Some(state) = &catalog_state {
            // Use stored ref price for baseline grid stability
            match state.ref_price_at_listing.to_string().parse::<f64>() {
                Ok(price) if price > 0.0 => (price, state.listed_at.div_euclid(1000)),
                Ok(price) => {
                    anyhow::bail!(
                        "Invalid stored ref_price {} for {}/{}",
                        price,
                        underlying,
                        expiry_info.code
                    );
                }
                Err(e) => {
                    anyhow::bail!(
                        "Failed to parse stored ref_price '{}' for {}/{}: {}",
                        state.ref_price_at_listing,
                        underlying,
                        expiry_info.code,
                        e
                    );
                }
            }
        } else {
            // New market, use current spot as ref price
            (spot_price, chrono::Utc::now().timestamp())
        };

        // Generate strike set based on ref price (for baseline stability)
        let strike_set = generate_strike_set_at_time(
            underlying,
            ref_price,
            expiry_timestamp,
            &self.config.strike_selection,
            strike_reference_timestamp_secs,
        )?;

        debug!(
            "Generated {} strikes for {}-{} (ref_price={}, policy={:?})",
            strike_set.strikes.len(),
            underlying,
            expiry_info.code,
            ref_price,
            strike_set.policy
        );

        // Create market if it doesn't exist
        if !market_exists {
            info!("Creating market {}-{}", underlying, expiry_info.code);

            // Get first strike safely
            let first_strike = strike_set.strikes.first().copied().ok_or_else(|| {
                anyhow::anyhow!(
                    "No strikes generated for {}-{}",
                    underlying,
                    expiry_info.code
                )
            })?;

            // Create market via engine channel
            self.create_market(underlying, expiry_timestamp, first_strike, true)
                .await
                .context("Failed to create market")?;

            counter!("catalog_manager_markets_created_total").increment(1);

            // Insert catalog state (using YYYYMMDD code)
            self.diesel_db
                .insert_catalog_listing_state(
                    underlying,
                    expiry_code,
                    spot_price,
                    self.config.version as i32,
                )
                .await?;
        }

        // Get existing instruments for this market (using YYYYMMDD code)
        let existing_strikes = self
            .diesel_db
            .get_existing_strikes(underlying, expiry_code)
            .await?;

        // Build the union of baseline grid strikes and already-existing
        // strikes (which includes prior extension strikes). This ensures
        // that if an extension strike had only one side created, the
        // missing call/put is retried here rather than falling through
        // the cracks: the extension planner filters by strike value, so
        // a partially-present extension strike is excluded from
        // plan.new_strikes.
        let mut all_strikes = strike_set.strikes.clone();
        for &es in &existing_strikes {
            if !all_strikes.iter().any(|&s| (s - es).abs() < 1e-6) {
                all_strikes.push(es);
            }
        }
        all_strikes.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));

        let mut instruments_created = 0;
        let mut instruments_failed = 0;
        for &strike in &all_strikes {
            for is_call in [true, false] {
                let symbol = format_symbol(underlying, expiry_info.code, strike, is_call);

                if !self.diesel_db.instrument_exists(&symbol).await? {
                    match self
                        .create_market(underlying, expiry_timestamp, strike, is_call)
                        .await
                    {
                        Ok(()) => {
                            instruments_created += 1;
                        }
                        Err(e) => {
                            warn!("Failed to create instrument {}: {}; continuing", symbol, e);
                            instruments_failed += 1;
                        }
                    }
                }
            }
        }

        if instruments_created > 0 || instruments_failed > 0 {
            info!(
                "Created {} instruments for {}-{} ({} failed)",
                instruments_created, underlying, expiry_info.code, instruments_failed
            );
            counter!("catalog_manager_instruments_created_total").increment(instruments_created);
            if instruments_failed > 0 {
                counter!("catalog_manager_instrument_creation_failures_total")
                    .increment(instruments_failed);
            }
        }

        // Check for extension if policy is enabled
        if self.config.extension_policy.enabled {
            if let Some(state) = &catalog_state {
                // Check cooldown
                let now_ms = chrono::Utc::now().timestamp_millis();
                let cooldown_ms = (self.config.extension_policy.cooldown_secs * 1000) as i64;
                let last_ext = state.last_extension_at.unwrap_or(0);

                if now_ms - last_ext >= cooldown_ms {
                    // Plan extension
                    let plan = plan_extension_at_time(
                        ExtensionRequest {
                            underlying,
                            current_spot: spot_price,
                            ref_price_at_listing: ref_price,
                            expiry_timestamp,
                            existing_strikes: &existing_strikes,
                            reference_timestamp_secs: strike_reference_timestamp_secs,
                        },
                        &self.config.strike_selection,
                        &self.config.extension_policy,
                    )?;

                    if plan.needs_extension {
                        info!(
                            "Extension needed for {}-{}: {:?}",
                            underlying, expiry_info.code, plan.reason
                        );

                        let mut ext_created = 0;
                        for &strike in &plan.new_strikes {
                            for is_call in [true, false] {
                                let symbol =
                                    format_symbol(underlying, expiry_info.code, strike, is_call);

                                if !self.diesel_db.instrument_exists(&symbol).await? {
                                    match self
                                        .create_market(
                                            underlying,
                                            expiry_timestamp,
                                            strike,
                                            is_call,
                                        )
                                        .await
                                    {
                                        Ok(()) => {
                                            ext_created += 1;
                                        }
                                        Err(e) => {
                                            warn!(
                                                "Failed to create extension instrument {}: {}; continuing",
                                                symbol, e
                                            );
                                        }
                                    }
                                }
                            }
                        }

                        if ext_created > 0 {
                            info!(
                                "Extension created {} instruments for {}-{}",
                                ext_created, underlying, expiry_info.code
                            );
                            counter!("catalog_manager_extensions_total").increment(1);
                            counter!("catalog_manager_instruments_created_total")
                                .increment(ext_created);

                            // Update extension state (using YYYYMMDD code)
                            self.diesel_db
                                .update_extension_state(underlying, expiry_code, spot_price)
                                .await?;
                        }
                    }
                } else {
                    debug!(
                        "Extension cooldown active for {}-{} ({} ms remaining)",
                        underlying,
                        expiry_info.code,
                        cooldown_ms - (now_ms - last_ext)
                    );
                }
            }
        }

        // Sync `trading_mode` from catalog config to DB for every
        // underlying, including the default case. Gating on `!= default`
        // would skip the rollback scenario where an underlying flips
        // from `rfq` / `orderbook|rfq` back to plain `orderbook`: the
        // catalog config would carry the new value but `instruments.
        // trading_mode` would remain stale, and execution gating
        // (`allows_rfq` / `allows_orderbook`) would keep the old mode.
        // The `UPDATE ... WHERE trading_mode != $1` clause already
        // makes the write a no-op on rows that already match, so
        // running this unconditionally has no extra write cost.
        //
        // KNOWN LIMITATION: the engine's in-memory
        // `instrument_trading_modes` map is only populated on
        // recovery + CreateMarket paths and is NOT refreshed from
        // this DB update. After a catalog mode change here, the
        // running engine keeps applying the old mode until restart,
        // so admission for orders / RFQs against the mutated
        // instruments can diverge from the new catalog config. When
        // we detect that a row was actually changed, emit a loud
        // WARN so operators know a restart is needed to pick up the
        // new mode. TODO(CALL-RFQ-MODE-LIVE): plumb a Notify from
        // here into the engine main loop to refresh
        // `instrument_trading_modes` from DB on change.
        let mut any_row_changed = false;
        for (underlying, config) in &self.config.underlyings {
            let mode_str = config.trading_mode.as_db_str();
            match self
                .diesel_db
                .update_trading_mode_count(underlying, &mode_str)
                .await
            {
                Ok(rows) => {
                    if rows > 0 {
                        any_row_changed = true;
                        tracing::info!(
                            underlying,
                            trading_mode = %mode_str,
                            rows_changed = rows,
                            "trading_mode changed in DB; publishing live update to engine via notify channel"
                        );
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        underlying,
                        trading_mode = %mode_str,
                        "Failed to sync trading_mode: {}",
                        e
                    );
                }
            }
        }

        // If anything actually changed, publish the full current
        // underlying -> TradingModes map on the watch channel. We always
        // send the full map (not a delta) so the engine has no state
        // machine to reconcile -- it just replaces its view with what
        // the catalog says.
        if any_row_changed {
            if let Some(sender) = &self.trading_mode_notify {
                let current: TradingModeUpdate = self
                    .config
                    .underlyings
                    .iter()
                    .map(|(u, c)| (u.clone(), c.trading_mode))
                    .collect();
                // send_replace is the right watch primitive -- we want
                // the latest value, and we don't care if no receivers
                // are currently observing (it'll still be stored).
                let _ = sender.send(current);
            }
        }

        Ok(())
    }

    /// Get spot price from oracle
    async fn get_spot_price(&self, underlying: &str) -> Option<f64> {
        let oracle = self.mark_price_oracles.get(underlying)?;
        oracle.get_spot_price().await
    }

    /// Create a market/instrument via engine channel
    async fn create_market(
        &self,
        underlying: &str,
        expiry: i64,
        strike: f64,
        is_call: bool,
    ) -> Result<()> {
        use hypercall_types::{Market, MarketAction, OptionType};
        use rust_decimal::Decimal;

        let expiry_code = timestamp_to_code(expiry)
            .ok_or_else(|| anyhow::anyhow!("Invalid expiry timestamp: {}", expiry))?;
        let symbol = format_symbol(underlying, expiry_code, strike, is_call);

        let option_type = if is_call {
            OptionType::Call
        } else {
            OptionType::Put
        };

        let strike_decimal = Decimal::from_f64_retain(strike)
            .ok_or_else(|| anyhow::anyhow!("Invalid strike: {}", strike))?;

        let market = Market {
            symbol: symbol.clone(),
            underlying: underlying.to_string(),
            expiry: expiry_code as u64, // Send YYYYMMDD code, not Unix timestamp
            strike: strike_decimal,
            option_type,
        };

        let message = MarketActionMessage {
            market,
            action: MarketAction::CreateMarket,
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
        };

        // Create a one-shot channel for the response
        let (response_tx, mut response_rx) = mpsc::channel(1);

        let request = MarketRequest {
            message,
            response_tx,
        };

        self.market_sender
            .send(request)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to send market request: {}", e))?;

        // Wait for response (with timeout)
        match tokio::time::timeout(Duration::from_secs(10), response_rx.recv()).await {
            Ok(Some(response)) => {
                use hypercall_types::MarketUpdateStatus;
                match response.status {
                    MarketUpdateStatus::MarketCreated => {
                        // Success - instrument created
                    }
                    MarketUpdateStatus::MarketAlreadyExists => {
                        // Idempotent - market already exists, this is fine
                    }
                    MarketUpdateStatus::MarketCreationFailed => {
                        let error_msg = if let Some(ref reason) = response.reason {
                            format!("Market creation failed for {}: {}", symbol, reason)
                        } else {
                            format!("Market creation failed for {} (no reason provided)", symbol)
                        };
                        return Err(anyhow::anyhow!(error_msg));
                    }
                    status => {
                        return Err(anyhow::anyhow!(
                            "Unexpected market creation status {:?} for {}",
                            status,
                            symbol
                        ));
                    }
                }
            }
            Ok(None) => {
                return Err(anyhow::anyhow!(
                    "Market response channel closed for {}",
                    symbol
                ));
            }
            Err(_) => {
                return Err(anyhow::anyhow!(
                    "Timeout waiting for market creation response for {}",
                    symbol
                ));
            }
        }

        Ok(())
    }
}

#[async_trait::async_trait]
impl crate::shared::service::Service for CatalogManager {
    fn name(&self) -> &'static str {
        "CatalogManager"
    }

    fn owner(&self) -> crate::shared::service::ServiceOwner {
        crate::shared::service::ServiceOwner::Engine
    }

    async fn run(
        self: std::sync::Arc<Self>,
        mut shutdown: crate::shared::ShutdownRx,
    ) -> anyhow::Result<()> {
        info!(
            "Starting CatalogManager with interval={}s, version={}",
            self.interval_secs, self.config.version
        );

        let mut tick_interval = interval(Duration::from_secs(self.interval_secs));

        if let Err(e) = self.reconcile_tick().await {
            error!("Initial reconciliation failed: {}", e);
            counter!("catalog_manager_errors_total").increment(1);
        }

        loop {
            tokio::select! {
                _ = tick_interval.tick() => {
                    if let Err(e) = self.reconcile_tick().await {
                        error!("Reconciliation tick failed: {}", e);
                        counter!("catalog_manager_errors_total").increment(1);
                    }
                }
                _ = shutdown.recv() => {
                    info!("CatalogManager received shutdown signal");
                    break;
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // =========================================================================
    // format_symbol tests
    // =========================================================================

    #[test]
    fn test_format_symbol_basic_call() {
        assert_eq!(
            format_symbol("BTC", 20260110, 95000.0, true),
            "BTC-20260110-95000-C"
        );
    }

    #[test]
    fn test_format_symbol_basic_put() {
        assert_eq!(
            format_symbol("ETH", 20260227, 4000.0, false),
            "ETH-20260227-4000-P"
        );
    }

    #[test]
    fn test_format_symbol_with_decimal_strike() {
        // Strike with one decimal place
        assert_eq!(
            format_symbol("BTC", 20260315, 97500.5, true),
            "BTC-20260315-97500.5-C"
        );

        // Strike with two decimal places
        assert_eq!(
            format_symbol("ETH", 20260401, 3456.25, false),
            "ETH-20260401-3456.25-P"
        );
    }

    #[test]
    fn test_format_symbol_rounds_integer_strikes() {
        // Strike that should round cleanly
        assert_eq!(
            format_symbol("BTC", 20260110, 100000.0, true),
            "BTC-20260110-100000-C"
        );

        // Strike with tiny floating point error should still format as integer
        assert_eq!(
            format_symbol("BTC", 20260110, 100000.0000001, true),
            "BTC-20260110-100000-C"
        );
    }

    #[test]
    fn test_format_symbol_trailing_zeros_stripped() {
        // 3500.50 should become 3500.5, not 3500.50
        assert_eq!(
            format_symbol("ETH", 20260110, 3500.50, true),
            "ETH-20260110-3500.5-C"
        );
    }

    #[test]
    fn test_format_symbol_small_strikes() {
        // Small strike values (ETH-like)
        assert_eq!(
            format_symbol("ETH", 20260110, 3000.0, true),
            "ETH-20260110-3000-C"
        );
        assert_eq!(
            format_symbol("ETH", 20260110, 3250.0, false),
            "ETH-20260110-3250-P"
        );
    }

    #[test]
    fn test_format_symbol_large_strikes() {
        // Large strike values (BTC-like)
        assert_eq!(
            format_symbol("BTC", 20260110, 150000.0, true),
            "BTC-20260110-150000-C"
        );
        assert_eq!(
            format_symbol("BTC", 20260110, 200000.0, false),
            "BTC-20260110-200000-P"
        );
    }

    // =========================================================================
    // timestamp_to_code tests
    // =========================================================================

    #[test]
    fn test_timestamp_to_code_basic() {
        // 2026-01-10 00:00:00 UTC
        let code = timestamp_to_code(1768003200);
        assert_eq!(code, Some(20260110));
    }

    #[test]
    fn test_timestamp_to_code_with_time() {
        // 2026-01-31 08:00:00 UTC - typical 8am expiry
        let code = timestamp_to_code(1769846400);
        assert_eq!(code, Some(20260131));
    }

    #[test]
    fn test_timestamp_to_code_end_of_day() {
        // 2026-02-28 23:59:59 UTC
        let code = timestamp_to_code(1772323199);
        assert_eq!(code, Some(20260228));
    }

    #[test]
    fn test_timestamp_to_code_leap_year() {
        // 2028-02-29 (leap year) 08:00:00 UTC
        let code = timestamp_to_code(1835424000);
        assert_eq!(code, Some(20280229));
    }

    #[test]
    fn test_timestamp_to_code_year_boundaries() {
        // Dec 31, 2025 23:59:59 UTC
        let code_dec = timestamp_to_code(1767225599);
        assert_eq!(code_dec, Some(20251231));

        // Jan 1, 2026 00:00:00 UTC
        let code_jan = timestamp_to_code(1767225600);
        assert_eq!(code_jan, Some(20260101));
    }

    #[test]
    fn test_timestamp_to_code_various_months() {
        // Various months to ensure correct padding
        // 2026-03-15
        assert_eq!(timestamp_to_code(1773532800), Some(20260315));
        // 2026-12-01
        assert_eq!(timestamp_to_code(1796083200), Some(20261201));
    }

    // =========================================================================
    // CatalogManager construction tests
    // =========================================================================

    fn create_test_config() -> CatalogConfig {
        use catalog_manager::*;

        let mut underlyings = HashMap::new();
        underlyings.insert(
            "BTC".to_string(),
            UnderlyingConfig {
                vol_source: "deribit".to_string(),
                hl_symbol: None,
                trading_mode: hypercall_types::TradingModes::ORDERBOOK,
                max_expiry_code: None,
                expiry_time_utc: None,
                schedule: None,
            },
        );
        underlyings.insert(
            "ETH".to_string(),
            UnderlyingConfig {
                vol_source: "deribit".to_string(),
                hl_symbol: None,
                trading_mode: hypercall_types::TradingModes::ORDERBOOK,
                max_expiry_code: None,
                expiry_time_utc: None,
                schedule: None,
            },
        );

        CatalogConfig {
            version: 1,
            expiry: ExpiryConfig {
                expiry_time_utc: "08:00".to_string(),
                schedule: ExpiryScheduleConfig {
                    daily_count: 2,
                    weekly_count: 4,
                    monthly_count: 3,
                    weekdays_only: false,
                },
            },
            underlyings,
            collateral: HashMap::from([
                (
                    "BTC_PERP".to_string(),
                    HyperliquidAssetConfig::Perp(PerpCollateralConfig {
                        asset_id: 0,
                        underlying: "BTC".to_string(),
                    }),
                ),
                (
                    "ETH_PERP".to_string(),
                    HyperliquidAssetConfig::Perp(PerpCollateralConfig {
                        asset_id: 1,
                        underlying: "ETH".to_string(),
                    }),
                ),
                (
                    "USDC".to_string(),
                    HyperliquidAssetConfig::Stablecoin(StablecoinCollateralConfig { token_id: 0 }),
                ),
            ]),
            strike_selection: StrikeSelectionConfig {
                deribit_table_assets: vec![
                    "BTC".to_string(),
                    "ETH".to_string(),
                    "SOL".to_string(),
                    "AVAX".to_string(),
                    "XRP".to_string(),
                    "TRX".to_string(),
                ],
                deribit_region_steps: DeribitRegionStepsConfig {
                    atm: 3,
                    outer: 4,
                    wings: 3,
                },
                occ_fallback_side_count: 8,
            },
            extension_policy: ExtensionPolicyConfig {
                enabled: true,
                ensure_min_strikes_per_side: 3,
                ensure_atm_within_pct: 0.05,
                cooldown_secs: 3600,
                max_total_strikes_per_expiry: 30,
                min_spot_move_pct: 0.05,
            },
            observability: ObservabilityConfig::default(),
            vol_oracles: VolOracleCatalogConfig {
                providers: HashMap::from([(
                    "fixed_test".to_string(),
                    VolOracleProviderConfig::Fixed(FixedVolOracleProviderConfig { iv: 0.5 }),
                )]),
                routes: HashMap::from([
                    ("BTC".to_string(), vec!["fixed_test".to_string()]),
                    ("ETH".to_string(), vec!["fixed_test".to_string()]),
                ]),
            },
        }
    }

    #[test]
    fn test_config_expiry_times() {
        let config = create_test_config();
        let times = config.expiry_times().unwrap();
        assert_eq!(
            times.for_underlying("BTC"),
            hypercall_types::ExpiryTime { hour: 8, minute: 0 }
        );
    }

    #[test]
    fn test_config_underlyings() {
        let config = create_test_config();

        assert!(config.underlyings.contains_key("BTC"));
        assert!(config.underlyings.contains_key("ETH"));

        let btc = config.underlyings.get("BTC").unwrap();
        assert_eq!(btc.vol_source, "deribit");

        let eth = config.underlyings.get("ETH").unwrap();
        assert_eq!(eth.trading_mode, hypercall_types::TradingModes::ORDERBOOK);
    }

    #[test]
    fn test_config_extension_policy() {
        let config = create_test_config();

        assert!(config.extension_policy.enabled);
        assert_eq!(config.extension_policy.ensure_min_strikes_per_side, 3);
        assert!((config.extension_policy.ensure_atm_within_pct - 0.05).abs() < 1e-6);
        assert_eq!(config.extension_policy.cooldown_secs, 3600);
    }

    // =========================================================================
    // Integration with other modules
    // =========================================================================

    #[test]
    fn test_expiry_schedule_generation_integration() {
        let config = create_test_config();
        let times = config.expiry_times().unwrap();

        let schedule = generate_expiry_schedule_at_date(
            chrono::Utc::now().date_naive(),
            config.expiry.schedule.daily_count,
            config.expiry.schedule.weekly_count,
            config.expiry.schedule.monthly_count,
            config.expiry.schedule.weekdays_only,
            times.for_underlying("BTC"),
        )
        .unwrap();

        // Should have daily + weekly + monthly expiries
        let min_expiries = config.expiry.schedule.daily_count + config.expiry.schedule.weekly_count;
        assert!(schedule.expiries.len() >= min_expiries);

        // All expiries should be in the future
        let now = chrono::Utc::now().timestamp();
        for expiry in &schedule.expiries {
            assert!(expiry.timestamp > now);
        }
    }

    #[test]
    fn test_strike_generation_integration() {
        use catalog_manager::generate_strike_set;

        let config = create_test_config();
        // Generate strikes for BTC at $100,000 spot
        let strike_set = generate_strike_set(
            "BTC",
            100000.0,
            chrono::Utc::now().timestamp() + 30 * 86400,
            &config.strike_selection,
        )
        .unwrap();

        // Strikes should be in order
        for window in strike_set.strikes.windows(2) {
            assert!(window[0] < window[1]);
        }

        // ATM strike should exist
        let has_atm = strike_set
            .strikes
            .iter()
            .any(|&s| (s - 100000.0).abs() < 1e-6);
        assert!(has_atm);
    }

    #[test]
    fn test_strike_set_for_eth() {
        use catalog_manager::generate_strike_set;

        let config = create_test_config();
        // Generate strikes for ETH at $3,500 spot
        let strike_set = generate_strike_set(
            "ETH",
            3500.0,
            chrono::Utc::now().timestamp() + 30 * 86400,
            &config.strike_selection,
        )
        .unwrap();

        // Monthly ETH table uses $50/$100/$200 intervals, so every strike
        // remains aligned to $50.
        for &strike in &strike_set.strikes {
            let remainder = strike % 50.0;
            assert!(
                remainder.abs() < 1e-6 || (50.0 - remainder).abs() < 1e-6,
                "Strike {} not aligned to tick 50",
                strike
            );
        }
    }

    // =========================================================================
    // Symbol parsing round-trip tests
    // =========================================================================

    #[test]
    fn test_format_symbol_roundtrip() {
        // Test that we can format and parse back consistently
        let test_cases = [
            ("BTC", 20260131, 95000.0, true),
            ("ETH", 20260228, 3500.0, false),
            ("BTC", 20260315, 105000.5, true),
            ("ETH", 20260401, 3250.0, false),
        ];

        for (underlying, expiry_code, strike, is_call) in test_cases {
            let symbol = format_symbol(underlying, expiry_code, strike, is_call);

            // Verify symbol contains all components
            assert!(symbol.starts_with(underlying));
            assert!(symbol.contains(&expiry_code.to_string()));
            assert!(symbol.ends_with(if is_call { "-C" } else { "-P" }));
        }
    }

    #[test]
    fn test_timestamp_code_consistency() {
        // Test various timestamps produce valid YYYYMMDD codes
        let timestamps = [
            1769846400i64, // 2026-01-31 08:00:00 UTC
            1772265600,    // 2026-02-28 08:00:00 UTC
            1774857600,    // 2026-03-27 08:00:00 UTC
            1777536000,    // 2026-04-24 08:00:00 UTC
        ];

        for ts in timestamps {
            let code = timestamp_to_code(ts).expect("valid timestamp");

            // Verify code is valid YYYYMMDD
            let year = code / 10000;
            let month = (code / 100) % 100;
            let day = code % 100;

            assert!(
                (2025..=2030).contains(&year),
                "Year {} out of expected range",
                year
            );
            assert!((1..=12).contains(&month), "Month {} invalid", month);
            assert!((1..=31).contains(&day), "Day {} invalid", day);
        }
    }

    // =========================================================================
    // Edge case tests
    // =========================================================================

    #[test]
    fn test_format_symbol_very_small_decimal() {
        // Very small decimal that should not affect output
        let symbol = format_symbol("BTC", 20260110, 100000.00000001, true);
        assert_eq!(symbol, "BTC-20260110-100000-C");
    }

    #[test]
    fn test_format_symbol_negative_strike_handled() {
        // This shouldn't happen in practice, but test robustness
        let symbol = format_symbol("BTC", 20260110, -100.0, true);
        assert!(symbol.contains("-100"));
    }

    #[test]
    fn test_timestamp_to_code_handles_edge_timestamps() {
        // Unix epoch start (1970-01-01 00:00:00 UTC)
        let code = timestamp_to_code(0);
        assert_eq!(code, Some(19700101));

        // Timestamp 1 second after epoch
        let code = timestamp_to_code(1);
        assert_eq!(code, Some(19700101));

        // Just before epoch (1969-12-31 23:59:59) - negative timestamps are valid
        let code = timestamp_to_code(-1);
        assert_eq!(code, Some(19691231));

        // Far-future timestamp that overflows - should return None
        // chrono's from_timestamp returns None for values outside valid range
        let code = timestamp_to_code(i64::MAX);
        assert!(code.is_none(), "Should return None for overflow timestamp");
    }
}