Multi-Currency Portfolio Valuation System Design¶

services/portfolio-service/
├── src/
│   ├── main.py                            # FastAPI application entry point
│   ├── portfolio/
│   │   ├── multi_currency_portfolio.py    # Multi-currency portfolio management
│   │   ├── position_manager.py            # Position tracking and management
│   │   ├── cash_manager.py                # Cash balance management
│   │   └── asset_allocator.py             # Asset allocation tracking
│   ├── valuation/
│   │   ├── multi_currency_valuation.py    # Cross-currency valuation engine
│   │   ├── nav_calculator.py              # NAV calculation engine
│   │   ├── pnl_calculator.py              # P&L calculation across currencies
│   │   └── risk_analyzer.py               # Currency risk analysis
│   ├── forex/
│   │   ├── forex_service.py               # Exchange rate management
│   │   ├── rate_provider.py               # Rate provider integration
│   │   ├── rate_cache.py                  # Rate caching and management
│   │   └── rate_validator.py              # Rate validation and quality control
│   ├── api/
│   │   ├── portfolio_api.py               # Portfolio management endpoints
│   │   ├── valuation_api.py               # Valuation and NAV endpoints
│   │   └── forex_api.py                   # Exchange rate endpoints
│   ├── models/
│   │   ├── portfolio_model.py             # Portfolio data models
│   │   ├── currency_model.py              # Currency and rate models
│   │   ├── valuation_model.py             # Valuation result models
│   │   └── risk_model.py                  # Risk analysis 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, Tuple
from datetime import datetime
import uuid

class MultiCurrencyPortfolio:
    def __init__(self):
        # Portfolio structure: account_id -> currency -> position_data
        self.portfolios = defaultdict(lambda: defaultdict(lambda: {
            "volume": 0.0,
            "avg_cost": 0.0,
            "market_value": 0.0,
            "unrealized_pnl": 0.0,
            "last_update": None
        }))

        # Cash balances: account_id -> currency -> cash_amount
        self.cash_balances = defaultdict(lambda: defaultdict(float))

        # Transaction history
        self.transactions = []

        # Supported currencies
        self.supported_currencies = {
            "USD", "EUR", "GBP", "JPY", "HKD", "CNY", "SGD",
            "BTC", "ETH", "USDT", "USDC", "BNB", "ADA", "DOT"
        }

    def update_position(self, account_id: str, currency: str, volume: float, 
                       price: float, timestamp: datetime = None) -> bool:
        """Update position for specific currency"""
        if currency not in self.supported_currencies:
            return False

        if timestamp is None:
            timestamp = datetime.now()

        position = self.portfolios[account_id][currency]
        old_volume = position["volume"]

        # Update volume
        position["volume"] += volume

        # Update average cost
        if position["volume"] != 0:
            if volume > 0:  # Buy
                total_cost = (old_volume * position["avg_cost"]) + (volume * price)
                position["avg_cost"] = total_cost / position["volume"]
            # For sells, average cost remains the same

        # Update timestamp
        position["last_update"] = timestamp

        # Record transaction
        transaction = {
            "transaction_id": str(uuid.uuid4()),
            "account_id": account_id,
            "currency": currency,
            "volume": volume,
            "price": price,
            "timestamp": timestamp,
            "type": "buy" if volume > 0 else "sell"
        }
        self.transactions.append(transaction)

        return True

    def update_cash(self, account_id: str, currency: str, amount: float) -> bool:
        """Update cash balance for specific currency"""
        if currency not in self.supported_currencies:
            return False

        self.cash_balances[account_id][currency] += amount
        return True

    def get_position(self, account_id: str, currency: str) -> Dict:
        """Get position for specific currency"""
        return self.portfolios[account_id][currency].copy()

    def get_cash_balance(self, account_id: str, currency: str) -> float:
        """Get cash balance for specific currency"""
        return self.cash_balances[account_id][currency]

    def get_account_positions(self, account_id: str) -> Dict:
        """Get all positions for an account"""
        return dict(self.portfolios[account_id])

    def get_account_cash(self, account_id: str) -> Dict:
        """Get all cash balances for an account"""
        return dict(self.cash_balances[account_id])

    def get_total_volume(self, account_id: str, currency: str) -> float:
        """Get total volume (position + cash) for a currency"""
        position_volume = self.portfolios[account_id][currency]["volume"]
        cash_volume = self.cash_balances[account_id][currency]
        return position_volume + cash_volume

    def calculate_unrealized_pnl(self, account_id: str, currency: str, 
                                current_price: float) -> float:
        """Calculate unrealized P&L for a currency position"""
        position = self.portfolios[account_id][currency]
        if position["volume"] == 0:
            return 0.0

        market_value = position["volume"] * current_price
        cost_basis = position["volume"] * position["avg_cost"]
        unrealized_pnl = market_value - cost_basis

        # Update position
        position["market_value"] = market_value
        position["unrealized_pnl"] = unrealized_pnl

        return unrealized_pnl

    def get_transaction_history(self, account_id: str = None, 
                               currency: str = None, limit: int = 100) -> List[Dict]:
        """Get transaction history with optional filters"""
        transactions = self.transactions

        if account_id:
            transactions = [t for t in transactions if t["account_id"] == account_id]

        if currency:
            transactions = [t for t in transactions if t["currency"] == currency]

        # Sort by timestamp (newest first) and limit
        transactions.sort(key=lambda x: x["timestamp"], reverse=True)
        return transactions[:limit]

    def get_portfolio_summary(self, account_id: str) -> Dict:
        """Get portfolio summary for an account"""
        positions = self.get_account_positions(account_id)
        cash_balances = self.get_account_cash(account_id)

        summary = {
            "account_id": account_id,
            "total_currencies": len(set(list(positions.keys()) + list(cash_balances.keys()))),
            "total_positions": len(positions),
            "total_cash_currencies": len(cash_balances),
            "last_update": datetime.now()
        }

        return summary

