Async Monte Carlo Portfolio Risk Engine

Introduction

30 Second Summary

Every trading day, risk teams at hedge funds need to answer one question before markets open: "What's the worst that could happen?" They simulate thousands of possible futures to find out.

In this project, you will build an async Monte Carlo portfolio risk engine in Python. It fetches real market data, simulates 10,000 correlated return paths, and outputs a structured risk report with VaR, CVaR, and per-asset risk attribution.

What You'll Build

A multi-asset portfolio risk engine that fetches historical stock prices asynchronously, calibrates a correlated Geometric Brownian Motion model, and runs 10,000 Monte Carlo simulations to compute comprehensive risk metrics.

By the end of this project, you'll have:

• A modular Python package (risk_engine/) with separate modules for data fetching, simulation, risk metrics, and orchestration.
• A YAML configuration file defining portfolio composition, simulation parameters, and risk thresholds.
• An async orchestrator that coordinates the full pipeline from market data ingestion to a formatted console risk report.
• Secret Mission: A pytest test suite that validates stochastic simulation correctness using statistical assertions.

Are there any prerequisites?

You'll need basic Python knowledge (variables, functions, classes) and Python 3.12+ installed on your machine. No finance background is required.

Set Up Project Structure and Models

With your environment ready, it is time to lay the architectural foundation for the risk engine. A professional quantitative system separates concerns into distinct modules so each piece can be tested, replaced, and reasoned about independently.

In this step, you will create the modular package layout, define the configuration that drives the entire pipeline, and build the type-hinted data models that structure information as it flows from raw market data to a final risk report.

In this step, get ready to:

• Create the modular package layout with placeholder module files.
• Define the YAML configuration file with portfolio and simulation parameters.
• Build type-hinted dataclass models and a configuration loader.

Create the package directory and module files

Python treats a folder as a package when it contains an __init__.py file. You will create the risk_engine/ package folder with empty module files for each piece of the pipeline.

• Create the package folder and all module files by running these commands:

mkdir risk_engine
touch risk_engine/__init__.py
touch risk_engine/models.py
touch risk_engine/config.py
touch risk_engine/data.py
touch risk_engine/simulation.py
touch risk_engine/risk.py
touch risk_engine/report.py
touch risk_engine/pipeline.py
touch risk_engine/__main__.py

What does this structure do?

Each file handles one responsibility in the pipeline. models.py holds data shapes, data.py fetches market prices, simulation.py runs Monte Carlo paths, risk.py computes metrics, and pipeline.py orchestrates everything.

The __main__.py file lets you run the entire engine with python -m risk_engine.

• Confirm the files exist by running:

ls risk_engine/

You should see all nine .py files listed in the output.

• Create the requirements.txt file to pin your project dependencies by pasting this content into a new file called requirements.txt in your project root:

numpy==2.4.6
pandas==3.0.3
scipy==1.18.0
aiohttp==3.14.1
yfinance==1.5.1
pyyaml==6.0.2
pytest==9.1.1

Why pin exact versions?

Pinning versions ensures every developer (or your future self) gets the same behaviour. Quantitative code is especially sensitive to numerical differences across library versions.

• Save requirements.txt.

Seeing a 'No such file or directory' error?

Make sure your terminal is inside the monte-carlo-risk-engine/ folder. Run pwd to check your current location.

Help me figure out if I'm in the wrong directory

Write the YAML configuration file

The configuration file is the single source of truth for what the engine does. It defines which stocks to analyse, how much capital to simulate, and how many Monte Carlo paths to generate.

• Create a file called config.yaml in your project root (inside monte-carlo-risk-engine/, next to requirements.txt) and paste the following:

portfolio:
  tickers:
    - AAPL
    - MSFT
    - GOOGL
    - AMZN
    - JPM
  weights:
    - 0.25
    - 0.25
    - 0.20
    - 0.15
    - 0.15
  notional: 1000000

simulation:
  lookback_days: 504
  num_simulations: 10000
  time_horizon_days: 10
  confidence_levels:
    - 0.95
    - 0.99
  random_seed: 42

logging:
  level: INFO

What do these parameters control?

• tickers and weights define a $1M portfolio split across five stocks (weights must sum to 1.0).
• lookback_days: 504 means the engine fetches about 2 years of historical prices for calibration.
• num_simulations: 10000 generates 10,000 correlated return paths for statistical confidence.
• confidence_levels sets the VaR/CVaR thresholds at 95% and 99%.
• random_seed: 42 makes the simulation reproducible across runs.

• Save config.yaml.

YAML indentation error later?

YAML is whitespace-sensitive. Make sure you use spaces (not tabs) for indentation, and that list items are aligned under their parent key.

Help me check for YAML formatting problems

Define the data models and configuration loader

Every piece of data flowing through the pipeline has a defined shape. Dataclasses give you typed, validated containers that catch configuration errors early instead of letting them silently corrupt simulation results downstream.

• Open risk_engine/__init__.py and paste this single line to mark the package:

"""Async Monte Carlo Portfolio Risk Engine."""

• Save __init__.py.
• Open risk_engine/models.py and paste the following imports and the PortfolioConfig dataclass:

"""Data models for the risk engine."""

from dataclasses import dataclass, field
from typing import Any

import numpy as np
import pandas as pd


@dataclass
class PortfolioConfig:
    """Portfolio configuration loaded from YAML."""

    tickers: list[str]
    weights: list[float]
    notional: float
    lookback_days: int
    num_simulations: int
    time_horizon_days: int
    confidence_levels: list[float]
    random_seed: int

    def __post_init__(self) -> None:
        if abs(sum(self.weights) - 1.0) > 1e-6:
            raise ValueError(f"Weights must sum to 1.0, got {sum(self.weights):.6f}")
        if len(self.tickers) != len(self.weights):
            raise ValueError("Tickers and weights must have same length")

What does PortfolioConfig do?

This dataclass holds every parameter the engine needs. The __post_init__ method runs automatically after creation and validates that portfolio weights sum to 1.0 and that each ticker has a corresponding weight.

Catching these errors at config load time prevents a silent miscalculation from running 10,000 simulations on bad inputs.

• Below PortfolioConfig in the same file, add the three output data models:

@dataclass
class ValidationResult:
    """Result of data validation."""

    is_valid: bool
    ticker: str
    issues: list[str] = field(default_factory=list)


@dataclass
class SimulationResult:
    """Output of Monte Carlo simulation."""

    terminal_values: np.ndarray  # shape: (num_simulations,)
    pnl_distribution: np.ndarray  # shape: (num_simulations,)
    elapsed_seconds: float


@dataclass
class RiskMetrics:
    """Computed risk metrics for the portfolio."""

    var_95: float
    var_99: float
    cvar_95: float
    cvar_99: float
    annualized_volatility: float
    annualized_sharpe: float
    max_drawdown: float
    risk_contributions: dict[str, float]

What do these models represent?

• ValidationResult tracks whether fetched market data passes quality checks.
• SimulationResult holds the 10,000 simulated terminal values and P&L distribution.
• RiskMetrics is the final output containing VaR, CVaR, Sharpe ratio, max drawdown, and per-asset risk contributions.

• Save models.py.
• Open risk_engine/config.py and paste the imports and structured logging setup:

"""Configuration loader."""

import logging
import json
from pathlib import Path
from typing import Any

import yaml

from risk_engine.models import PortfolioConfig


def setup_logging(log_level: str = "INFO") -> None:
    """Configure structured JSON logging."""
    class JsonFormatter(logging.Formatter):
        def format(self, record: logging.LogRecord) -> str:
            log_entry = {
                "timestamp": self.formatTime(record),
                "level": record.levelname,
                "module": record.module,
                "message": record.getMessage(),
            }
            if record.exc_info:
                log_entry["exception"] = self.formatException(record.exc_info)
            return json.dumps(log_entry)

    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    logging.root.handlers = [handler]
    logging.root.setLevel(getattr(logging, log_level.upper()))

Why structured JSON logging?

Production risk systems emit logs as JSON so monitoring tools can parse them automatically. Each log line includes a timestamp, severity level, source module, and message in a machine-readable format.

• Below setup_logging in the same file, add the YAML configuration loader:

