Skip to content

API Reference

Auto-generated documentation from source code docstrings.

Domain Layer

Value Objects

portfolio_risk_engine.domain.value_objects.ticker

Ticker dataclass

Financial asset ticker symbol.

Represents the identifier used on exchanges (e.g., AAPL, MSFT, BRK.B).

The ticker is normalized to uppercase and validated using a simple regex.

Source code in src/portfolio_risk_engine/domain/value_objects/ticker.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@dataclass(frozen=True)
class Ticker:
    """
    Financial asset ticker symbol.

    Represents the identifier used on exchanges
    (e.g., AAPL, MSFT, BRK.B).

    The ticker is normalized to uppercase and validated
    using a simple regex.
    """

    value: str

    def __post_init__(self) -> None:
        v = self.value.strip().upper()

        if not v:
            raise ValueError("Ticker cannot be empty.")

        if not re.fullmatch(r"[A-Z0-9\.\-]{1,10}", v):
            raise ValueError(f"Invalid ticker: {v}")

        object.__setattr__(self, "value", v)

portfolio_risk_engine.domain.value_objects.currency

Currency dataclass

Value object representing a currency using a 3-letter ISO code.

The code is normalized to uppercase and validated to ensure it follows the ISO 4217 format (e.g., USD, EUR, JPY).

Source code in src/portfolio_risk_engine/domain/value_objects/currency.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@dataclass(frozen=True)
class Currency:
    """
    Value object representing a currency using a 3-letter ISO code.

    The code is normalized to uppercase and validated to ensure it
    follows the ISO 4217 format (e.g., USD, EUR, JPY).
    """

    code: str

    def __post_init__(self) -> None:
        """
        Normalize and validate the currency code after initialization.
        """
        v = self.code.strip().upper()

        if len(v) != 3 or not v.isalpha():
            raise ValueError("Currency must be a 3-letter ISO code.")

        object.__setattr__(self, "code", v)

__post_init__()

Normalize and validate the currency code after initialization.

Source code in src/portfolio_risk_engine/domain/value_objects/currency.py 
15
16
17
18
19
20
21
22
23
24
def __post_init__(self) -> None:
    """
    Normalize and validate the currency code after initialization.
    """
    v = self.code.strip().upper()

    if len(v) != 3 or not v.isalpha():
        raise ValueError("Currency must be a 3-letter ISO code.")

    object.__setattr__(self, "code", v)

portfolio_risk_engine.domain.value_objects.weight

Weight dataclass

Portfolio weight value object.

Represents the allocation of an asset in a portfolio.

The value must be between 0 and 1 inclusive.

Source code in src/portfolio_risk_engine/domain/value_objects/weight.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
@dataclass(frozen=True)
class Weight:
    """
    Portfolio weight value object.

    Represents the allocation of an asset in a portfolio.

    The value must be between 0 and 1 inclusive.
    """

    value: float

    def __post_init__(self) -> None:
        """
        Validate the weight value.
        """
        if not (0.0 <= self.value <= 1.0):
            raise ValueError("Weight must be between 0 and 1.")

__post_init__()

Validate the weight value.

Source code in src/portfolio_risk_engine/domain/value_objects/weight.py 
16
17
18
19
20
21
def __post_init__(self) -> None:
    """
    Validate the weight value.
    """
    if not (0.0 <= self.value <= 1.0):
        raise ValueError("Weight must be between 0 and 1.")

portfolio_risk_engine.domain.value_objects.date_range

Models

portfolio_risk_engine.domain.models.asset

portfolio_risk_engine.domain.models.position

portfolio_risk_engine.domain.models.portfolio

portfolio_risk_engine.domain.models.historical_prices

portfolio_risk_engine.domain.models.historical_returns

portfolio_risk_engine.domain.models.market_parameters

portfolio_risk_engine.domain.models.gbm_model

MultivariateGBM dataclass

Multivariate Geometric Brownian Motion model with pre-computed Cholesky factor.