import asyncio
import aiohttp
from typing import Dict, Optional, List
from datetime import datetime, timedelta
import json

class ForexService:
    def __init__(self, base_currency: str = "USD", cache_duration: int = 60):
        self.base_currency = base_currency
        self.cache_duration = cache_duration

        # Rate storage
        self.rates = {}
        self.rate_cache = {}
        self.last_update = {}

        # Rate providers
        self.providers = {
            "crypto": ["binance", "coinbase", "kraken"],
            "forex": ["forex_api", "alpha_vantage", "fixer"],
            "stablecoins": ["internal"]
        }

        # Default rates (fallback)
        self.default_rates = {
            "USD": 1.0,
            "EUR": 0.85,
            "GBP": 0.73,
            "JPY": 110.0,
            "HKD": 7.8,
            "CNY": 6.4,
            "SGD": 1.35,
            "BTC": 65000.0,
            "ETH": 3200.0,
            "USDT": 1.0,
            "USDC": 1.0,
            "BNB": 400.0,
            "ADA": 1.2,
            "DOT": 25.0
        }

        # Initialize rates
        self._initialize_rates()

    def _initialize_rates(self):
        """Initialize rates with default values"""
        for currency, rate in self.default_rates.items():
            self.rates[currency] = rate
            self.last_update[currency] = datetime.now()

    async def get_rate(self, currency: str, base_currency: str = None) -> float:
        """Get exchange rate for currency pair"""
        if base_currency is None:
            base_currency = self.base_currency

        if currency == base_currency:
            return 1.0

        # Check cache
        cache_key = f"{currency}_{base_currency}"
        if self._is_rate_cached(cache_key):
            return self.rate_cache[cache_key]["rate"]

        # Get rate from providers
        rate = await self._fetch_rate(currency, base_currency)

        # Cache rate
        self._cache_rate(cache_key, rate)

        return rate

    async def _fetch_rate(self, currency: str, base_currency: str) -> float:
        """Fetch rate from providers"""
        # Try crypto providers first for crypto currencies
        if currency in ["BTC", "ETH", "BNB", "ADA", "DOT"]:
            rate = await self._fetch_crypto_rate(currency, base_currency)
            if rate > 0:
                return rate

        # Try forex providers for fiat currencies
        if currency in ["EUR", "GBP", "JPY", "HKD", "CNY", "SGD"]:
            rate = await self._fetch_forex_rate(currency, base_currency)
            if rate > 0:
                return rate

        # Try stablecoin providers
        if currency in ["USDT", "USDC"]:
            rate = await self._fetch_stablecoin_rate(currency, base_currency)
            if rate > 0:
                return rate

        # Fallback to default rate
        return self._calculate_default_rate(currency, base_currency)

    async def _fetch_crypto_rate(self, currency: str, base_currency: str) -> float:
        """Fetch crypto rate from providers"""
        for provider in self.providers["crypto"]:
            try:
                if provider == "binance":
                    rate = await self._fetch_binance_rate(currency, base_currency)
                    if rate > 0:
                        return rate
                # Add other crypto providers here
            except Exception as e:
                print(f"Error fetching from {provider}: {e}")
                continue

        return 0.0

    async def _fetch_binance_rate(self, currency: str, base_currency: str) -> float:
        """Fetch rate from Binance API"""
        try:
            symbol = f"{currency}{base_currency}"
            url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"

            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    if response.status == 200:
                        data = await response.json()
                        return float(data["price"])
        except Exception as e:
            print(f"Error fetching Binance rate: {e}")

        return 0.0

    async def _fetch_forex_rate(self, currency: str, base_currency: str) -> float:
        """Fetch forex rate from providers"""
        # Implementation for forex rate providers
        # This would integrate with forex API providers
        return 0.0

    async def _fetch_stablecoin_rate(self, currency: str, base_currency: str) -> float:
        """Fetch stablecoin rate"""
        # Stablecoins are typically pegged to USD
        if base_currency == "USD":
            return 1.0
        else:
            # Convert through USD
            usd_rate = await self.get_rate(currency, "USD")
            base_usd_rate = await self.get_rate(base_currency, "USD")
            return usd_rate / base_usd_rate if base_usd_rate > 0 else 0.0

    def _calculate_default_rate(self, currency: str, base_currency: str) -> float:
        """Calculate rate using default values"""
        if currency in self.default_rates and base_currency in self.default_rates:
            currency_rate = self.default_rates[currency]
            base_rate = self.default_rates[base_currency]
            return currency_rate / base_rate if base_rate > 0 else 0.0
        return 0.0

    def _is_rate_cached(self, cache_key: str) -> bool:
        """Check if rate is cached and valid"""
        if cache_key not in self.rate_cache:
            return False

        cache_time = self.rate_cache[cache_key]["timestamp"]
        return (datetime.now() - cache_time).seconds < self.cache_duration

    def _cache_rate(self, cache_key: str, rate: float):
        """Cache exchange rate"""
        self.rate_cache[cache_key] = {
            "rate": rate,
            "timestamp": datetime.now()
        }

    async def update_all_rates(self):
        """Update all exchange rates"""
        currencies = list(self.default_rates.keys())

        for currency in currencies:
            if currency != self.base_currency:
                rate = await self.get_rate(currency, self.base_currency)
                if rate > 0:
                    self.rates[currency] = rate
                    self.last_update[currency] = datetime.now()

    def get_rate_history(self, currency: str, base_currency: str = None, 
                        hours: int = 24) -> List[Dict]:
        """Get rate history for currency pair"""
        # This would implement rate history tracking
        # For now, return empty list
        return []

    def get_rate_summary(self) -> Dict:
        """Get summary of all rates"""
        return {
            "base_currency": self.base_currency,
            "total_currencies": len(self.rates),
            "last_update": max(self.last_update.values()) if self.last_update else None,
            "rates": self.rates.copy()
        }