def load_config(config_path: Path) -> PortfolioConfig:
    """Load portfolio configuration from YAML file."""
    logger = logging.getLogger(__name__)

    if not config_path.exists():
        raise FileNotFoundError(f"Config file not found: {config_path}")

    with open(config_path, "r") as f:
        raw = yaml.safe_load(f)

    logger.info(f"Loaded configuration from {config_path}")

    portfolio = raw["portfolio"]
    simulation = raw["simulation"]

    return PortfolioConfig(
        tickers=portfolio["tickers"],
        weights=portfolio["weights"],
        notional=portfolio["notional"],
        lookback_days=simulation["lookback_days"],
        num_simulations=simulation["num_simulations"],
        time_horizon_days=simulation["time_horizon_days"],
        confidence_levels=simulation["confidence_levels"],
        random_seed=simulation["random_seed"],
    )

What does load_config do?

This function reads the config.yaml file you created, parses it with yaml.safe_load, and maps the raw dictionary into a validated PortfolioConfig dataclass.

If the config file is missing or the weights do not sum to 1.0, it raises a clear error immediately instead of failing silently mid-simulation.

• Save config.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files: __init__.py, models.py, and config.py.

ⓧ I'd like to double check the full code

Here is the complete models.py:

"""Data models for the risk engine."""

from dataclasses import dataclass, field
from typing import Any

import numpy as np
import pandas as pd


@dataclass
class PortfolioConfig:
    """Portfolio configuration loaded from YAML."""

    tickers: list[str]
    weights: list[float]
    notional: float
    lookback_days: int
    num_simulations: int
    time_horizon_days: int
    confidence_levels: list[float]
    random_seed: int

    def __post_init__(self) -> None:
        if abs(sum(self.weights) - 1.0) > 1e-6:
            raise ValueError(f"Weights must sum to 1.0, got {sum(self.weights):.6f}")
        if len(self.tickers) != len(self.weights):
            raise ValueError("Tickers and weights must have same length")


@dataclass
class ValidationResult:
    """Result of data validation."""

    is_valid: bool
    ticker: str
    issues: list[str] = field(default_factory=list)


@dataclass
class SimulationResult:
    """Output of Monte Carlo simulation."""

    terminal_values: np.ndarray  # shape: (num_simulations,)
    pnl_distribution: np.ndarray  # shape: (num_simulations,)
    elapsed_seconds: float


@dataclass
class RiskMetrics:
    """Computed risk metrics for the portfolio."""

    var_95: float
    var_99: float
    cvar_95: float
    cvar_99: float
    annualized_volatility: float
    annualized_sharpe: float
    max_drawdown: float
    risk_contributions: dict[str, float]

Here is the complete config.py:

"""Configuration loader."""

import logging
import json
from pathlib import Path
from typing import Any

import yaml

from risk_engine.models import PortfolioConfig


def setup_logging(log_level: str = "INFO") -> None:
    """Configure structured JSON logging."""
    class JsonFormatter(logging.Formatter):
        def format(self, record: logging.LogRecord) -> str:
            log_entry = {
                "timestamp": self.formatTime(record),
                "level": record.levelname,
                "module": record.module,
                "message": record.getMessage(),
            }
            if record.exc_info:
                log_entry["exception"] = self.formatException(record.exc_info)
            return json.dumps(log_entry)

    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    logging.root.handlers = [handler]
    logging.root.setLevel(getattr(logging, log_level.upper()))


def load_config(config_path: Path) -> PortfolioConfig:
    """Load portfolio configuration from YAML file."""
    logger = logging.getLogger(__name__)

    if not config_path.exists():
        raise FileNotFoundError(f"Config file not found: {config_path}")

    with open(config_path, "r") as f:
        raw = yaml.safe_load(f)

    logger.info(f"Loaded configuration from {config_path}")

    portfolio = raw["portfolio"]
    simulation = raw["simulation"]

    return PortfolioConfig(
        tickers=portfolio["tickers"],
        weights=portfolio["weights"],
        notional=portfolio["notional"],
        lookback_days=simulation["lookback_days"],
        num_simulations=simulation["num_simulations"],
        time_horizon_days=simulation["time_horizon_days"],
        confidence_levels=simulation["confidence_levels"],
        random_seed=simulation["random_seed"],
    )

• Verify everything is wired correctly by running this command in your terminal:

python -c "from risk_engine.models import PortfolioConfig; print('OK')"

You should see OK printed with no errors.

Seeing a ModuleNotFoundError?

• Make sure your terminal is in the monte-carlo-risk-engine/ folder (the parent of risk_engine/), not inside risk_engine/ itself.
• Confirm that risk_engine/__init__.py exists. Without it, Python does not recognise the folder as a package.
• Check that your virtual environment is still activated. You should see (.venv) at the start of your terminal prompt.

Help me debug this ModuleNotFoundError

Your project foundation is locked in. Next up, you will build the async data pipeline that fetches real market prices and validates them for simulation quality.

Build the Async Data Pipeline

Your project skeleton is in place with dataclass models and a config loader ready to go. Now you need real market data flowing into the engine.

A risk engine is only as good as its data. In this step, you will build an async data pipeline that fetches historical stock prices for all five portfolio tickers concurrently, validates the data for quality issues, and transforms raw prices into the daily log returns your simulation engine will consume.

In this step, get ready to:

• Build the DataFetcher class that downloads price data for all tickers concurrently using asyncio.
• Build the DataValidator class that checks data quality and completeness.
• Wire both into a compute_returns function and verify the full pipeline end-to-end.

Build the DataFetcher class

Fetching data for five tickers one at a time is slow because each network request blocks the next. The asyncio library lets you run all five downloads concurrently, cutting total fetch time dramatically.

• In VS Code, open risk_engine/data.py from the Explorer sidebar.
• Replace any existing content with the following code:

"""Async data fetching and validation."""

import asyncio
import logging
from datetime import datetime, timedelta

import numpy as np
import pandas as pd
import yfinance as yf

from risk_engine.models import PortfolioConfig, ValidationResult


logger = logging.getLogger(__name__)


class DataFetcher:
    """Fetches historical price data asynchronously."""

    def __init__(self, config: PortfolioConfig) -> None:
        self.config = config

    async def fetch_all(self) -> dict[str, pd.DataFrame]:
        """Fetch price data for all tickers concurrently."""
        logger.info(f"Fetching data for {len(self.config.tickers)} tickers")

        loop = asyncio.get_event_loop()
        tasks = [
            loop.run_in_executor(None, self._fetch_single, ticker)
            for ticker in self.config.tickers
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        prices: dict[str, pd.DataFrame] = {}
        for ticker, result in zip(self.config.tickers, results):
            if isinstance(result, Exception):
                logger.error(f"Failed to fetch {ticker}: {result}")
                raise RuntimeError(f"Data fetch failed for {ticker}") from result
            logger.info(f"Fetched {len(result)} rows for {ticker}")
            prices[ticker] = result

        return prices

    def _fetch_single(self, ticker: str) -> pd.DataFrame:
        """Fetch historical data for a single ticker."""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=int(self.config.lookback_days * 1.5))

        data = yf.download(
            ticker,
            start=start_date.strftime("%Y-%m-%d"),
            end=end_date.strftime("%Y-%m-%d"),
            progress=False,
            auto_adjust=True,
        )

        if data.empty:
            raise ValueError(f"No data returned for {ticker}")

        return data

What does this code do?

The DataFetcher class wraps the yfinance library and uses asyncio's thread pool executor to download all ticker data concurrently.

• fetch_all() dispatches one _fetch_single call per ticker into the default thread pool using run_in_executor. All five downloads run at the same time.
• asyncio.gather() collects results and surfaces any exceptions. If a ticker fails, the pipeline stops with a clear error.
• _fetch_single() requests 1.5x the configured lookback days to ensure enough trading days remain after weekends and holidays are excluded.

Why asyncio instead of threading directly?

The data fetching is I/O-bound (network requests to Yahoo Finance). The CPU-bound simulation later uses vectorized NumPy, which releases the GIL.

Using asyncio with run_in_executor bridges the synchronous yfinance API into an async pipeline. This is the standard pattern on hedge fund risk desks: async for data ingestion, vectorized NumPy for computation.

• Save data.py.

You won't see output yet. The next chunk adds validation before you test the full pipeline.

Build the DataValidator class

Raw market data often has gaps from exchange closures, data feed outages, or corporate actions. A production pipeline must reject data that would corrupt downstream calculations.