Source code in src/portfolio_risk_engine/domain/models/gbm_model.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@dataclass(frozen=True)
class MultivariateGBM:
    """Multivariate Geometric Brownian Motion model with pre-computed Cholesky factor."""

    market_parameters: MarketParameters
    cholesky_factor: tuple[tuple[float, ...], ...]

    def __post_init__(self) -> None:
        n = len(self.market_parameters.tickers)

        if len(self.cholesky_factor) != n:
            raise ValueError(
                f"cholesky_factor rows ({len(self.cholesky_factor)}) "
                f"must match number of tickers ({n})."
            )

        for i, row in enumerate(self.cholesky_factor):
            if len(row) != n:
                raise ValueError(
                    f"cholesky_factor row {i} has {len(row)} columns, expected {n}."
                )

portfolio_risk_engine.domain.models.simulation_result

portfolio_risk_engine.domain.models.portfolio_risk_metrics

PortfolioRiskMetrics dataclass

Risk metrics computed from Monte Carlo simulation. VaR and ES use loss-positive convention.

Source code in src/portfolio_risk_engine/domain/models/portfolio_risk_metrics.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
@dataclass(frozen=True)
class PortfolioRiskMetrics:
    """Risk metrics computed from Monte Carlo simulation. VaR and ES use loss-positive convention."""

    mean_return: float
    volatility: float
    var_95: float
    var_99: float
    es_95: float
    es_99: float

Ports

portfolio_risk_engine.domain.ports.market_data_provider

portfolio_risk_engine.domain.ports.monte_carlo_engine

Services

portfolio_risk_engine.domain.services.cholesky

cholesky(matrix)

Pure-Python Cholesky decomposition. Returns lower-triangular L such that L @ L^T == matrix.

Source code in src/portfolio_risk_engine/domain/services/cholesky.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def cholesky(
    matrix: tuple[tuple[float, ...], ...],
) -> tuple[tuple[float, ...], ...]:
    """Pure-Python Cholesky decomposition. Returns lower-triangular L such that L @ L^T == matrix."""
    n = len(matrix)
    if n == 0:
        raise ValueError("Matrix must be non-empty.")

    for row in matrix:
        if len(row) != n:
            raise ValueError("Matrix must be square.")

    lower: list[list[float]] = [[0.0] * n for _ in range(n)]

    for i in range(n):
        for j in range(i + 1):
            s = sum(lower[i][k] * lower[j][k] for k in range(j))
            if i == j:
                val = matrix[i][i] - s
                if val <= 0:
                    raise ValueError("Matrix is not positive definite.")
                lower[i][j] = math.sqrt(val)
            else:
                lower[i][j] = (matrix[i][j] - s) / lower[j][j]

    return tuple(tuple(row) for row in lower)

Application Layer

Use Cases

portfolio_risk_engine.application.use_cases.fetch_market_data

portfolio_risk_engine.application.use_cases.compute_log_returns

portfolio_risk_engine.application.use_cases.estimate_market_parameters

portfolio_risk_engine.application.use_cases.run_monte_carlo

portfolio_risk_engine.application.use_cases.compute_portfolio_risk

Infrastructure Layer

Market Data

portfolio_risk_engine.infrastructure.market_data.yahoo_finance_market_data_provider

Simulation Engines

portfolio_risk_engine.infrastructure.simulation.cpu_monte_carlo_engine

CpuMonteCarloEngine

NumPy-based CPU implementation of the MonteCarloEngine port.

