Struct PortfolioCache 

pub struct PortfolioCache {

    service: Arc<PortfolioServiceImpl>,
    db: Arc<DatabaseHandler>,
    settlement: Arc<dyn SettlementWriter>,
    last_processed_seq: Arc<RwLock<i64>>,
    projection_barrier: Arc<Mutex<()>>,
    subscribers: Arc<RwLock<HashMap<WalletAddress, HashMap<u64, UnboundedSender<PortfolioUpdate>>>>>,
    next_subscriber_id: Arc<RwLock<u64>>,
    span_margin_service: Arc<RwLock<Option<Arc<SpanMarginService>>>>,
    greeks_cache: Arc<RwLock<Option<Arc<GreeksCache>>>>,
    risk_account_builder: Arc<RwLock<Option<Arc<RiskAccountBuilder>>>>,
    tier_cache: Arc<RwLock<Option<Arc<TierCache>>>>,
    order_snapshot: Arc<RwLock<Option<Arc<dyn OrderSnapshotProvider>>>>,
    standard_margin_service: Arc<RwLock<Option<Arc<StandardMarginService>>>>,
    standard_account_builder: Arc<RwLock<Option<Arc<StandardAccountBuilder>>>>,
    sync_status: Arc<SyncStatus>,
}

Fields

service: Arc<PortfolioServiceImpl>

db: Arc<DatabaseHandler>

settlement: Arc<dyn SettlementWriter>

last_processed_seq: Arc<RwLock<i64>>

projection_barrier: Arc<Mutex<()>>

subscribers: Arc<RwLock<HashMap<WalletAddress, HashMap<u64, UnboundedSender<PortfolioUpdate>>>>>

next_subscriber_id: Arc<RwLock<u64>>

span_margin_service: Arc<RwLock<Option<Arc<SpanMarginService>>>>

Optional SPAN margin service for real-time margin updates

greeks_cache: Arc<RwLock<Option<Arc<GreeksCache>>>>

Optional Greeks cache for spot price lookups

risk_account_builder: Arc<RwLock<Option<Arc<RiskAccountBuilder>>>>

Optional RiskAccountBuilder for computing open orders IM

tier_cache: Arc<RwLock<Option<Arc<TierCache>>>>

Optional tier cache for margin mode lookups.

order_snapshot: Arc<RwLock<Option<Arc<dyn OrderSnapshotProvider>>>>

Optional order snapshot provider for open-order calculations.

standard_margin_service: Arc<RwLock<Option<Arc<StandardMarginService>>>>

Optional standard margin service for Standard mode calculations.

standard_account_builder: Arc<RwLock<Option<Arc<StandardAccountBuilder>>>>

Optional standard account builder for Standard mode account assembly.

sync_status: Arc<SyncStatus>

Portfolio sync status for readiness gating. API/engine should not report “ready” until this is Ready.

Implementations

impl PortfolioCache

pub fn new(db: Arc<DatabaseHandler>) -> Self

pub fn sync_status(&self) -> Arc<SyncStatus>

Get the sync status for readiness checks.

API endpoints and WebSocket handlers should check this before serving requests that depend on portfolio state.

pub async fn lock_projection_barrier(&self) -> OwnedMutexGuard<()>

pub fn is_ready(&self) -> bool

Check if portfolio is ready to serve requests.

pub fn sync_state(&self) -> SyncState

Get the current sync state.

pub fn set_catching_up(&self)

Mark portfolio as catching up (called after snapshot restore).

pub fn set_ready(&self)

Mark portfolio as ready (called after catchup replay completes).

pub async fn set_margin_dependencies( &self, span_margin_service: Arc<SpanMarginService>, greeks_cache: Arc<GreeksCache>, risk_account_builder: Arc<RiskAccountBuilder>, tier_cache: Arc<TierCache>, order_snapshot: Arc<dyn OrderSnapshotProvider>, standard_margin_service: Arc<StandardMarginService>, standard_account_builder: Arc<StandardAccountBuilder>, )

Set margin-related dependencies for shared REST/WS margin calculations.

pub async fn compute_wallet_margin_snapshot( &self, wallet: &WalletAddress, ) -> Result<WalletMarginSnapshot>

async fn compute_wallet_margin_snapshot_for_mode( &self, wallet: &WalletAddress, margin_mode: MarginMode, ) -> Result<WalletMarginSnapshot>

async fn compute_portfolio_mode_snapshot( &self, wallet: &WalletAddress, ) -> Result<WalletMarginSnapshot>

pub async fn compute_pm_risk_grid_data( &self, wallet: &WalletAddress, ) -> Result<PmRiskGridData>

async fn compute_standard_mode_snapshot( &self, wallet: &WalletAddress, ) -> Result<WalletMarginSnapshot>

