Clustering is a practical approach in finance for grouping assets with similar return profiles, helping you understand which assets behave alike and thus manage portfolio risk more effectively. By applying clustering algorithms such as **KMeans**, you can identify patterns and relationships among assets that might not be obvious at first glance. This is especially valuable for risk management, as you can diversify your portfolio by ensuring assets from different clusters are included, reducing the chance of simultaneous losses.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib

class TestTask(unittest.TestCase):
    def test_returns_dict_with_all_assets(self):
        import user_code
        importlib.reload(user_code)
        asset_returns = {
            "AAPL": [0.01, 0.02, -0.01, 0.03],
            "GOOG": [0.015, 0.025, -0.005, 0.027],
            "TSLA": [-0.02, -0.03, 0.01, -0.025],
            "JPM": [0.005, 0.007, 0.002, 0.008],
        }
        result = user_code.cluster_asset_returns(asset_returns)
        _dynamic_test(
            self,
            isinstance(result, dict),
            "Function returns a dictionary.",
            f"Function should return a dictionary, got {type(result)}."
        )
        _dynamic_test(
            self,
            set(result.keys()) == set(asset_returns.keys()),
            "All asset names are present in the result.",
            f"Returned keys: {result.keys()} do not match input keys: {asset_returns.keys()}"
        )
    def test_labels_are_0_or_1(self):
        import user_code
        importlib.reload(user_code)
        asset_returns = {
            "AAPL": [0.01, 0.02, -0.01, 0.03],
            "GOOG": [0.015, 0.025, -0.005, 0.027],
            "TSLA": [-0.02, -0.03, 0.01, -0.025],
            "JPM": [0.005, 0.007, 0.002, 0.008],
        }
        result = user_code.cluster_asset_returns(asset_returns)
        _dynamic_test(
            self,
            all(label in [0, 1] for label in result.values()),
            "All cluster labels are 0 or 1.",
            f"Cluster labels must be 0 or 1, got: {set(result.values())}"
        )
    def test_similar_assets_in_same_cluster(self):
        import user_code
        importlib.reload(user_code)
        # AAPL and GOOG have similar returns, TSLA is different
        asset_returns = {
            "AAPL": [0.01, 0.02, -0.01, 0.03],
            "GOOG": [0.015, 0.025, -0.005, 0.027],
            "TSLA": [-0.02, -0.03, 0.01, -0.025],
            "JPM": [0.005, 0.007, 0.002, 0.008],
        }
        result = user_code.cluster_asset_returns(asset_returns)
        _dynamic_test(
            self,
            result is not None and result["AAPL"] == result["GOOG"],
            "AAPL and GOOG assigned to the same cluster.",
            f"Expected AAPL and GOOG to be in the same cluster, got: {result}"
        )
    def test_works_with_min_two_assets(self):
        import user_code
        importlib.reload(user_code)
        asset_returns = {
            "X": [0.01, 0.02, 0.03, 0.04],
            "Y": [-0.01, -0.02, -0.03, -0.04],
        }
        result = user_code.cluster_asset_returns(asset_returns)
        _dynamic_test(
            self,
            isinstance(result, dict) and set(result.keys()) == set(asset_returns.keys()),
            "Function works with two assets.",
            f"Expected keys: {asset_returns.keys()}, got: {result.keys() if isinstance(result, dict) else result}"
        )
        _dynamic_test(
            self,
            all(label in [0, 1] for label in result.values()),
            "Cluster labels are 0 or 1 for two assets.",
            f"Cluster labels must be 0 or 1, got: {set(result.values())}"
        )
    def test_works_with_larger_input(self):
        import user_code
        importlib.reload(user_code)
        asset_returns = {
            f"Asset{i}": [i * 0.01, i * 0.02, i * 0.03, i * 0.04] for i in range(1, 10)
        }
        result = user_code.cluster_asset_returns(asset_returns)
        _dynamic_test(
            self,
            isinstance(result, dict) and set(result.keys()) == set(asset_returns.keys()),
            "Function works with larger input.",
            f"Expected keys: {asset_returns.keys()}, got: {result.keys() if isinstance(result, dict) else result}"
        )
        _dynamic_test(
            self,
            all(label in [0, 1] for label in result.values()),
            "Cluster labels are 0 or 1 for larger input.",
            f"Cluster labels must be 0 or 1, got: {set(result.values())}"
        )

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

Explore how Python empowers FinTech solutions by automating financial analysis, modeling, and data-driven decision making. This course blends practical coding with real-world finance scenarios, guiding you through essential FinTech concepts and hands-on challenges.

Learn to analyze, manipulate, and visualize financial data using Python, focusing on core FinTech data workflows.

Apply Python to assess financial risk and optimize investment portfolios using statistical and mathematical techniques.

Apply machine learning techniques to financial data for prediction, classification, and decision support in FinTech applications.

Challenge: Cluster Asset Returns

解答