• In risk_engine/data.py, add the following DataValidator class directly below the DataFetcher class:

class DataValidator:
    """Validates fetched market data."""

    MAX_GAP_DAYS: int = 5
    MIN_HISTORY_DAYS: int = 252

    def validate(self, prices: dict[str, pd.DataFrame]) -> list[ValidationResult]:
        """Validate all ticker data."""
        results: list[ValidationResult] = []

        for ticker, df in prices.items():
            issues: list[str] = []

            if len(df) < self.MIN_HISTORY_DAYS:
                issues.append(
                    f"Insufficient history: {len(df)} days (need {self.MIN_HISTORY_DAYS})"
                )

            if "Close" in df.columns:
                close = df["Close"]
            else:
                close = df.iloc[:, 0]

            if (close <= 0).any():
                issues.append("Contains non-positive prices")

            date_diffs = pd.Series(df.index).diff().dt.days
            max_gap = date_diffs.max()
            if max_gap is not None and max_gap > self.MAX_GAP_DAYS:
                issues.append(f"Gap of {max_gap} days exceeds maximum {self.MAX_GAP_DAYS}")

            result = ValidationResult(
                is_valid=len(issues) == 0,
                ticker=ticker,
                issues=issues,
            )
            results.append(result)

            if result.is_valid:
                logger.info(f"Validation passed for {ticker}")
            else:
                logger.warning(f"Validation issues for {ticker}: {issues}")

        return results

What does this code do?

The DataValidator runs three checks on every ticker's DataFrame before it enters the simulation:

• Sufficient history: at least 252 trading days (one full year) are needed to estimate stable return statistics.
• Non-positive prices: negative or zero prices indicate data corruption and would produce invalid log returns.
• Gap detection: a gap exceeding 5 calendar days suggests missing data rather than a normal weekend.

Each ticker gets a ValidationResult dataclass with a pass/fail flag and a list of specific issues. Structured logging records validation outcomes for audit trails.

• Save data.py.

You won't see output yet. The next chunk adds the returns computation and a pipeline test.

Wire the pipeline together and test

The fetcher and validator handle raw DataFrames, but the simulation engine needs daily log returns. Log returns are additive across time, which makes them the correct input for Geometric Brownian Motion simulation.

• In risk_engine/data.py, add the following compute_returns function below the DataValidator class:

def compute_returns(prices: dict[str, pd.DataFrame]) -> pd.DataFrame:
    """Compute daily log returns from price data."""
    close_prices = pd.DataFrame(
        {ticker: df["Close"].squeeze() if "Close" in df.columns else df.iloc[:, 0]
         for ticker, df in prices.items()}
    )

    close_prices = close_prices.dropna()
    log_returns = np.log(close_prices / close_prices.shift(1)).dropna()

    logger.info(f"Computed log returns: {log_returns.shape[0]} days x {log_returns.shape[1]} assets")
    return log_returns

Why log returns instead of simple returns?

Simple returns (price change / previous price) cannot be summed across time. Log returns can: the sum of daily log returns equals the total period log return.

Log returns also prevent impossible negative prices when simulating GBM paths, because exponentiating a log return always produces a positive value.

• Save data.py.
• Test the full data pipeline by running this command in your terminal:

python -c "
import asyncio
from risk_engine.config import load_config, setup_logging
from risk_engine.data import DataFetcher, DataValidator, compute_returns
from pathlib import Path

setup_logging('INFO')
config = load_config(Path('config.yaml'))

async def test_pipeline():
    fetcher = DataFetcher(config)
    prices = await fetcher.fetch_all()
    validator = DataValidator()
    results = validator.validate(prices)
    log_returns = compute_returns(prices)
    print(f'Shape: {log_returns.shape}')

asyncio.run(test_pipeline())
"

You should see structured JSON log lines showing successful fetches for each ticker, validation passes, and a final line reporting the shape of your log returns DataFrame (approximately 500 days x 5 assets).

Seeing a connection or download error?

Make sure your computer has an active internet connection. The yfinance library downloads data from Yahoo Finance servers.

If you see a ticker-specific error, check that the ticker symbols in config.yaml are valid NYSE/NASDAQ symbols (AAPL, MSFT, GOOGL, AMZN, JPM).

If you see an ImportError, confirm your virtual environment is activated and all packages from requirements.txt are installed.

help me debug my data pipeline error

✔️ Awesome, I've got everything!

Great. Double check you have saved risk_engine/data.py with all three components: DataFetcher, DataValidator, and compute_returns.

ⓧ I'd like to double check the full code

Here is the complete risk_engine/data.py file:

"""Async data fetching and validation."""

import asyncio
import logging
from datetime import datetime, timedelta

import numpy as np
import pandas as pd
import yfinance as yf

from risk_engine.models import PortfolioConfig, ValidationResult


logger = logging.getLogger(__name__)


class DataFetcher:
    """Fetches historical price data asynchronously."""

    def __init__(self, config: PortfolioConfig) -> None:
        self.config = config

    async def fetch_all(self) -> dict[str, pd.DataFrame]:
        """Fetch price data for all tickers concurrently."""
        logger.info(f"Fetching data for {len(self.config.tickers)} tickers")

        loop = asyncio.get_event_loop()
        tasks = [
            loop.run_in_executor(None, self._fetch_single, ticker)
            for ticker in self.config.tickers
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        prices: dict[str, pd.DataFrame] = {}
        for ticker, result in zip(self.config.tickers, results):
            if isinstance(result, Exception):
                logger.error(f"Failed to fetch {ticker}: {result}")
                raise RuntimeError(f"Data fetch failed for {ticker}") from result
            logger.info(f"Fetched {len(result)} rows for {ticker}")
            prices[ticker] = result

        return prices

    def _fetch_single(self, ticker: str) -> pd.DataFrame:
        """Fetch historical data for a single ticker."""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=int(self.config.lookback_days * 1.5))

        data = yf.download(
            ticker,
            start=start_date.strftime("%Y-%m-%d"),
            end=end_date.strftime("%Y-%m-%d"),
            progress=False,
            auto_adjust=True,
        )

        if data.empty:
            raise ValueError(f"No data returned for {ticker}")

        return data


class DataValidator:
    """Validates fetched market data."""

    MAX_GAP_DAYS: int = 5
    MIN_HISTORY_DAYS: int = 252

    def validate(self, prices: dict[str, pd.DataFrame]) -> list[ValidationResult]:
        """Validate all ticker data."""
        results: list[ValidationResult] = []

        for ticker, df in prices.items():
            issues: list[str] = []

            if len(df) < self.MIN_HISTORY_DAYS:
                issues.append(
                    f"Insufficient history: {len(df)} days (need {self.MIN_HISTORY_DAYS})"
                )

            if "Close" in df.columns:
                close = df["Close"]
            else:
                close = df.iloc[:, 0]

            if (close <= 0).any():
                issues.append("Contains non-positive prices")

            date_diffs = pd.Series(df.index).diff().dt.days
            max_gap = date_diffs.max()
            if max_gap is not None and max_gap > self.MAX_GAP_DAYS:
                issues.append(f"Gap of {max_gap} days exceeds maximum {self.MAX_GAP_DAYS}")

            result = ValidationResult(
                is_valid=len(issues) == 0,
                ticker=ticker,
                issues=issues,
            )
            results.append(result)

            if result.is_valid:
                logger.info(f"Validation passed for {ticker}")
            else:
                logger.warning(f"Validation issues for {ticker}: {issues}")

        return results


def compute_returns(prices: dict[str, pd.DataFrame]) -> pd.DataFrame:
    """Compute daily log returns from price data."""
    close_prices = pd.DataFrame(
        {ticker: df["Close"].squeeze() if "Close" in df.columns else df.iloc[:, 0]
         for ticker, df in prices.items()}
    )

    close_prices = close_prices.dropna()
    log_returns = np.log(close_prices / close_prices.shift(1)).dropna()

    logger.info(f"Computed log returns: {log_returns.shape[0]} days x {log_returns.shape[1]} assets")
    return log_returns

Your data pipeline is pulling real market data, validating it, and computing the log returns your simulation engine needs. Next up, you will build the Monte Carlo simulation engine that generates 10,000 correlated return paths from these calibrated statistics.

Build the Monte Carlo Simulation Engine