Source code in src/portfolio_risk_engine/infrastructure/simulation/cpu_monte_carlo_engine.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class CpuMonteCarloEngine:
    """NumPy-based CPU implementation of the MonteCarloEngine port."""

    def __init__(self, seed: int | None = None) -> None:
        self._rng = np.random.default_rng(seed)

    def simulate(
        self,
        model: MultivariateGBM,
        initial_prices: tuple[float, ...],
        num_simulations: int,
        time_horizon_days: int,
    ) -> MonteCarloSimulationResult:
        params = model.market_parameters
        n = len(params.tickers)
        T = time_horizon_days / params.annualization_factor

        drift = np.array(params.drift_vector)
        variances = np.array([params.covariance_matrix[i][i] for i in range(n)])
        S0 = np.array(initial_prices)
        L = np.array(model.cholesky_factor)

        # Independent standard normals: shape (n, num_simulations)
        Z = self._rng.standard_normal((n, num_simulations))

        # Correlated normals via Cholesky: shape (n, num_simulations)
        correlated_Z = L @ Z

        # GBM terminal price: S_T = S_0 * exp((mu - sigma^2/2)*T + sqrt(T)*L*Z)
        drift_adj = (drift - 0.5 * variances) * T
        log_returns = drift_adj[:, np.newaxis] + np.sqrt(T) * correlated_Z
        terminal_prices_array = S0[:, np.newaxis] * np.exp(log_returns)

        terminal_prices = {}
        for i, ticker in enumerate(params.tickers):
            terminal_prices[ticker] = tuple(terminal_prices_array[i].tolist())

        return MonteCarloSimulationResult(
            tickers=params.tickers,
            initial_prices=initial_prices,
            terminal_prices=terminal_prices,
            num_simulations=num_simulations,
            time_horizon_days=time_horizon_days,
        )

portfolio_risk_engine.infrastructure.simulation.gpu_monte_carlo_engine

GpuMonteCarloEngine

CuPy-based GPU implementation of the MonteCarloEngine port.

Source code in src/portfolio_risk_engine/infrastructure/simulation/gpu_monte_carlo_engine.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class GpuMonteCarloEngine:
    """CuPy-based GPU implementation of the MonteCarloEngine port."""

    def __init__(self, seed: int | None = None) -> None:
        if not _CUPY_AVAILABLE:
            raise RuntimeError(
                "CuPy is not installed. Install with: pip install cupy-cuda12x"
            )
        try:
            cp.cuda.runtime.getDeviceCount()
        except (RuntimeError, cp.cuda.runtime.CUDARuntimeError) as e:
            raise RuntimeError(f"No CUDA-capable GPU detected: {e}") from e
        self._seed = seed

    def simulate(
        self,
        model: MultivariateGBM,
        initial_prices: tuple[float, ...],
        num_simulations: int,
        time_horizon_days: int,
    ) -> MonteCarloSimulationResult:
        params = model.market_parameters
        n = len(params.tickers)
        T = time_horizon_days / params.annualization_factor

        # Transfer to GPU
        drift = cp.array(params.drift_vector, dtype=cp.float64)
        variances = cp.array(
            [params.covariance_matrix[i][i] for i in range(n)], dtype=cp.float64
        )
        S0 = cp.array(initial_prices, dtype=cp.float64)
        L = cp.array(model.cholesky_factor, dtype=cp.float64)

        # Generate random normals on GPU
        rng = cp.random.RandomState(self._seed)
        Z = rng.standard_normal((n, num_simulations), dtype=cp.float64)

        # Correlated normals via Cholesky: shape (n, num_simulations)
        correlated_Z = L @ Z

        # GBM terminal price: S_T = S_0 * exp((mu - sigma^2/2)*T + sqrt(T)*L*Z)
        drift_adj = (drift - 0.5 * variances) * T
        log_returns = drift_adj[:, cp.newaxis] + cp.sqrt(T) * correlated_Z
        terminal_prices_array = S0[:, cp.newaxis] * cp.exp(log_returns)

        # Transfer back to CPU and convert to pure Python types
        terminal_prices_cpu = cp.asnumpy(terminal_prices_array)

        terminal_prices = {}
        for i, ticker in enumerate(params.tickers):
            terminal_prices[ticker] = tuple(terminal_prices_cpu[i].tolist())

        return MonteCarloSimulationResult(
            tickers=params.tickers,
            initial_prices=initial_prices,
            terminal_prices=terminal_prices,
            num_simulations=num_simulations,
            time_horizon_days=time_horizon_days,
        )

portfolio_risk_engine.infrastructure.simulation.gpu_accelerated_pipeline

End-to-end GPU simulation + risk computation.

