75. Portfolio Valuation System Design¶

services/portfolio-service/
├── src/
│   ├── main.py                    # FastAPI application entry point
│   ├── portfolio/
│   │   ├── position_manager.py    # Individual position management
│   │   ├── account_valuation.py   # Single account NAV calculation
│   │   ├── portfolio_valuation.py # Multi-account portfolio aggregation
│   │   ├── risk_metrics.py        # Portfolio risk calculations
│   │   └── valuation_api.py       # NAV calculation endpoints
│   ├── api/
│   │   ├── portfolio_api.py       # Portfolio management endpoints
│   │   └── websocket_api.py       # Real-time WebSocket endpoints
│   ├── models/
│   │   ├── portfolio_model.py     # Portfolio data models
│   │   └── valuation_model.py     # NAV calculation models
│   ├── config.py                  # Configuration management
│   └── requirements.txt           # Python dependencies
├── Dockerfile                     # Container definition
└── docker-compose.yml             # Local development setup

from collections import defaultdict
from typing import Dict, List, Optional
from datetime import datetime
import asyncio

class AccountValuation:
    def __init__(self, account_id: str):
        self.account_id = account_id
        # symbol -> {"volume", "avg_price", "unrealized_pnl", "realized_pnl"}
        self.positions = defaultdict(lambda: {
            "volume": 0.0, 
            "avg_price": 0.0,
            "unrealized_pnl": 0.0,
            "realized_pnl": 0.0,
            "last_updated": datetime.now()
        })
        self.cash = 0.0  # Available cash
        self.margin_used = 0.0  # Used margin (for futures)
        self.market_prices = {}  # symbol -> latest market price
        self.currency = "USD"  # Base currency
        self.last_nav_calculation = datetime.now()

    def update_position(self, symbol: str, volume: float, avg_price: float, 
                       realized_pnl: float = 0.0):
        """Update position with new volume and average price"""
        if volume == 0:
            # Position closed, remove from tracking
            if symbol in self.positions:
                del self.positions[symbol]
        else:
            self.positions[symbol]["volume"] = volume
            self.positions[symbol]["avg_price"] = avg_price
            self.positions[symbol]["realized_pnl"] = realized_pnl
            self.positions[symbol]["last_updated"] = datetime.now()

        self._recalculate_unrealized_pnl(symbol)

    def update_market_price(self, symbol: str, price: float):
        """Update market price for a symbol"""
        self.market_prices[symbol] = price
        self._recalculate_unrealized_pnl(symbol)

    def update_cash(self, delta_cash: float):
        """Update available cash balance"""
        self.cash += delta_cash

    def update_margin(self, margin_used: float):
        """Update used margin (for futures accounts)"""
        self.margin_used = margin_used

    def _recalculate_unrealized_pnl(self, symbol: str):
        """Recalculate unrealized P&L for a position"""
        if symbol in self.positions and symbol in self.market_prices:
            position = self.positions[symbol]
            current_price = self.market_prices[symbol]
            avg_price = position["avg_price"]
            volume = position["volume"]

            # Calculate unrealized P&L
            if volume > 0:  # Long position
                unrealized_pnl = (current_price - avg_price) * volume
            else:  # Short position
                unrealized_pnl = (avg_price - current_price) * abs(volume)

            self.positions[symbol]["unrealized_pnl"] = unrealized_pnl

    def calculate_nav(self) -> Dict:
        """Calculate Net Asset Value (NAV) for the account"""
        total_equity = self.cash

        # Add position values
        total_unrealized_pnl = 0.0
        total_realized_pnl = 0.0

        for symbol, position in self.positions.items():
            if symbol in self.market_prices:
                current_price = self.market_prices[symbol]
                volume = position["volume"]
                position_value = current_price * abs(volume)
                total_equity += position_value
                total_unrealized_pnl += position["unrealized_pnl"]
                total_realized_pnl += position["realized_pnl"]
            else:
                # Use average price if no market price available
                position_value = position["avg_price"] * abs(position["volume"])
                total_equity += position_value

        self.last_nav_calculation = datetime.now()

        return {
            "account_id": self.account_id,
            "total_equity": total_equity,
            "available_cash": self.cash,
            "margin_used": self.margin_used,
            "total_unrealized_pnl": total_unrealized_pnl,
            "total_realized_pnl": total_realized_pnl,
            "position_count": len(self.positions),
            "currency": self.currency,
            "last_updated": self.last_nav_calculation
        }

    def get_positions_summary(self) -> Dict:
        """Get summary of all positions"""
        positions_summary = {}
        total_position_value = 0.0

        for symbol, position in self.positions.items():
            current_price = self.market_prices.get(symbol, position["avg_price"])
            position_value = current_price * abs(position["volume"])
            total_position_value += position_value

            positions_summary[symbol] = {
                "volume": position["volume"],
                "avg_price": position["avg_price"],
                "current_price": current_price,
                "position_value": position_value,
                "unrealized_pnl": position["unrealized_pnl"],
                "realized_pnl": position["realized_pnl"],
                "last_updated": position["last_updated"]
            }

        return {
            "positions": positions_summary,
            "total_position_value": total_position_value,
            "position_count": len(self.positions)
        }

