hypercall/portfolio/

portfolio_service_impl.rs

use crate::portfolio::HypercorePositionUpdate;
use crate::portfolio::{
    canonical_perp_symbol, PortfolioBalance, PortfolioChange, PortfolioError, PortfolioService,
    PositionChange, PositionData,
};
use crate::read_cache::tier::TierCache;
use crate::rsm::ledger::Ledger;
use crate::rsm::MarginMode;
use crate::shared::order_types::ParsedSymbol;
use crate::snapshot::error::SnapshotError;
use crate::snapshot::traits::Snapshotable;
use anyhow::Result;
use async_trait::async_trait;
use hypercall_engine::{
    calculate_fill_accounting as calculate_engine_fill_accounting, fill_premium_delta,
    EnginePosition, FillAccountingContext, FillAccountingPosition, FillCashSettlement,
};
use hypercall_types::api_models::{Portfolio, Position, PositionWithMetrics};
use hypercall_types::EngineMessage;
use hypercall_types::FillAccounting;
use hypercall_types::Side;
use hypercall_types::{to_human_readable_decimal, WalletAddress};
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};

/// Implementation of PortfolioService.
///
/// Manages in-memory portfolio state with pure state mutations.
/// Tests can still attach a Ledger to exercise legacy PortfolioService APIs,
/// but production balance reads are owned by EngineSnapshot.balance_ledger.
///
/// Fill persistence is journal-owned. This service only applies the in-memory
/// position projection for already-authoritative fill events.
pub struct PortfolioServiceImpl {
    /// In-memory portfolio state keyed by account address.
    portfolios: Arc<RwLock<HashMap<WalletAddress, PortfolioBalance>>>,
    /// Optional legacy ledger used by direct PortfolioService tests.
    /// Production cache paths do not attach a DB-backed ledger here.
    ledger: Arc<RwLock<Option<Arc<dyn Ledger + Send + Sync>>>>,
    /// Optional tier cache for checking margin mode.
    /// Required for premium settlement in Standard margin mode.
    tier_cache: Arc<RwLock<Option<Arc<TierCache>>>>,
    /// Last-seen timestamp per (wallet, symbol) for perp position updates.
    /// Prevents stale REST-polled updates from overwriting fresher WS-fed data.
    perp_position_timestamps: Arc<RwLock<HashMap<(WalletAddress, String), u64>>>,
}

fn calculate_position_metrics(position: Position) -> PositionWithMetrics {
    let notional_value = position.amount * position.entry_price;
    let margin_ratio = if notional_value.abs() > dec!(0) {
        position.margin_posted / notional_value.abs()
    } else {
        dec!(0)
    };

    PositionWithMetrics {
        position,
        notional_value,
        maintenance_margin: dec!(0),
        liquidation_price: dec!(0),
        margin_ratio,
    }
}

