As you have learned, **mean-variance optimization** is a cornerstone of modern portfolio theory. It allows you to select asset weights that maximize expected return for a given level of risk, or minimize risk for a given expected return. However, in practice, the analytical solution can become complex, especially with more assets and constraints. This is where **simulation-based approaches**—such as Monte Carlo simulations—become powerful. By generating thousands of random portfolios and calculating their risk and return, you can visualize the opportunity set and identify optimal portfolios based on your chosen criteria, such as the Sharpe ratio.

import numpy as np

# Example expected annual returns for three assets
expected_returns = np.array([0.08, 0.12, 0.10])  # Asset A, B, C

# Example standard deviations (annualized)
std_devs = np.array([0.15, 0.20, 0.18])

# Example correlation matrix
correlation_matrix = np.array([
    [1.0, 0.3, 0.5],
    [0.3, 1.0, 0.4],
    [0.5, 0.4, 1.0]
])

To proceed, you will first need to construct the covariance matrix using the standard deviations and correlation matrix. Then, simulate 10,000 random portfolios by generating random weights for each asset (ensuring all weights are non-negative and sum to one—no short selling). For each simulated portfolio, calculate the expected return as the weighted sum of asset returns, and the risk as the portfolio's standard deviation using the covariance matrix. Finally, plot all simulated portfolios on a scatter plot of risk (x-axis) versus return (y-axis), and highlight the portfolio with the highest Sharpe ratio (assuming a risk-free rate of zero).


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import numpy as np
import sys
import io

