
import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import numpy as np
import pandas as pd

class TestTask(unittest.TestCase):
    def test_metrics_simple(self):
        import user_code
        importlib.reload(user_code)
        # Simple 2-asset, 5-day example
        prices = pd.DataFrame({
            'AAPL': [100, 101, 102, 103, 104],
            'MSFT': [200, 202, 204, 206, 208]
        })
        weights = [0.5, 0.5]
        res = user_code.calculate_portfolio_metrics(prices, weights)
        _dynamic_test(
            self,
            res is not None and isinstance(res, dict),
            "Function returns a dictionary",
            f"Expected dict, got: {type(res)}"
        )
        # Calculate expected values
        daily_returns = prices.pct_change().dropna()
        # Portfolio daily returns
        port_daily = daily_returns @ np.array(weights)
        mean_daily = np.mean(port_daily)
        expected_annual_return = mean_daily * 252
        cov = daily_returns.cov()
        port_var = np.array(weights).T @ cov.values @ np.array(weights)
        annual_vol = np.sqrt(port_var) * np.sqrt(252)
        sharpe = expected_annual_return / annual_vol if annual_vol != 0 else np.nan
        _dynamic_test(
            self,
            np.isclose(res['expected_annual_return'], expected_annual_return, rtol=1e-6),
            "Expected annual return is correct",
            f"Expected {expected_annual_return}, got {res['expected_annual_return']}"
        )
        _dynamic_test(
            self,
            np.isclose(res['annual_volatility'], annual_vol, rtol=1e-6),
            "Annual volatility is correct",
            f"Expected {annual_vol}, got {res['annual_volatility']}"
        )
        _dynamic_test(
            self,
            np.isclose(res['sharpe_ratio'], sharpe, rtol=1e-6),
            "Sharpe ratio is correct",
            f"Expected {sharpe}, got {res['sharpe_ratio']}"
        )

    def test_metrics_single_asset(self):
        import user_code
        importlib.reload(user_code)
        # Single asset
        prices = pd.DataFrame({
            'AAPL': [50, 51, 52, 53, 54]
        })
        weights = [1.0]
        res = user_code.calculate_portfolio_metrics(prices, weights)
        daily_returns = prices.pct_change().dropna().iloc[:,0]
        mean_daily = np.mean(daily_returns)
        expected_annual_return = mean_daily * 252
        annual_vol = np.std(daily_returns, ddof=1) * np.sqrt(252)
        sharpe = expected_annual_return / annual_vol if annual_vol != 0 else np.nan
        _dynamic_test(
            self,
            np.isclose(res['expected_annual_return'], expected_annual_return, rtol=1e-6),
            "Expected annual return is correct for single asset",
            f"Expected {expected_annual_return}, got {res['expected_annual_return']}"
        )
        _dynamic_test(
            self,
            np.isclose(res['annual_volatility'], annual_vol, rtol=1e-6),
            "Annual volatility is correct for single asset",
            f"Expected {annual_vol}, got {res['annual_volatility']}"
        )
        _dynamic_test(
            self,
            np.isclose(res['sharpe_ratio'], sharpe, rtol=1e-6),
            "Sharpe ratio is correct for single asset",
            f"Expected {sharpe}, got {res['sharpe_ratio']}"
        )

    def test_weights_sum_to_one(self):
        import user_code
        importlib.reload(user_code)
        # 3-asset, weights sum to 1
        prices = pd.DataFrame({
            'A': [10, 11, 12, 13, 14],
            'B': [20, 19, 18, 17, 16],
            'C': [30, 32, 31, 33, 35]
        })
        weights = [0.2, 0.5, 0.3]
        res = user_code.calculate_portfolio_metrics(prices, weights)
        _dynamic_test(
            self,
            isinstance(res, dict) and all(k in res for k in ['expected_annual_return','annual_volatility','sharpe_ratio']),
            "Function returns all required keys",
            f"Missing keys in result: {res}"
        )
        # No exception or nan for valid data
        _dynamic_test(
            self,
            not any(pd.isna(v) for v in res.values()),
            "No NaN values in result for valid data",
            f"Got NaN in result: {res}"
        )

    def test_zero_volatility(self):
        import user_code
        importlib.reload(user_code)
        # All prices constant, volatility should be zero, Sharpe should be nan or inf
        prices = pd.DataFrame({
            'AAPL': [100,100,100,100,100],
            'MSFT': [200,200,200,200,200]
        })
        weights = [0.5, 0.5]
        res = user_code.calculate_portfolio_metrics(prices, weights)
        _dynamic_test(
            self,
            res['annual_volatility'] == 0,
            "Annual volatility is zero for constant prices",
            f"Expected volatility 0, got {res['annual_volatility']}"
        )
        _dynamic_test(
            self,
            np.isnan(res['sharpe_ratio']) or np.isinf(res['sharpe_ratio']),
            "Sharpe ratio is nan or inf when volatility is zero",
            f"Expected NaN or inf, got {res['sharpe_ratio']}"
        )

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

Unlock the power of Python to analyze financial data, evaluate investment opportunities, and automate key investor workflows. This course guides you through practical, real-world scenarios using Python, focusing on data analysis, visualization, and quantitative techniques relevant to investors.

Learn how to use Python to analyze and visualize financial data, focusing on the core skills investors need to interpret market trends and make informed decisions.

Dive deeper into key investment metrics and learn how to analyze and visualize portfolios using Python.

Apply advanced Python techniques to automate investment analysis and leverage statistical and machine learning tools for deeper insights.

Challenge: Automate Portfolio Metrics Calculation

解答