rusty_ems/

position_manager.rs

//! Position Manager for tracking and managing futures positions
//!
//! This module provides a centralized position management system that:
//! - Tracks all open positions across exchanges
//! - Updates positions from exchange reports
//! - Calculates aggregate risk metrics
//! - Provides position queries by symbol/venue

use anyhow::Result;
use async_trait::async_trait;
use flume::Sender;
use parking_lot::RwLock;
use rust_decimal::Decimal;
use rusty_common::{SmartString, collections::FxHashMap};
use rusty_model::{
    position::{FuturesPosition, PositionUpdate},
    types::PositionId,
    venues::Venue,
};
use std::sync::Arc;

/// Event emitted when position state changes
#[derive(Debug, Clone)]
pub enum PositionEvent {
    /// New position opened
    PositionOpened {
        /// Unique identifier for the position
        position_id: PositionId,
        /// Exchange venue where the position was opened
        venue: Venue,
        /// Trading symbol for the position
        symbol: SmartString,
    },
    /// Position updated
    PositionUpdated {
        /// Unique identifier for the position
        position_id: PositionId,
        /// Exchange venue where the position is held
        venue: Venue,
        /// Trading symbol for the position
        symbol: SmartString,
        /// Previous position amount before update
        old_amount: Decimal,
        /// New position amount after update
        new_amount: Decimal,
    },
    /// Position closed
    PositionClosed {
        /// Unique identifier for the position
        position_id: PositionId,
        /// Exchange venue where the position was held
        venue: Venue,
        /// Trading symbol for the position
        symbol: SmartString,
        /// Realized profit/loss from closing the position
        realized_pnl: Decimal,
    },
}

/// Trait for position management
#[async_trait]
pub trait PositionManager: Send + Sync {
    /// Update or create a position from exchange update
    async fn update_position(&self, update: PositionUpdate) -> Result<()>;

    /// Get position by ID
    async fn get_position(&self, position_id: &PositionId) -> Option<FuturesPosition>;

    /// Get all positions for a symbol on a venue
    async fn get_positions_by_symbol(&self, venue: Venue, symbol: &str) -> Vec<FuturesPosition>;

    /// Get all open positions
    async fn get_all_positions(&self) -> Vec<FuturesPosition>;

    /// Get total unrealized `PnL` across all positions
    async fn get_total_unrealized_pnl(&self) -> Decimal;

    /// Get total realized `PnL` across all positions
    async fn get_total_realized_pnl(&self) -> Decimal;

    /// Clear all positions (useful for testing)
    async fn clear_all_positions(&self);
}

/// Type alias for the symbol index map to avoid complex type warnings
type SymbolIndex = Arc<RwLock<FxHashMap<(Venue, SmartString), Vec<PositionId>>>>;

/// Default implementation of position manager
pub struct DefaultPositionManager {
    /// Position storage: `position_id` -> position
    positions: Arc<RwLock<FxHashMap<PositionId, FuturesPosition>>>,

    /// Index by venue and symbol for fast lookups
    symbol_index: SymbolIndex,

    /// Event channel for position updates
    event_sender: Option<Sender<PositionEvent>>,
}

impl DefaultPositionManager {
    /// Create a new position manager
    #[must_use]
    pub fn new(event_sender: Option<Sender<PositionEvent>>) -> Self {
        Self {
            positions: Arc::new(RwLock::new(FxHashMap::default())),
            symbol_index: Arc::new(RwLock::new(FxHashMap::default())),
            event_sender,
        }
    }

    /// Send position event if channel is configured
    fn send_event(&self, event: PositionEvent) {
        if let Some(sender) = &self.event_sender
            && let Err(e) = sender.send(event)
        {
            log::warn!("Failed to send position event: {e}");
        }
    }