Keeps all data on GPU throughout the pipeline. Only 6 scalar floats are transferred back to CPU. No intermediate tuple conversion.

Architecture: this is an infrastructure-level optimization. Domain models and application use cases are not modified.

GpuAcceleratedPipeline

Simulation + risk in a single GPU pass — zero tuple allocation.

Source code in src/portfolio_risk_engine/infrastructure/simulation/gpu_accelerated_pipeline.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class GpuAcceleratedPipeline:
    """Simulation + risk in a single GPU pass — zero tuple allocation."""

    def __init__(self, seed: int | None = None) -> None:
        if not _CUPY_AVAILABLE:
            raise RuntimeError(
                "CuPy is not installed. Install with: pip install cupy-cuda12x"
            )
        try:
            cp.cuda.runtime.getDeviceCount()
        except (RuntimeError, cp.cuda.runtime.CUDARuntimeError) as e:
            raise RuntimeError(f"No CUDA-capable GPU detected: {e}") from e
        self._seed = seed

    def run(
        self,
        market_params: MarketParameters,
        initial_prices: tuple[float, ...],
        weights: tuple[float, ...],
        num_simulations: int,
        time_horizon_days: int,
    ) -> PortfolioRiskMetrics:
        n = len(market_params.tickers)
        T = time_horizon_days / market_params.annualization_factor

        # Transfer inputs to GPU (small, N-sized vectors + NxN matrix)
        drift = cp.array(market_params.drift_vector, dtype=cp.float64)
        cov = cp.array(market_params.covariance_matrix, dtype=cp.float64)
        variances = cp.array(
            [market_params.covariance_matrix[i][i] for i in range(n)], dtype=cp.float64
        )
        S0 = cp.array(initial_prices, dtype=cp.float64)
        w = cp.array(weights, dtype=cp.float64)

        # Cholesky on GPU
        L = cp.linalg.cholesky(cov)

        # --- Simulation (stays on GPU) ---
        rng = cp.random.RandomState(self._seed)
        Z = rng.standard_normal((n, num_simulations), dtype=cp.float64)
        correlated_Z = L @ Z

        drift_adj = (drift - 0.5 * variances) * T
        log_returns = drift_adj[:, cp.newaxis] + cp.sqrt(T) * correlated_Z
        terminal_prices = S0[:, cp.newaxis] * cp.exp(log_returns)

        # --- Risk computation (stays on GPU) ---
        asset_returns = terminal_prices / S0[:, cp.newaxis] - 1.0  # (n, num_sims)
        portfolio_returns = w @ asset_returns  # (num_sims,)
        losses = -portfolio_returns

        mean_return = float(cp.mean(portfolio_returns))
        volatility = float(cp.std(portfolio_returns, ddof=1))
        var_95 = float(cp.percentile(losses, 95))
        var_99 = float(cp.percentile(losses, 99))
        es_95 = float(cp.mean(losses[losses >= var_95]))
        es_99 = float(cp.mean(losses[losses >= var_99]))

        return PortfolioRiskMetrics(
            mean_return=mean_return,
            volatility=volatility,
            var_95=var_95,
            var_99=var_99,
            es_95=es_95,
            es_99=es_99,
        )

    def run_with_summary(
        self,
        market_params: MarketParameters,
        initial_prices: tuple[float, ...],
        weights: tuple[float, ...],
        num_simulations: int,
        time_horizon_days: int,
    ) -> tuple[PortfolioRiskMetrics, dict[str, float]]:
        """Like run(), but also returns per-ticker mean terminal prices."""
        n = len(market_params.tickers)
        T = time_horizon_days / market_params.annualization_factor

        drift = cp.array(market_params.drift_vector, dtype=cp.float64)
        cov = cp.array(market_params.covariance_matrix, dtype=cp.float64)
        variances = cp.array(
            [market_params.covariance_matrix[i][i] for i in range(n)], dtype=cp.float64
        )
        S0 = cp.array(initial_prices, dtype=cp.float64)
        w = cp.array(weights, dtype=cp.float64)

        L = cp.linalg.cholesky(cov)

        rng = cp.random.RandomState(self._seed)
        Z = rng.standard_normal((n, num_simulations), dtype=cp.float64)
        correlated_Z = L @ Z

        drift_adj = (drift - 0.5 * variances) * T
        log_returns = drift_adj[:, cp.newaxis] + cp.sqrt(T) * correlated_Z
        terminal_prices = S0[:, cp.newaxis] * cp.exp(log_returns)

        # Per-ticker summary (N floats, cheap)
        mean_terminal = cp.mean(terminal_prices, axis=1)
        summary = {}
        for i, ticker in enumerate(market_params.tickers):
            summary[ticker.value] = float(mean_terminal[i])

        # Risk
        asset_returns = terminal_prices / S0[:, cp.newaxis] - 1.0
        portfolio_returns = w @ asset_returns
        losses = -portfolio_returns

        mean_return = float(cp.mean(portfolio_returns))
        volatility = float(cp.std(portfolio_returns, ddof=1))
        var_95 = float(cp.percentile(losses, 95))
        var_99 = float(cp.percentile(losses, 99))
        es_95 = float(cp.mean(losses[losses >= var_95]))
        es_99 = float(cp.mean(losses[losses >= var_99]))

        metrics = PortfolioRiskMetrics(
            mean_return=mean_return,
            volatility=volatility,
            var_95=var_95,
            var_99=var_99,
            es_95=es_95,
            es_99=es_99,
        )
        return metrics, summary