fn snapshot_open_sell_positions( wallet: &WalletAddress, order_snapshot: &dyn OrderSnapshotProvider, ) -> Result<Vec<OpenSellPositionInfo>>

fn snapshot_open_buy_premium( wallet: &WalletAddress, order_snapshot: &dyn OrderSnapshotProvider, ) -> Result<Decimal>

pub async fn publish_margin_update(&self, wallet: &WalletAddress)

Compute and publish margin update for a wallet to all subscribers. This is called after fills to provide real-time margin updates.

async fn try_publish_margin_update(&self, wallet: &WalletAddress) -> Result<()>

async fn prune_applied_settlement_ghost_positions( &self, wallet: &WalletAddress, ) -> Result<usize>

pub async fn publish_margin_updates_for_subscribers(&self) -> usize

Compute and publish margin updates for all wallets with active portfolio subscribers.

Returns the number of unique wallets attempted.

async fn has_portfolio_subscribers(&self, wallet: &WalletAddress) -> bool

async fn compute_wallet_live_greeks_snapshot( &self, wallet: &WalletAddress, ) -> Result<(Vec<PositionGreeksLeg>, Option<PortfolioGreeksAggregate>)>

pub async fn publish_greeks_update(&self, wallet: &WalletAddress)

Compute and publish a portfolio Greeks update for a wallet.

pub async fn publish_greeks_updates_for_subscribers(&self) -> usize

Compute and publish greeks updates for all wallets with active portfolio subscribers.

Returns the number of unique wallets attempted.

pub fn get_service(&self) -> Arc<PortfolioServiceImpl>

Get the underlying PortfolioService for wiring into other components.

This returns the same service instance that PortfolioCache uses, allowing UnifiedEngine to read executed state from the canonical source.

pub async fn reprice_all_wallets(&self) -> usize

Per-position repricing: update each position’s market price individually.

Reprice all wallets that have positions. Called once at startup after all dependencies (greeks cache, oracles) are wired, to fix stale UPNL=0 from journal replay.

async fn refresh_live_market_prices_for_wallet( &self, wallet: &WalletAddress, ) -> Result<usize>

Unlike the old batch approach (refresh_wallet_theoretical_option_upnl), this continues on individual failures so one instrument’s IV failure doesn’t poison the entire wallet’s UPNL.

async fn margin_mode_for_wallet( &self, wallet: &WalletAddress, ) -> Result<MarginMode>

pub async fn publish_position_updates(&self, wallet: &WalletAddress)

Publish repriced positions for a subscribed wallet.

pub async fn publish_position_updates_for_subscribers(&self) -> usize

Publish repriced position updates for every wallet with active subscribers.

async fn build_enriched_positions( &self, wallet: &WalletAddress, ) -> Result<Vec<PositionWithMetrics>>

async fn build_enriched_position_update( &self, wallet: &WalletAddress, symbol: &str, ) -> Result<PositionWithMetrics>

pub async fn initialize(&self) -> Result<i64>

Initialize portfolios from the latest snapshot and return the next engine_commands.command_id that must be replayed into memory.

pub async fn handle_hypercore_position_update( &self, update: HypercorePositionUpdate, )

Handle hypercore position updates from the HypercorePositionService.

Routes to set_hypercore_position (for snapshots) or apply_hypercore_position_update (for diffs) based on the snapshot flag.

pub async fn handle_engine_message(&self, message: EngineMessage, seq: i64)

Handle incoming engine messages to update portfolio state.

Fill handling is memory-only. Durable fills, trades, and fill-driven ledger rows are owned by the journal persistence path.

pub(crate) async fn handle_engine_message_under_barrier( &self, message: EngineMessage, seq: i64, ) -> PendingNotifications

pub(crate) async fn handle_order_filled_under_barrier( &self, fill: &Fill, seq: i64, )

pub(crate) async fn handle_position_expired_under_barrier( &self, expiry_msg: &PositionExpiredMessage, seq: Option<i64>, ) -> PendingNotifications

pub(crate) async fn handle_replayed_position_expired_projection_under_barrier( &self, expiry_msg: &PositionExpiredMessage, ) -> PendingNotifications

pub(crate) async fn handle_option_custody_delta( &self, wallet: WalletAddress, symbol: String, quantity_delta: Decimal, seq: Option<i64>, )

pub(crate) async fn handle_option_custody_delta_under_barrier( &self, wallet: WalletAddress, symbol: String, quantity_delta: Decimal, seq: Option<i64>, ) -> PendingNotifications

pub async fn replay_option_custody_delta( &self, wallet: WalletAddress, symbol: String, quantity_delta: Decimal, seq: i64, )

pub(crate) async fn send_fill_notifications(&self, fill: &Fill)

Send fill notifications. Call AFTER releasing the projection barrier so these non-critical operations don’t block other state mutations.