    /// Update symbol index when position changes
    fn update_symbol_index(
        &self,
        venue: Venue,
        symbol: &str,
        position_id: PositionId,
        is_closed: bool,
    ) {
        let mut index = self.symbol_index.write();
        let key = (venue, SmartString::from(symbol));

        if is_closed {
            // Remove from index if position is closed
            if let Some(positions) = index.get_mut(&key) {
                positions.retain(|&id| id != position_id);
                if positions.is_empty() {
                    index.remove(&key);
                }
            }
        } else {
            // Add to index if not already present
            let positions = index.entry(key).or_default();
            if !positions.contains(&position_id) {
                positions.push(position_id);
            }
        }
    }
}

#[async_trait]
impl PositionManager for DefaultPositionManager {
    async fn update_position(&self, update: PositionUpdate) -> Result<()> {
        let mut positions = self.positions.write();

        // Check if position already exists
        if let Some(existing_position) = positions.get_mut(&update.position_id) {
            // Update existing position
            let old_amount = existing_position.amount;

            existing_position.update(
                update.amount,
                update.entry_price,
                update.breakeven_price,
                update.unrealized_pnl,
                update.realized_pnl,
                update.isolated_wallet,
            );

            // Check if position was closed
            if existing_position.is_closed() {
                self.update_symbol_index(update.venue, &update.symbol, update.position_id, true);
                self.send_event(PositionEvent::PositionClosed {
                    position_id: update.position_id,
                    venue: update.venue,
                    symbol: update.symbol,
                    realized_pnl: update.realized_pnl,
                });
            } else {
                self.send_event(PositionEvent::PositionUpdated {
                    position_id: update.position_id,
                    venue: update.venue,
                    symbol: update.symbol.clone(),
                    old_amount,
                    new_amount: update.amount,
                });
            }
        } else if !update.amount.is_zero() {
            // Create new position only if amount is non-zero
            let position = FuturesPosition::new(
                update.venue,
                &update.symbol,
                update.side,
                update.amount,
                update.entry_price,
                update.margin_type,
            );

            // Update with the additional fields from the update
            let mut position = position;
            position.id = update.position_id; // Use the stable ID from the update
            position.update(
                update.amount,
                update.entry_price,
                update.breakeven_price,
                update.unrealized_pnl,
                update.realized_pnl,
                update.isolated_wallet,
            );

            positions.insert(update.position_id, position);
            self.update_symbol_index(update.venue, &update.symbol, update.position_id, false);

            self.send_event(PositionEvent::PositionOpened {
                position_id: update.position_id,
                venue: update.venue,
                symbol: update.symbol,
            });
        }

        Ok(())
    }

    async fn get_position(&self, position_id: &PositionId) -> Option<FuturesPosition> {
        self.positions.read().get(position_id).cloned()
    }

    async fn get_positions_by_symbol(&self, venue: Venue, symbol: &str) -> Vec<FuturesPosition> {
        let positions = self.positions.read();
        let index = self.symbol_index.read();

        let key = (venue, SmartString::from(symbol));
        if let Some(position_ids) = index.get(&key) {
            position_ids
                .iter()
                .filter_map(|id| positions.get(id).cloned())
                .collect()
        } else {
            Vec::new()
        }
    }

    async fn get_all_positions(&self) -> Vec<FuturesPosition> {
        self.positions
            .read()
            .values()
            .filter(|p| !p.is_closed())
            .cloned()
            .collect()
    }

    async fn get_total_unrealized_pnl(&self) -> Decimal {
        self.positions
            .read()
            .values()
            .filter(|p| !p.is_closed())
            .map(|p| p.unrealized_pnl)
            .sum()
    }

    async fn get_total_realized_pnl(&self) -> Decimal {
        self.positions.read().values().map(|p| p.realized_pnl).sum()
    }