from typing import Dict, List, Optional, Tuple
from datetime import datetime
import asyncio

class MultiCurrencyValuation:
    def __init__(self, portfolio, forex_service, base_currency: str = "USD"):
        self.portfolio = portfolio
        self.forex_service = forex_service
        self.base_currency = base_currency

    async def calculate_account_nav(self, account_id: str) -> Dict:
        """Calculate Net Asset Value for an account"""
        positions = self.portfolio.get_account_positions(account_id)
        cash_balances = self.portfolio.get_account_cash(account_id)

        nav_breakdown = {
            "positions": {},
            "cash": {},
            "total_nav": 0.0,
            "currency_exposure": {}
        }

        # Calculate position values
        for currency, position in positions.items():
            if position["volume"] != 0:
                rate = await self.forex_service.get_rate(currency, self.base_currency)
                market_value = position["volume"] * rate
                nav_breakdown["positions"][currency] = {
                    "volume": position["volume"],
                    "avg_cost": position["avg_cost"],
                    "market_value": market_value,
                    "unrealized_pnl": position["unrealized_pnl"],
                    "rate": rate
                }
                nav_breakdown["total_nav"] += market_value

        # Calculate cash values
        for currency, cash_amount in cash_balances.items():
            if cash_amount != 0:
                rate = await self.forex_service.get_rate(currency, self.base_currency)
                cash_value = cash_amount * rate
                nav_breakdown["cash"][currency] = {
                    "amount": cash_amount,
                    "value": cash_value,
                    "rate": rate
                }
                nav_breakdown["total_nav"] += cash_value

        # Calculate currency exposure
        nav_breakdown["currency_exposure"] = self._calculate_currency_exposure(
            nav_breakdown["positions"], nav_breakdown["cash"], nav_breakdown["total_nav"]
        )

        return nav_breakdown

    def _calculate_currency_exposure(self, positions: Dict, cash: Dict, 
                                   total_nav: float) -> Dict:
        """Calculate currency exposure percentages"""
        exposure = {}

        # Calculate position exposure
        for currency, pos_data in positions.items():
            exposure[currency] = pos_data["market_value"] / total_nav if total_nav > 0 else 0

        # Calculate cash exposure
        for currency, cash_data in cash.items():
            if currency in exposure:
                exposure[currency] += cash_data["value"] / total_nav if total_nav > 0 else 0
            else:
                exposure[currency] = cash_data["value"] / total_nav if total_nav > 0 else 0

        return exposure

    async def calculate_portfolio_nav(self, account_ids: List[str]) -> Dict:
        """Calculate total NAV across multiple accounts"""
        total_nav = 0.0
        account_navs = {}
        portfolio_exposure = {}

        for account_id in account_ids:
            account_nav = await self.calculate_account_nav(account_id)
            account_navs[account_id] = account_nav
            total_nav += account_nav["total_nav"]

        # Calculate portfolio-wide currency exposure
        if total_nav > 0:
            for account_nav in account_navs.values():
                for currency, exposure in account_nav["currency_exposure"].items():
                    if currency not in portfolio_exposure:
                        portfolio_exposure[currency] = 0
                    portfolio_exposure[currency] += exposure * (account_nav["total_nav"] / total_nav)

        return {
            "total_nav": total_nav,
            "account_count": len(account_ids),
            "account_navs": account_navs,
            "portfolio_exposure": portfolio_exposure,
            "base_currency": self.base_currency,
            "timestamp": datetime.now()
        }

    async def calculate_pnl(self, account_id: str, start_date: datetime, 
                           end_date: datetime) -> Dict:
        """Calculate P&L for a specific period"""
        # Get transactions in the period
        transactions = self.portfolio.get_transaction_history(account_id)
        period_transactions = [
            t for t in transactions 
            if start_date <= t["timestamp"] <= end_date
        ]

        pnl_breakdown = {
            "realized_pnl": {},
            "unrealized_pnl": {},
            "total_pnl": {},
            "currency_breakdown": {}
        }

        # Calculate realized P&L from transactions
        for transaction in period_transactions:
            currency = transaction["currency"]
            if currency not in pnl_breakdown["realized_pnl"]:
                pnl_breakdown["realized_pnl"][currency] = 0.0

            if transaction["type"] == "sell":
                # Calculate realized P&L for sell transactions
                position = self.portfolio.get_position(account_id, currency)
                if position["avg_cost"] > 0:
                    realized_pnl = (transaction["price"] - position["avg_cost"]) * abs(transaction["volume"])
                    pnl_breakdown["realized_pnl"][currency] += realized_pnl

        # Calculate unrealized P&L for current positions
        positions = self.portfolio.get_account_positions(account_id)
        for currency, position in positions.items():
            if position["volume"] != 0:
                rate = await self.forex_service.get_rate(currency, self.base_currency)
                current_value = position["volume"] * rate
                cost_basis = position["volume"] * position["avg_cost"] * rate
                unrealized_pnl = current_value - cost_basis

                pnl_breakdown["unrealized_pnl"][currency] = unrealized_pnl

        # Calculate total P&L
        for currency in set(list(pnl_breakdown["realized_pnl"].keys()) + 
                           list(pnl_breakdown["unrealized_pnl"].keys())):
            realized = pnl_breakdown["realized_pnl"].get(currency, 0.0)
            unrealized = pnl_breakdown["unrealized_pnl"].get(currency, 0.0)
            pnl_breakdown["total_pnl"][currency] = realized + unrealized

        return pnl_breakdown

    async def get_currency_risk_metrics(self, account_id: str) -> Dict:
        """Calculate currency risk metrics"""
        nav_breakdown = await self.calculate_account_nav(account_id)
        exposure = nav_breakdown["currency_exposure"]

        risk_metrics = {
            "currency_concentration": max(exposure.values()) if exposure else 0,
            "currency_diversification": len(exposure),
            "base_currency_exposure": exposure.get(self.base_currency, 0),
            "foreign_currency_exposure": sum(v for k, v in exposure.items() 
                                           if k != self.base_currency),
            "high_exposure_currencies": [k for k, v in exposure.items() if v > 0.2]
        }

        return risk_metrics

