rusty_ems/

feeder_integration.rs

//! Integration with rusty-feeder for market data
//!
//! This module provides utilities to integrate the Execution Management System (EMS)
//! with the rusty-feeder crate for efficient market data access.

use std::sync::Arc;

use anyhow::{Result, anyhow};
use async_trait::async_trait;
use log::info;
use parking_lot::RwLock;
use quanta::Clock;
use rust_decimal::Decimal;

use rusty_common::collections::FxHashMap;
use rusty_model::{
    data::{MarketTrade, OrderBook},
    instruments::InstrumentId,
    venues::Venue,
};

/// Type alias for a shared orderbook
type SharedOrderBook = OrderBook;

// Optional dependency on rusty-feeder
#[cfg(feature = "feeder")]
mod with_feeder {
    use super::{
        Arc, Clock, FxHashMap, InstrumentId, MarketTrade, Result, RwLock, SharedOrderBook, info,
    };

    use anyhow::Context;
    use log::{debug, warn};
    use rusty_common::SmartString;
    use rusty_feeder::Provider;
    use smallvec::SmallVec;

    /// A market data provider that uses rusty-feeder for connectivity
    pub struct FeederMarketDataProvider<P: Provider> {
        /// The underlying provider from rusty-feeder
        provider: P,

        /// Shared clock instance
        clock: Clock,

        /// Map of instrument IDs to orderbooks for quick access
        orderbooks: Arc<RwLock<FxHashMap<InstrumentId, SharedOrderBook>>>,

        /// Recent trades for each instrument
        trades: Arc<RwLock<FxHashMap<InstrumentId, Vec<MarketTrade>>>>,

        /// Maximum trade history to keep per instrument
        max_trade_history: usize,
    }

    impl<P: Provider> FeederMarketDataProvider<P> {
        /// Create a new market data provider using the given rusty-feeder provider
        #[must_use]
        pub fn new(provider: P) -> Self {
            Self {
                provider,
                clock: Clock::new(),
                orderbooks: Arc::new(RwLock::new(FxHashMap::default())),
                trades: Arc::new(RwLock::new(FxHashMap::default())),
                max_trade_history: 1000, // Store up to 1000 recent trades per instrument
            }
        }

        /// Start receiving market data for the specified instruments
        pub async fn start(&self, instruments: &[InstrumentId]) -> Result<()> {
            // Only subscribe to the venue of the current provider
            let venue = self.provider.venue();

            // Collect symbols for this provider's venue
            let symbols: SmallVec<[SmartString; 8]> = instruments
                .iter()
                .filter(|instr| instr.venue == venue)
                .map(|instr| instr.symbol.clone())
                .collect();

            if !symbols.is_empty() {
                // Subscribe to orderbook data
                let mut orderbook_rx = self
                    .provider
                    .subscribe_orderbook(symbols.clone())
                    .await
                    .context("Failed to subscribe to orderbook data")?;

                // Clone shared data for the task
                let _orderbooks = self.orderbooks.clone();
                let venue_clone = venue;

                // Make a copy of symbols to move into the task
                let symbols_copy = symbols.to_vec();

                // Start processing orderbook updates
                tokio::spawn(async move {
                    info!(
                        "Starting orderbook processing for {} symbols on {}",
                        symbols_copy.len(),
                        venue_clone.to_str()
                    );

                    while let Some(_depth_msg) = orderbook_rx.recv().await {
                        // Process the depth message (specifics depend on the provider's message type)
                        // In a real implementation, we'd extract the symbol, bids, and asks from the message
                        // and update the relevant orderbook
                        debug!(
                            "Received orderbook update for venue: {}",
                            venue_clone.to_str()
                        );
                    }

                    warn!(
                        "Orderbook processing stopped for venue: {}",
                        venue_clone.to_str()
                    );
                });

                // Subscribe to trade data
                let mut trade_rx = self
                    .provider
                    .subscribe_trades(symbols.clone())
                    .await
                    .context("Failed to subscribe to trade data")?;

                // Clone shared data for the task
                let _trades = self.trades.clone();
                let venue_clone2 = venue;
                let _max_history = self.max_trade_history;

                // Make another copy of symbols for the trade task
                let symbols_copy2 = symbols.to_vec();

                // Start processing trade updates
                tokio::spawn(async move {
                    info!(
                        "Starting trade processing for {} symbols on {}",
                        symbols_copy2.len(),
                        venue_clone2.to_str()
                    );

                    while let Some(_trade_msg) = trade_rx.recv().await {
                        // Process the trade message (specifics depend on the provider's message type)
                        // In a real implementation, we'd extract the symbol and trade details
                        // and update the trade history
                        debug!("Received trade update for venue: {}", venue_clone2.to_str());
                    }

                    warn!(
                        "Trade processing stopped for venue: {}",
                        venue_clone2.to_str()
                    );
                });
            }

            Ok(())
        }

        /// Get the latest orderbook for an instrument
        pub fn get_orderbook(&self, instrument: &InstrumentId) -> Option<SharedOrderBook> {
            self.orderbooks.read().get(instrument).cloned()
        }

        /// Get the recent trades for an instrument
        pub fn get_recent_trades(
            &self,
            instrument: &InstrumentId,
            limit: Option<usize>,
        ) -> Vec<MarketTrade> {
            let trades_map = self.trades.read();
            let limit = limit.unwrap_or(10).min(self.max_trade_history);

            if let Some(instrument_trades) = trades_map.get(instrument) {
                let start_idx = instrument_trades.len().saturating_sub(limit);
                instrument_trades[start_idx..].to_vec()
            } else {
                Vec::new()
            }
        }
    }
}

