As you approach the challenge of building a compliance data dashboard, recall how **pandas** enables you to efficiently analyze transaction data and how **matplotlib** helps you visualize patterns that may be relevant for compliance monitoring. These tools are essential for summarizing large datasets, identifying anomalies, and communicating findings to stakeholders in a compliance context.

import pandas as pd
import matplotlib.pyplot as plt

# Hardcoded transaction data
data = {
    "id": [1, 2, 3, 4, 5, 6, 7, 8],
    "amount": [1200, 500, 3200, 1500, 700, 2200, 4000, 800],
    "account_id": ["A01", "A02", "A01", "A03", "A02", "A01", "A03", "A02"]
}
df = pd.DataFrame(data)

# Calculate summary statistics
total_amount = df["amount"].sum()
average_amount = df["amount"].mean()
max_amount = df["amount"].max()

print("Total transaction amount:", total_amount)
print("Average transaction amount:", average_amount)
print("Maximum transaction amount:", max_amount)

# Aggregate by account for bar chart
account_totals = df.groupby("account_id")["amount"].sum()
account_totals.plot(kind="bar", title="Total Transaction Amounts per Account")
plt.xlabel("Account ID")
plt.ylabel("Total Amount")
plt.tight_layout()
plt.show()

# Identify anomalies: transactions > 2 std dev above mean
mean_amount = df["amount"].mean()
std_amount = df["amount"].std()
anomaly_threshold = mean_amount + 2 * std_amount
anomalies = df[df["amount"] > anomaly_threshold]

print("Transactions above anomaly threshold (2 std dev):")
print(anomalies)

To structure your compliance dashboard, follow a clear sequence: start by loading and summarizing your transaction data, using **pandas** to calculate totals and averages. Next, use **matplotlib** to visualize aggregated results, such as total transactions per account, to spot trends or concentrations. Finally, apply statistical analysis to flag any transactions that are unusually large—those that are more than two standard deviations above the mean—helping you quickly detect potential compliance risks.


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

class TestTask(unittest.TestCase):
    def setUp(self):
        self.expected_data = {
            "id": [1, 2, 3, 4, 5, 6, 7, 8],
            "amount": [1200, 500, 3200, 1500, 700, 2200, 4000, 800],
            "account_id": ["A01", "A02", "A01", "A03", "A02", "A01", "A03", "A02"]
        }
        self.df = pd.DataFrame(self.expected_data)

    def test_total_amount(self):
        import user_code
        importlib.reload(user_code)
        actual = getattr(user_code, 'total_amount', None)
        expected = self.df["amount"].sum()
        _dynamic_test(
            self,
            isinstance(actual, (int, float)) and actual == expected,
            "Total transaction amount is correct.",
            f"Expected total_amount={expected}, got {actual}.",
        )

    def test_average_amount(self):
        import user_code
        importlib.reload(user_code)
        actual = getattr(user_code, 'average_amount', None)
        expected = self.df["amount"].mean()
        _dynamic_test(
            self,
            isinstance(actual, (int, float)) and actual == expected,
            "Average transaction amount is correct.",
            f"Expected average_amount={expected}, got {actual}.",
        )

    def test_max_amount(self):
        import user_code
        importlib.reload(user_code)
        actual = getattr(user_code, 'max_amount', None)
        expected = self.df["amount"].max()
        _dynamic_test(
            self,
            isinstance(actual, (int, float)) and actual == expected,
            "Maximum transaction amount is correct.",
            f"Expected max_amount={expected}, got {actual}.",
        )

    def test_account_totals(self):
        import user_code
        importlib.reload(user_code)
        actual = getattr(user_code, 'account_totals', None)
        expected = self.df.groupby("account_id")["amount"].sum()
        _dynamic_test(
            self,
            isinstance(actual, pd.Series) and actual.equals(expected),
            "Account totals Series is correct.",
            f"Expected account_totals=\\n{expected}\\ngot:\\n{actual}",
        )

    def test_anomalies(self):
        import user_code
        importlib.reload(user_code)
        actual = getattr(user_code, 'anomalies', None)
        mean_amount = self.df["amount"].mean()
        std_amount = self.df["amount"].std()
        threshold = mean_amount + 2 * std_amount
        expected = self.df[self.df["amount"] > threshold]
        _dynamic_test(
            self,
            isinstance(actual, pd.DataFrame) and actual.equals(expected),
            "Anomalies DataFrame is correct.",
            f"Expected anomalies=\\n{expected}\\ngot:\\n{actual}",
        )

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

Focused on advanced automation and monitoring of compliance in MLOps using Python. Covers rule-based compliance systems, anomaly detection, and regulatory validation most relevant to security and governance in machine learning operations.

Challenge: Compliance Data Dashboard

解答