impl PortfolioServiceImpl {
    /// Create a new portfolio service.
    pub fn new() -> Self {
        Self {
            portfolios: Arc::new(RwLock::new(HashMap::new())),
            ledger: Arc::new(RwLock::new(None)),
            tier_cache: Arc::new(RwLock::new(None)),
            perp_position_timestamps: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Set the legacy ledger used by direct PortfolioService tests.
    pub async fn set_ledger(&self, ledger: Arc<dyn Ledger + Send + Sync>) {
        let mut l = self.ledger.write().await;
        *l = Some(ledger);
    }

    /// Set the tier cache for checking margin mode.
    ///
    /// When set, option fills will apply premium settlement for Standard margin mode.
    /// Premium is debited on buy, credited on sell - directly affecting USDC balance.
    pub async fn set_tier_cache(&self, tier_cache: Arc<TierCache>) {
        let mut tc = self.tier_cache.write().await;
        *tc = Some(tier_cache);
    }

    /// Apply option premium settlement for Standard margin mode accounts.
    ///
    /// In Standard margin mode, option premium is settled in USDC at fill time:
    /// - Buy option: USDC balance decreases by premium (debit)
    /// - Sell option: USDC balance increases by premium (credit)
    ///
    /// In Portfolio margin mode, this does nothing (premium is financed via margin).
    async fn apply_option_premium_if_standard(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
        side: &Side,
        price: Decimal,
        size: Decimal,
    ) -> Result<(), PortfolioError> {
        if ParsedSymbol::from_symbol(symbol).is_err() {
            return Ok(());
        }

        let ledger = self.ledger.read().await.clone();
        let Some(ledger) = ledger else {
            debug!(
                "No ledger configured, skipping premium settlement lookup for {}",
                wallet
            );
            return Ok(());
        };

        if !self
            .wallet_uses_standard_option_premium(wallet, symbol)
            .await?
        {
            debug!(
                "Skipping premium settlement for {} - Portfolio mode",
                wallet
            );
            return Ok(());
        }

        let premium_delta = fill_premium_delta(*side, price, size);

        ledger
            .apply_premium(wallet, premium_delta)
            .await
            .map_err(|e| {
                error!(
                    "CRITICAL: Failed to apply option premium {} to ledger for wallet {}: {:?}",
                    premium_delta, wallet, e
                );
                PortfolioError::LedgerError(format!(
                    "Failed to apply option premium {} for {}: {:?}",
                    premium_delta, wallet, e
                ))
            })?;
        info!(
            "Applied option premium {} to ledger for wallet {} (symbol: {}, side: {:?}, price: {}, size: {})",
            premium_delta, wallet, symbol, side, price, size
        );

        Ok(())
    }

    /// Reset portfolio state from snapshots (for initialization from DB).
    ///
    /// Replaces all in-memory state with the provided snapshots.
    pub async fn reset_from_snapshots(
        &self,
        snapshots: HashMap<WalletAddress, PortfolioBalance>,
    ) -> Result<()> {
        let mut portfolios = self.portfolios.write().await;
        *portfolios = snapshots;
        Ok(())
    }

    /// Get all portfolios as a snapshot (for persistence).
    ///
    /// Returns a clone of all in-memory portfolio state.
    pub async fn all_portfolios(&self) -> HashMap<WalletAddress, PortfolioBalance> {
        let portfolios = self.portfolios.read().await;
        portfolios.clone()
    }

    /// Get the internal PortfolioBalance for a specific account.
    ///
    /// Returns None if account doesn't exist.
    /// This is for risk/margin calculations - use get_portfolio() for API responses.
    pub async fn get_portfolio_balance(&self, account: &WalletAddress) -> Option<PortfolioBalance> {
        let portfolios = self.portfolios.read().await;
        portfolios.get(account).cloned()
    }

    /// Update market prices for unrealized PnL calculations.
    ///
    /// Not an event - ephemeral mark-to-market updates.
    pub async fn update_market_prices(&self, prices: HashMap<String, Decimal>) {
        let mut portfolios = self.portfolios.write().await;

        for (wallet, portfolio) in portfolios.iter_mut() {
            for (symbol, position) in portfolio.positions.iter_mut() {
                if let Some(&market_price) = prices.get(symbol) {
                    let new_unrealized_pnl =
                        (market_price - position.entry_price) * position.amount;
                    debug!(
                        wallet = %wallet,
                        symbol,
                        amount = %position.amount,
                        entry_price = %position.entry_price,
                        market_price = %market_price,
                        previous_unrealized_pnl = %position.unrealized_pnl,
                        new_unrealized_pnl = %new_unrealized_pnl,
                        "Applied repriced market price to portfolio position"
                    );
                    position.unrealized_pnl = new_unrealized_pnl;
                }
            }
        }
    }

    /// Handle a fill event for a specific wallet (acquires lock internally).
    ///
    /// This is for backward compatibility with tests. For atomic taker+maker fills,
    /// use `apply_fill_locked` directly under a single lock.
    ///
    /// Note: Uses trade_id=0 which means idempotency is disabled for test calls.
    /// Each call will be processed (not skipped).
    ///
    /// # Returns
    /// The realized PnL from closing positions in this fill.
    #[allow(dead_code)] // Used by tests
    async fn handle_fill(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
        side: Side,
        price: Decimal,
        quantity: Decimal,
    ) -> Decimal {
        static TEST_TRADE_ID: std::sync::atomic::AtomicU64 =
            std::sync::atomic::AtomicU64::new(1_000_000_000);
        let trade_id = TEST_TRADE_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

        let mut portfolios = self.portfolios.write().await;
        Self::apply_fill_locked(
            &mut portfolios,
            wallet,
            trade_id,
            symbol,
            side,
            price,
            quantity,
        )
    }

    /// Apply a fill to a wallet's portfolio without acquiring the lock.
    ///
    /// The caller must hold `self.portfolios.write()` and pass the mutable reference.
    /// This allows atomic application of taker+maker fills under a single lock.
    ///
    /// # Idempotency
    /// If `trade_id` has already been applied to this wallet, the fill is skipped.
    /// This makes replay safe - applying the same fill twice is a no-op.
    /// Update position for a fill and return the realized PnL.
    ///
    /// This is a pure in-memory operation used by both `apply_fill_locked`
    /// (for event processing) and `apply_fill_to_memory` (for memory-only updates).
    ///
    /// # Arguments
    /// * `portfolio` - The portfolio to update
    /// * `symbol` - Trading symbol
    /// * `side` - Buy or Sell
    /// * `price` - Fill price
    /// * `quantity` - Fill quantity (positive)
    ///
    /// # Returns
    /// The realized PnL from closing positions in this fill.
    fn update_position_for_fill(
        portfolio: &mut PortfolioBalance,
        symbol: &str,
        side: &Side,
        price: Decimal,
        quantity: Decimal,
    ) -> Decimal {
        let position = portfolio
            .positions
            .entry(symbol.to_string())
            .or_insert_with(|| PositionData {
                symbol: symbol.to_string(),
                amount: dec!(0),
                entry_price: dec!(0),
                margin_posted: dec!(0),
                realized_pnl: dec!(0),
                unrealized_pnl: dec!(0),
            });

        let signed_quantity = match side {
            Side::Buy => quantity,
            Side::Sell => -quantity,
        };
        let engine_position = EnginePosition {
            quantity: position.amount,
            entry_price: position.entry_price,
        };
        let transition =
            EnginePosition::fill_transition(Some(&engine_position), signed_quantity, price);

        // Track realized PnL
        if transition.realized_pnl != dec!(0) {
            position.realized_pnl += transition.realized_pnl;
        }
        position.amount = transition.quantity;
        position.entry_price = transition.entry_price;

        // UPNL is mark-to-market and must be refreshed from theoretical marks.
        // Do not use fill price as a mark proxy.
        if position.amount == dec!(0) {
            position.unrealized_pnl = dec!(0);
        }

        // Clean up zero positions (using small threshold for Decimal)
        if position.amount.abs() < dec!(0.00000001) {
            portfolio.positions.remove(symbol);
        }

        transition.realized_pnl
    }

    ///
    /// # Returns
    /// The realized PnL from closing positions in this fill.
    /// - Positive: profit from closing
    /// - Negative: loss from closing
    /// - Zero: no position was closed (pure open/add), OR fill was skipped (idempotent)
    fn apply_fill_locked(
        portfolios: &mut HashMap<WalletAddress, PortfolioBalance>,
        wallet: &WalletAddress,
        _trade_id: u64,
        symbol: &str,
        side: Side,
        price: Decimal,
        quantity: Decimal,
    ) -> Decimal {
        let portfolio = portfolios
            .entry(*wallet)
            .or_insert_with(|| PortfolioBalance {
                positions: HashMap::new(),
                total_margin_used: dec!(0),
            });

        // NOTE: Fill idempotency is handled at the DB level via
        // ON CONFLICT DO NOTHING in apply_fill_with_ledger_sync.
        // This method is only called when DB insert succeeds.

        Self::update_position_for_fill(portfolio, symbol, &side, price, quantity)
    }

    /// Apply realized PnL to the legacy test ledger if configured.
    ///
    /// This is called after handle_fill when there's a non-zero realized PnL.
    ///
    /// # Errors
    /// Returns `PortfolioError::LedgerError` if the ledger update fails.
    /// Production fill persistence is journal-owned; runtime collateral comes
    /// from EngineSnapshot.balance_ledger.
    async fn apply_realized_pnl_to_ledger(
        &self,
        wallet: &WalletAddress,
        realized_pnl: Decimal,
    ) -> Result<(), PortfolioError> {
        if realized_pnl == dec!(0) {
            return Ok(());
        }

        let ledger = self.ledger.read().await;
        if let Some(ref ledger) = *ledger {
            ledger.apply_pnl(wallet, realized_pnl).await.map_err(|e| {
                error!(
                    "CRITICAL: Failed to apply realized PnL {} to ledger for wallet {}: {:?}",
                    realized_pnl, wallet, e
                );
                PortfolioError::LedgerError(format!(
                    "Failed to apply realized PnL {} for {}: {:?}",
                    realized_pnl, wallet, e
                ))
            })?;

            debug!(
                "Applied realized PnL {} to ledger for wallet {}",
                realized_pnl, wallet
            );
        }
        Ok(())
    }

    // ===== Crash-Safe Fill Processing Methods =====

    /// Calculate the accounting delta that would result from applying a fill.
    ///
    /// This is a pure calculation with no side effects.
    /// Used to determine what to write to the ledger in the atomic DB transaction.
    ///
    /// IMPORTANT: For Standard margin mode on OPTIONS, we use premium settlement only.
    /// The premium flow (buy = debit, sell = credit) already captures the P&L:
    /// - Buy at $1000 → debit $1000
    /// - Sell at $700 → credit $700
    /// - Net = -$300 (loss is implicit in premium flow)
    ///
    /// For Portfolio margin or perps, we use realized PnL instead.
    pub async fn calculate_fill_accounting(
        &self,
        fill: &hypercall_types::Fill,
    ) -> Result<FillAccounting, PortfolioError> {
        let taker_uses_premium = self
            .wallet_uses_standard_option_premium(&fill.taker_wallet_address, &fill.symbol)
            .await?;
        let maker_uses_premium = self
            .wallet_uses_standard_option_premium(&fill.maker_wallet_address, &fill.symbol)
            .await?;
        let portfolios = self.portfolios.read().await;
        Ok(Self::calculate_fill_accounting_against_portfolios(
            &portfolios,
            fill,
            taker_uses_premium,
            maker_uses_premium,
        ))
    }

    fn calculate_fill_accounting_against_portfolios(
        portfolios: &HashMap<WalletAddress, PortfolioBalance>,
        fill: &hypercall_types::Fill,
        taker_uses_premium: bool,
        maker_uses_premium: bool,
    ) -> FillAccounting {
        calculate_engine_fill_accounting(
            fill,
            FillAccountingContext {
                taker_position: Self::fill_accounting_position_for_wallet(
                    portfolios,
                    &fill.taker_wallet_address,
                    &fill.symbol,
                ),
                maker_position: Self::fill_accounting_position_for_wallet(
                    portfolios,
                    &fill.maker_wallet_address,
                    &fill.symbol,
                ),
                taker_cash_settlement: Self::cash_settlement_for_fill(taker_uses_premium),
                maker_cash_settlement: Self::cash_settlement_for_fill(maker_uses_premium),
            },
        )
    }

    fn fill_accounting_position_for_wallet(
        portfolios: &HashMap<WalletAddress, PortfolioBalance>,
        wallet: &WalletAddress,
        symbol: &str,
    ) -> Option<FillAccountingPosition> {
        portfolios
            .get(wallet)
            .and_then(|portfolio| portfolio.positions.get(symbol))
            .map(|position| FillAccountingPosition {
                quantity: position.amount,
                entry_price: position.entry_price,
            })
    }

    fn cash_settlement_for_fill(uses_premium: bool) -> FillCashSettlement {
        if uses_premium {
            FillCashSettlement::OptionPremium
        } else {
            FillCashSettlement::RealizedPnl
        }
    }

    /// Returns whether this fill should settle option premium through the premium ledger.
    async fn wallet_uses_standard_option_premium(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
    ) -> Result<bool, PortfolioError> {
        if ParsedSymbol::from_symbol(symbol).is_err() {
            return Ok(false);
        }

        let tc = self.tier_cache.read().await;
        let margin_mode = match tc.as_ref() {
            Some(tier_cache) => tier_cache.get_margin_mode(wallet).await.map_err(|error| {
                PortfolioError::InternalError(format!(
                    "failed to load margin mode for premium settlement wallet {}: {}",
                    wallet, error
                ))
            })?,
            None => {
                return Err(PortfolioError::InternalError(format!(
                    "TierCache not configured for option premium settlement wallet {}",
                    wallet
                )));
            }
        };
        Ok(margin_mode == MarginMode::Standard)
    }

    /// Get position change for notification, under lock.
    ///
    /// Returns a PositionChange struct capturing the current state of the position.
    /// If position doesn't exist (closed), returns zeroed values.
    fn get_position_change_locked(
        portfolios: &HashMap<WalletAddress, PortfolioBalance>,
        wallet: &WalletAddress,
        symbol: &str,
    ) -> PositionChange {
        if let Some(portfolio) = portfolios.get(wallet) {
            if let Some(pos) = portfolio.positions.get(symbol) {
                return PositionChange {
                    symbol: symbol.to_string(),
                    amount: pos.amount,
                    entry_price: pos.entry_price,
                    margin_posted: pos.margin_posted,
                    realized_pnl: pos.realized_pnl,
                    unrealized_pnl: pos.unrealized_pnl,
                };
            }
        }
        // Position closed or doesn't exist
        PositionChange {
            symbol: symbol.to_string(),
            amount: dec!(0),
            entry_price: dec!(0),
            margin_posted: dec!(0),
            realized_pnl: dec!(0),
            unrealized_pnl: dec!(0),
        }
    }

    /// Apply a fill's position changes to in-memory state only.
    ///
    /// This does NOT update the durable ledger. Live fills are journal-authoritative,
    /// and restart replay also uses this path for position-only reconstruction.
    pub async fn apply_fill_to_memory(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
        side: &Side,
        price: Decimal,
        quantity: Decimal,
    ) {
        let mut portfolios = self.portfolios.write().await;

        let portfolio = portfolios
            .entry(*wallet)
            .or_insert_with(|| PortfolioBalance {
                positions: HashMap::new(),
                total_margin_used: dec!(0),
            });

        // Use shared helper for position update logic
        Self::update_position_for_fill(portfolio, symbol, side, price, quantity);

        debug!(
            "Applied fill to memory: wallet={}, symbol={}, side={:?}, price={}, qty={}",
            wallet, symbol, side, price, quantity
        );
    }

    /// Apply a fill's position changes for both taker and maker atomically under a single write lock.
    pub async fn apply_fill_to_memory_both_sides(
        &self,
        taker_wallet: &WalletAddress,
        maker_wallet: &WalletAddress,
        symbol: &str,
        taker_side: &Side,
        maker_side: &Side,
        price: Decimal,
        quantity: Decimal,
    ) {
        let mut portfolios = self.portfolios.write().await;
        let portfolio = portfolios
            .entry(*taker_wallet)
            .or_insert_with(|| PortfolioBalance {
                positions: HashMap::new(),
                total_margin_used: dec!(0),
            });
        Self::update_position_for_fill(portfolio, symbol, taker_side, price, quantity);

        let portfolio = portfolios
            .entry(*maker_wallet)
            .or_insert_with(|| PortfolioBalance {
                positions: HashMap::new(),
                total_margin_used: dec!(0),
            });
        Self::update_position_for_fill(portfolio, symbol, maker_side, price, quantity);
    }

    pub async fn apply_option_custody_delta(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
        quantity_delta: Decimal,
    ) -> PortfolioChange {
        let mut portfolios = self.portfolios.write().await;
        let total_margin_used = {
            let portfolio = portfolios.entry(*wallet).or_default();
            let position = portfolio
                .positions
                .entry(symbol.to_string())
                .or_insert_with(|| PositionData {
                    symbol: symbol.to_string(),
                    amount: dec!(0),
                    entry_price: dec!(0),
                    margin_posted: dec!(0),
                    realized_pnl: dec!(0),
                    unrealized_pnl: dec!(0),
                });

            let old_amount = position.amount;
            let old_entry = position.entry_price;
            let new_amount = old_amount + quantity_delta;

            if new_amount == dec!(0) {
                portfolio.positions.remove(symbol);
            } else {
                position.amount = new_amount;
                position.entry_price = if old_amount > dec!(0) {
                    old_entry
                } else if new_amount < dec!(0) {
                    old_entry
                } else {
                    dec!(0)
                };
            }
            portfolio.total_margin_used
        };

        let position_change = Self::get_position_change_locked(&portfolios, wallet, symbol);
        PortfolioChange {
            wallet: *wallet,
            position_changes: vec![position_change],
            balance_change: None,
            total_margin_used,
        }
    }
}

impl Default for PortfolioServiceImpl {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl PortfolioService for PortfolioServiceImpl {
    async fn get_portfolio(&self, account: &WalletAddress) -> Portfolio {
        let portfolios = self.portfolios.read().await;

        if let Some(cached) = portfolios.get(account) {
            // Convert cached data to API Portfolio format
            let positions: Vec<PositionWithMetrics> = cached
                .positions
                .values()
                .map(|pos| {
                    let position = Position {
                        wallet_address: *account,
                        symbol: pos.symbol.clone(),
                        amount: pos.amount,
                        entry_price: pos.entry_price,
                        margin_posted: pos.margin_posted,
                        realized_pnl: pos.realized_pnl,
                        unrealized_pnl: pos.unrealized_pnl,
                        updated_at: chrono::Utc::now(),
                    };
                    calculate_position_metrics(position)
                })
                .collect();

            Portfolio {
                wallet_address: *account,
                positions,
                total_margin_used: cached.total_margin_used,
                available_balance: dec!(0), // Computed by handler using equity - IM
                span_margin: None,          // Filled in by HTTP handler from engine
                margin_mode: "standard".to_string(), // Default, filled by HTTP handler
                margin_summary: None,       // Filled by HTTP handler
            }
        } else {
            // Return empty portfolio (lazy account creation)
            Portfolio {
                wallet_address: *account,
                positions: Vec::new(),
                total_margin_used: dec!(0),
                available_balance: dec!(0),
                span_margin: None,
                margin_mode: "standard".to_string(), // Default, filled by HTTP handler
                margin_summary: None,                // Filled by HTTP handler
            }
        }
    }

    async fn get_portfolio_balance(&self, account: &WalletAddress) -> Option<PortfolioBalance> {
        // Delegate to the inherent method
        PortfolioServiceImpl::get_portfolio_balance(self, account).await
    }

    async fn all_portfolios(&self) -> HashMap<WalletAddress, PortfolioBalance> {
        // Delegate to the inherent method
        PortfolioServiceImpl::all_portfolios(self).await
    }

    // WARNING: The OrderFilled branch below uses apply_realized_pnl_to_ledger()
    // and apply_option_premium_if_standard(), both of which write to InMemoryLedger
    // ONLY (no DB persistence). This is safe because the live fill path in
    // PortfolioCache uses apply_fill_state_mutation() instead (which syncs memory
    // AFTER the journal writes to DB). Do NOT use apply_event(OrderFilled) for
    // real fills (e.g., catchup replay) without adding DB persistence first.
    // See: CALL-500 post-mortem — same class of bug (in-memory-only mutation).
    async fn apply_event(
        &self,
        event: &EngineMessage,
    ) -> Result<Vec<PortfolioChange>, PortfolioError> {
        match event {
            EngineMessage::OrderFilled { fill, .. } => {
                let size_human = to_human_readable_decimal(&fill.symbol, fill.size);

                // Compute maker side
                let maker_side = match fill.taker_side {
                    Side::Buy => Side::Sell,
                    Side::Sell => Side::Buy,
                };

                // === ATOMIC TAKER + MAKER PORTFOLIO UPDATE ===
                // Apply both fills under a single lock to ensure readers never see
                // a state where taker is updated but maker is not.
                // Idempotency: if trade_id already applied to a wallet, that side is skipped.
                // Also capture position state for notifications.
                let (
                    taker_realized_pnl,
                    maker_realized_pnl,
                    taker_change,
                    maker_change,
                    taker_margin,
                    maker_margin,
                ) = {
                    let mut portfolios = self.portfolios.write().await;

                    let taker_pnl = Self::apply_fill_locked(
                        &mut portfolios,
                        &fill.taker_wallet_address,
                        fill.trade_id,
                        &fill.symbol,
                        fill.taker_side,
                        fill.price,
                        size_human,
                    );

                    let maker_pnl = Self::apply_fill_locked(
                        &mut portfolios,
                        &fill.maker_wallet_address,
                        fill.trade_id,
                        &fill.symbol,
                        maker_side,
                        fill.price,
                        size_human,
                    );

                    // Capture position state for notifications
                    let taker_pos = Self::get_position_change_locked(
                        &portfolios,
                        &fill.taker_wallet_address,
                        &fill.symbol,
                    );
                    let maker_pos = Self::get_position_change_locked(
                        &portfolios,
                        &fill.maker_wallet_address,
                        &fill.symbol,
                    );

                    // Capture margin used
                    let taker_margin = portfolios
                        .get(&fill.taker_wallet_address)
                        .map(|b| b.total_margin_used)
                        .unwrap_or(dec!(0));
                    let maker_margin = portfolios
                        .get(&fill.maker_wallet_address)
                        .map(|b| b.total_margin_used)
                        .unwrap_or(dec!(0));

                    (
                        taker_pnl,
                        maker_pnl,
                        taker_pos,
                        maker_pos,
                        taker_margin,
                        maker_margin,
                    )
                };
                // Lock released here

                // Apply taker's realized PnL to ledger (only on position closes)
                // CRITICAL: Ledger failure halts processing
                self.apply_realized_pnl_to_ledger(&fill.taker_wallet_address, taker_realized_pnl)
                    .await?;

                // Apply option premium for Standard margin mode (taker)
                // In Standard mode: Buy debits USDC, Sell credits USDC
                self.apply_option_premium_if_standard(
                    &fill.taker_wallet_address,
                    &fill.symbol,
                    &fill.taker_side,
                    fill.price,
                    size_human,
                )
                .await?;

                // Apply maker's realized PnL to ledger (only on position closes)
                // CRITICAL: Ledger failure halts processing
                self.apply_realized_pnl_to_ledger(&fill.maker_wallet_address, maker_realized_pnl)
                    .await?;

                // Apply option premium for Standard margin mode (maker)
                // In Standard mode: Buy debits USDC, Sell credits USDC
                self.apply_option_premium_if_standard(
                    &fill.maker_wallet_address,
                    &fill.symbol,
                    &maker_side,
                    fill.price,
                    size_human,
                )
                .await?;

                // Return changes for both taker and maker
                let mut changes = Vec::with_capacity(2);

                changes.push(PortfolioChange {
                    wallet: fill.taker_wallet_address,
                    position_changes: vec![taker_change],
                    balance_change: None,
                    total_margin_used: taker_margin,
                });

                changes.push(PortfolioChange {
                    wallet: fill.maker_wallet_address,
                    position_changes: vec![maker_change],
                    balance_change: None,
                    total_margin_used: maker_margin,
                });

                return Ok(changes);
            }
            EngineMessage::PositionExpired(expiry_msg) => {
                let wallet = &expiry_msg.wallet_address;
                let symbol = &expiry_msg.symbol;

                info!(
                    "PortfolioService: Removing expired position for wallet={}, symbol={}",
                    wallet, symbol
                );

                {
                    let mut portfolios = self.portfolios.write().await;

                    if let Some(portfolio) = portfolios.get_mut(wallet) {
                        if portfolio.positions.remove(symbol).is_some() {
                            info!(
                                "PortfolioService: Removed expired position {} for wallet {}",
                                symbol, wallet
                            );
                        } else {
                            warn!(
                                "PortfolioService: Position {} not found for wallet {} during expiry",
                                symbol, wallet
                            );
                        }
                    } else {
                        warn!(
                            "PortfolioService: Portfolio not found for wallet {} during expiry",
                            wallet
                        );
                    }
                }
            }
            _ => {
                // Other event types not handled yet (OrderInfo, L2Snapshot, etc.)
            }
        }
        Ok(vec![])
    }

    async fn apply_hypercore_position_update(&self, update: &HypercorePositionUpdate) {
        // For incremental updates, use same logic as set_hypercore_position
        self.set_hypercore_position(update).await;
    }

    async fn set_hypercore_position(&self, update: &HypercorePositionUpdate) {
        let account_normalized = update.account.to_lowercase();

        // Parse wallet address before acquiring lock - fail gracefully on malformed input
        let wallet_address = match WalletAddress::from_str(&account_normalized) {
            Ok(addr) => addr,
            Err(e) => {
                tracing::warn!(
                    "Skipping hypercore position update: invalid account address '{}' for coin {}: {}",
                    update.account,
                    update.coin,
                    e
                );
                return;
            }
        };

        let position_key = canonical_perp_symbol(&update.coin);

        // Reject stale updates: if we already have a newer timestamp for this
        // (wallet, symbol) pair, skip. This prevents REST-polled fallback data
        // from overwriting fresher WS-fed Hydromancer data.
        {
            let mut ts_map = self.perp_position_timestamps.write().await;
            let ts_key = (wallet_address, position_key.clone());
            if let Some(&prev_ts) = ts_map.get(&ts_key) {
                if update.timestamp < prev_ts {
                    debug!(
                        "Dropping stale perp update for {}/{}: ts {} < prev {}",
                        update.account, position_key, update.timestamp, prev_ts
                    );
                    return;
                }
            }
            ts_map.insert(ts_key, update.timestamp);
        }

        let mut portfolios = self.portfolios.write().await;

        let portfolio = portfolios
            .entry(wallet_address)
            .or_insert_with(|| PortfolioBalance {
                positions: HashMap::new(),
                total_margin_used: dec!(0),
            });
        let raw_position_key = update.coin.trim().to_ascii_uppercase();

        if raw_position_key != position_key {
            portfolio.positions.remove(&raw_position_key);
        }
        if update.coin != raw_position_key && update.coin != position_key {
            portfolio.positions.remove(&update.coin);
        }

        if update.size == 0.0 {
            // Position closed - remove from map
            portfolio.positions.remove(&position_key);
            debug!(
                "Removed hypercore position {} for account {}",
                position_key, update.account
            );
        } else {
            // Set/update position - convert f64 to Decimal
            portfolio.positions.insert(
                position_key.clone(),
                PositionData {
                    symbol: position_key.clone(),
                    amount: Decimal::from_f64_retain(update.size).unwrap_or(Decimal::ZERO),
                    entry_price: Decimal::from_f64_retain(update.entry_price)
                        .unwrap_or(Decimal::ZERO),
                    margin_posted: dec!(0), // Hypercore positions have their own margin
                    realized_pnl: dec!(0),  // Not tracked for hypercore positions
                    unrealized_pnl: Decimal::from_f64_retain(update.unrealized_pnl)
                        .unwrap_or(Decimal::ZERO),
                },
            );
            debug!(
                "Set hypercore position {} for account {}: size={}, entry_price={}, unrealized_pnl={}",
                position_key, update.account, update.size, update.entry_price, update.unrealized_pnl
            );
        }
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    async fn calculate_fill_accounting(
        &self,
        fill: &hypercall_types::Fill,
    ) -> Result<FillAccounting, PortfolioError> {
        // Delegate to the inherent method
        PortfolioServiceImpl::calculate_fill_accounting(self, fill).await
    }

    async fn apply_fill_to_memory(
        &self,
        wallet: &WalletAddress,
        symbol: &str,
        side: &Side,
        price: Decimal,
        quantity: Decimal,
    ) {
        // Delegate to the inherent method
        PortfolioServiceImpl::apply_fill_to_memory(self, wallet, symbol, side, price, quantity)
            .await
    }

    async fn remove_expired_position(&self, wallet: &WalletAddress, symbol: &str) {
        let mut portfolios = self.portfolios.write().await;

        if let Some(portfolio) = portfolios.get_mut(wallet) {
            if portfolio.positions.remove(symbol).is_some() {
                info!(
                    "PortfolioService: Removed expired position {} for wallet {} (replay cleanup, no ledger credit)",
                    symbol, wallet
                );
            } else {
                debug!(
                    "PortfolioService: Position {} already removed for wallet {} (replay cleanup)",
                    symbol, wallet
                );
            }
        } else {
            debug!(
                "PortfolioService: Portfolio not found for wallet {} during replay cleanup",
                wallet
            );
        }
    }
}

// Implement Snapshotable for use with generic snapshot infrastructure
#[async_trait]
impl Snapshotable for PortfolioServiceImpl {
    type Key = WalletAddress;
    type State = PortfolioBalance;

    async fn list_all(&self) -> Result<HashMap<Self::Key, Self::State>, SnapshotError> {
        Ok(self.all_portfolios().await)
    }

    async fn restore(&self, key: &Self::Key, state: Self::State) -> Result<(), SnapshotError> {
        let mut portfolios = self.portfolios.write().await;
        portfolios.insert(*key, state);
        Ok(())
    }

    async fn clear_all(&self) -> Result<(), SnapshotError> {
        let mut portfolios = self.portfolios.write().await;
        portfolios.clear();
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::Result;
    use hypercall_types::Fill;
    use hypercall_types::MarginMode as TypeMarginMode;
    use hypercall_types::WalletAddress;
    use hypercall_types::CONTRACT_UNIT_MULTIPLIER;

    use hypercall_types::wallet_address::test_wallet;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// Auto-incrementing trade ID for test isolation
    static NEXT_TRADE_ID: AtomicU64 = AtomicU64::new(1);

    fn create_service() -> PortfolioServiceImpl {
        PortfolioServiceImpl::new()
    }

    struct NoopTierDb;

    impl hypercall_db::TierReader for NoopTierDb {
        fn get_margin_mode_sync(&self, _wallet: &WalletAddress) -> Result<TypeMarginMode> {
            Ok(TypeMarginMode::Standard)
        }

        fn get_existing_margin_mode_sync(
            &self,
            _wallet: &WalletAddress,
        ) -> Result<Option<TypeMarginMode>> {
            Ok(None)
        }

        fn get_tier_defaults_sync(
            &self,
            _tier_name: &str,
        ) -> Result<Option<hypercall_db::TierDefaultsRecord>> {
            Ok(None)
        }

        fn get_user_tier_sync(
            &self,
            _wallet: &WalletAddress,
        ) -> Result<Option<hypercall_db::UserTierRecord>> {
            Ok(None)
        }

        fn get_all_user_tiers_sync(&self) -> Result<Vec<hypercall_db::UserTierRecord>> {
            Ok(Vec::new())
        }
    }

    impl hypercall_db::TierWriter for NoopTierDb {
        fn save_user_tier_sync(&self, _update: &hypercall_db::UserTierUpdate) -> Result<()> {
            Ok(())
        }

        fn set_margin_mode_sync(
            &self,
            _wallet: &WalletAddress,
            _margin_mode: TypeMarginMode,
        ) -> Result<i64> {
            Ok(1)
        }

        fn insert_margin_mode_if_missing_sync(
            &self,
            _wallet: &WalletAddress,
            _margin_mode: TypeMarginMode,
        ) -> Result<Option<i64>> {
            Ok(Some(1))
        }

        fn delete_user_tier_sync(&self, _wallet: &WalletAddress) -> Result<()> {
            Ok(())
        }
    }

    async fn attach_tier_cache(
        service: &PortfolioServiceImpl,
        modes: &[(WalletAddress, TypeMarginMode)],
    ) {
        let tier_cache = Arc::new(TierCache::new(Arc::new(NoopTierDb)).unwrap());
        for (wallet, mode) in modes {
            tier_cache.set_margin_mode_in_memory(wallet, *mode).await;
        }
        service.set_tier_cache(tier_cache).await;
    }

    fn fill_from_event(event: &EngineMessage) -> Fill {
        match event {
            EngineMessage::OrderFilled { fill, .. } => fill.clone(),
            _ => panic!("expected OrderFilled event"),
        }
    }

    #[tokio::test]
    async fn option_custody_delta_updates_read_model_without_cash_effects() {
        let service = create_service();
        let wallet = test_wallet(231);
        let symbol = "BTC-20261231-100000-C";

        service
            .apply_option_custody_delta(&wallet, symbol, dec!(2))
            .await;
        let balance = service
            .get_portfolio_balance(&wallet)
            .await
            .expect("portfolio should exist after option deposit");
        let position = balance
            .positions
            .get(symbol)
            .expect("position should exist");
        assert_eq!(position.amount, dec!(2));
        assert_eq!(position.entry_price, dec!(0));

        service
            .apply_option_custody_delta(&wallet, symbol, dec!(-1))
            .await;
        let balance = service
            .get_portfolio_balance(&wallet)
            .await
            .expect("portfolio should still exist after partial withdrawal");
        assert_eq!(balance.positions[symbol].amount, dec!(1));

        service
            .apply_option_custody_delta(&wallet, symbol, dec!(-1))
            .await;
        let balance = service
            .get_portfolio_balance(&wallet)
            .await
            .expect("portfolio should remain for empty account");
        assert!(
            !balance.positions.contains_key(symbol),
            "withdrawal to zero should remove the read-model position"
        );
    }

    /// Create a fill with an auto-incremented trade_id (for unique fills)
    fn create_fill(
        wallet: WalletAddress,
        symbol: &str,
        side: Side,
        price: f64,
        quantity: f64,
    ) -> EngineMessage {
        let trade_id = NEXT_TRADE_ID.fetch_add(1, Ordering::Relaxed);
        create_fill_with_trade_id(trade_id, wallet, symbol, side, price, quantity)
    }

    /// Create a fill with a specific trade_id (for idempotency testing)
    fn create_fill_with_trade_id(
        trade_id: u64,
        wallet: WalletAddress,
        symbol: &str,
        side: Side,
        price: f64,
        quantity: f64,
    ) -> EngineMessage {
        let price_dec = Decimal::from_f64_retain(price).unwrap_or(dec!(0));
        let size_dec =
            Decimal::from_f64_retain(quantity * CONTRACT_UNIT_MULTIPLIER).unwrap_or(dec!(0));
        let fill = Fill {
            trade_id,
            taker_order_id: 0,
            maker_order_id: 0,
            symbol: symbol.to_string(),
            price: price_dec,
            size: size_dec,
            taker_side: side,
            taker_wallet_address: wallet,
            maker_wallet_address: test_wallet(200),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };
        EngineMessage::OrderFilled {
            accounting: hypercall_engine::FillAccounting::from_fill(&fill),
            fill,
        }
    }

    #[tokio::test]
    async fn calculate_fill_accounting_standard_buy_uses_premium_only() {
        let service = create_service();
        let taker = test_wallet(1);
        let maker = test_wallet(200);
        attach_tier_cache(
            &service,
            &[
                (taker, TypeMarginMode::Standard),
                (maker, TypeMarginMode::Standard),
            ],
        )
        .await;
        let fill = fill_from_event(&create_fill_with_trade_id(
            1001,
            taker,
            "BTC-20261231-100000-C",
            Side::Buy,
            100.0,
            2.0,
        ));

        let accounting = service.calculate_fill_accounting(&fill).await.unwrap();

        assert_eq!(accounting.taker_premium_delta(), dec!(-200));
        assert_eq!(accounting.maker_premium_delta(), dec!(200));
        assert_eq!(accounting.taker_ledger_residual_delta(), Decimal::ZERO);
        assert_eq!(accounting.maker_ledger_residual_delta(), Decimal::ZERO);
        assert_eq!(accounting.taker_net_cash_delta(), dec!(-200));
        assert_eq!(accounting.maker_net_cash_delta(), dec!(200));
    }

    #[tokio::test]
    async fn calculate_fill_accounting_standard_sell_uses_premium_only() {
        let service = create_service();
        let taker = test_wallet(2);
        let maker = test_wallet(200);
        attach_tier_cache(
            &service,
            &[
                (taker, TypeMarginMode::Standard),
                (maker, TypeMarginMode::Standard),
            ],
        )
        .await;
        let fill = fill_from_event(&create_fill_with_trade_id(
            1002,
            taker,
            "BTC-20261231-100000-C",
            Side::Sell,
            100.0,
            2.0,
        ));

        let accounting = service.calculate_fill_accounting(&fill).await.unwrap();

        assert_eq!(accounting.taker_premium_delta(), dec!(200));
        assert_eq!(accounting.maker_premium_delta(), dec!(-200));
        assert_eq!(accounting.taker_ledger_residual_delta(), Decimal::ZERO);
        assert_eq!(accounting.maker_ledger_residual_delta(), Decimal::ZERO);
        assert_eq!(accounting.taker_net_cash_delta(), dec!(200));
        assert_eq!(accounting.maker_net_cash_delta(), dec!(-200));
    }

    #[tokio::test]
    async fn calculate_fill_accounting_standard_close_preserves_reported_pnl() {
        let service = create_service();
        let taker = test_wallet(5);
        let maker = test_wallet(200);
        attach_tier_cache(
            &service,
            &[
                (taker, TypeMarginMode::Standard),
                (maker, TypeMarginMode::Standard),
            ],
        )
        .await;
        service
            .apply_fill_to_memory(
                &taker,
                "BTC-20261231-100000-C",
                &Side::Buy,
                dec!(100),
                dec!(2),
            )
            .await;
        let fill = fill_from_event(&create_fill_with_trade_id(
            1005,
            taker,
            "BTC-20261231-100000-C",
            Side::Sell,
            70.0,
            1.0,
        ));

        let accounting = service.calculate_fill_accounting(&fill).await.unwrap();

        assert_eq!(accounting.taker_realized_pnl, dec!(-30));
        assert_eq!(accounting.taker_premium_delta(), dec!(70));
        assert_eq!(accounting.taker_ledger_residual_delta(), Decimal::ZERO);
        assert_eq!(accounting.taker_net_cash_delta(), dec!(70));
    }

    #[tokio::test]
    async fn calculate_fill_accounting_portfolio_mode_carries_realized_pnl() {
        let service = create_service();
        let taker = test_wallet(3);
        let maker = test_wallet(200);
        attach_tier_cache(
            &service,
            &[
                (taker, TypeMarginMode::Portfolio),
                (maker, TypeMarginMode::Portfolio),
            ],
        )
        .await;
        service
            .apply_fill_to_memory(
                &taker,
                "BTC-20261231-100000-C",
                &Side::Buy,
                dec!(100),
                dec!(2),
            )
            .await;
        let fill = fill_from_event(&create_fill_with_trade_id(
            1003,
            taker,
            "BTC-20261231-100000-C",
            Side::Sell,
            70.0,
            1.0,
        ));

        let accounting = service.calculate_fill_accounting(&fill).await.unwrap();

        assert_eq!(accounting.taker_premium_delta(), Decimal::ZERO);
        assert_eq!(accounting.taker_realized_pnl, dec!(-30));
        assert_eq!(accounting.taker_ledger_residual_delta(), dec!(-30));
        assert_eq!(accounting.taker_net_cash_delta(), dec!(-30));
        assert_eq!(accounting.maker_net_cash_delta(), Decimal::ZERO);
    }

    #[tokio::test]
    async fn calculate_fill_accounting_without_position_has_zero_realized_pnl() {
        let service = create_service();
        let fill = fill_from_event(&create_fill_with_trade_id(
            1004,
            test_wallet(4),
            "BTC-PERP",
            Side::Sell,
            70.0,
            1.0,
        ));

        let accounting = service.calculate_fill_accounting(&fill).await.unwrap();

        assert_eq!(accounting, hypercall_types::FillAccounting::zero(1004));
    }

    #[tokio::test]
    async fn test_handle_buy_fill() {
        let service = create_service();

        let event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&event).await.unwrap();

        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.wallet_address, test_wallet(1));
        assert_eq!(portfolio.positions.len(), 1);
        let position = &portfolio.positions[0].position;
        assert_eq!(position.symbol, "BTC-CALL-100000");
        assert_eq!(position.amount, dec!(10));
        assert_eq!(position.entry_price, dec!(1000));
    }

    #[tokio::test]
    async fn test_handle_sell_fill() {
        let service = create_service();

        // First buy some
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Then sell half
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1100.0, 5.0);
        service.apply_event(&sell_event).await.unwrap();

        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);
        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(5)); // 10 - 5
    }