### 2. Portfolio Valuation Module (`portfolio_valuation.py`)

**Purpose**: Aggregates multiple accounts into comprehensive portfolio view

**Key Functions**:
- **Account Aggregation**: Multi-account portfolio consolidation
- **Cross-Exchange Summary**: Unified view across multiple exchanges
- **Currency Normalization**: Standardized currency reporting
- **Portfolio Metrics**: Total NAV, allocation, risk metrics

**Implementation**:
```python
from typing import Dict, List, Optional
from datetime import datetime
import asyncio

class PortfolioValuation:
    def __init__(self):
        self.accounts = {}  # account_id -> AccountValuation
        self.account_groups = {}  # group_id -> [account_ids]
        self.base_currency = "USD"
        self.last_portfolio_calculation = datetime.now()

    def add_account(self, account_id: str, account_obj):
        """Add an account to the portfolio"""
        self.accounts[account_id] = account_obj

    def remove_account(self, account_id: str):
        """Remove an account from the portfolio"""
        if account_id in self.accounts:
            del self.accounts[account_id]

    def create_account_group(self, group_id: str, account_ids: List[str]):
        """Create a logical group of accounts"""
        self.account_groups[group_id] = account_ids

    def update_account_nav(self, account_id: str) -> Dict:
        """Update and return NAV for a specific account"""
        if account_id not in self.accounts:
            raise ValueError(f"Account {account_id} not found")

        return self.accounts[account_id].calculate_nav()

    def calculate_total_nav(self) -> Dict:
        """Calculate total NAV across all accounts"""
        total_nav = 0.0
        total_cash = 0.0
        total_unrealized_pnl = 0.0
        total_realized_pnl = 0.0
        total_margin_used = 0.0
        account_details = {}

        for account_id, account in self.accounts.items():
            nav_data = account.calculate_nav()
            account_nav = nav_data["total_equity"]

            total_nav += account_nav
            total_cash += nav_data["available_cash"]
            total_unrealized_pnl += nav_data["total_unrealized_pnl"]
            total_realized_pnl += nav_data["total_realized_pnl"]
            total_margin_used += nav_data["margin_used"]

            account_details[account_id] = {
                "nav": account_nav,
                "cash": nav_data["available_cash"],
                "unrealized_pnl": nav_data["total_unrealized_pnl"],
                "realized_pnl": nav_data["total_realized_pnl"],
                "margin_used": nav_data["margin_used"],
                "position_count": nav_data["position_count"],
                "last_updated": nav_data["last_updated"]
            }

        self.last_portfolio_calculation = datetime.now()

        return {
            "total_nav": total_nav,
            "total_cash": total_cash,
            "total_unrealized_pnl": total_unrealized_pnl,
            "total_realized_pnl": total_realized_pnl,
            "total_margin_used": total_margin_used,
            "account_count": len(self.accounts),
            "base_currency": self.base_currency,
            "last_updated": self.last_portfolio_calculation,
            "accounts": account_details
        }

    def calculate_group_nav(self, group_id: str) -> Dict:
        """Calculate NAV for a specific account group"""
        if group_id not in self.account_groups:
            raise ValueError(f"Account group {group_id} not found")

        group_accounts = self.account_groups[group_id]
        group_nav = 0.0
        group_cash = 0.0
        group_unrealized_pnl = 0.0
        group_realized_pnl = 0.0
        group_margin_used = 0.0
        account_details = {}

        for account_id in group_accounts:
            if account_id in self.accounts:
                nav_data = self.accounts[account_id].calculate_nav()
                account_nav = nav_data["total_equity"]

                group_nav += account_nav
                group_cash += nav_data["available_cash"]
                group_unrealized_pnl += nav_data["total_unrealized_pnl"]
                group_realized_pnl += nav_data["total_realized_pnl"]
                group_margin_used += nav_data["margin_used"]

                account_details[account_id] = {
                    "nav": account_nav,
                    "cash": nav_data["available_cash"],
                    "unrealized_pnl": nav_data["total_unrealized_pnl"],
                    "realized_pnl": nav_data["total_realized_pnl"],
                    "margin_used": nav_data["margin_used"],
                    "position_count": nav_data["position_count"],
                    "last_updated": nav_data["last_updated"]
                }

        return {
            "group_id": group_id,
            "total_nav": group_nav,
            "total_cash": group_cash,
            "total_unrealized_pnl": group_unrealized_pnl,
            "total_realized_pnl": group_realized_pnl,
            "total_margin_used": group_margin_used,
            "account_count": len(account_details),
            "base_currency": self.base_currency,
            "last_updated": datetime.now(),
            "accounts": account_details
        }

    def get_portfolio_allocation(self) -> Dict:
        """Get portfolio allocation analysis"""
        total_nav = 0.0
        allocation_data = {}

        # Calculate total NAV first
        for account_id, account in self.accounts.items():
            nav_data = account.calculate_nav()
            total_nav += nav_data["total_equity"]

        if total_nav == 0:
            return {"total_nav": 0, "allocations": {}}

        # Calculate allocation percentages
        for account_id, account in self.accounts.items():
            nav_data = account.calculate_nav()
            account_nav = nav_data["total_equity"]
            allocation_percentage = (account_nav / total_nav) * 100

            allocation_data[account_id] = {
                "nav": account_nav,
                "allocation_percentage": allocation_percentage,
                "cash_percentage": (nav_data["available_cash"] / total_nav) * 100,
                "position_count": nav_data["position_count"]
            }

        return {
            "total_nav": total_nav,
            "allocations": allocation_data,
            "last_updated": datetime.now()
        }

### 3. Portfolio API Module (`api/portfolio_api.py`)

**Purpose**: Provides REST API endpoints for portfolio management

**Implementation**:
```python
from fastapi import APIRouter, HTTPException, WebSocket, WebSocketDisconnect
from typing import Dict, List, Optional
from portfolio.portfolio_valuation import PortfolioValuation
from portfolio.account_valuation import AccountValuation
import json