/// Interface for market data providers
#[async_trait]
pub trait MarketDataProvider: Send + Sync + 'static {
    /// Initialize the provider
    async fn init(&mut self) -> Result<()>;

    /// Subscribe to market data for the specified instruments
    async fn subscribe(&self, instruments: &[InstrumentId]) -> Result<()>;

    /// Get the best bid and ask for an instrument
    async fn get_best_bid_ask(&self, instrument: &InstrumentId) -> Result<(Decimal, Decimal)>;

    /// Get the latest trade price for an instrument
    async fn get_latest_trade_price(&self, instrument: &InstrumentId) -> Result<Decimal>;

    /// Get the entire orderbook for an instrument
    async fn get_orderbook(&self, instrument: &InstrumentId) -> Result<OrderBook>;

    /// Get recent trades for an instrument
    async fn get_recent_trades(
        &self,
        instrument: &InstrumentId,
        limit: Option<usize>,
    ) -> Result<Vec<MarketTrade>>;
}

#[cfg(feature = "feeder")]
pub use with_feeder::FeederMarketDataProvider;

/// A standalone market data provider that uses the EMS's own connectivity
/// This is used when the feeder feature is not enabled
pub struct StandaloneMarketDataProvider {
    /// Venue this provider connects to
    venue: Venue,

    /// REST API base URL
    api_base_url: String,

    /// WebSocket base URL
    ws_base_url: String,

    /// API key for authentication
    api_key: Option<String>,

    /// API secret for authentication
    api_secret: Option<String>,

    /// Shared clock instance
    clock: Clock,

    /// Map of instrument IDs to orderbooks for quick access
    orderbooks: Arc<RwLock<FxHashMap<InstrumentId, OrderBook>>>,

    /// Recent trades for each instrument
    trades: Arc<RwLock<FxHashMap<InstrumentId, Vec<MarketTrade>>>>,
}

impl StandaloneMarketDataProvider {
    /// Create a new standalone market data provider
    #[must_use]
    pub fn new(
        venue: Venue,
        api_base_url: String,
        ws_base_url: String,
        api_key: Option<String>,
        api_secret: Option<String>,
    ) -> Self {
        Self {
            venue,
            api_base_url,
            ws_base_url,
            api_key,
            api_secret,
            clock: Clock::new(),
            orderbooks: Arc::new(RwLock::new(FxHashMap::default())),
            trades: Arc::new(RwLock::new(FxHashMap::default())),
        }
    }
}

#[async_trait]
impl MarketDataProvider for StandaloneMarketDataProvider {
    async fn init(&mut self) -> Result<()> {
        // In a real implementation, we'd initialize API clients here
        Ok(())
    }

    async fn subscribe(&self, instruments: &[InstrumentId]) -> Result<()> {
        // Filter instruments by venue
        let instruments: Vec<&InstrumentId> = instruments
            .iter()
            .filter(|i| i.venue == self.venue)
            .collect();

        if instruments.is_empty() {
            return Ok(());
        }

        // In a real implementation, we'd establish WebSocket connections
        // and subscribe to market data
        info!(
            "Subscribing to {} instruments on {}",
            instruments.len(),
            self.venue.to_str()
        );

        Ok(())
    }

    async fn get_best_bid_ask(&self, instrument: &InstrumentId) -> Result<(Decimal, Decimal)> {
        if instrument.venue != self.venue {
            return Err(anyhow!("Instrument venue mismatch"));
        }

        // In a real implementation, we'd return data from our cached orderbook
        // For now, return dummy values
        Ok((
            Decimal::new(10000, 2), // 100.00
            Decimal::new(10001, 2), // 100.01
        ))
    }

    async fn get_latest_trade_price(&self, instrument: &InstrumentId) -> Result<Decimal> {
        if instrument.venue != self.venue {
            return Err(anyhow!("Instrument venue mismatch"));
        }

        // In a real implementation, we'd return data from our cached trades
        // For now, return a dummy value
        Ok(Decimal::new(10000, 2)) // 100.00
    }

    async fn get_orderbook(&self, instrument: &InstrumentId) -> Result<OrderBook> {
        if instrument.venue != self.venue {
            return Err(anyhow!("Instrument venue mismatch"));
        }

        // Check if we have a cached orderbook
        if let Some(orderbook) = self.orderbooks.read().get(instrument) {
            return Ok(orderbook.clone());
        }

        // In a real implementation, we'd fetch the orderbook via REST API if not cached
        Err(anyhow!(
            "Orderbook not available for instrument: {}",
            instrument.symbol
        ))
    }

    async fn get_recent_trades(
        &self,
        instrument: &InstrumentId,
        limit: Option<usize>,
    ) -> Result<Vec<MarketTrade>> {
        if instrument.venue != self.venue {
            return Err(anyhow!("Instrument venue mismatch"));
        }

        let limit = limit.unwrap_or(10);

        // Check if we have cached trades
        if let Some(trades) = self.trades.read().get(instrument) {
            let start_idx = trades.len().saturating_sub(limit);
            return Ok(trades[start_idx..].to_vec());
        }

        // In a real implementation, we'd fetch trades via REST API if not cached
        Err(anyhow!(
            "Recent trades not available for instrument: {}",
            instrument.symbol
        ))
    }
}