    #[tokio::test]
    async fn test_handle_full_position_close() {
        let service = create_service();

        // Buy position
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Sell entire position
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1100.0, 10.0);
        service.apply_event(&sell_event).await.unwrap();

        // Check position was closed
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 0);
        assert_eq!(portfolio.total_margin_used, dec!(0));
    }

    #[tokio::test]
    async fn test_handle_short_selling() {
        let service = create_service();

        // Short sell (sell without existing position)
        let event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1000.0, 10.0);
        service.apply_event(&event).await.unwrap();

        // Check short position was created
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);
        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(-10)); // Negative for short

        // NOTE: margin_posted and total_margin_used are deprecated (always 0.0)
        // Real margin comes from SPAN via SpanMarginSummary
    }

    #[tokio::test]
    async fn test_buy_to_close_short_partial() {
        let service = create_service();

        // Short sell 10 contracts at 1000
        let short_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1000.0, 10.0);
        service.apply_event(&short_event).await.unwrap();

        // Buy 5 contracts at 900 to close half the short (profit)
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 900.0, 5.0);
        service.apply_event(&buy_event).await.unwrap();

        // Check internal state directly for realized PnL
        let portfolios = service.all_portfolios().await;
        let wallet_portfolio = portfolios.get(&test_wallet(1)).unwrap();
        let position = wallet_portfolio.positions.get("BTC-CALL-100000").unwrap();

        // Should have -5 contracts remaining (closed 5 of the 10 short)
        assert_eq!(position.amount, dec!(-5));

        // Realized PnL should be (1000 - 900) * 5 = 500 (profit on closing short at lower price)
        assert_eq!(position.realized_pnl, dec!(500));

        // NOTE: margin_posted is deprecated - real margin comes from SPAN
    }

    #[tokio::test]
    async fn test_extend_short_position() {
        let service = create_service();

        // Short sell 10 contracts at 1000
        let short_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1000.0, 10.0);
        service.apply_event(&short_event).await.unwrap();

        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(-10));
        assert_eq!(position.entry_price, dec!(1000));

        // Sell 5 more contracts at 1100 (extending the short)
        let extend_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1100.0, 5.0);
        service.apply_event(&extend_event).await.unwrap();

        // Check position
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        let position = &portfolio.positions[0].position;

        // Should have -15 contracts total
        assert_eq!(position.amount, dec!(-15));

        // Entry price should be weighted average: (1000*10 + 1100*5) / 15 = 1033.33...
        let expected_entry = (dec!(1000) * dec!(10) + dec!(1100) * dec!(5)) / dec!(15);
        assert!((position.entry_price - expected_entry).abs() < dec!(0.01));

        // NOTE: margin_posted is deprecated - real margin comes from SPAN
    }

    #[tokio::test]
    async fn test_buy_to_close_short_full() {
        let service = create_service();

        // Short sell 10 contracts at 1000
        let short_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1000.0, 10.0);
        service.apply_event(&short_event).await.unwrap();

        // Buy 10 contracts at 1100 to close the entire short (loss)
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1100.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Check portfolio
        let portfolio = service.get_portfolio(&test_wallet(1)).await;

        // Position should be closed
        assert_eq!(portfolio.positions.len(), 0);
    }

    #[tokio::test]
    async fn test_sell_to_close_long_with_profit() {
        let service = create_service();

        // Buy 10 contracts at 1000
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Sell 5 contracts at 1200 (profit)
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1200.0, 5.0);
        service.apply_event(&sell_event).await.unwrap();

        // Check internal state directly for realized PnL
        let portfolios = service.all_portfolios().await;
        let wallet_portfolio = portfolios.get(&test_wallet(1)).unwrap();
        let position = wallet_portfolio.positions.get("BTC-CALL-100000").unwrap();

        // Should have 5 contracts remaining
        assert_eq!(position.amount, dec!(5));

        // Realized PnL should be (1200 - 1000) * 5 = 1000 (profit)
        assert_eq!(position.realized_pnl, dec!(1000));
    }

    #[tokio::test]
    async fn test_sell_to_close_long_with_loss() {
        let service = create_service();

        // Buy 10 contracts at 1000
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Sell 10 contracts at 800 (loss)
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 800.0, 10.0);
        service.apply_event(&sell_event).await.unwrap();

        // Check portfolio - position should be closed
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 0);
    }

    #[tokio::test]
    async fn test_multiple_fills_cumulative_realized_pnl() {
        let service = create_service();

        // Buy 10 contracts at 1000
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Sell 5 contracts at 1100 (profit of 500)
        let sell1_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1100.0, 5.0);
        service.apply_event(&sell1_event).await.unwrap();

        {
            let portfolios = service.all_portfolios().await;
            let wallet_portfolio = portfolios.get(&test_wallet(1)).unwrap();
            let position = wallet_portfolio.positions.get("BTC-CALL-100000").unwrap();
            assert_eq!(position.realized_pnl, dec!(500));
        }

        // Sell 3 more contracts at 1200 (profit of 600)
        let sell2_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1200.0, 3.0);
        service.apply_event(&sell2_event).await.unwrap();

        {
            let portfolios = service.all_portfolios().await;
            let wallet_portfolio = portfolios.get(&test_wallet(1)).unwrap();
            let position = wallet_portfolio.positions.get("BTC-CALL-100000").unwrap();

            // Cumulative realized PnL should be 500 + 600 = 1100
            assert_eq!(position.realized_pnl, dec!(1100));
            assert_eq!(position.amount, dec!(2)); // 10 - 5 - 3 = 2 remaining
        }
    }

    #[tokio::test]
    async fn test_multiple_fills_weighted_average() {
        let service = create_service();

        // Multiple buys at different prices
        let buy1_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy1_event).await.unwrap();

        let buy2_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1100.0, 10.0);
        service.apply_event(&buy2_event).await.unwrap();

        // Check weighted average price
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(20));
        assert_eq!(position.entry_price, dec!(1050)); // (1000*10 + 1100*10) / 20
    }

    #[tokio::test]
    async fn test_order_filled_message_quantity_is_converted_from_contract_units() {
        let service = create_service();

        // Create a fill message with contract units (like real production)
        let fill = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-20260131-100000-C".to_string(),
            price: dec!(2000),      // $2000 per contract
            size: dec!(10_000_000), // 10 contracts in CONTRACT UNITS
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(2),
            fee: dec!(1),
            is_taker: true,
            timestamp: chrono::Utc::now().timestamp() as u64,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        // Process through apply_event
        service
            .apply_event(&EngineMessage::OrderFilled {
                accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                fill,
            })
            .await
            .unwrap();

        // Verify taker portfolio
        let portfolio = service.get_portfolio(&test_wallet(1)).await;

        // Verify position quantity
        assert_eq!(portfolio.positions.len(), 1);
        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(10), "Position should be 10 contracts");
    }

    #[tokio::test]
    async fn test_flip_long_to_short() {
        let service = create_service();

        // Buy 10 contracts (long position)
        service
            .handle_fill(
                &test_wallet(1),
                "BTC-CALL-100000",
                Side::Buy,
                dec!(1000),
                dec!(10),
            )
            .await;

        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);
        assert_eq!(portfolio.positions[0].position.amount, dec!(10));

        // Sell 15 contracts (close 10 long, open 5 short)
        service
            .handle_fill(
                &test_wallet(1),
                "BTC-CALL-100000",
                Side::Sell,
                dec!(1100),
                dec!(15),
            )
            .await;

        // Check position after flip
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);

        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(-5), "Should have 5 short contracts");
        assert_eq!(
            position.entry_price,
            dec!(1100),
            "Entry price should be the flip price"
        );

        // NOTE: margin_posted is deprecated - real margin comes from SPAN
    }

    #[tokio::test]
    async fn test_flip_short_to_long() {
        let service = create_service();

        // Sell 10 contracts (short position)
        service
            .handle_fill(
                &test_wallet(1),
                "BTC-CALL-100000",
                Side::Sell,
                dec!(1000),
                dec!(10),
            )
            .await;

        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);
        assert_eq!(portfolio.positions[0].position.amount, dec!(-10));

        // Buy 15 contracts (close 10 short, open 5 long)
        service
            .handle_fill(
                &test_wallet(1),
                "BTC-CALL-100000",
                Side::Buy,
                dec!(900),
                dec!(15),
            )
            .await;

        // Check position after flip
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);

        let position = &portfolio.positions[0].position;
        assert_eq!(position.amount, dec!(5), "Should have 5 long contracts");
        assert_eq!(
            position.entry_price,
            dec!(900),
            "Entry price should be the flip price"
        );

        // NOTE: margin_posted is deprecated - real margin comes from SPAN
    }

    // ============================================================================
    // Ledger integration tests
    // ============================================================================

    #[tokio::test]
    async fn test_ledger_not_touched_on_open() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = PortfolioServiceImpl::new();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        // Set initial ledger balance
        ledger
            .set_balance(&test_wallet(1), dec!(10000))
            .await
            .unwrap();

        // Open a position (buy call)
        let event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&event).await.unwrap();

        // Ledger balance should NOT change (opening position doesn't touch ledger)
        let balance = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance,
            dec!(10000),
            "Opening position should not change ledger balance"
        );
    }

    #[tokio::test]
    async fn test_ledger_increases_on_profitable_close() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = PortfolioServiceImpl::new();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        // Set initial ledger balance
        ledger
            .set_balance(&test_wallet(1), dec!(10000))
            .await
            .unwrap();

        // Buy 10 contracts at 1000
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Balance unchanged after buy
        let balance_after_buy = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance_after_buy,
            dec!(10000),
            "Balance should be unchanged after opening"
        );

        // Sell 10 contracts at 1500 (profit of 500 per contract = 5000 total)
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1500.0, 10.0);
        service.apply_event(&sell_event).await.unwrap();

        // Ledger balance should increase by realized profit
        let balance_after_sell = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance_after_sell,
            dec!(15000),
            "Balance should increase by realized profit (5000)"
        );
    }

    #[tokio::test]
    async fn test_ledger_decreases_on_losing_close() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = PortfolioServiceImpl::new();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        // Set initial ledger balance
        ledger
            .set_balance(&test_wallet(1), dec!(10000))
            .await
            .unwrap();

        // Buy 10 contracts at 1000
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Sell 10 contracts at 800 (loss of 200 per contract = 2000 total)
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 800.0, 10.0);
        service.apply_event(&sell_event).await.unwrap();

        // Ledger balance should decrease by realized loss
        let balance_after_sell = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance_after_sell,
            dec!(8000),
            "Balance should decrease by realized loss (2000)"
        );
    }

    #[tokio::test]
    async fn test_ledger_on_short_close() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = PortfolioServiceImpl::new();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        // Set initial ledger balance
        ledger
            .set_balance(&test_wallet(1), dec!(10000))
            .await
            .unwrap();

        // Short sell 10 contracts at 1000
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1000.0, 10.0);
        service.apply_event(&sell_event).await.unwrap();

        // Balance unchanged after short (opening)
        let balance_after_short = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance_after_short,
            dec!(10000),
            "Balance should be unchanged after opening short"
        );

        // Buy to close at 900 (profit of 100 per contract = 1000 total)
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 900.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Ledger balance should increase by realized profit from closing short
        let balance_after_close = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance_after_close,
            dec!(11000),
            "Balance should increase by profit from closing short (1000)"
        );
    }

    #[tokio::test]
    async fn test_partial_close_applies_partial_pnl() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = PortfolioServiceImpl::new();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        // Set initial ledger balance
        ledger
            .set_balance(&test_wallet(1), dec!(10000))
            .await
            .unwrap();

        // Buy 10 contracts at 1000
        let buy_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Buy, 1000.0, 10.0);
        service.apply_event(&buy_event).await.unwrap();

        // Sell 5 contracts at 1200 (profit of 200 per contract = 1000 total)
        let sell_event = create_fill(test_wallet(1), "BTC-CALL-100000", Side::Sell, 1200.0, 5.0);
        service.apply_event(&sell_event).await.unwrap();

        // Ledger balance should increase by partial realized profit
        let balance = ledger.get_balance(&test_wallet(1)).await.unwrap();
        assert_eq!(
            balance,
            dec!(11000),
            "Balance should increase by partial profit (1000)"
        );

        // Verify position is partially closed
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);
        assert_eq!(
            portfolio.positions[0].position.amount,
            dec!(5),
            "Should have 5 remaining contracts"
        );
    }

    // ============================================================================
    // Idempotency tests
    // ============================================================================
    // NOTE: Fill idempotency is now handled at the DB level via
    // ON CONFLICT (trade_id, wallet_address, is_taker) DO NOTHING in
    // apply_fill_with_ledger_sync. See portfolio_cache_tests for DB-level
    // idempotency tests. The in-memory apply_fill* methods no longer
    // deduplicate - they assume the caller has already checked DB state.
    // ============================================================================

    #[tokio::test]
    async fn test_multiple_fills_accumulate() {
        let service = create_service();

        // Apply two different fills with different trade_ids
        let fill1 = create_fill_with_trade_id(
            1,
            test_wallet(1),
            "BTC-CALL-100000",
            Side::Buy,
            1000.0,
            10.0,
        );
        let fill2 = create_fill_with_trade_id(
            2,
            test_wallet(1),
            "BTC-CALL-100000",
            Side::Buy,
            1000.0,
            10.0,
        );

        service.apply_event(&fill1).await.unwrap();
        service.apply_event(&fill2).await.unwrap();

        // Both should be applied
        let portfolio = service.get_portfolio(&test_wallet(1)).await;
        assert_eq!(portfolio.positions.len(), 1);
        assert_eq!(
            portfolio.positions[0].position.amount,
            dec!(20),
            "Should have 20 contracts (both fills applied)"
        );
    }

    // ============================================================================
    // Position Expiration Tests
    // ============================================================================

    use hypercall_types::PositionExpiredMessage;

    fn create_position_expired(
        wallet: WalletAddress,
        symbol: &str,
        position_size: Decimal,
        settlement_price: Decimal,
        settlement_value: Decimal,
    ) -> EngineMessage {
        EngineMessage::PositionExpired(PositionExpiredMessage {
            wallet_address: wallet,
            margin_mode: crate::rsm::MarginMode::Standard,
            symbol: symbol.to_string(),
            position_size,
            settlement_price,
            settlement_value,
            settlement_entry_price: None,
            cost_basis: None,
            net_pnl: None,
            timestamp: 12345,
        })
    }

    #[tokio::test]
    async fn test_position_expiry_removes_position() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = create_service();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        let wallet = test_wallet(1);
        let symbol = "BTC-CALL-100000";

        // Set initial ledger balance
        ledger.set_balance(&wallet, dec!(10000)).await.unwrap();

        // Open a position via fill
        let fill = create_fill(wallet, symbol, Side::Buy, 500.0, 10.0);
        service.apply_event(&fill).await.unwrap();
        let balance_after_fill = ledger.get_balance(&wallet).await.unwrap();

        // Verify position exists
        let portfolio = service.get_portfolio(&wallet).await;
        assert_eq!(portfolio.positions.len(), 1);
        assert_eq!(portfolio.positions[0].position.symbol, symbol);
        assert_eq!(portfolio.positions[0].position.amount, dec!(10));

        // Apply position expiration
        let expiry = create_position_expired(
            wallet,
            symbol,
            dec!(10),    // position_size
            dec!(5000),  // settlement_price (intrinsic value)
            dec!(50000), // settlement_value (5000 * 10)
        );
        service.apply_event(&expiry).await.unwrap();

        // Position should be removed
        let portfolio_after = service.get_portfolio(&wallet).await;
        assert_eq!(
            portfolio_after.positions.len(),
            0,
            "Position should be removed after expiry"
        );

        // Settlement accounting is DB-owned; PortfolioService only removes the position.
        let balance = ledger.get_balance(&wallet).await.unwrap();
        assert_eq!(
            balance, balance_after_fill,
            "PortfolioService expiry projection must not apply settlement cash"
        );
    }

    #[tokio::test]
    async fn test_position_expiry_nonexistent_position_is_noop() {
        let service = create_service();
        let wallet = test_wallet(1);
        let symbol = "BTC-CALL-100000";

        // Don't create any position first - just expire a non-existent one
        let expiry = create_position_expired(wallet, symbol, dec!(10), dec!(5000), dec!(50000));

        // This should NOT error - it's a no-op for non-existent positions
        service.apply_event(&expiry).await.unwrap();

        // Portfolio should remain empty
        let portfolio = service.get_portfolio(&wallet).await;
        assert!(portfolio.positions.is_empty());
    }

    #[tokio::test]
    async fn test_position_expiry_zero_settlement_no_ledger_call() {
        let service = create_service();
        let wallet = test_wallet(1);
        let symbol = "BTC-CALL-100000";

        // Create a position
        let fill = create_fill(wallet, symbol, Side::Buy, 500.0, 10.0);
        service.apply_event(&fill).await.unwrap();

        // Expire with zero settlement (OTM option)
        let expiry = create_position_expired(
            wallet,
            symbol,
            dec!(10),
            dec!(0), // OTM - no intrinsic value
            dec!(0), // Zero settlement
        );
        service.apply_event(&expiry).await.unwrap();

        // Position removed even with zero settlement
        let portfolio = service.get_portfolio(&wallet).await;
        assert!(portfolio.positions.is_empty());
    }

    #[tokio::test]
    async fn test_position_expiry_replay_idempotency() {
        use crate::rsm::ledger::InMemoryLedger;

        let service = create_service();
        let ledger = Arc::new(InMemoryLedger::new());
        service.set_ledger(ledger.clone()).await;

        let wallet = test_wallet(1);
        let symbol = "BTC-CALL-100000";

        // Set initial ledger balance
        ledger.set_balance(&wallet, dec!(10000)).await.unwrap();

        // Open a position via fill
        let fill = create_fill(wallet, symbol, Side::Buy, 500.0, 10.0);
        service.apply_event(&fill).await.unwrap();
        let balance_after_fill = ledger.get_balance(&wallet).await.unwrap();

        // Apply position expiration first time. PortfolioService projection
        // removes positions only; settlement accounting is DB-owned.
        let expiry = create_position_expired(
            wallet,
            symbol,
            dec!(10),    // position_size
            dec!(5000),  // settlement_price
            dec!(50000), // settlement_value
        );
        service.apply_event(&expiry).await.unwrap();

        // Verify first application worked
        let balance_after_first = ledger.get_balance(&wallet).await.unwrap();
        assert_eq!(
            balance_after_first, balance_after_fill,
            "PortfolioService expiry projection must not apply settlement cash"
        );

        // Apply the SAME expiration event again (simulating event replay)
        // This should be a no-op because position no longer exists
        service.apply_event(&expiry).await.unwrap();

        // Balance should NOT change - idempotency via position_removed check
        let balance_after_replay = ledger.get_balance(&wallet).await.unwrap();
        assert_eq!(
            balance_after_replay, balance_after_fill,
            "Replay should not change ledger balance"
        );

        // Position should still be removed
        let portfolio = service.get_portfolio(&wallet).await;
        assert!(portfolio.positions.is_empty());
    }
}