router = APIRouter()
portfolio = PortfolioValuation()

# REST API Endpoints

@router.get("/portfolio/accounts")
async def get_all_accounts():
    """Get list of all accounts in the portfolio"""
    return {
        "accounts": list(portfolio.accounts.keys()),
        "account_count": len(portfolio.accounts)
    }

@router.get("/portfolio/account/{account_id}")
async def get_account_nav(account_id: str):
    """Get NAV for a specific account"""
    try:
        nav = portfolio.update_account_nav(account_id)
        return nav
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

@router.get("/portfolio/account/{account_id}/positions")
async def get_account_positions(account_id: str):
    """Get positions for a specific account"""
    if account_id not in portfolio.accounts:
        raise HTTPException(status_code=404, detail="Account not found")

    account = portfolio.accounts[account_id]
    return account.get_positions_summary()

@router.get("/portfolio/total")
async def get_total_nav():
    """Get total portfolio NAV"""
    return portfolio.calculate_total_nav()

@router.get("/portfolio/summary")
async def get_portfolio_summary():
    """Get comprehensive portfolio summary"""
    total_nav = portfolio.calculate_total_nav()
    allocation = portfolio.get_portfolio_allocation()

    return {
        "portfolio_summary": total_nav,
        "allocation": allocation,
        "timestamp": datetime.now()
    }

