Enterprise Risk Management System

1┌─────────────────────────────────────────────────────┐
2│                  Risk Dashboard                     │
3│              (React + TypeScript)                   │
4└──────────────────┬──────────────────────────────────┘
5                   │ REST API
6┌──────────────────▼──────────────────────────────────┐
7│           API Gateway (Kong)                        │
8└──────────────────┬──────────────────────────────────┘
9                   │ gRPC
10        ┌──────────┼───────────┬───────────┐
11        │          │           │           │
12    ┌───▼───┐  ┌───▼────┐  ┌───▼────┐  ┌───▼────┐
13    │Credit │  │ Market │  │ Liqui- │  │ Opera- │
14    │ Risk  │  │  Risk  │  │ dity   │  │ tional │
15    │Service│  │Service │  │Service │  │  Risk  │
16    │(Rust) │  │(Python)│  │(Python)│  │(Python)│
17    └───┬───┘  └───┬────┘  └───┬────┘  └ ──┬────┘
18        │          │           │           │
19        └──────────┴───────────┴───────────┘
20                   │
21            ┌──────▼──────┐
22            │Apache Kafka │
23            │ (Streaming) │
24            └──────┬──────┘
25                   │
26        ┌──────────┴─────────┐
27        │                    │
28   ┌────▼────┐         ┌─────▼─────┐
29   │TimeScale│         │ Redis     │
30   │   DB    │         │ (Cache)   │
31   └─────────┘         └───────────┘
32

1use rayon::prelude::*;
2use serde::{Deserialize, Serialize};
3
4#[derive(Debug, Serialize, Deserialize)]
5struct CreditExposure {
6    counterparty_id: String,
7    exposure: f64,
8    cvа: f64,        // Credit Value Adjustment
9    pd: f64,         // Probability of Default
10    lgd: f64,        // Loss Given Default
11    ead: f64,        // Exposure at Default
12}
13
14impl CreditRiskEngine {
15    /// Calculate portfolio credit risk using parallel processing
16    pub fn calculate_portfolio_risk(
17        &self,
18        positions: &[Position],
19    ) -> PortfolioRisk {
20        // Parallel calculation across positions
21        let exposures: Vec<CreditExposure> = positions
22            .par_iter()
23            .map(|pos| self.calculate_exposure(pos))
24            .collect();
25        
26        // Aggregate with correlation adjustments
27        self.aggregate_with_correlation(exposures)
28    }
29    
30    /// Monte Carlo simulation for tail risk
31    pub fn simulate_tail_risk(
32        &self,
33        scenarios: usize,
34    ) -> TailRiskMetrics {
35        (0..scenarios)
36            .into_par_iter()
37            .map(|_| self.simulate_default_scenario())
38            .fold(
39                || TailRiskMetrics::default(),
40                |acc, result| acc.merge(result),
41            )
42            .reduce(
43                || TailRiskMetrics::default(),
44                |a, b| a.merge(b),
45            )
46    }
47}
48