    async fn clear_all_positions(&self) {
        self.positions.write().clear();
        self.symbol_index.write().clear();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusty_model::position::{MarginType, PositionSide};

    #[tokio::test]
    async fn test_position_creation() -> Result<()> {
        let manager = DefaultPositionManager::new(None);

        let update = PositionUpdate {
            position_id: PositionId::new(),
            venue: Venue::Binance,
            symbol: "BTCUSDT".into(),
            side: PositionSide::Long,
            amount: Decimal::from(1),
            entry_price: Decimal::from(50000),
            breakeven_price: Decimal::from(50000),
            unrealized_pnl: Decimal::ZERO,
            realized_pnl: Decimal::ZERO,
            margin_type: MarginType::Cross,
            isolated_wallet: Decimal::ZERO,
            timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
        };

        manager.update_position(update.clone()).await?;

        let position = manager.get_position(&update.position_id).await;
        assert!(position.is_some());

        let position = position.unwrap();
        assert_eq!(position.symbol.as_str(), "BTCUSDT");
        assert_eq!(position.amount, Decimal::from(1));

        Ok(())
    }

    #[tokio::test]
    async fn test_position_update() -> Result<()> {
        let manager = DefaultPositionManager::new(None);

        let position_id = PositionId::new();

        // Initial position
        let update1 = PositionUpdate {
            position_id,
            venue: Venue::Binance,
            symbol: "BTCUSDT".into(),
            side: PositionSide::Long,
            amount: Decimal::from(1),
            entry_price: Decimal::from(50000),
            breakeven_price: Decimal::from(50000),
            unrealized_pnl: Decimal::ZERO,
            realized_pnl: Decimal::ZERO,
            margin_type: MarginType::Cross,
            isolated_wallet: Decimal::ZERO,
            timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
        };

        manager.update_position(update1).await?;

        // Update position
        let update2 = PositionUpdate {
            position_id,
            venue: Venue::Binance,
            symbol: "BTCUSDT".into(),
            side: PositionSide::Long,
            amount: Decimal::from(2),
            entry_price: Decimal::from(51000),
            breakeven_price: Decimal::from(51000),
            unrealized_pnl: Decimal::from(1000),
            realized_pnl: Decimal::ZERO,
            margin_type: MarginType::Cross,
            isolated_wallet: Decimal::ZERO,
            timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
        };

        manager.update_position(update2).await?;

        let position = manager.get_position(&position_id).await;
        assert!(position.is_some());

        let position = position.unwrap();
        assert_eq!(position.amount, Decimal::from(2));
        assert_eq!(position.entry_price, Decimal::from(51000));
        assert_eq!(position.unrealized_pnl, Decimal::from(1000));

        Ok(())
    }

    #[tokio::test]
    async fn test_position_closure() -> Result<()> {
        let (event_tx, event_rx) = flume::unbounded();
        let manager = DefaultPositionManager::new(Some(event_tx));

        let position_id = PositionId::new();

        // Open position
        let update1 = PositionUpdate {
            position_id,
            venue: Venue::Binance,
            symbol: "BTCUSDT".into(),
            side: PositionSide::Long,
            amount: Decimal::from(1),
            entry_price: Decimal::from(50000),
            breakeven_price: Decimal::from(50000),
            unrealized_pnl: Decimal::ZERO,
            realized_pnl: Decimal::ZERO,
            margin_type: MarginType::Cross,
            isolated_wallet: Decimal::ZERO,
            timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
        };

        manager.update_position(update1).await?;

        // Verify position opened event
        let event = event_rx.recv_async().await?;
        assert!(matches!(event, PositionEvent::PositionOpened { .. }));

        // Close position
        let update2 = PositionUpdate {
            position_id,
            venue: Venue::Binance,
            symbol: "BTCUSDT".into(),
            side: PositionSide::Long,
            amount: Decimal::ZERO,
            entry_price: Decimal::from(50000),
            breakeven_price: Decimal::from(50000),
            unrealized_pnl: Decimal::ZERO,
            realized_pnl: Decimal::from(1000),
            margin_type: MarginType::Cross,
            isolated_wallet: Decimal::ZERO,
            timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
        };

        manager.update_position(update2).await?;

        // Verify position closed event
        let event = event_rx.recv_async().await?;
        match event {
            PositionEvent::PositionClosed { realized_pnl, .. } => {
                assert_eq!(realized_pnl, Decimal::from(1000));
            }
            _ => panic!("Expected PositionClosed event"),
        }

        // Verify position is closed
        let all_positions = manager.get_all_positions().await;
        assert!(all_positions.is_empty());

        Ok(())
    }

    #[tokio::test]
    async fn test_get_positions_by_symbol() -> Result<()> {
        let manager = DefaultPositionManager::new(None);

        // Create multiple positions
        for i in 0..3 {
            let update = PositionUpdate {
                position_id: PositionId::new(),
                venue: Venue::Binance,
                symbol: "BTCUSDT".into(),
                side: PositionSide::Long,
                amount: Decimal::from(i + 1),
                entry_price: Decimal::from(50000),
                breakeven_price: Decimal::from(50000),
                unrealized_pnl: Decimal::ZERO,
                realized_pnl: Decimal::ZERO,
                margin_type: MarginType::Cross,
                isolated_wallet: Decimal::ZERO,
                timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
            };

            manager.update_position(update).await?;
        }

        // Create position for different symbol
        let update = PositionUpdate {
            position_id: PositionId::new(),
            venue: Venue::Binance,
            symbol: "ETHUSDT".into(),
            side: PositionSide::Short,
            amount: Decimal::from(5),
            entry_price: Decimal::from(3000),
            breakeven_price: Decimal::from(3000),
            unrealized_pnl: Decimal::ZERO,
            realized_pnl: Decimal::ZERO,
            margin_type: MarginType::Cross,
            isolated_wallet: Decimal::ZERO,
            timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
        };

        manager.update_position(update).await?;

        // Test retrieval by symbol
        let btc_positions = manager
            .get_positions_by_symbol(Venue::Binance, "BTCUSDT")
            .await;
        assert_eq!(btc_positions.len(), 3);

        let eth_positions = manager
            .get_positions_by_symbol(Venue::Binance, "ETHUSDT")
            .await;
        assert_eq!(eth_positions.len(), 1);

        let empty_positions = manager
            .get_positions_by_symbol(Venue::Binance, "XRPUSDT")
            .await;
        assert!(empty_positions.is_empty());

        Ok(())
    }

    #[tokio::test]
    async fn test_pnl_calculations() -> Result<()> {
        let manager = DefaultPositionManager::new(None);

        // Create positions with different PnL
        let positions = [
            (Decimal::from(1000), Decimal::from(500)), // unrealized, realized
            (Decimal::from(-500), Decimal::from(200)),
            (Decimal::from(2000), Decimal::from(-100)),
        ];

        for (i, (unrealized, realized)) in positions.iter().enumerate() {
            let update = PositionUpdate {
                position_id: PositionId::new(),
                venue: Venue::Binance,
                symbol: format!("TEST{i}").into(),
                side: PositionSide::Long,
                amount: Decimal::from(1),
                entry_price: Decimal::from(50000),
                breakeven_price: Decimal::from(50000),
                unrealized_pnl: *unrealized,
                realized_pnl: *realized,
                margin_type: MarginType::Cross,
                isolated_wallet: Decimal::ZERO,
                timestamp_ns: rusty_common::time::get_epoch_timestamp_ns(),
            };

            manager.update_position(update).await?;
        }

        let total_unrealized = manager.get_total_unrealized_pnl().await;
        assert_eq!(total_unrealized, Decimal::from(2500)); // 1000 - 500 + 2000

        let total_realized = manager.get_total_realized_pnl().await;
        assert_eq!(total_realized, Decimal::from(600)); // 500 + 200 - 100

        Ok(())
    }
}