Fraud Detection Case Study Interview

from datetime import datetime, timedelta
from typing import List, Dict, Set, Tuple
from dataclasses import dataclass
import itertools

class Transaction:
    transactionId: str
    transactionAmount: float
    sender: str
    recipient: str
    transactedOn: datetime

MINIMUM_TRANSFERS = 5
MINIMUM_RECIPIENTS = 3
WINDOW_MINUTES = 10


def check_for_multiple_receipients(transactions: List[Transaction]) -> Dict[str, Dict]:
    flagged: Dict[str, Dict] = {}
    transaction_by_sender: Dict[str, List[Transaction]] = {}
    for transaction in transactions:
        transaction_by_sender.setdefault(transaction.sender, []).append(transaction)
    for sender, transactions in transaction_by_sender.items():
        transactions.sort(key=lambda t: t.transactedOn)
        window_delta = timedelta(minutes=WINDOW_MINUTES)
        for i, start_transaction in enumerate(transactions):
            starting_window = start_transaction.transactedOn
            ending_window = starting_window + window_delta
            windowed = [t for t in transactions if starting_window <= t.transactedOn <= ending_window]
            total_transfers = len(windowed)
            recipients: Set[str] = {t.recipient for t in windowed}
            if total_transfers >= MINIMUM_TRANSFERS and len(recipients) >= MINIMUM_recipients:
                flagged[sender] = {
                    "starting_window": starting_window.isoformat(),
                    "ending_window": ending_window.isoformat(),
                    "number_of_transfers": total_transfers,
                    "recipients": list(recipients),
                    "transactions": [t.transactionId for t in windowed[:10]],
                }
                break
    return flagged


def hardcoded_transactions() -> List[Transaction]:
    raw = [
        ("transaction1", "A", "R1", 100.0, "2025-11-24T08:00:00"),
        ("transaction2", "A", "R2", 50.0, "2025-11-24T08:01:00"),
        ("transaction3", "A", "R3", 75.0, "2025-11-24T08:02:30"),
        ("transaction4", "A", "R2", 125.0, "2025-11-24T08:03:00"),
        ("transaction5", "A", "R4", 20.0, "2025-11-24T08:05:00"),
        ("transaction6", "B", "R9", 12.0, "2025-11-24T07:00:00"),
        ("transaction7", "B", "R9", 13.0, "2025-11-24T09:00:00"),
        ("transaction8", "C", "R8", 10.0, "2025-11-24T08:00:00"),
        ("transaction9", "C", "R8", 10.0, "2025-11-24T08:02:00"),
        ("transaction10","C", "R8", 10.0, "2025-11-24T08:05:00"),
        ("transaction11","C", "R8", 10.0, "2025-11-24T08:06:00"),
        ("transaction12","C", "R8", 10.0, "2025-11-24T08:07:00"),
    ]
    return [Transaction(transactionId=t[0], sender=t[1], recipient=t[2], transactionAmount=t[3], transactedOn=parse_iso(t[4])) for t in raw]


def main():
    transactions = hardcoded_transactions()
    flagged = detect_rapid_multi_recipient(transactions)
    if not flagged:
        print("No rapid multi-recipient senders detected.")
    else:
        print("Flagged senders (rapid multi-recipient):")
        for sender, details in flagged.items():
            print(f"- Sender: {sender}")
            print(f"  Window: {details['starting_window']} → {details['ending_window']}")
            print(f"  Transfer count: {details['number_of_transfers']}")
            print(f"  Unique recipients: {details['recipients']}")
            print(f"  Sample transaction ids: {details['transactions']}")


if __name__ == "__main__":
    main()