class TestTask(unittest.TestCase):
    def setUp(self):
        import user_code
        importlib.reload(user_code)
        self.expected_returns = np.array([0.08, 0.12, 0.10])
        self.std_devs = np.array([0.15, 0.20, 0.18])
        self.correlation_matrix = np.array([
            [1.0, 0.3, 0.5],
            [0.3, 1.0, 0.4],
            [0.5, 0.4, 1.0]
        ])

    def test_covariance_matrix(self):
        import user_code
        importlib.reload(user_code)
        # Reconstruct covariance matrix from solution
        cov_matrix = np.outer(self.std_devs, self.std_devs) * self.correlation_matrix
        # Try to extract covariance matrix from user function
        # We'll monkey-patch np.show_cov_matrix in the user's function to capture the value, or check by re-running
        # Instead, run the function and check if the covariance matrix is constructed correctly
        # We'll patch np.dot to check if the correct matrix is used
        # Instead, we'll just call the function and check if it runs without error and outputs plausible results
        try:
            func = getattr(user_code, 'simulate_portfolios', None)
            _dynamic_test(self, func is not None, "Function simulate_portfolios exists", "Function simulate_portfolios is missing")
            # Call the function with a small number of portfolios for speed
            weights, portfolio = func(self.expected_returns, self.std_devs, self.correlation_matrix, num_portfolios=5)
            # If it runs, check if the results are plausible
            _dynamic_test(self, isinstance(weights, np.ndarray) or isinstance(weights, list), "Function returns weights as array or list", "Returned weights are not array or list")
            _dynamic_test(self, isinstance(portfolio, np.ndarray) or isinstance(portfolio, list), "Function returns portfolio as array or list", "Returned portfolio is not array or list")
        except Exception as e:
            _dynamic_test(self, False, "Function runs without error", f"Function raised an error: {e}")

    def test_weights_sum_to_one_and_nonnegative(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'simulate_portfolios', None)
        _dynamic_test(self, func is not None, "Function simulate_portfolios exists", "Function simulate_portfolios is missing")
        # Call the function with a small number of portfolios for speed
        weights, portfolio = func(self.expected_returns, self.std_devs, self.correlation_matrix, num_portfolios=20)
        # Check returned optimal weights
        weights = np.array(weights)
        _dynamic_test(self, np.all(weights >= 0), "Optimal weights are non-negative", f"Some optimal weights are negative: {weights}")
        _dynamic_test(self, np.isclose(np.sum(weights), 1.0), "Optimal weights sum to one", f"Optimal weights do not sum to one: {weights} (sum={np.sum(weights)})")

    def test_expected_return_and_risk_calculation(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'simulate_portfolios', None)
        _dynamic_test(self, func is not None, "Function simulate_portfolios exists", "Function simulate_portfolios is missing")
        # Call with 1 portfolio, check formula
        weights = np.array([0.5, 0.3, 0.2])
        cov_matrix = np.outer(self.std_devs, self.std_devs) * self.correlation_matrix
        expected_return = np.dot(weights, self.expected_returns)
        expected_var = np.dot(weights.T, np.dot(cov_matrix, weights))
        expected_std = np.sqrt(expected_var)
        # Patch numpy random to return our weights
        np_random_rand = np.random.rand
        np.random.rand = lambda n: weights.copy()
        try:
            opt_weights, opt_portfolio = func(self.expected_returns, self.std_devs, self.correlation_matrix, num_portfolios=1)
            # opt_portfolio: [return, risk, sharpe]
            _dynamic_test(self, np.isclose(opt_portfolio[0], expected_return), "Expected return is calculated correctly", f"Expected return: {expected_return}, got: {opt_portfolio[0]}")
            _dynamic_test(self, np.isclose(opt_portfolio[1], expected_std), "Risk (std) is calculated correctly", f"Expected std: {expected_std}, got: {opt_portfolio[1]}")
        finally:
            np.random.rand = np_random_rand

    def test_sharpe_ratio_calculation(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'simulate_portfolios', None)
        _dynamic_test(self, func is not None, "Function simulate_portfolios exists", "Function simulate_portfolios is missing")
        # Call with 1 portfolio, check formula
        weights = np.array([0.3, 0.3, 0.4])
        cov_matrix = np.outer(self.std_devs, self.std_devs) * self.correlation_matrix
        expected_return = np.dot(weights, self.expected_returns)
        expected_var = np.dot(weights.T, np.dot(cov_matrix, weights))
        expected_std = np.sqrt(expected_var)
        expected_sharpe = expected_return / expected_std if expected_std > 0 else 0
        # Patch numpy random to return our weights
        np_random_rand = np.random.rand
        np.random.rand = lambda n: weights.copy()
        try:
            opt_weights, opt_portfolio = func(self.expected_returns, self.std_devs, self.correlation_matrix, num_portfolios=1)
            _dynamic_test(self, np.isclose(opt_portfolio[2], expected_sharpe), "Sharpe ratio is calculated correctly", f"Expected Sharpe: {expected_sharpe}, got: {opt_portfolio[2]}")
        finally:
            np.random.rand = np_random_rand

    def test_plot_and_highlight(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'simulate_portfolios', None)
        _dynamic_test(self, func is not None, "Function simulate_portfolios exists", "Function simulate_portfolios is missing")
        # Patch plt.scatter and plt.show to check if called
        import matplotlib.pyplot as plt
        scatter_calls = []
        show_called = [False]
        orig_scatter = plt.scatter
        orig_show = plt.show
        def fake_scatter(*args, **kwargs):
            scatter_calls.append((args, kwargs))
            return None
        def fake_show(*args, **kwargs):
            show_called[0] = True
            return None
        plt.scatter = fake_scatter
        plt.show = fake_show
        try:
            func(self.expected_returns, self.std_devs, self.correlation_matrix, num_portfolios=20)
            _dynamic_test(self, len(scatter_calls) >= 2, "Scatter plot is called at least twice (portfolios and optimal)", f"Scatter plot not called enough times: {len(scatter_calls)}")
            _dynamic_test(self, show_called[0], "Plot show is called", "Plot show() was not called")
        finally:
            plt.scatter = orig_scatter
            plt.show = orig_show

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


test_main.py

Master the essential Python skills needed for modern financial analysis. This course guides financial analysts through practical Python applications, from data manipulation and visualization to risk assessment and portfolio optimization, using real-world financial scenarios.

Learn how to use Python and pandas to structure, clean, and analyze financial data for actionable insights.

Master the art of visualizing financial data and uncovering trends using Python's powerful plotting libraries.

Apply Python to assess financial risk and optimize investment portfolios using real-world scenarios.

Challenge: Optimize a Three-Asset Portfolio

解答