{
  "account_id": "acc_12345",
  "currency": "BTC",
  "position_data": {
    "volume": 2.5,
    "avg_cost": 45000.0,
    "market_value": 162500.0,
    "unrealized_pnl": 22500.0,
    "last_update": "2024-12-20T10:30:15.123Z"
  },
  "cash_balance": 1000.0,
  "total_volume": 2.5
}

{
  "currency_pair": "BTC_USD",
  "rate": 65000.0,
  "timestamp": "2024-12-20T10:30:15.123Z",
  "provider": "binance",
  "bid": 64995.0,
  "ask": 65005.0,
  "volume_24h": 1500000000.0
}

{
  "account_id": "acc_12345",
  "total_nav_usd": 250000.0,
  "base_currency": "USD",
  "currency_breakdown": {
    "USD": {
      "value": 50000.0,
      "percentage": 20.0
    },
    "BTC": {
      "value": 162500.0,
      "percentage": 65.0
    },
    "ETH": {
      "value": 37500.0,
      "percentage": 15.0
    }
  },
  "currency_exposure": {
    "USD": 0.2,
    "BTC": 0.65,
    "ETH": 0.15
  },
  "risk_metrics": {
    "currency_concentration": 0.65,
    "currency_diversification": 3,
    "foreign_currency_exposure": 0.8
  },
  "timestamp": "2024-12-20T10:30:15.123Z"
}