run_with_summary(market_params, initial_prices, weights, num_simulations, time_horizon_days)

Like run(), but also returns per-ticker mean terminal prices.

Source code in src/portfolio_risk_engine/infrastructure/simulation/gpu_accelerated_pipeline.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def run_with_summary(
    self,
    market_params: MarketParameters,
    initial_prices: tuple[float, ...],
    weights: tuple[float, ...],
    num_simulations: int,
    time_horizon_days: int,
) -> tuple[PortfolioRiskMetrics, dict[str, float]]:
    """Like run(), but also returns per-ticker mean terminal prices."""
    n = len(market_params.tickers)
    T = time_horizon_days / market_params.annualization_factor

    drift = cp.array(market_params.drift_vector, dtype=cp.float64)
    cov = cp.array(market_params.covariance_matrix, dtype=cp.float64)
    variances = cp.array(
        [market_params.covariance_matrix[i][i] for i in range(n)], dtype=cp.float64
    )
    S0 = cp.array(initial_prices, dtype=cp.float64)
    w = cp.array(weights, dtype=cp.float64)

    L = cp.linalg.cholesky(cov)

    rng = cp.random.RandomState(self._seed)
    Z = rng.standard_normal((n, num_simulations), dtype=cp.float64)
    correlated_Z = L @ Z

    drift_adj = (drift - 0.5 * variances) * T
    log_returns = drift_adj[:, cp.newaxis] + cp.sqrt(T) * correlated_Z
    terminal_prices = S0[:, cp.newaxis] * cp.exp(log_returns)

    # Per-ticker summary (N floats, cheap)
    mean_terminal = cp.mean(terminal_prices, axis=1)
    summary = {}
    for i, ticker in enumerate(market_params.tickers):
        summary[ticker.value] = float(mean_terminal[i])

    # Risk
    asset_returns = terminal_prices / S0[:, cp.newaxis] - 1.0
    portfolio_returns = w @ asset_returns
    losses = -portfolio_returns

    mean_return = float(cp.mean(portfolio_returns))
    volatility = float(cp.std(portfolio_returns, ddof=1))
    var_95 = float(cp.percentile(losses, 95))
    var_99 = float(cp.percentile(losses, 99))
    es_95 = float(cp.mean(losses[losses >= var_95]))
    es_99 = float(cp.mean(losses[losses >= var_99]))

    metrics = PortfolioRiskMetrics(
        mean_return=mean_return,
        volatility=volatility,
        var_95=var_95,
        var_99=var_99,
        es_95=es_95,
        es_99=es_99,
    )
    return metrics, summary