Automated Banking Transaction Pipeline

import hashlib
import json
from datetime import datetime
from typing import List, Dict, Tuple

class TransactionValidator:
    def __init__(self, db_connection, fraud_detector):
        self.db = db_connection
        self.fraud_detector = fraud_detector
        self.validation_rules = self._load_rules()
    
    def validate_transaction(self, transaction: Dict) -> Tuple[bool, List[str]]:
        """Multi-layer validation before processing"""
        errors = []
        
        # Layer 1: Schema validation
        if not self._validate_schema(transaction):
            errors.append("Invalid transaction schema")
            return False, errors
        
        # Layer 2: Amount validation
        amount = float(transaction['amount'])
        if amount <= 0:
            errors.append("Amount must be positive")
        if amount > 1_000_000:
            errors.append("Amount exceeds daily limit")
        
        # Layer 3: Account validation
        sender_account = self.db.get_account(transaction['sender_id'])
        if not sender_account or sender_account['status'] != 'active':
            errors.append("Invalid sender account")
        
        if sender_account['balance'] < amount:
            errors.append("Insufficient funds")
        
        # Layer 4: Fraud detection
        fraud_score = self.fraud_detector.score_transaction(transaction)
        if fraud_score > 0.85:
            errors.append(f"High fraud risk (score: {fraud_score})")
        
        # Layer 5: Regulatory checks
        if self._is_suspicious_pattern(transaction):
            errors.append("Suspicious transaction pattern detected")
        
        return len(errors) == 0, errors
    
    def _validate_schema(self, transaction):
        """Ensure all required fields exist"""
        required = ['sender_id', 'recipient_id', 'amount', 'currency', 'timestamp']
        return all(field in transaction for field in required)
    
    def _is_suspicious_pattern(self, transaction):
        """Detect money laundering and other patterns"""
        recipient_id = transaction['recipient_id']
        recent_txns = self.db.get_recent_transactions(recipient_id, hours=24)
        
        # Flag if multiple large transactions to same recipient
        large_txn_count = sum(1 for t in recent_txns if float(t['amount']) > 50_000)
        return large_txn_count >= 3

class BatchProcessor:
    def __init__(self, db, batch_size=1000):
        self.db = db
        self.batch_size = batch_size
    
    def process_batch(self, transactions: List[Dict]) -> Dict:
        """Process transactions in batches for efficiency"""
        results = {
            'processed': 0,
            'failed': 0,
            'failed_transactions': [],
            'duration_seconds': 0
        }
        
        start_time = datetime.now()
        
        # Batch insert reduces database round-trips
        valid_txns = [t for t in transactions if t.get('valid')]
        
        for i in range(0, len(valid_txns), self.batch_size):
            batch = valid_txns[i:i + self.batch_size]
            
            try:
                # Atomic batch insert with transaction rollback on failure
                with self.db.transaction():
                    for txn in batch:
                        self.db.record_transaction(txn)
                        # Update account balances
                        self.db.debit_account(txn['sender_id'], txn['amount'])
                        self.db.credit_account(txn['recipient_id'], txn['amount'])
                
                results['processed'] += len(batch)
            except Exception as e:
                results['failed'] += len(batch)
                results['failed_transactions'].extend([t['id'] for t in batch])
                # Send to dead-letter queue for manual review
                self.db.send_to_dlq(batch, str(e))
        
        results['duration_seconds'] = (datetime.now() - start_time).total_seconds()
        return results

from jinja2 import Template
from weasyprint import HTML, CSS
from io import BytesIO

class ComplianceReportGenerator:
    def __init__(self, template_dir):
        self.template_dir = template_dir
    
    def generate_ctf_report(self, transactions_data: List[Dict]) -> bytes:
        """Generate Suspicious Activity Report (CTF) for regulatory compliance"""
        
        # Aggregate suspicious transactions
        suspicious = [t for t in transactions_data if t['risk_score'] > 0.7]
        total_amount = sum(t['amount'] for t in suspicious)
        
        template_str = """
        <html>
            <head>
                <style>
                    body { font-family: Arial, sans-serif; margin: 20px; }
                    .header { text-align: center; font-weight: bold; margin-bottom: 20px; }
                    table { width: 100%; border-collapse: collapse; }
                    th, td { border: 1px solid #000; padding: 8px; text-align: left; }
                    th { background-color: #f0f0f0; }
                </style>
            </head>
            <body>
                <div class="header">
                    Suspicious Activity Report (CTF)
                    Report Date: {{ report_date }}
                </div>
                
                <h2>Summary</h2>
                <p>Total Suspicious Transactions: {{ suspicious_count }}</p>
                <p>Total Amount: ${{ total_amount | format_currency }}</p>
                
                <h2>Transaction Details</h2>
                <table>
                    <tr>
                        <th>Transaction ID</th>
                        <th>Sender</th>
                        <th>Recipient</th>
                        <th>Amount</th>
                        <th>Risk Score</th>
                        <th>Reason</th>
                    </tr>
                    {% for txn in suspicious_transactions %}
                    <tr>
                        <td>{{ txn.id }}</td>
                        <td>{{ txn.sender_id }}</td>
                        <td>{{ txn.recipient_id }}</td>
                        <td>${{ txn.amount | format_currency }}</td>
                        <td>{{ (txn.risk_score * 100) | int }}%</td>
                        <td>{{ txn.reason }}</td>
                    </tr>
                    {% endfor %}
                </table>
                
                <p style="margin-top: 40px; font-size: 10px; color: #666;">
                    Generated by Automated Compliance System
                    Signature: {{ digital_signature }}
                </p>
            </body>
        </html>
        """
        
        template = Template(template_str)
        html_content = template.render(
            report_date=datetime.now().strftime('%Y-%m-%d'),
            suspicious_count=len(suspicious),
            suspicious_transactions=suspicious,
            total_amount=total_amount,
            digital_signature=self._generate_signature(suspicious)
        )
        
        # Convert to PDF
        pdf_bytes = HTML(string=html_content).write_pdf()
        return pdf_bytes
    
    def _generate_signature(self, data):
        """Generate digital signature for compliance"""
        content = json.dumps(data, sort_keys=True, default=str)
        return hashlib.sha256(content.encode()).hexdigest()[:16]