Market Data → Exchange Rate Updates → Rate Validation → Rate Cache
    ↓
Portfolio Updates → Position Reconciliation → NAV Calculation → Risk Analysis
    ↓
Real-time Valuation → Currency Exposure → Performance Tracking → Risk Monitoring

GET    /api/v1/portfolio/{account_id}/positions     # Get all positions
GET    /api/v1/portfolio/{account_id}/cash          # Get cash balances
POST   /api/v1/portfolio/{account_id}/position      # Update position
POST   /api/v1/portfolio/{account_id}/cash          # Update cash balance
GET    /api/v1/portfolio/{account_id}/summary       # Get portfolio summary

GET    /api/v1/portfolio/{account_id}/nav           # Get account NAV
GET    /api/v1/portfolio/nav                        # Get portfolio NAV
GET    /api/v1/portfolio/{account_id}/pnl           # Get P&L calculation
GET    /api/v1/portfolio/{account_id}/risk          # Get risk metrics

GET    /api/v1/forex/rate/{currency_pair}           # Get exchange rate
GET    /api/v1/forex/rates                           # Get all rates
GET    /api/v1/forex/history/{currency_pair}        # Get rate history
POST   /api/v1/forex/update                          # Force rate update

/ws/portfolio/{account_id}/nav                       # Real-time NAV updates
/ws/portfolio/{account_id}/positions                 # Real-time position updates
/ws/forex/rates                                       # Real-time rate updates
/ws/portfolio/risk                                    # Real-time risk updates

Multi-Currency Portfolio → Position Management → NAV Calculation → Risk Analysis