Lære Challenge: Compliance Rule Engine | Advanced Compliance Monitoring

Seksjon 3. Kapittel 5

single

Sveip for å vise menyen

In previous chapters, you learned how to automate transaction screening, flag suspicious patterns, and validate regulatory requirements. You also explored rule-based compliance systems, generating compliance reports, and visualizing advanced metrics. Now, you will bring these skills together by constructing a compliance rule engine that can process transactions, apply multiple rules, and report (as well as visualize) violations. This challenge will reinforce your understanding of how to build flexible, auditable compliance tools using Python.


              12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
            
import seaborn as sns
import matplotlib.pyplot as plt
from collections import defaultdict
from datetime import datetime, timedelta

# Sample transactions
transactions = [
    {"id": 1, "amount": 12000, "account_id": "A001", "timestamp": "2024-06-01T09:00:00"},
    {"id": 2, "amount": 4500, "account_id": "A002", "timestamp": "2024-06-01T09:05:00"},
    {"id": 3, "amount": 200, "account_id": "A003", "timestamp": "2024-06-01T09:07:00"},
    {"id": 4, "amount": 9900, "account_id": "A001", "timestamp": "2024-06-01T09:10:00"},
    {"id": 5, "amount": 15000, "account_id": "A004", "timestamp": "2024-06-01T09:15:00"},
    {"id": 6, "amount": 500, "account_id": "A002", "timestamp": "2024-06-01T09:17:00"},
    {"id": 7, "amount": 7000, "account_id": "A005", "timestamp": "2024-06-01T09:18:00"},
    {"id": 8, "amount": 11000, "account_id": "A006", "timestamp": "2024-06-01T09:19:00"},
    {"id": 9, "amount": 100, "account_id": "A003", "timestamp": "2024-06-01T09:20:00"},
    {"id": 10, "amount": 300, "account_id": "A002", "timestamp": "2024-06-01T09:21:00"},
]

# Blacklisted accounts
blacklist = {"A004", "A007"}

# Rule: Amount threshold
def rule_amount_threshold(tx, threshold=10000):
    if tx["amount"] > threshold:
        return True, f"Amount exceeds threshold of {threshold}"
    return False, ""

# Rule: Blacklist
def rule_blacklist(tx, blacklist_set):
    if tx["account_id"] in blacklist_set:
        return True, "Account is blacklisted"
    return False, ""

# Rule: Frequency within short period
def rule_frequency(transactions, window_minutes=10, max_tx=2):
    violations = []
    txs_by_account = defaultdict(list)
    for tx in transactions:
        txs_by_account[tx["account_id"]].append(tx)
    for account, txs in txs_by_account.items():
        txs_sorted = sorted(txs, key=lambda x: x["timestamp"])
        for i in range(len(txs_sorted)):
            count = 1
            t0 = datetime.fromisoformat(txs_sorted[i]["timestamp"])
            for j in range(i+1, len(txs_sorted)):
                t1 = datetime.fromisoformat(txs_sorted[j]["timestamp"])
                if (t1 - t0) <= timedelta(minutes=window_minutes):
                    count += 1
            if count > max_tx:
                for k in range(i, i+count):
                    if k < len(txs_sorted):
                        violations.append({
                            "id": txs_sorted[k]["id"],
                            "rule": "Frequency",
                            "reason": f"More than {max_tx} transactions in {window_minutes} minutes"
                        })
                break
    return violations

# Collect violations
def collect_violations(transactions, blacklist):
    violations = []
    for tx in transactions:
        violated, reason = rule_amount_threshold(tx)
        if violated:
            violations.append({"id": tx["id"], "rule": "AmountThreshold", "reason": reason})
        violated, reason = rule_blacklist(tx, blacklist)
        if violated:
            violations.append({"id": tx["id"], "rule": "Blacklist", "reason": reason})
    freq_violations = rule_frequency(transactions)
    violations.extend(freq_violations)
    return violations

# Reporting
def print_report(violations):
    print("Compliance Violations Report")
    for v in violations:
        print(f"Transaction {v['id']}: Rule - {v['rule']} | Reason - {v['reason']}")

# Visualization
def visualize_violations(violations):
    rule_counts = defaultdict(int)
    for v in violations:
        rule_counts[v["rule"]] += 1
    rules = list(rule_counts.keys())
    counts = list(rule_counts.values())
    sns.barplot(x=rules, y=counts)
    plt.title("Number of Violations per Rule")
    plt.xlabel("Rule")
    plt.ylabel("Violations")
    plt.show()

# Usage
violations = collect_violations(transactions, blacklist)
print_report(violations)
visualize_violations(violations)

When building a compliance rule engine, organize your code so each rule is a clear, separate function. This makes it easy to add, remove, or adjust rules as regulations change. Keep data preparation, rule checking, reporting, and visualization in distinct steps. This modular approach not only improves readability and maintainability, but also allows for better testing and auditing of each rule. By collecting all violations into a single list and producing both textual and visual summaries, you create a transparent and extensible compliance monitoring system.

Oppgave

Swipe to start coding

Implement a compliance rule engine that processes the provided list of transactions and applies three rules: an amount threshold, a blacklist check, and a frequency check for transactions from the same account within a short period. The engine should collect all violations, report them, and visualize the number of violations per rule.

The rule_amount_threshold function must return a tuple indicating if the threshold was violated and a reason.
The rule_blacklist function must return a tuple indicating if the account is blacklisted and a reason.
The rule_frequency function must return a list of violation dictionaries for accounts exceeding the transaction frequency within the specified window.
The collect_violations function must apply all rules to the transactions and return a combined list of violation dictionaries.
The final output must include a printed report and a seaborn bar chart showing the number of violations per rule.

Løsning

Bytt til skrivebordet for virkelighetspraksisFortsett der du er med et av alternativene nedenfor

Alt var klart?

Takk for tilbakemeldingene dine!

Seksjon 3. Kapittel 5

single

Spør AI

Spør om hva du vil, eller prøv ett av de foreslåtte spørsmålene for å starte chatten vår