Your data pipeline can now fetch and validate historical prices for all five portfolio assets. The next question is: what could happen to this portfolio over the next 10 trading days?

To answer that, you will build a Monte Carlo simulation engine that generates 10,000 possible future return paths. It uses Geometric Brownian Motion to model stock price dynamics and Cholesky decomposition to preserve the correlation structure between assets.

In this step, get ready to:

• Build the MonteCarloEngine class with a GBM calibration method.
• Implement the Cholesky-decomposed simulation that generates 10,000 correlated return paths in a single vectorized operation.
• Add structured logging that reports simulation timing and path statistics.

Set up the MonteCarloEngine class with GBM calibration

Before you can simulate future prices, you need to calibrate the model from historical data. Calibration extracts the drift (expected return) and volatility for each asset from the log returns your data pipeline computed.

The standard financial convention is to annualize these parameters. You multiply daily mean returns by 252 (trading days per year) and multiply the daily covariance matrix by 252.

• In risk_engine/simulation.py, replace the placeholder content with the following imports, class definition, and calibration method:

"""Monte Carlo simulation engine."""

import logging
import time

import numpy as np
import pandas as pd

from risk_engine.models import PortfolioConfig, SimulationResult


logger = logging.getLogger(__name__)


class MonteCarloEngine:
    """Generates correlated return paths using GBM and Cholesky decomposition."""

    def __init__(self, config: PortfolioConfig) -> None:
        self.config = config

    def calibrate(self, log_returns: pd.DataFrame) -> tuple[np.ndarray, np.ndarray]:
        """Calibrate GBM parameters from historical returns.

        Returns:
            Tuple of (annualized_mean, covariance_matrix)
        """
        daily_mean = log_returns.mean().values
        daily_cov = log_returns.cov().values

        annualized_mean = daily_mean * 252
        annualized_cov = daily_cov * 252

        logger.info(
            f"Calibrated GBM: mean range [{annualized_mean.min():.4f}, {annualized_mean.max():.4f}], "
            f"vol range [{np.sqrt(np.diag(annualized_cov)).min():.4f}, {np.sqrt(np.diag(annualized_cov)).max():.4f}]"
        )

        return annualized_mean, annualized_cov

What does this code do?

The MonteCarloEngine class takes a PortfolioConfig and provides methods for calibration and simulation.

• log_returns.mean().values computes the average daily return for each asset. Multiplying by 252 gives the annualized expected return.
• log_returns.cov().values computes the covariance matrix of daily returns. Multiplying by 252 annualizes it, capturing how asset returns move together over a year.
• The structured log line reports the calibrated ranges so you can sanity-check the parameters before simulation.

Why annualize by 252?

There are approximately 252 trading days in a year. Financial conventions express returns and volatility on an annual basis so that different assets and strategies are comparable regardless of the observation frequency.

Do not save the file yet. You will add the simulation method next.

Build the Cholesky-decomposed simulation method

Stock returns are correlated. If you simulated each asset independently, you would miss diversification effects and produce unrealistic portfolio outcomes. Cholesky decomposition factors the covariance matrix into a lower-triangular matrix that transforms independent random samples into correlated ones.

• In the same file, add the simulate method directly below the calibrate method:

    def simulate(
        self,
        annualized_mean: np.ndarray,
        annualized_cov: np.ndarray,
    ) -> SimulationResult:
        """Run Monte Carlo simulation using Cholesky decomposition.

        Generates correlated return paths for the portfolio.
        """
        start_time = time.perf_counter()

        rng = np.random.default_rng(self.config.random_seed)
        num_assets = len(self.config.tickers)
        dt = self.config.time_horizon_days / 252

        # Cholesky decomposition for correlated samples
        cholesky = np.linalg.cholesky(annualized_cov)

        # Generate uncorrelated standard normal samples
        # Shape: (num_simulations, num_assets)
        z = rng.standard_normal((self.config.num_simulations, num_assets))

        # Apply Cholesky to correlate the samples
        correlated_z = z @ cholesky.T

        # GBM: S_T = S_0 * exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z)
        vols = np.sqrt(np.diag(annualized_cov))
        drift = (annualized_mean - 0.5 * vols**2) * dt
        diffusion = correlated_z * np.sqrt(dt)

        # Log returns over the horizon
        log_returns_sim = drift + diffusion

        # Portfolio returns using weights
        weights = np.array(self.config.weights)
        portfolio_returns = log_returns_sim @ weights

        # Terminal values and P&L
        terminal_values = self.config.notional * np.exp(portfolio_returns)
        pnl_distribution = terminal_values - self.config.notional

        elapsed = time.perf_counter() - start_time

        logger.info(
            f"Simulation complete: {self.config.num_simulations} paths in {elapsed:.3f}s | "
            f"Mean terminal: ${terminal_values.mean():,.0f} | "
            f"Std terminal: ${terminal_values.std():,.0f}"
        )

        return SimulationResult(
            terminal_values=terminal_values,
            pnl_distribution=pnl_distribution,
            elapsed_seconds=elapsed,
        )

What does this simulation do?

This method generates 10,000 possible portfolio outcomes in a single vectorized matrix operation. Here is the flow:

• np.random.default_rng(self.config.random_seed) creates a reproducible random number generator. Using the same seed always produces the same simulation results.
• np.linalg.cholesky(annualized_cov) decomposes the covariance matrix into L such that L @ L.T reconstructs the original. Multiplying independent samples by L.T introduces the correct correlation structure.
• The GBM formula applies drift (expected return minus a volatility correction) and diffusion (random shock scaled by volatility) to produce log returns over the time horizon.
• portfolio_returns = log_returns_sim @ weights combines all assets into a single portfolio return per simulation path.
• terminal_values converts log returns back to dollar amounts. pnl_distribution is simply the terminal value minus the starting notional.

Why vectorized NumPy instead of a loop?

A Python for-loop over 10,000 paths would take seconds. NumPy operations run in compiled C under the hood, so the entire (10000, 5) matrix is computed in a single pass. This is the same approach production risk systems use to hit their daily computation deadlines.

• Save risk_engine/simulation.py.

Seeing a syntax error?

• Make sure the simulate method is indented inside the MonteCarloEngine class at the same level as calibrate.
• Check that the from risk_engine.models import PortfolioConfig, SimulationResult import at the top includes both names.
• Confirm you have import time in the imports section. The time.perf_counter() call requires it.

Help me debug a syntax error in my MonteCarloEngine simulation.py file.

Verify the simulation engine

The structured logging you added reports calibration ranges, simulation timing, and terminal value statistics. You can confirm everything works by running a quick integration test from the terminal.

• Verify the module imports correctly by running this command in your terminal:

python -c "from risk_engine.simulation import MonteCarloEngine; print('MonteCarloEngine ready')"

You should see MonteCarloEngine ready printed without any import errors.

Getting a ModuleNotFoundError?

• Make sure your virtual environment is activated. You should see (.venv) in your terminal prompt.
• Confirm you are running the command from the monte-carlo-risk-engine/ project root, not from inside the risk_engine/ folder.
• Check that risk_engine/__init__.py exists. Python needs this file to treat the folder as a package.

Help me fix a ModuleNotFoundError when importing MonteCarloEngine.

• Run a full simulation test to confirm the engine produces results with timing logs by running:

python -c "
import numpy as np
import pandas as pd
from risk_engine.models import PortfolioConfig
from risk_engine.simulation import MonteCarloEngine
from risk_engine.config import setup_logging

setup_logging('INFO')

