Struct PortfolioServiceImpl 

pub struct PortfolioServiceImpl {
    portfolios: Arc<RwLock<HashMap<WalletAddress, PortfolioBalance>>>,
    ledger: Arc<RwLock<Option<Arc<dyn Ledger + Send + Sync>>>>,
    tier_cache: Arc<RwLock<Option<Arc<TierCache>>>>,
    perp_position_timestamps: Arc<RwLock<HashMap<(WalletAddress, String), u64>>>,
}

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.

Fields

portfolios: Arc<RwLock<HashMap<WalletAddress, PortfolioBalance>>>

In-memory portfolio state keyed by account address.

ledger: Arc<RwLock<Option<Arc<dyn Ledger + Send + Sync>>>>

Optional legacy ledger used by direct PortfolioService tests. Production cache paths do not attach a DB-backed ledger here.

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

Optional tier cache for checking margin mode. Required for premium settlement in Standard margin mode.

perp_position_timestamps: Arc<RwLock<HashMap<(WalletAddress, String), u64>>>

Last-seen timestamp per (wallet, symbol) for perp position updates. Prevents stale REST-polled updates from overwriting fresher WS-fed data.

Implementations

impl PortfolioServiceImpl

pub fn new() -> Self

Create a new portfolio service.

pub async fn set_ledger(&self, ledger: Arc<dyn Ledger + Send + Sync>)

Set the legacy ledger used by direct PortfolioService tests.

pub async fn set_tier_cache(&self, tier_cache: Arc<TierCache>)

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.

async fn apply_option_premium_if_standard( &self, wallet: &WalletAddress, symbol: &str, side: &Side, price: Decimal, size: Decimal, ) -> Result<(), PortfolioError>

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).

pub async fn reset_from_snapshots( &self, snapshots: HashMap<WalletAddress, PortfolioBalance>, ) -> Result<()>

Reset portfolio state from snapshots (for initialization from DB).

Replaces all in-memory state with the provided snapshots.

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

Get all portfolios as a snapshot (for persistence).

Returns a clone of all in-memory portfolio state.

pub async fn get_portfolio_balance( &self, account: &WalletAddress, ) -> Option<PortfolioBalance>

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 update_market_prices(&self, prices: HashMap<String, Decimal>)

Update market prices for unrealized PnL calculations.

Not an event - ephemeral mark-to-market updates.

async fn handle_fill( &self, wallet: &WalletAddress, symbol: &str, side: Side, price: Decimal, quantity: Decimal, ) -> Decimal

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.

fn update_position_for_fill( portfolio: &mut PortfolioBalance, symbol: &str, side: &Side, price: Decimal, quantity: Decimal, ) -> Decimal

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 apply_fill_locked( portfolios: &mut HashMap<WalletAddress, PortfolioBalance>, wallet: &WalletAddress, _trade_id: u64, symbol: &str, side: Side, price: Decimal, quantity: Decimal, ) -> Decimal

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)

async fn apply_realized_pnl_to_ledger( &self, wallet: &WalletAddress, realized_pnl: Decimal, ) -> Result<(), PortfolioError>

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.

pub async fn calculate_fill_accounting( &self, fill: &Fill, ) -> Result<FillAccounting, PortfolioError>

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.

fn calculate_fill_accounting_against_portfolios( portfolios: &HashMap<WalletAddress, PortfolioBalance>, fill: &Fill, taker_uses_premium: bool, maker_uses_premium: bool, ) -> FillAccounting

fn fill_accounting_position_for_wallet( portfolios: &HashMap<WalletAddress, PortfolioBalance>, wallet: &WalletAddress, symbol: &str, ) -> Option<FillAccountingPosition>

fn cash_settlement_for_fill(uses_premium: bool) -> FillCashSettlement

async fn wallet_uses_standard_option_premium( &self, wallet: &WalletAddress, symbol: &str, ) -> Result<bool, PortfolioError>

Returns whether this fill should settle option premium through the premium ledger.

fn get_position_change_locked( portfolios: &HashMap<WalletAddress, PortfolioBalance>, wallet: &WalletAddress, symbol: &str, ) -> PositionChange

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.

pub async fn apply_fill_to_memory( &self, wallet: &WalletAddress, symbol: &str, side: &Side, price: Decimal, quantity: Decimal, )

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_both_sides( &self, taker_wallet: &WalletAddress, maker_wallet: &WalletAddress, symbol: &str, taker_side: &Side, maker_side: &Side, price: Decimal, quantity: Decimal, )

Apply a fill’s position changes for both taker and maker atomically under a single write lock.

pub async fn apply_option_custody_delta( &self, wallet: &WalletAddress, symbol: &str, quantity_delta: Decimal, ) -> PortfolioChange

Trait Implementations

impl Default for PortfolioServiceImpl

fn default() -> Self

Returns the “default value” for a type.

impl PortfolioService for PortfolioServiceImpl

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

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

Get current portfolio state for an account (API format).

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

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

Get internal balance state for risk/margin calculations.

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

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

Get all portfolios as a snapshot.

fn apply_event<'life0, 'life1, 'async_trait>( &'life0 self, event: &'life1 EngineMessage, ) -> Pin<Box<dyn Future<Output = Result<Vec<PortfolioChange>, PortfolioError>> + Send + 'async_trait>>

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

Apply an event from the matching engine.

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

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

Apply perp position update from Hypercore Position Service (incremental diff).

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

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

Set perp position from Hypercore Position Service (snapshot).

fn as_any(&self) -> &dyn Any

Returns a reference to self as Any for downcasting.

fn calculate_fill_accounting<'life0, 'life1, 'async_trait>( &'life0 self, fill: &'life1 Fill, ) -> Pin<Box<dyn Future<Output = Result<FillAccounting, PortfolioError>> + Send + 'async_trait>>

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

Calculate the accounting delta that would result from applying a fill.

fn apply_fill_to_memory<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, wallet: &'life1 WalletAddress, symbol: &'life2 str, side: &'life3 Side, price: Decimal, quantity: Decimal, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

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

Apply a fill’s position changes to in-memory state only.

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

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

Remove an expired position from memory without applying a cash delta.

impl Snapshotable for PortfolioServiceImpl

type Key = WalletAddress

The key type for identifying entities (e.g., WalletAddress)

type State = PortfolioBalance

The state type for each keyed entity (e.g., wallet -> balance)

fn list_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<HashMap<Self::Key, Self::State>, SnapshotError>> + Send + 'async_trait>>

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

List all states keyed by identifier.

fn restore<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 Self::Key, state: Self::State, ) -> Pin<Box<dyn Future<Output = Result<(), SnapshotError>> + Send + 'async_trait>>

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

Restore a single state entry.

fn clear_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), SnapshotError>> + Send + 'async_trait>>

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

Clear all state before restore.