In banking, transaction data often arrives with missing values and duplicate records, which can hinder accurate analysis and reporting. As you work with financial DataFrames, it's crucial to ensure that the data is clean, consistent, and ready for downstream processing. Your task is to take a DataFrame containing transaction records, some of which have missing amounts and duplicate entries, and prepare it for further use by addressing these common data quality issues.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import pandas as pd
import sys
from io import StringIO

class TestTask(unittest.TestCase):
    def setUp(self):
        self.data = {
            'TransactionID': [101, 102, 103, 104, 102],
            'Date': ['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-04', '2023-01-02'],
            'Amount': [150.0, None, 200.0, None, None]
        }
        self.df = pd.DataFrame(self.data)
        self.expected_data = {
            'TransactionID': [101, 102, 103, 104],
            'Date': ['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-04'],
            'Amount': [150.0, 0.0, 200.0, 0.0]
        }
        self.expected_df = pd.DataFrame(self.expected_data)

    def test_missing_amounts_filled_with_zero(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.clean_transaction_data(self.df.copy())
        missing_filled = result['Amount'].isna().sum() == 0 and (result['Amount'] == 0).sum() >= 1
        _dynamic_test(
            self,
            missing_filled,
            "All missing values in Amount column are filled with zero.",
            "Missing values in Amount column were not filled with zero correctly."
        )

    def test_duplicates_removed(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.clean_transaction_data(self.df.copy())
        # After removing duplicates, should match expected rows
        no_duplicates = result.drop_duplicates().shape[0] == self.expected_df.shape[0]
        _dynamic_test(
            self,
            no_duplicates,
            "All duplicate rows are removed from the DataFrame.",
            f"Expected {self.expected_df.shape[0]} rows after removing duplicates, got {result.shape[0]}."
        )

    def test_amount_column_is_float(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.clean_transaction_data(self.df.copy())
        is_float = result['Amount'].map(type).eq(float).all()
        _dynamic_test(
            self,
            is_float,
            "All values in the Amount column are of type float.",
            "Not all values in the Amount column are of type float."
        )

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

A practical Python course tailored for banking professionals, focusing on real-world financial data analysis, risk assessment, and automation tasks. Learn how Python can streamline banking operations, analyze financial data, and solve common problems faced by bankers.

Learn how to use Python to analyze, manipulate, and visualize financial data relevant to banking operations.

Apply Python to assess credit risk, analyze loan portfolios, and perform key calculations relevant to banking risk management.

Explore how Python can automate banking tasks and help detect fraudulent activities using data analysis techniques.

Challenge: Clean Transaction Data

解答