config = PortfolioConfig(
    tickers=['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'JPM'],
    weights=[0.25, 0.25, 0.20, 0.15, 0.15],
    notional=1_000_000,
    lookback_days=504,
    num_simulations=10_000,
    time_horizon_days=10,
    confidence_levels=[0.95, 0.99],
    random_seed=42,
)

# Create synthetic log returns for testing
rng = np.random.default_rng(99)
fake_returns = pd.DataFrame(
    rng.normal(0.0004, 0.015, (252, 5)),
    columns=config.tickers
)

engine = MonteCarloEngine(config)
mean, cov = engine.calibrate(fake_returns)
result = engine.simulate(mean, cov)
print(f'Terminal values shape: {result.terminal_values.shape}')
print(f'P&L distribution shape: {result.pnl_distribution.shape}')
print(f'Elapsed: {result.elapsed_seconds:.3f}s')
"

You should see structured JSON log lines showing the calibrated GBM parameters, then the simulation completion message with timing, mean terminal value, and standard deviation. The final print statements confirm the output shapes are (10000,) for both arrays.

What should I see?

The JSON log lines confirm each stage completed. The calibration log shows annualized mean and volatility ranges. The simulation log reports how many paths were generated, how long it took (typically under 0.1 seconds for 10,000 paths), and the dollar mean and standard deviation of terminal portfolio values.

The exact dollar values depend on the synthetic random data, but you should see a mean terminal value near $1,000,000 (the notional) and a standard deviation in the tens of thousands.

✔️ Awesome, I've got everything!

Great. Double check you have saved risk_engine/simulation.py.

ⓧ I'd like to double check the full code

Here is the complete risk_engine/simulation.py file for comparison:

"""Monte Carlo simulation engine."""

import logging
import time

import numpy as np
import pandas as pd

from risk_engine.models import PortfolioConfig, SimulationResult


logger = logging.getLogger(__name__)


class MonteCarloEngine:
    """Generates correlated return paths using GBM and Cholesky decomposition."""

    def __init__(self, config: PortfolioConfig) -> None:
        self.config = config

    def calibrate(self, log_returns: pd.DataFrame) -> tuple[np.ndarray, np.ndarray]:
        """Calibrate GBM parameters from historical returns.

        Returns:
            Tuple of (annualized_mean, covariance_matrix)
        """
        daily_mean = log_returns.mean().values
        daily_cov = log_returns.cov().values

        annualized_mean = daily_mean * 252
        annualized_cov = daily_cov * 252

        logger.info(
            f"Calibrated GBM: mean range [{annualized_mean.min():.4f}, {annualized_mean.max():.4f}], "
            f"vol range [{np.sqrt(np.diag(annualized_cov)).min():.4f}, {np.sqrt(np.diag(annualized_cov)).max():.4f}]"
        )

        return annualized_mean, annualized_cov

    def simulate(
        self,
        annualized_mean: np.ndarray,
        annualized_cov: np.ndarray,
    ) -> SimulationResult:
        """Run Monte Carlo simulation using Cholesky decomposition.

        Generates correlated return paths for the portfolio.
        """
        start_time = time.perf_counter()

        rng = np.random.default_rng(self.config.random_seed)
        num_assets = len(self.config.tickers)
        dt = self.config.time_horizon_days / 252

        # Cholesky decomposition for correlated samples
        cholesky = np.linalg.cholesky(annualized_cov)

        # Generate uncorrelated standard normal samples
        # Shape: (num_simulations, num_assets)
        z = rng.standard_normal((self.config.num_simulations, num_assets))

        # Apply Cholesky to correlate the samples
        correlated_z = z @ cholesky.T

        # GBM: S_T = S_0 * exp((mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z)
        vols = np.sqrt(np.diag(annualized_cov))
        drift = (annualized_mean - 0.5 * vols**2) * dt
        diffusion = correlated_z * np.sqrt(dt)

        # Log returns over the horizon
        log_returns_sim = drift + diffusion

        # Portfolio returns using weights
        weights = np.array(self.config.weights)
        portfolio_returns = log_returns_sim @ weights

        # Terminal values and P&L
        terminal_values = self.config.notional * np.exp(portfolio_returns)
        pnl_distribution = terminal_values - self.config.notional

        elapsed = time.perf_counter() - start_time

        logger.info(
            f"Simulation complete: {self.config.num_simulations} paths in {elapsed:.3f}s | "
            f"Mean terminal: ${terminal_values.mean():,.0f} | "
            f"Std terminal: ${terminal_values.std():,.0f}"
        )

        return SimulationResult(
            terminal_values=terminal_values,
            pnl_distribution=pnl_distribution,
            elapsed_seconds=elapsed,
        )

Your simulation engine now generates 10,000 correlated portfolio outcomes in a fraction of a second. Next, you will build the risk calculator that turns this distribution into actionable metrics like VaR, CVaR, and per-asset risk attribution.

Compute Risk Metrics and Attribution

Your simulation engine now generates 10,000 correlated return paths. The raw output is a numpy array of P&L values. A risk desk doesn't report raw arrays to portfolio managers.

In this step, you'll build the RiskCalculator class that transforms raw simulation output into the metrics a hedge fund actually reports: Value at Risk, Expected Shortfall, Sharpe ratio, and per-asset risk attribution.

In this step, get ready to:

• Build VaR and CVaR computation at 95% and 99% confidence levels.
• Add portfolio performance metrics including Sharpe ratio, volatility, and max drawdown.
• Implement Euler decomposition for component risk attribution.

Build the RiskCalculator with VaR and CVaR

VaR answers: "What's the most I can lose at a given confidence level?" CVaR (also called Expected Shortfall) answers the follow-up: "If I breach VaR, how bad does it get on average?"

• Switch back to risk_engine/risk.py in VS Code.
• Replace the empty file contents with the following imports, class definition, and VaR/CVaR method:

"""Risk metrics calculation and attribution."""

import logging

import numpy as np
import pandas as pd

from risk_engine.models import PortfolioConfig, RiskMetrics, SimulationResult


logger = logging.getLogger(__name__)


class RiskCalculator:
    """Computes VaR, CVaR, and portfolio risk metrics."""

    def __init__(self, config: PortfolioConfig) -> None:
        self.config = config

    def compute_var_cvar(self, pnl: np.ndarray) -> tuple[dict[str, float], dict[str, float]]:
        """Compute VaR and CVaR at configured confidence levels."""
        var_results: dict[str, float] = {}
        cvar_results: dict[str, float] = {}

        for level in self.config.confidence_levels:
            percentile = (1 - level) * 100
            var = -np.percentile(pnl, percentile)
            cvar = -pnl[pnl <= -var].mean() if (pnl <= -var).any() else var

            var_results[f"{int(level * 100)}%"] = var
            cvar_results[f"{int(level * 100)}%"] = cvar

            logger.info(f"VaR {int(level*100)}%: ${var:,.0f} | CVaR {int(level*100)}%: ${cvar:,.0f}")

        return var_results, cvar_results

What does this code do?

The RiskCalculator takes the same PortfolioConfig that every other module uses, keeping the design consistent.

• VaR at 95% means: "In 95% of scenarios, losses stay below this number." It's computed as the negative of the 5th percentile of P&L.
• CVaR averages only the losses that breach the VaR threshold. This captures tail severity, which VaR alone ignores.
• The method loops over both 0.95 and 0.99 from the config's confidence_levels list, so you get both thresholds in one pass.

Why both VaR and CVaR?

VaR tells you the threshold loss at a confidence level. CVaR tells you the average loss beyond that threshold. Basel III replaced VaR with CVaR as the regulatory standard because VaR ignores tail severity. Hedge funds report both.

• Save risk_engine/risk.py.
• You won't see a change yet. The next chunk adds portfolio metrics that complete the class.

Seeing a red underline on the imports?

Make sure your virtual environment is activated. VS Code may not resolve numpy or pandas if the interpreter path points outside .venv. Check the Python interpreter in the bottom-right of VS Code.

Help me fix import resolution errors in VS Code

Add portfolio performance metrics

VaR and CVaR measure tail risk from the simulation. But risk managers also need backward-looking performance context: how volatile has this portfolio actually been, what's the risk-adjusted return, and what's the worst historical drawdown?

• In risk_engine/risk.py, add the following method directly below the compute_var_cvar method (still inside the RiskCalculator class):

    def compute_portfolio_metrics(self, log_returns: pd.DataFrame) -> dict[str, float]:
        """Compute annualized portfolio performance metrics."""
        weights = np.array(self.config.weights)
        portfolio_returns = (log_returns.values @ weights)

        annualized_return = portfolio_returns.mean() * 252
        annualized_vol = portfolio_returns.std() * np.sqrt(252)
        sharpe = annualized_return / annualized_vol if annualized_vol > 0 else 0.0

        # Max drawdown
        cumulative = np.exp(np.cumsum(portfolio_returns))
        running_max = np.maximum.accumulate(cumulative)
        drawdowns = (cumulative - running_max) / running_max
        max_drawdown = drawdowns.min()

        return {
            "annualized_return": annualized_return,
            "annualized_volatility": annualized_vol,
            "sharpe_ratio": sharpe,
            "max_drawdown": max_drawdown,
        }

What does this code do?

• log_returns.values @ weights computes the weighted portfolio return for each day using matrix multiplication.
• Annualization follows the standard convention: multiply daily mean by 252 (trading days), multiply daily std by sqrt(252).
• The Sharpe ratio is annualized return divided by annualized volatility. It measures return per unit of risk.
• Max drawdown finds the worst peak-to-trough decline by tracking np.maximum.accumulate (the running all-time high) and measuring how far cumulative returns drop below it.

• Save risk_engine/risk.py.
• You won't see a change yet. The next chunk adds attribution and the wiring method that ties everything together.

Indentation error?

This method must be indented one level inside the RiskCalculator class (4 spaces from the left margin). If you pasted it at the module level, Python will not associate it with the class.

Help me fix the indentation in my RiskCalculator class

Add risk attribution and wire everything together

Knowing total portfolio VaR is useful. Knowing which position is responsible for most of that risk is actionable. The Euler decomposition method splits total portfolio risk into additive per-asset contributions that sum to 100%.

• Add the following compute_risk_attribution method below compute_portfolio_metrics:

    def compute_risk_attribution(
        self,
        annualized_cov: np.ndarray,
    ) -> dict[str, float]:
        """Compute component risk attribution using Euler decomposition.

        Each asset's contribution = w_i * (Sigma @ w)_i / sigma_p
        """
        weights = np.array(self.config.weights)

        # Portfolio variance
        port_variance = weights @ annualized_cov @ weights
        port_vol = np.sqrt(port_variance)

        # Marginal contribution to risk
        marginal = annualized_cov @ weights

        # Component contribution
        component = weights * marginal / port_vol

        # Percentage contribution
        pct_contribution = component / port_vol

        attribution = {
            ticker: float(pct_contribution[i])
            for i, ticker in enumerate(self.config.tickers)
        }

        logger.info(f"Risk attribution: {attribution}")
        return attribution

What is Euler decomposition?

Euler decomposition is the standard approach on hedge fund risk desks. It is the only decomposition method that is both additive (contributions sum to total risk) and homogeneous (doubling a position doubles its contribution).

• annualized_cov @ weights gives each asset's marginal contribution to portfolio variance.
• Multiplying by weights and dividing by portfolio volatility converts marginal contributions into actual risk shares.
• Dividing again by port_vol gives percentage contributions that sum to 1.0 (100%).

• Now add the compute_all method below compute_risk_attribution to wire all three calculations together:

    def compute_all(
        self,
        simulation_result: SimulationResult,
        log_returns: pd.DataFrame,
        annualized_cov: np.ndarray,
    ) -> RiskMetrics:
        """Compute all risk metrics."""
        var_results, cvar_results = self.compute_var_cvar(simulation_result.pnl_distribution)
        portfolio_metrics = self.compute_portfolio_metrics(log_returns)
        attribution = self.compute_risk_attribution(annualized_cov)

        return RiskMetrics(
            var_95=var_results.get("95%", 0.0),
            var_99=var_results.get("99%", 0.0),
            cvar_95=cvar_results.get("95%", 0.0),
            cvar_99=cvar_results.get("99%", 0.0),
            annualized_volatility=portfolio_metrics["annualized_volatility"],
            annualized_sharpe=portfolio_metrics["sharpe_ratio"],
            max_drawdown=portfolio_metrics["max_drawdown"],
            risk_contributions=attribution,
        )

What does compute_all do?

This is the orchestration method. It calls the three computation methods in sequence and packs their outputs into a single RiskMetrics dataclass. The pipeline (in the next step) will call only this one method to get everything it needs for the final report.

• Save risk_engine/risk.py.
• Verify the module loads correctly by running this command in your terminal:

python -c "from risk_engine.risk import RiskCalculator; print('OK')"

• You should see OK printed with no errors.

Seeing a ModuleNotFoundError or ImportError?

• Make sure you are running the command from the monte-carlo-risk-engine/ root directory (the one containing the risk_engine/ folder).
• Check that risk_engine/__init__.py exists. Without it, Python does not recognize the folder as a package.
• Verify the imports at the top of risk.py match exactly: from risk_engine.models import PortfolioConfig, RiskMetrics, SimulationResult.

Help me debug this import error in risk_engine/risk.py

✔️ Awesome, I've got everything!

Great. Double check you've saved risk_engine/risk.py and the import test printed OK.

ⓧ I'd like to double check the full code

"""Risk metrics calculation and attribution."""

import logging

import numpy as np
import pandas as pd

from risk_engine.models import PortfolioConfig, RiskMetrics, SimulationResult


logger = logging.getLogger(__name__)


class RiskCalculator:
    """Computes VaR, CVaR, and portfolio risk metrics."""

    def __init__(self, config: PortfolioConfig) -> None:
        self.config = config

    def compute_var_cvar(self, pnl: np.ndarray) -> tuple[dict[str, float], dict[str, float]]:
        """Compute VaR and CVaR at configured confidence levels."""
        var_results: dict[str, float] = {}
        cvar_results: dict[str, float] = {}

        for level in self.config.confidence_levels:
            percentile = (1 - level) * 100
            var = -np.percentile(pnl, percentile)
            cvar = -pnl[pnl <= -var].mean() if (pnl <= -var).any() else var

            var_results[f"{int(level * 100)}%"] = var
            cvar_results[f"{int(level * 100)}%"] = cvar

            logger.info(f"VaR {int(level*100)}%: ${var:,.0f} | CVaR {int(level*100)}%: ${cvar:,.0f}")

        return var_results, cvar_results

    def compute_portfolio_metrics(self, log_returns: pd.DataFrame) -> dict[str, float]:
        """Compute annualized portfolio performance metrics."""
        weights = np.array(self.config.weights)
        portfolio_returns = (log_returns.values @ weights)

        annualized_return = portfolio_returns.mean() * 252
        annualized_vol = portfolio_returns.std() * np.sqrt(252)
        sharpe = annualized_return / annualized_vol if annualized_vol > 0 else 0.0

        # Max drawdown
        cumulative = np.exp(np.cumsum(portfolio_returns))
        running_max = np.maximum.accumulate(cumulative)
        drawdowns = (cumulative - running_max) / running_max
        max_drawdown = drawdowns.min()

        return {
            "annualized_return": annualized_return,
            "annualized_volatility": annualized_vol,
            "sharpe_ratio": sharpe,
            "max_drawdown": max_drawdown,
        }

    def compute_risk_attribution(
        self,
        annualized_cov: np.ndarray,
    ) -> dict[str, float]:
        """Compute component risk attribution using Euler decomposition.

        Each asset's contribution = w_i * (Sigma @ w)_i / sigma_p
        """
        weights = np.array(self.config.weights)

        # Portfolio variance
        port_variance = weights @ annualized_cov @ weights
        port_vol = np.sqrt(port_variance)

        # Marginal contribution to risk
        marginal = annualized_cov @ weights

        # Component contribution
        component = weights * marginal / port_vol

        # Percentage contribution
        pct_contribution = component / port_vol

        attribution = {
            ticker: float(pct_contribution[i])
            for i, ticker in enumerate(self.config.tickers)
        }

        logger.info(f"Risk attribution: {attribution}")
        return attribution

    def compute_all(
        self,
        simulation_result: SimulationResult,
        log_returns: pd.DataFrame,
        annualized_cov: np.ndarray,
    ) -> RiskMetrics:
        """Compute all risk metrics."""
        var_results, cvar_results = self.compute_var_cvar(simulation_result.pnl_distribution)
        portfolio_metrics = self.compute_portfolio_metrics(log_returns)
        attribution = self.compute_risk_attribution(annualized_cov)

        return RiskMetrics(
            var_95=var_results.get("95%", 0.0),
            var_99=var_results.get("99%", 0.0),
            cvar_95=cvar_results.get("95%", 0.0),
            cvar_99=cvar_results.get("99%", 0.0),
            annualized_volatility=portfolio_metrics["annualized_volatility"],
            annualized_sharpe=portfolio_metrics["sharpe_ratio"],
            max_drawdown=portfolio_metrics["max_drawdown"],
            risk_contributions=attribution,
        )

Your risk engine now knows how to measure tail risk, evaluate performance, and attribute risk to individual positions. Next up, you'll build the async orchestrator that wires the entire pipeline together and formats a polished risk report for the console.

Wire the Async Orchestrator and Run the Engine

You now have all the computational pieces in place. The data fetcher grabs historical prices, the simulation engine generates correlated paths, and the risk calculator produces VaR, CVaR, and attribution metrics.

The missing layer is orchestration. On a real risk desk, these components run in a defined sequence where each stage depends on the previous stage's output. In this step, you will build the pipeline that wires everything together and produces a formatted risk report.

In this step, get ready to:

• Build a formatted console risk report with section headers and a per-asset bar chart.
• Build an async pipeline orchestrator that coordinates all six stages end-to-end.
• Add a CLI entry point and run the full engine against live market data.

Build the formatted risk report

A risk report is only useful if a human can read it at a glance. The report module takes the raw RiskMetrics dataclass and formats it into labelled sections with aligned numbers and a visual bar chart for risk attribution.

• In VS Code, open risk_engine/report.py and replace its contents with the following code:

"""Formatted risk report output."""

from risk_engine.models import PortfolioConfig, RiskMetrics, SimulationResult


def print_risk_report(
    config: PortfolioConfig,
    metrics: RiskMetrics,
    simulation: SimulationResult,
) -> None:
    """Print a formatted risk report to console."""
    print("\n" + "=" * 60)
    print("  MONTE CARLO PORTFOLIO RISK REPORT")
    print("=" * 60)

    print(f"\n  Portfolio: {', '.join(config.tickers)}")
    print(f"  Notional: ${config.notional:,.0f}")
    print(f"  Simulations: {config.num_simulations:,}")
    print(f"  Horizon: {config.time_horizon_days} days")
    print(f"  Simulation time: {simulation.elapsed_seconds:.3f}s")

    print("\n" + "-" * 60)
    print("  VALUE AT RISK (VaR)")
    print("-" * 60)
    print(f"  95% VaR:  ${metrics.var_95:>12,.0f}  ({metrics.var_95/config.notional*100:.2f}%)")
    print(f"  99% VaR:  ${metrics.var_99:>12,.0f}  ({metrics.var_99/config.notional*100:.2f}%)")

    print("\n" + "-" * 60)
    print("  EXPECTED SHORTFALL (CVaR)")
    print("-" * 60)
    print(f"  95% CVaR: ${metrics.cvar_95:>12,.0f}  ({metrics.cvar_95/config.notional*100:.2f}%)")
    print(f"  99% CVaR: ${metrics.cvar_99:>12,.0f}  ({metrics.cvar_99/config.notional*100:.2f}%)")

    print("\n" + "-" * 60)
    print("  PORTFOLIO METRICS")
    print("-" * 60)
    print(f"  Annualized Volatility: {metrics.annualized_volatility*100:>8.2f}%")
    print(f"  Sharpe Ratio:          {metrics.annualized_sharpe:>8.3f}")
    print(f"  Max Drawdown:          {metrics.max_drawdown*100:>8.2f}%")

    print("\n" + "-" * 60)
    print("  RISK ATTRIBUTION (% of total portfolio risk)")
    print("-" * 60)
    for ticker, contrib in metrics.risk_contributions.items():
        bar = "#" * int(abs(contrib) * 40)
        print(f"  {ticker:<6} {contrib*100:>6.1f}%  {bar}")

    print("\n" + "=" * 60)
    print()

What does this code do?

The print_risk_report function takes three inputs (config, metrics, simulation result) and prints a structured report with five sections.

• The header block shows portfolio composition, notional value, and simulation parameters.
• VaR and CVaR sections display both dollar amounts and percentages of notional, right-aligned for readability.
• The risk attribution section draws a hash-bar chart where each # character represents roughly 2.5% of total portfolio risk.

• Save risk_engine/report.py.

You won't see output yet. The report function will be called by the pipeline in the next chunk.

Build the async pipeline orchestrator

The pipeline is the conductor that runs each stage in sequence. It loads configuration, fetches and validates data, calibrates the model, simulates paths, computes risk, and prints the report. Each stage logs its progress so you can trace the full execution.

• Open risk_engine/pipeline.py and replace its contents with the following code:

"""Async orchestrator for the risk pipeline."""

import asyncio
import logging
from pathlib import Path

from risk_engine.config import load_config, setup_logging
from risk_engine.data import DataFetcher, DataValidator, compute_returns
from risk_engine.models import PortfolioConfig, RiskMetrics
from risk_engine.report import print_risk_report
from risk_engine.risk import RiskCalculator
from risk_engine.simulation import MonteCarloEngine


logger = logging.getLogger(__name__)


class RiskPipeline:
    """Orchestrates the full risk calculation pipeline."""

    def __init__(self, config_path: Path) -> None:
        self.config_path = config_path
        self.config: PortfolioConfig | None = None

    async def run(self) -> RiskMetrics:
        """Execute the full pipeline."""
        # Load configuration
        self.config = load_config(self.config_path)
        logger.info("Pipeline started")

        # Stage 1: Fetch data
        logger.info("Stage 1: Fetching market data")
        fetcher = DataFetcher(self.config)
        prices = await fetcher.fetch_all()

        # Stage 2: Validate data
        logger.info("Stage 2: Validating data")
        validator = DataValidator()
        validation_results = validator.validate(prices)

        failed = [r for r in validation_results if not r.is_valid]
        if failed:
            raise RuntimeError(
                f"Validation failed for: {[r.ticker for r in failed]}. "
                f"Issues: {[r.issues for r in failed]}"
            )

        # Stage 3: Compute returns and calibrate
        logger.info("Stage 3: Calibrating model")
        log_returns = compute_returns(prices)
        engine = MonteCarloEngine(self.config)
        annualized_mean, annualized_cov = engine.calibrate(log_returns)

        # Stage 4: Run simulation
        logger.info("Stage 4: Running Monte Carlo simulation")
        simulation_result = engine.simulate(annualized_mean, annualized_cov)

        # Stage 5: Compute risk metrics
        logger.info("Stage 5: Computing risk metrics")
        calculator = RiskCalculator(self.config)
        metrics = calculator.compute_all(simulation_result, log_returns, annualized_cov)

        # Stage 6: Report
        print_risk_report(self.config, metrics, simulation_result)

        logger.info("Pipeline complete")
        return metrics

What does this code do?

The RiskPipeline class is the single entry point for the entire risk calculation workflow.

• The run() method is async because the data fetching stage uses asyncio to download from Yahoo Finance concurrently.
• Six stages run sequentially. Each stage logs its start so you can trace timing and identify bottlenecks.
• If any ticker fails validation, the pipeline raises a RuntimeError with details about which tickers failed and why. This prevents running simulations on bad data.

• Save risk_engine/pipeline.py.

You won't see a change yet. The next chunk adds the entry point that calls this pipeline.

Add the entry point and run the engine

Python's __main__.py convention lets you run an entire package as a script with python -m risk_engine. This file sets up logging, instantiates the pipeline, and wraps the execution in error handling so failures produce a clean log instead of a raw traceback.

• Open risk_engine/__main__.py and replace its contents with the following code:

"""Entry point for the risk engine."""

import asyncio
import logging
import sys
from pathlib import Path

from risk_engine.config import setup_logging
from risk_engine.pipeline import RiskPipeline


def main() -> None:
    """Run the risk engine."""
    setup_logging(log_level="INFO")
    logger = logging.getLogger(__name__)

    config_path = Path("config.yaml")

    if not config_path.exists():
        logger.error(f"Configuration file not found: {config_path}")
        sys.exit(1)

    try:
        pipeline = RiskPipeline(config_path)
        asyncio.run(pipeline.run())
    except KeyboardInterrupt:
        logger.info("Pipeline interrupted by user")
    except Exception as e:
        logger.error(f"Pipeline failed: {e}", exc_info=True)
        sys.exit(1)


if __name__ == "__main__":
    main()

What does this code do?

• setup_logging(log_level="INFO") activates the structured JSON log formatter you built in config.py. Every log line from the pipeline will be valid JSON.
• asyncio.run(pipeline.run()) creates an event loop and runs the async pipeline to completion.
• The try/except block catches any unhandled exception, logs it with full stack trace (exc_info=True), and exits with code 1 so that automated systems know the run failed.

• Save risk_engine/__main__.py.
• Run the full pipeline by executing this command in your terminal:

python -m risk_engine

You should see structured JSON log lines for each stage (fetching data, validating, calibrating, simulating, computing risk), followed by the formatted risk report with VaR, CVaR, portfolio metrics, and a risk attribution bar chart.

Seeing an error?

• If you see a ModuleNotFoundError, make sure your virtual environment is activated and that you ran pip install -r requirements.txt in a previous step.
• If you see a file-not-found error for config.yaml, make sure you are running the command from the monte-carlo-risk-engine/ root directory, not from inside risk_engine/.
• If a ticker fetch fails (network timeout or invalid symbol), check your internet connection and verify the tickers in config.yaml are valid stock symbols.

Why isn't my risk engine running successfully?

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for each file created in this step.

"""Formatted risk report output."""

from risk_engine.models import PortfolioConfig, RiskMetrics, SimulationResult


def print_risk_report(
    config: PortfolioConfig,
    metrics: RiskMetrics,
    simulation: SimulationResult,
) -> None:
    """Print a formatted risk report to console."""
    print("\n" + "=" * 60)
    print("  MONTE CARLO PORTFOLIO RISK REPORT")
    print("=" * 60)

    print(f"\n  Portfolio: {', '.join(config.tickers)}")
    print(f"  Notional: ${config.notional:,.0f}")
    print(f"  Simulations: {config.num_simulations:,}")
    print(f"  Horizon: {config.time_horizon_days} days")
    print(f"  Simulation time: {simulation.elapsed_seconds:.3f}s")

    print("\n" + "-" * 60)
    print("  VALUE AT RISK (VaR)")
    print("-" * 60)
    print(f"  95% VaR:  ${metrics.var_95:>12,.0f}  ({metrics.var_95/config.notional*100:.2f}%)")
    print(f"  99% VaR:  ${metrics.var_99:>12,.0f}  ({metrics.var_99/config.notional*100:.2f}%)")

    print("\n" + "-" * 60)
    print("  EXPECTED SHORTFALL (CVaR)")
    print("-" * 60)
    print(f"  95% CVaR: ${metrics.cvar_95:>12,.0f}  ({metrics.cvar_95/config.notional*100:.2f}%)")
    print(f"  99% CVaR: ${metrics.cvar_99:>12,.0f}  ({metrics.cvar_99/config.notional*100:.2f}%)")

    print("\n" + "-" * 60)
    print("  PORTFOLIO METRICS")
    print("-" * 60)
    print(f"  Annualized Volatility: {metrics.annualized_volatility*100:>8.2f}%")
    print(f"  Sharpe Ratio:          {metrics.annualized_sharpe:>8.3f}")
    print(f"  Max Drawdown:          {metrics.max_drawdown*100:>8.2f}%")

    print("\n" + "-" * 60)
    print("  RISK ATTRIBUTION (% of total portfolio risk)")
    print("-" * 60)
    for ticker, contrib in metrics.risk_contributions.items():
        bar = "#" * int(abs(contrib) * 40)
        print(f"  {ticker:<6} {contrib*100:>6.1f}%  {bar}")

    print("\n" + "=" * 60)
    print()

"""Async orchestrator for the risk pipeline."""

import asyncio
import logging
from pathlib import Path

from risk_engine.config import load_config, setup_logging
from risk_engine.data import DataFetcher, DataValidator, compute_returns
from risk_engine.models import PortfolioConfig, RiskMetrics
from risk_engine.report import print_risk_report
from risk_engine.risk import RiskCalculator
from risk_engine.simulation import MonteCarloEngine


logger = logging.getLogger(__name__)


class RiskPipeline:
    """Orchestrates the full risk calculation pipeline."""

    def __init__(self, config_path: Path) -> None:
        self.config_path = config_path
        self.config: PortfolioConfig | None = None

    async def run(self) -> RiskMetrics:
        """Execute the full pipeline."""
        # Load configuration
        self.config = load_config(self.config_path)
        logger.info("Pipeline started")

        # Stage 1: Fetch data
        logger.info("Stage 1: Fetching market data")
        fetcher = DataFetcher(self.config)
        prices = await fetcher.fetch_all()

        # Stage 2: Validate data
        logger.info("Stage 2: Validating data")
        validator = DataValidator()
        validation_results = validator.validate(prices)

        failed = [r for r in validation_results if not r.is_valid]
        if failed:
            raise RuntimeError(
                f"Validation failed for: {[r.ticker for r in failed]}. "
                f"Issues: {[r.issues for r in failed]}"
            )

        # Stage 3: Compute returns and calibrate
        logger.info("Stage 3: Calibrating model")
        log_returns = compute_returns(prices)
        engine = MonteCarloEngine(self.config)
        annualized_mean, annualized_cov = engine.calibrate(log_returns)

        # Stage 4: Run simulation
        logger.info("Stage 4: Running Monte Carlo simulation")
        simulation_result = engine.simulate(annualized_mean, annualized_cov)

        # Stage 5: Compute risk metrics
        logger.info("Stage 5: Computing risk metrics")
        calculator = RiskCalculator(self.config)
        metrics = calculator.compute_all(simulation_result, log_returns, annualized_cov)

        # Stage 6: Report
        print_risk_report(self.config, metrics, simulation_result)

        logger.info("Pipeline complete")
        return metrics

"""Entry point for the risk engine."""

import asyncio
import logging
import sys
from pathlib import Path

from risk_engine.config import setup_logging
from risk_engine.pipeline import RiskPipeline


def main() -> None:
    """Run the risk engine."""
    setup_logging(log_level="INFO")
    logger = logging.getLogger(__name__)

    config_path = Path("config.yaml")

    if not config_path.exists():
        logger.error(f"Configuration file not found: {config_path}")
        sys.exit(1)

    try:
        pipeline = RiskPipeline(config_path)
        asyncio.run(pipeline.run())
    except KeyboardInterrupt:
        logger.info("Pipeline interrupted by user")
    except Exception as e:
        logger.error(f"Pipeline failed: {e}", exc_info=True)
        sys.exit(1)


if __name__ == "__main__":
    main()

Secret mission

Add a Pytest Test Suite

Your risk engine works, but how do you prove it? Monte Carlo outputs are random by definition. Add a pytest test suite that uses statistical assertions to verify your simulation correctness, risk metric invariants, and data validation logic.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources for future extension or delete them entirely.

Resources you used:

• Local project directory monte-carlo-risk-engine/ on your Desktop.
• Python virtual environment .venv with installed pip packages (numpy, pandas, scipy, aiohttp, yfinance, pyyaml, pytest).

Keep everything running

No action needed. Choose this if you plan to extend the engine with scenario analysis, portfolio optimization, or backtesting.

• Your project directory and virtual environment stay on your Desktop ready to use.
• To reactivate the environment later, navigate to the project folder and run source .venv/bin/activate (macOS/Linux) or .venv\Scripts\activate (Windows).

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files stay intact on disk.

• Deactivate the virtual environment by running:

deactivate

• Your project files remain in ~/Desktop/monte-carlo-risk-engine/ for when you return.

Delete - I don't want to use this again

Remove all project files and the virtual environment to start fresh.

• Deactivate the virtual environment first by running:

deactivate

• Delete the entire project directory (macOS/Linux) by running:

cd ~/Desktop
rm -rf monte-carlo-risk-engine

• On Windows, run:

cd ~\Desktop
rmdir /s /q monte-carlo-risk-engine

• This removes the project folder, the virtual environment, and all installed pip packages.

Well Done!

Well Done!

Nice work! You just built a production-grade async Monte Carlo portfolio risk engine from scratch in Python.

You've learned how to:

• Built a Monte Carlo simulation engine using Geometric Brownian Motion and Cholesky decomposition to generate 10,000 correlated return paths across a multi-asset portfolio.
• Computed Value at Risk and CVaR (Expected Shortfall) at 95% and 99% confidence levels, plus component risk attribution showing each asset's contribution to total portfolio risk.
• Designed a YAML-configured, async pipeline with OOP architecture, type hints, structured JSON logging, and error handling that fetches and validates real market data concurrently.
• Secret Mission: Added a pytest test suite that validates stochastic simulation correctness using statistical assertions.

Ready to quiz yourself?