Automating transaction screening is a core use case for Python in compliance operations. By leveraging Python’s ability to process lists and apply logic, you can efficiently check transactions against multiple rules—such as amount thresholds and blacklisted accounts—without manual review. This approach ensures consistent, fast, and auditable screening of financial activity, helping you meet regulatory requirements and reduce the risk of oversight.

def screen_transactions(transactions, blacklist, threshold):
    flagged = []
    for tx in transactions:
        # Write your code here
        pass
    return flagged

transactions = [
    {"id": 1, "amount": 12000, "account_id": "A123", "timestamp": "2024-06-01T10:00:00"},
    {"id": 2, "amount": 500, "account_id": "B456", "timestamp": "2024-06-01T11:00:00"},
    {"id": 3, "amount": 25000, "account_id": "X999", "timestamp": "2024-06-01T12:00:00"},
    {"id": 4, "amount": 7000, "account_id": "C789", "timestamp": "2024-06-01T13:00:00"},
    {"id": 5, "amount": 300, "account_id": "Y888", "timestamp": "2024-06-01T14:00:00"},
]

blacklist = ["X999", "Y888"]
threshold = 10000

flagged = screen_transactions(transactions, blacklist, threshold)
print(flagged)

To build your screening function, structure your logic so that you loop through each transaction, check if it violates the amount threshold or appears in the blacklist, and collect any flagged transactions with the appropriate reason. For each violation, append a new dictionary with the transaction `id` and the reason for flagging to your results list. This modular approach makes it easy to extend screening logic as compliance requirements evolve.


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

class TestTask(unittest.TestCase):
    def test_flag_amount_rule(self):
        import user_code
        importlib.reload(user_code)
        txs = [
            {"id": 1, "amount": 12001, "account_id": "A123", "timestamp": "2024-06-01T10:00:00"},
            {"id": 2, "amount": 9999, "account_id": "B456", "timestamp": "2024-06-01T11:00:00"}
        ]
        blacklist = []
        threshold = 10000
        flagged = user_code.screen_transactions(txs, blacklist, threshold)
        expected = [{"id": 1, "reason": "amount"}]
        _dynamic_test(self,
            flagged == expected,
            "Correctly flags only transactions exceeding threshold.",
            f"Expected: {expected}, got: {flagged}")

    def test_flag_blacklist_rule(self):
        import user_code
        importlib.reload(user_code)
        txs = [
            {"id": 3, "amount": 500, "account_id": "BLK1", "timestamp": "2024-06-01T12:00:00"},
            {"id": 4, "amount": 800, "account_id": "SAFE", "timestamp": "2024-06-01T13:00:00"}
        ]
        blacklist = ["BLK1"]
        threshold = 10000
        flagged = user_code.screen_transactions(txs, blacklist, threshold)
        expected = [{"id": 3, "reason": "blacklist"}]
        _dynamic_test(self,
            flagged == expected,
            "Correctly flags only blacklisted account transactions.",
            f"Expected: {expected}, got: {flagged}")

    def test_flag_both_rules(self):
        import user_code
        importlib.reload(user_code)
        txs = [
            {"id": 5, "amount": 20000, "account_id": "BLK2", "timestamp": "2024-06-01T14:00:00"}
        ]
        blacklist = ["BLK2"]
        threshold = 10000
        flagged = user_code.screen_transactions(txs, blacklist, threshold)
        expected = [
            {"id": 5, "reason": "amount"},
            {"id": 5, "reason": "blacklist"}
        ]
        _dynamic_test(self,
            flagged == expected,
            "Correctly flags both amount and blacklist violations for a single transaction.",
            f"Expected: {expected}, got: {flagged}")

    def test_flag_none(self):
        import user_code
        importlib.reload(user_code)
        txs = [
            {"id": 6, "amount": 100, "account_id": "SAFE", "timestamp": "2024-06-01T15:00:00"}
        ]
        blacklist = ["BLK3"]
        threshold = 10000
        flagged = user_code.screen_transactions(txs, blacklist, threshold)
        expected = []
        _dynamic_test(self,
            flagged == expected,
            "Returns empty list when no violations are present.",
            f"Expected: {expected}, got: {flagged}")

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

Learn how to leverage Python to automate, analyze, and streamline compliance tasks. This course is tailored for compliance officers seeking to enhance their efficiency and accuracy using Python, focusing on real-world scenarios such as monitoring transactions, detecting anomalies, and generating compliance reports.

Explore how Python can automate routine compliance tasks, such as flagging suspicious transactions and validating regulatory requirements.

Learn how to analyze compliance data using Python and visualize findings to support audits and reporting.

Delve into advanced Python techniques for compliance, including rule-based systems, simple machine learning, and reporting.

Challenge: Automated Transaction Screening

Lösung