1import numpy as np
2from scipy import stats
3from typing import Dict, List
4import asyncio
5
6class MarketRiskEngine:
7    """Calculate VaR, Expected Shortfall, and stress scenarios"""
8    
9    async def calculate_var(
10        self,
11        portfolio: Portfolio,
12        confidence: float = 0.99,
13        horizon_days: int = 1
14    ) -> Dict[str, float]:
15        """
16        Calculate Value at Risk using multiple methods:
17        - Historical simulation
18        - Parametric (variance-covariance)
19        - Monte Carlo
20        """
21        # Run calculations in parallel
22        historical, parametric, monte_carlo = await asyncio.gather(
23            self._historical_var(portfolio, confidence),
24            self._parametric_var(portfolio, confidence),
25            self._monte_carlo_var(portfolio, confidence, horizon_days)
26        )
27        
28        return {
29            'historical_var': historical,
30            'parametric_var': parametric,
31            'monte_carlo_var': monte_carlo,
32            'expected_shortfall': self._calculate_es(
33                portfolio, confidence
34            ),
35        }
36    
37    async def stress_test(
38        self,
39        portfolio: Portfolio,
40        scenarios: List[StressScenario]
41    ) -> List[StressResult]:
42        """
43        Apply stress scenarios (2008 crisis, COVID-19, etc.)
44        """
45        tasks = [
46            self._apply_scenario(portfolio, scenario)
47            for scenario in scenarios
48        ]
49        return await asyncio.gather(*tasks)
50    
51    def _monte_carlo_var(
52        self,
53        portfolio: Portfolio,
54        confidence: float,
55        horizon_days: int,
56        num_simulations: int = 100_000
57    ) -> float:
58        """Monte Carlo VaR using variance reduction techniques"""
59        # Use antithetic variates for variance reduction
60        returns = self._simulate_returns_antithetic(
61            portfolio, 
62            num_simulations, 
63            horizon_days
64        )
65        
66        # Calculate portfolio P&L
67        pnl = portfolio.value * returns
68        
69        # VaR is the loss at confidence level
70        var = -np.percentile(pnl, (1 - confidence) * 100)
71        
72        return float(var)
73

1from kafka import KafkaConsumer, KafkaProducer
2from confluent_kafka import Consumer
3import json
4
5class RiskEventProcessor:
6    """Process risk events in real-time"""
7    
8    def __init__(self):
9        self.consumer = KafkaConsumer(
10            'trades',
11            'positions',
12            'market-data',
13            bootstrap_servers=['kafka:9092'],
14            group_id='risk-processor',
15            enable_auto_commit=True,
16            auto_offset_reset='latest',
17            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
18        )
19        
20        self.producer = KafkaProducer(
21            bootstrap_servers=['kafka:9092'],
22            value_serializer=lambda v: json.dumps(v).encode('utf-8')
23        )
24    
25    async def process_events(self):
26        """Process events and update risk metrics in real-time"""
27        async for message in self.consumer:
28            event = message.value
29            
30            # Calculate incremental risk impact
31            risk_delta = await self.calculate_risk_delta(event)
32            
33            # Update risk metrics
34            await self.update_risk_metrics(risk_delta)
35            
36            # Publish updated risk
37            self.producer.send('risk-updates', risk_delta)
38            
39            # Check limits and alert if breached
40            if self.is_limit_breached(risk_delta):
41                await self.send_alert(risk_delta)
42

1-- Hypertable for intraday risk snapshots
2CREATE TABLE risk_snapshots (
3    timestamp TIMESTAMPTZ NOT NULL,
4    desk_id VARCHAR(50) NOT NULL,
5    risk_type VARCHAR(50) NOT NULL,
6    metric_name VARCHAR(100) NOT NULL,
7    value DOUBLE PRECISION,
8    currency VARCHAR(3),
9    metadata JSONB
10);
11
12SELECT create_hypertable(
13    'risk_snapshots',
14    'timestamp',
15    chunk_time_interval => INTERVAL '1 day'
16);
17
18-- Continuous aggregate for hourly risk
19CREATE MATERIALIZED VIEW hourly_risk
20WITH (timescaledb.continuous) AS
21SELECT
22    time_bucket('1 hour', timestamp) AS hour,
23    desk_id,
24    risk_type,
25    metric_name,
26    AVG(value) as avg_value,
27    MAX(value) as max_value,
28    MIN(value) as min_value,
29    STDDEV(value) as std_dev
30FROM risk_snapshots
31GROUP BY hour, desk_id, risk_type, metric_name;
32
33-- Compression policy (older data compressed automatically)
34SELECT add_compression_policy('risk_snapshots', INTERVAL '7 days');
35
36-- Retention policy (keep raw data for 2 years)
37SELECT add_retention_policy('risk_snapshots', INTERVAL '2 years');
38