rusty_ems/

instrument_registry.rs

//! Unified Instrument Registry for centralized instrument/order mapping
//!
//! This module provides a centralized system for tracking relationships between
//! order IDs, instrument IDs, and exchange-specific symbol formats across all venues.
//! It eliminates 'UNKNOWN' placeholders by maintaining bidirectional mappings.

use parking_lot::RwLock;
use rust_decimal::Decimal;
use rusty_common::{SmartString, collections::FxHashMap, time::get_epoch_timestamp_ns};
use rusty_model::{instruments::InstrumentId, venues::Venue};
use std::fmt::Write;
use std::sync::Arc;
use std::time::Duration;

/// Cache statistics for monitoring and debugging
#[derive(Debug, Clone, Default)]
pub struct CacheStats {
    /// Total number of cached order mappings
    pub order_mappings: usize,
    /// Total number of cached symbol normalizations
    pub symbol_mappings: usize,
    /// Number of cache hits
    pub hits: u64,
    /// Number of cache misses
    pub misses: u64,
    /// Cache hit ratio (0.0 - 1.0)
    pub hit_ratio: f64,
}

impl CacheStats {
    /// Update hit ratio after new stats
    pub fn update_hit_ratio(&mut self) {
        let total = self.hits + self.misses;
        self.hit_ratio = if total > 0 {
            self.hits as f64 / total as f64
        } else {
            0.0
        };
    }
}

/// Cached order mapping with TTL expiration
#[derive(Debug, Clone)]
struct OrderMapping {
    /// Associated instrument
    instrument_id: InstrumentId,
    /// Cache entry timestamp for TTL
    timestamp_ns: u64,
    /// Additional metadata
    metadata: Option<OrderMetadata>,
}

/// Additional order metadata for enhanced tracking
#[derive(Debug, Clone)]
pub struct OrderMetadata {
    /// Original order size
    pub size: Option<Decimal>,
    /// Order type (limit, market, etc.)
    pub order_type: Option<SmartString>,
    /// Order side (buy/sell)
    pub side: Option<SmartString>,
}

/// Cached symbol normalization
#[derive(Debug, Clone)]
struct SymbolMapping {
    /// Normalized symbol format
    normalized: SmartString,
    /// Cache timestamp
    timestamp_ns: u64,
}

/// Trait for unified instrument registry operations
pub trait InstrumentRegistry: Send + Sync {
    /// Cache bidirectional order ID to instrument mapping
    fn cache_order_mapping(
        &self,
        order_id: &str,
        instrument_id: InstrumentId,
        metadata: Option<OrderMetadata>,
    );

    /// Lookup instrument ID by order ID
    fn lookup_by_order_id(&self, order_id: &str) -> Option<InstrumentId>;

    /// Normalize exchange-specific symbol to standard format
    fn normalize_symbol(&self, exchange_symbol: &str, venue: Venue) -> SmartString;

    /// Remove order mapping (e.g., when order is fully filled)
    fn remove_mapping(&self, order_id: &str);

    /// Cleanup expired entries
    fn cleanup_expired(&self);

    /// Get cache statistics
    fn get_cache_stats(&self) -> CacheStats;

    /// Clear all mappings (for testing)
    fn clear_all(&self);
}

/// Default implementation of instrument registry
pub struct DefaultInstrumentRegistry {
    /// Order ID to instrument mapping
    order_mappings: Arc<RwLock<FxHashMap<SmartString, OrderMapping>>>,

    /// Exchange symbol to normalized symbol mapping
    symbol_mappings: Arc<RwLock<FxHashMap<(Venue, SmartString), SymbolMapping>>>,

    /// Cache statistics
    stats: Arc<RwLock<CacheStats>>,

    /// TTL for order mappings (default: 24 hours)
    order_ttl_ns: u64,

    /// TTL for symbol mappings (default: 1 hour)
    symbol_ttl_ns: u64,
}

impl Default for DefaultInstrumentRegistry {
    fn default() -> Self {
        Self::new(
            Duration::from_secs(24 * 60 * 60), // Order mapping TTL: 24 hours
            Duration::from_secs(60 * 60),      // Symbol mapping TTL: 1 hour
        )
    }
}