@router.get("/portfolio/group/{group_id}")
async def get_group_nav(group_id: str):
    """Get NAV for a specific account group"""
    try:
        return portfolio.calculate_group_nav(group_id)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

@router.get("/portfolio/allocation")
async def get_portfolio_allocation():
    """Get portfolio allocation analysis"""
    return portfolio.get_portfolio_allocation()

# WebSocket endpoints for real-time updates

@router.websocket("/ws/portfolio/account/{account_id}")
async def websocket_account_updates(websocket: WebSocket, account_id: str):
    """WebSocket endpoint for real-time account updates"""
    await websocket.accept()

    try:
        while True:
            # Send account NAV updates every 5 seconds
            if account_id in portfolio.accounts:
                nav_data = portfolio.update_account_nav(account_id)
                await websocket.send_text(json.dumps(nav_data))

            await asyncio.sleep(5)
    except WebSocketDisconnect:
        print(f"WebSocket disconnected for account {account_id}")

@router.websocket("/ws/portfolio/total")
async def websocket_portfolio_updates(websocket: WebSocket):
    """WebSocket endpoint for real-time portfolio updates"""
    await websocket.accept()

    try:
        while True:
            # Send portfolio total NAV updates every 10 seconds
            total_nav = portfolio.calculate_total_nav()
            await websocket.send_text(json.dumps(total_nav))

            await asyncio.sleep(10)
    except WebSocketDisconnect:
        print("Portfolio WebSocket disconnected")
## Data Architecture
## Data Architecture

### Portfolio Data Models