pub async fn replay_journal_fill(&self, fill: &Fill, seq: i64)

async fn apply_fill_state_mutation(&self, fill: &Fill, seq: i64)

Apply fill state mutations to portfolio memory. This is the barrier-protected part of fill processing: position updates and seq advancement. Does NOT send notifications – callers should use send_fill_notifications after releasing the barrier.

async fn apply_position_expired_state_mutation( &self, expiry_msg: &PositionExpiredMessage, seq: Option<i64>, ) -> PortfolioChange

pub(crate) async fn send_pending_notifications( &self, pending: PendingNotifications, )

Dispatch pending notification work that was collected under the barrier.

async fn send_change_notifications( &self, change: &PortfolioChange, timestamp: i64, )

Send portfolio update notifications from a PortfolioChange struct.

This is the centralized notification method - all portfolio state changes should go through here to ensure consistent WebSocket updates.

async fn get_position_and_margin_for_notification( &self, wallet: &WalletAddress, symbol: &str, ) -> (PositionChange, Decimal)

Get position state and margin for a wallet/symbol after a fill.

Returns both values from a single portfolio read to avoid redundant lock acquisitions. Used to construct PortfolioChange for notification after apply_fill_to_memory.

pub async fn subscribe( &self, account: WalletAddress, ) -> (u64, UnboundedReceiver<PortfolioUpdate>)

Subscribe to portfolio updates for a specific account

pub async fn unsubscribe(&self, account: &WalletAddress, subscriber_id: u64)

Unsubscribe a specific subscriber

pub async fn subscriber_count_for_wallet( &self, account: &WalletAddress, ) -> usize

Return the number of active subscribers for a wallet.

async fn notify_subscribers( &self, account: &WalletAddress, update: PortfolioUpdate, )

Notify subscribers of portfolio updates

pub async fn process_test_fill( &self, account: &WalletAddress, option_id: &str, side: &str, price: Decimal, volume: Decimal, )

pub async fn get_portfolio(&self, account: &WalletAddress) -> Result<Portfolio>

Get portfolio for a specific account

pub async fn get_portfolio_fail_closed_pm( &self, account: &WalletAddress, ) -> Result<Portfolio>

pub async fn has_live_position_symbol( &self, wallet: &WalletAddress, symbol: &str, ) -> bool

pub async fn has_portfolio(&self, account: &WalletAddress) -> bool

Check if a portfolio exists for an account.

pub async fn update_market_prices(&self, prices: HashMap<String, Decimal>)

Update market prices for unrealized PnL calculations

pub async fn get_last_processed_seq(&self) -> i64

Get the last processed portfolio sequence marker.

pub async fn capture_snapshot_state( &self, ) -> (i64, HashMap<WalletAddress, PortfolioBalance>)

Capture current (seq, portfolios) for tests and diagnostics.

pub async fn capture_portfolios_under_barrier( &self, _barrier_proof: &OwnedMutexGuard<()>, ) -> HashMap<WalletAddress, PortfolioBalance>

pub async fn get_all_portfolios( &self, ) -> HashMap<WalletAddress, PortfolioSummary>

Get all portfolios with margin info (for metrics collection).

Returns a map of wallet address to portfolio summary including positions and margin.

Trait Implementations

impl PortfolioCacheApi for PortfolioCache

fn compute_wallet_margin_snapshot<'life0, 'life1, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, ) -> Pin<Box<dyn Future<Output = Result<WalletMarginSnapshot>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn get_portfolio<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 WalletAddress, ) -> Pin<Box<dyn Future<Output = Result<Portfolio>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn get_portfolio_fail_closed_pm<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 WalletAddress, ) -> Pin<Box<dyn Future<Output = Result<Portfolio>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn compute_pm_risk_grid_data<'life0, 'life1, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, ) -> Pin<Box<dyn Future<Output = Result<PmRiskGridData>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn open_position_count<'life0, 'life1, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, ) -> Pin<Box<dyn Future<Output = usize> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn has_live_position_symbol<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, symbol: &'life2 str, ) -> Pin<Box<dyn Future<Output = bool> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn get_all_portfolios<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = HashMap<WalletAddress, PortfolioSummary>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn subscribe<'life0, 'async_trait>( &'life0 self, wallet: WalletAddress, ) -> Pin<Box<dyn Future<Output = (u64, UnboundedReceiver<PortfolioUpdate>)> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn unsubscribe<'life0, 'life1, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, subscriber_id: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn handle_engine_message<'life0, 'async_trait>( &'life0 self, message: EngineMessage, sequence: i64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn handle_hypercore_position_update<'life0, 'async_trait>( &'life0 self, update: HypercorePositionUpdate, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn publish_margin_update<'life0, 'life1, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,