impl DefaultInstrumentRegistry {
    /// Create new instrument registry with custom TTL settings
    #[must_use]
    pub fn new(order_ttl: Duration, symbol_ttl: Duration) -> Self {
        Self {
            order_mappings: Arc::new(RwLock::new(FxHashMap::default())),
            symbol_mappings: Arc::new(RwLock::new(FxHashMap::default())),
            stats: Arc::new(RwLock::new(CacheStats::default())),
            order_ttl_ns: order_ttl.as_nanos() as u64,
            symbol_ttl_ns: symbol_ttl.as_nanos() as u64,
        }
    }

    /// Check if timestamp is expired
    fn is_expired(&self, timestamp_ns: u64, ttl_ns: u64) -> bool {
        let now = get_epoch_timestamp_ns();
        now.saturating_sub(timestamp_ns) > ttl_ns
    }

    /// Record cache hit
    fn record_hit(&self) {
        let mut stats = self.stats.write();
        stats.hits += 1;
        stats.update_hit_ratio();
    }

    /// Record cache miss
    fn record_miss(&self) {
        let mut stats = self.stats.write();
        stats.misses += 1;
        stats.update_hit_ratio();
    }
}

impl InstrumentRegistry for DefaultInstrumentRegistry {
    fn cache_order_mapping(
        &self,
        order_id: &str,
        instrument_id: InstrumentId,
        metadata: Option<OrderMetadata>,
    ) {
        let mapping = OrderMapping {
            instrument_id,
            timestamp_ns: get_epoch_timestamp_ns(),
            metadata,
        };

        let mut mappings = self.order_mappings.write();
        mappings.insert(SmartString::from(order_id), mapping);

        // Update stats
        let mut stats = self.stats.write();
        stats.order_mappings = mappings.len();
    }

    fn lookup_by_order_id(&self, order_id: &str) -> Option<InstrumentId> {
        let mappings = self.order_mappings.read();

        if let Some(mapping) = mappings.get(order_id) {
            // Check if expired
            if self.is_expired(mapping.timestamp_ns, self.order_ttl_ns) {
                drop(mappings);
                self.remove_mapping(order_id);
                self.record_miss();
                return None;
            }

            self.record_hit();
            Some(mapping.instrument_id.clone())
        } else {
            self.record_miss();
            None
        }
    }

    fn normalize_symbol(&self, exchange_symbol: &str, venue: Venue) -> SmartString {
        let key = (venue, SmartString::from(exchange_symbol));

        // Check cache first
        {
            let mappings = self.symbol_mappings.read();
            if let Some(mapping) = mappings.get(&key)
                && !self.is_expired(mapping.timestamp_ns, self.symbol_ttl_ns)
            {
                self.record_hit();
                return mapping.normalized.clone();
            }
        }

        // Cache miss - compute normalization
        self.record_miss();
        let normalized = Self::compute_normalization(exchange_symbol, venue);

        // Cache the result
        let mapping = SymbolMapping {
            normalized: normalized.clone(),
            timestamp_ns: get_epoch_timestamp_ns(),
        };

        let mut mappings = self.symbol_mappings.write();
        mappings.insert(key, mapping);

        // Update stats
        let mut stats = self.stats.write();
        stats.symbol_mappings = mappings.len();

        normalized
    }

    fn remove_mapping(&self, order_id: &str) {
        let mut mappings = self.order_mappings.write();
        mappings.remove(order_id);

        // Update stats
        let mut stats = self.stats.write();
        stats.order_mappings = mappings.len();
    }

    fn cleanup_expired(&self) {
        let now = get_epoch_timestamp_ns();

        // Clean expired order mappings
        {
            let mut mappings = self.order_mappings.write();
            mappings.retain(|_, mapping| !self.is_expired(mapping.timestamp_ns, self.order_ttl_ns));
        }

        // Clean expired symbol mappings
        {
            let mut mappings = self.symbol_mappings.write();
            mappings
                .retain(|_, mapping| !self.is_expired(mapping.timestamp_ns, self.symbol_ttl_ns));
        }

        // Update stats
        let order_count = self.order_mappings.read().len();
        let symbol_count = self.symbol_mappings.read().len();

        let mut stats = self.stats.write();
        stats.order_mappings = order_count;
        stats.symbol_mappings = symbol_count;
    }

