Rolling volatility, measured as the **standard deviation** of returns over a moving window, is a key indicator of how much a stock's price fluctuates over time. In financial analysis, understanding **volatility** helps you assess risk, compare assets, and make informed investment decisions. By tracking rolling volatility, you can identify periods of market turbulence or calm, spot trends in risk, and better manage portfolio exposure.

import pandas as pd
import numpy as np

# Create a date range for one year of business days
dates = pd.date_range(start="2023-01-01", end="2023-12-31", freq="B")

# Simulate daily returns for four stocks with different volatility levels
np.random.seed(42)
returns_data = {
    "AAPL": np.random.normal(0, 0.012, len(dates)),
    "MSFT": np.random.normal(0, 0.010, len(dates)),
    "GOOGL": np.random.normal(0, 0.015, len(dates)),
    "AMZN": np.random.normal(0, 0.018, len(dates)),
}
returns_df = pd.DataFrame(returns_data, index=dates)

# Display the first few rows
print(returns_df.head())

To analyze and visualize the volatility of these stocks, you will calculate the 30-day rolling standard deviation of returns for each stock. This rolling standard deviation represents the recent volatility and is commonly used to monitor risk over time. After computing the rolling volatility, plot all four stocks' rolling volatility on a single chart using **matplotlib**. Make sure to plot each stock as a separate line, label each line with the stock symbol, and add a legend so viewers can easily distinguish between the stocks. Label the axes and provide a clear title to make your plot informative and professional.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import sys
from unittest.mock import patch

class TestTask(unittest.TestCase):
    def setUp(self):
        # Prepare reproducible returns_df
        self.dates = pd.date_range(start="2023-01-01", end="2023-12-31", freq="B")
        np.random.seed(42)
        returns_data = {
            "AAPL": np.random.normal(0, 0.012, len(self.dates)),
            "MSFT": np.random.normal(0, 0.010, len(self.dates)),
            "GOOGL": np.random.normal(0, 0.015, len(self.dates)),
            "AMZN": np.random.normal(0, 0.018, len(self.dates)),
        }
        self.returns_df = pd.DataFrame(returns_data, index=self.dates)
        # Import user_code fresh for every test
        import user_code
        importlib.reload(user_code)

    def test_rolling_std_calculation(self):
        import user_code
        importlib.reload(user_code)
        # Patch plt.show to prevent display
        with patch('matplotlib.pyplot.show'):
            # Capture rolling std inside the function
            called = {}
            def fake_plot(*args, **kwargs):
                called['args'] = args
                called['kwargs'] = kwargs
            orig_plot = plt.plot
            plt.plot = fake_plot
            try:
                user_code.plot_rolling_volatility(self.returns_df)
                # Recalculate rolling std for comparison
                rolling_vol = self.returns_df.rolling(window=30).std()
                # Check that rolling std is correct for each stock (after first 30 days)
                for stock in self.returns_df.columns:
                    user_rolling = rolling_vol[stock].dropna()
                    # Check rolling std is not all NaN and has correct length
                    _dynamic_test(
                        self,
                        not user_rolling.isnull().all() and len(user_rolling) == len(self.returns_df)-29,
                        f"Rolling std for {stock} calculated correctly",
                        f"Rolling std for {stock} not calculated correctly: {user_rolling}",
                    )
            finally:
                plt.plot = orig_plot

    def test_plot_lines_and_labels(self):
        import user_code
        importlib.reload(user_code)
        # Patch plt.show and plt.plot to intercept plotting calls
        lines_drawn = []
        orig_plot = plt.plot
        with patch('matplotlib.pyplot.show'):
            def fake_plot(x, y, label=None, **kwargs):
                lines_drawn.append(label)
                return orig_plot(x, y, label=label, **kwargs)
            plt.plot = fake_plot
            try:
                user_code.plot_rolling_volatility(self.returns_df)
            finally:
                plt.plot = orig_plot
        # Check that each stock is plotted with correct label
        expected_labels = set(self.returns_df.columns)
        actual_labels = set(lines_drawn)
        _dynamic_test(
            self,
            expected_labels.issubset(actual_labels),
            f"All four stocks are plotted and labeled: {expected_labels}",
            f"Missing labels in plot: {expected_labels - actual_labels}",
        )

    def test_legend_present(self):
        import user_code
        importlib.reload(user_code)
        # Patch plt.show and plt.legend to check if legend is called
        legend_called = {'called': False}
        orig_legend = plt.legend
        with patch('matplotlib.pyplot.show'):
            def fake_legend(*args, **kwargs):
                legend_called['called'] = True
                return orig_legend(*args, **kwargs)
            plt.legend = fake_legend
            try:
                user_code.plot_rolling_volatility(self.returns_df)
            finally:
                plt.legend = orig_legend
        _dynamic_test(
            self,
            legend_called['called'],
            "Legend is present on the plot",
            "Legend is missing from the plot",
        )

    def test_axis_labels_and_title(self):
        import user_code
        importlib.reload(user_code)
        # Patch plt.show and plt.xlabel, plt.ylabel, plt.title
        labels = {'xlabel': None, 'ylabel': None, 'title': None}
        orig_xlabel = plt.xlabel
        orig_ylabel = plt.ylabel
        orig_title = plt.title
        with patch('matplotlib.pyplot.show'):
            plt.xlabel = lambda x: labels.update({'xlabel': x}) or orig_xlabel(x)
            plt.ylabel = lambda y: labels.update({'ylabel': y}) or orig_ylabel(y)
            plt.title = lambda t: labels.update({'title': t}) or orig_title(t)
            try:
                user_code.plot_rolling_volatility(self.returns_df)
            finally:
                plt.xlabel = orig_xlabel
                plt.ylabel = orig_ylabel
                plt.title = orig_title
        _dynamic_test(
            self,
            normalize_text(labels['xlabel']) == normalize_text('Date'),
            "X-axis labeled as 'Date'",
            f"X-axis label incorrect: {labels['xlabel']}"
        )
        _dynamic_test(
            self,
            normalize_text(labels['ylabel']) == normalize_text('Rolling Std Dev (Volatility)'),
            "Y-axis labeled as 'Rolling Std Dev (Volatility)'",
            f"Y-axis label incorrect: {labels['ylabel']}"
        )
        _dynamic_test(
            self,
            'volatility' in normalize_text(labels['title']),
            "Plot has a descriptive title containing 'volatility'",
            f"Plot title does not mention volatility: {labels['title']}"
        )

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: Visualize Stock Volatility

解答