**Account Position Model**:
```json
{
  "account_id": "acc_12345",
  "symbol": "BTCUSDT",
  "volume": 1.5,
  "avg_price": 45000.00,
  "current_price": 46000.00,
  "unrealized_pnl": 1500.00,
  "realized_pnl": 500.00,
  "last_updated": "2024-12-20T10:30:15.123Z"
}

{
  "account_id": "acc_12345",
  "total_equity": 100000.50,
  "available_cash": 25000.00,
  "margin_used": 5000.00,
  "unrealized_pnl": 1500.00,
  "realized_pnl": 2500.00,
  "position_count": 8,
  "currency": "USD",
  "last_updated": "2024-12-20T10:30:15.123Z"
}

{
  "total_nav": 500000.00,
  "total_cash": 125000.00,
  "total_unrealized_pnl": 7500.00,
  "total_realized_pnl": 12500.00,
  "total_margin_used": 25000.00,
  "account_count": 5,
  "base_currency": "USD",
  "last_updated": "2024-12-20T10:30:15.123Z",
  "accounts": {
    "acc_12345": {
      "nav": 100000.50,
      "cash": 25000.00,
      "unrealized_pnl": 1500.00,
      "realized_pnl": 2500.00,
      "margin_used": 5000.00,
      "position_count": 8,
      "last_updated": "2024-12-20T10:30:15.123Z"
    }
  }
}

Trade Execution → Position Update → Account NAV Recalculation → Portfolio Aggregation → Frontend Display
    ↓
Market Data Feed → Price Update → Position Revaluation → NAV Update → Portfolio Update
    ↓
Time Series Storage → Historical NAV → Performance Analysis → Risk Reporting

import React, { useState, useEffect } from 'react';
import axios from 'axios';

interface PortfolioData {
  total_nav: number;
  total_cash: number;
  total_unrealized_pnl: number;
  total_realized_pnl: number;
  account_count: number;
  last_updated: string;
}

const PortfolioView: React.FC = () => {
  const [portfolioData, setPortfolioData] = useState<PortfolioData | null>(null);
  const [loading, setLoading] = useState(true);

  const fetchTotalNAV = async () => {
    try {
      const response = await axios.get('http://localhost:8000/api/portfolio/total');
      setPortfolioData(response.data);
      setLoading(false);
    } catch (error) {
      console.error('Error fetching portfolio data:', error);
    }
  };

  useEffect(() => {
    fetchTotalNAV();
    // Refresh every minute
    const interval = setInterval(fetchTotalNAV, 60000);
    return () => clearInterval(interval);
  }, []);

  if (loading) {
    return <div>Loading portfolio data...</div>;
  }

  return (
    <div className="p-6 bg-white rounded-lg shadow-lg">
      <h2 className="text-2xl font-bold mb-6">Portfolio Overview</h2>

      <div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-4 gap-4">
        <div className="bg-blue-50 p-4 rounded-lg">
          <h3 className="text-sm font-medium text-blue-600">Total NAV</h3>
          <p className="text-2xl font-bold text-blue-900">
            ${portfolioData?.total_nav.toFixed(2)}
          </p>
        </div>

        <div className="bg-green-50 p-4 rounded-lg">
          <h3 className="text-sm font-medium text-green-600">Available Cash</h3>
          <p className="text-2xl font-bold text-green-900">
            ${portfolioData?.total_cash.toFixed(2)}
          </p>
        </div>

        <div className="bg-yellow-50 p-4 rounded-lg">
          <h3 className="text-sm font-medium text-yellow-600">Unrealized P&L</h3>
          <p className={`text-2xl font-bold ${portfolioData?.total_unrealized_pnl >= 0 ? 'text-green-900' : 'text-red-900'}`}>
            ${portfolioData?.total_unrealized_pnl.toFixed(2)}
          </p>
        </div>

        <div className="bg-purple-50 p-4 rounded-lg">
          <h3 className="text-sm font-medium text-purple-600">Accounts</h3>
          <p className="text-2xl font-bold text-purple-900">
            {portfolioData?.account_count}
          </p>
        </div>
      </div>

      <div className="mt-6 text-sm text-gray-500">
        Last updated: {portfolioData?.last_updated}
      </div>
    </div>
  );
};

export default PortfolioView;

class PortfolioPerformanceTracker:
    def __init__(self, portfolio_valuation):
        self.portfolio = portfolio_valuation
        self.historical_nav = []
        self.performance_metrics = {}

    def record_nav_snapshot(self):
        """Record current NAV for performance tracking"""
        nav_data = self.portfolio.calculate_total_nav()
        self.historical_nav.append({
            "timestamp": datetime.now(),
            "nav": nav_data["total_nav"],
            "unrealized_pnl": nav_data["total_unrealized_pnl"],
            "realized_pnl": nav_data["total_realized_pnl"]
        })

    def calculate_performance_metrics(self, start_date: datetime, end_date: datetime):
        """Calculate performance metrics for a time period"""
        # Filter historical NAV data for the period
        period_data = [
            record for record in self.historical_nav
            if start_date <= record["timestamp"] <= end_date
        ]

        if len(period_data) < 2:
            return {}

        initial_nav = period_data[0]["nav"]
        final_nav = period_data[-1]["nav"]

        # Calculate metrics
        total_return = (final_nav - initial_nav) / initial_nav
        max_drawdown = self._calculate_max_drawdown(period_data)
        sharpe_ratio = self._calculate_sharpe_ratio(period_data)

        return {
            "total_return": total_return,
            "max_drawdown": max_drawdown,
            "sharpe_ratio": sharpe_ratio,
            "period_start": start_date,
            "period_end": end_date
        }