    fn get_cache_stats(&self) -> CacheStats {
        self.stats.read().clone()
    }

    fn clear_all(&self) {
        self.order_mappings.write().clear();
        self.symbol_mappings.write().clear();

        let mut stats = self.stats.write();
        stats.order_mappings = 0;
        stats.symbol_mappings = 0;
    }
}

impl DefaultInstrumentRegistry {
    /// Compute symbol normalization based on exchange format
    fn compute_normalization(exchange_symbol: &str, venue: Venue) -> SmartString {
        match venue {
            Venue::Binance => {
                // Binance: BTCUSDT -> BTC/USDT
                Self::normalize_binance_symbol(exchange_symbol)
            }
            Venue::Coinbase => {
                // Coinbase: BTC-USD -> BTC/USD
                SmartString::from(exchange_symbol.replace('-', "/"))
            }
            Venue::Bybit => {
                // Bybit: BTCUSDT -> BTC/USDT (same as Binance)
                Self::normalize_binance_symbol(exchange_symbol)
            }
            Venue::Upbit => {
                // Upbit: KRW-BTC -> BTC/KRW
                if let Some((quote, base)) = exchange_symbol.split_once('-') {
                    let mut s = SmartString::new();
                    write!(s, "{base}/{quote}").expect("Failed to format Upbit symbol");
                    s
                } else {
                    SmartString::from(exchange_symbol)
                }
            }
            Venue::Bithumb => {
                // Bithumb: BTC_KRW -> BTC/KRW
                SmartString::from(exchange_symbol.replace('_', "/"))
            }
            _ => {
                // Default: return as-is
                SmartString::from(exchange_symbol)
            }
        }
    }

    /// Normalize Binance/Bybit style symbols
    fn normalize_binance_symbol(symbol: &str) -> SmartString {
        // Common quote currencies (order by length desc for proper matching)
        const QUOTE_CURRENCIES: &[&str] = &[
            "USDT", "BUSD", "USDC", "TUSD", "DAI", "EUR", "GBP", "BRL", "TRY", "RUB", "BTC", "ETH",
            "BNB", "XRP", "ADA", "DOT", "SOL", "DOGE", "AVAX", "MATIC", "USD", "NGN", "UAH", "VAI",
            "BIDR", "BVND", "GYEN", "IDRT",
        ];

        for quote in QUOTE_CURRENCIES {
            if let Some(base) = symbol.strip_suffix(quote)
                && !base.is_empty()
            {
                let mut s = SmartString::new();
                write!(s, "{base}/{quote}").expect("Failed to format Binance symbol");
                return s;
            }
        }

        // Fallback: return as-is if no quote currency matched
        SmartString::from(symbol)
    }
}

/// Create a shared instrument registry instance
#[must_use]
pub fn create_shared_registry() -> Arc<dyn InstrumentRegistry> {
    Arc::new(DefaultInstrumentRegistry::default())
}

/// Create a shared instrument registry with custom TTL
#[must_use]
pub fn create_shared_registry_with_ttl(
    order_ttl: Duration,
    symbol_ttl: Duration,
) -> Arc<dyn InstrumentRegistry> {
    Arc::new(DefaultInstrumentRegistry::new(order_ttl, symbol_ttl))
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusty_model::venues::Venue;
    use std::time::Duration;

    #[test]
    fn test_symbol_normalization() {
        let registry = DefaultInstrumentRegistry::default();

        // Test Binance format
        assert_eq!(
            registry.normalize_symbol("BTCUSDT", Venue::Binance),
            "BTC/USDT"
        );
        assert_eq!(
            registry.normalize_symbol("ETHBTC", Venue::Binance),
            "ETH/BTC"
        );

        // Test Coinbase format
        assert_eq!(
            registry.normalize_symbol("BTC-USD", Venue::Coinbase),
            "BTC/USD"
        );
        assert_eq!(
            registry.normalize_symbol("ETH-EUR", Venue::Coinbase),
            "ETH/EUR"
        );

        // Test Bybit format (same as Binance)
        assert_eq!(
            registry.normalize_symbol("BTCUSDT", Venue::Bybit),
            "BTC/USDT"
        );

        // Test Upbit format
        assert_eq!(
            registry.normalize_symbol("KRW-BTC", Venue::Upbit),
            "BTC/KRW"
        );

        // Test Bithumb format
        assert_eq!(
            registry.normalize_symbol("BTC_KRW", Venue::Bithumb),
            "BTC/KRW"
        );
    }

    #[test]
    fn test_order_mapping_lifecycle() {
        let registry = DefaultInstrumentRegistry::default();
        let instrument_id = InstrumentId::new("BTC/USDT", Venue::Binance);

        // Test cache miss
        assert!(registry.lookup_by_order_id("order123").is_none());

        // Cache mapping
        registry.cache_order_mapping("order123", instrument_id.clone(), None);

        // Test cache hit
        assert_eq!(registry.lookup_by_order_id("order123"), Some(instrument_id));

        // Remove mapping
        registry.remove_mapping("order123");

        // Test cache miss after removal
        assert!(registry.lookup_by_order_id("order123").is_none());
    }

    #[test]
    fn test_cache_stats() {
        let registry = DefaultInstrumentRegistry::default();

        // Initial stats
        let stats = registry.get_cache_stats();
        assert_eq!(stats.hits, 0);
        assert_eq!(stats.misses, 0);
        assert_eq!(stats.hit_ratio, 0.0);

        // Generate some misses
        registry.lookup_by_order_id("nonexistent1");
        registry.lookup_by_order_id("nonexistent2");

        // Cache something
        let instrument_id = InstrumentId::new("BTC/USDT", Venue::Binance);
        registry.cache_order_mapping("order1", instrument_id, None);

        // Generate a hit
        registry.lookup_by_order_id("order1");

        let stats = registry.get_cache_stats();
        assert_eq!(stats.hits, 1);
        assert_eq!(stats.misses, 2);
        assert!((stats.hit_ratio - 0.333).abs() < 0.01); // ~33.3%
    }

    #[test]
    fn test_binance_symbol_normalization() {
        // Test common pairs
        assert_eq!(
            DefaultInstrumentRegistry::normalize_binance_symbol("BTCUSDT"),
            "BTC/USDT"
        );
        assert_eq!(
            DefaultInstrumentRegistry::normalize_binance_symbol("ETHBTC"),
            "ETH/BTC"
        );
        assert_eq!(
            DefaultInstrumentRegistry::normalize_binance_symbol("ADAUSDT"),
            "ADA/USDT"
        );

        // Test edge cases
        assert_eq!(
            DefaultInstrumentRegistry::normalize_binance_symbol("USDTBUSD"),
            "USDT/BUSD"
        );
        assert_eq!(
            DefaultInstrumentRegistry::normalize_binance_symbol("BTCEUR"),
            "BTC/EUR"
        );

        // Test unknown format (fallback)
        assert_eq!(
            DefaultInstrumentRegistry::normalize_binance_symbol("UNKNOWN"),
            "UNKNOWN"
        );
    }

    #[test]
    fn test_metadata_storage() {
        let registry = DefaultInstrumentRegistry::default();
        let instrument_id = InstrumentId::new("BTC/USDT", Venue::Binance);

        let metadata = OrderMetadata {
            size: Some(Decimal::new(1, 0)),
            order_type: Some("LIMIT".into()),
            side: Some("BUY".into()),
        };

        registry.cache_order_mapping("order123", instrument_id.clone(), Some(metadata));

        // Verify mapping exists
        assert_eq!(registry.lookup_by_order_id("order123"), Some(instrument_id));
    }

    #[test]
    fn test_ttl_expiration() {
        let registry = DefaultInstrumentRegistry::new(
            Duration::from_millis(1), // Very short TTL for testing
            Duration::from_millis(1),
        );

        let instrument_id = InstrumentId::new("BTC/USDT", Venue::Binance);
        registry.cache_order_mapping("order123", instrument_id, None);

        // Should exist initially
        assert!(registry.lookup_by_order_id("order123").is_some());

        // Wait for expiration
        std::thread::sleep(Duration::from_millis(2));

        // Should be expired and removed
        assert!(registry.lookup_by_order_id("order123").is_none());
    }
}