RegTech: Compliance Automation for Financial Services

from dataclasses import dataclass
from datetime import datetime
from enum import Enum

class CustomerRiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    BLOCKED = "blocked"

@dataclass
class Customer:
    customer_id: str
    first_name: str
    last_name: str
    date_of_birth: datetime
    address: str
    country: str
    document_type: str
    document_number: str

class KYCCheck:
    """
    KYC compliance workflow
    """
    
    def __init__(self):
        self.document_verifier = DocumentVerifier()
        self.watchlist_screener = WatchlistScreener()
        self.risk_calculator = RiskCalculator()
    
    def perform_kyc(self, customer: Customer) -> dict:
        """
        Perform complete KYC check
        """
        
        results = {
            'customer_id': customer.customer_id,
            'timestamp': datetime.utcnow(),
            'checks': {}
        }
        
        # 1. Document Verification
        doc_result = self._verify_document(customer)
        results['checks']['document_verification'] = doc_result
        
        # 2. Database Checks
        db_result = self._check_databases(customer)
        results['checks']['database_checks'] = db_result
        
        # 3. Sanctions Screening
        sanctions_result = self._screen_sanctions(customer)
        results['checks']['sanctions_screening'] = sanctions_result
        
        # 4. PEP Screening
        pep_result = self._screen_pep(customer)
        results['checks']['pep_screening'] = pep_result
        
        # 5. Adverse Media
        media_result = self._check_adverse_media(customer)
        results['checks']['adverse_media'] = media_result
        
        # 6. Calculate Overall Risk
        risk_score = self.risk_calculator.calculate(
            doc_result,
            db_result,
            sanctions_result,
            pep_result,
            media_result
        )
        results['risk_level'] = risk_score['risk_level']
        results['risk_factors'] = risk_score['factors']
        
        return results
    
    def _verify_document(self, customer: Customer) -> dict:
        """
        Verify identity documents
        """
        
        # Check document type
        valid_doc_types = ['passport', 'national_id', 'drivers_license']
        
        if customer.document_type.lower() not in valid_doc_types:
            return {
                'status': 'failed',
                'reason': 'Invalid document type'
            }
        
        # Verify document (would integrate with document verification service)
        verification_result = self.document_verifier.verify(
            document_type=customer.document_type,
            document_number=customer.document_number,
            country=customer.country,
            dob=customer.date_of_birth,
            name=f"{customer.first_name} {customer.last_name}"
        )
        
        return {
            'status': 'passed' if verification_result['valid'] else 'failed',
            'confidence': verification_result.get('confidence', 0),
            'reason': verification_result.get('reason')
        }
    
    def _check_databases(self, customer: Customer) -> dict:
        """
        Check against fraud and watchlist databases
        """
        
        # Check against internal fraud database
        internal_check = self._check_internal_fraud_db(customer)
        
        # Check against external databases
        external_check = self._check_external_databases(customer)
        
        return {
            'internal_fraud': internal_check,
            'external_databases': external_check,
            'overall_status': 'failed' if (internal_check['found'] or 
                                           external_check['found']) else 'passed'
        }
    
    def _screen_sanctions(self, customer: Customer) -> dict:
        """
        Screen against sanctions lists
        """
        
        result = self.watchlist_screener.screen(
            name=f"{customer.first_name} {customer.last_name}",
            country=customer.country,
            document_number=customer.document_number
        )
        
        return {
            'status': 'match' if result['matches'] else 'clear',
            'matches': result['matches'],
            'list_sources': result.get('sources', [])
        }
    
    def _screen_pep(self, customer: Customer) -> dict:
        """
        Screen against Politically Exposed Persons lists
        """
        
        pep_result = self.watchlist_screener.screen_pep(
            name=f"{customer.first_name} {customer.last_name}",
            country=customer.country
        )
        
        return {
            'status': 'match' if pep_result['matches'] else 'clear',
            'matches': pep_result['matches'],
            'pep_positions': pep_result.get('positions', [])
        }
    
    def _check_adverse_media(self, customer: Customer) -> dict:
        """
        Check for adverse media mentions
        """
        
        media_result = self.watchlist_screener.screen_media(
            name=f"{customer.first_name} {customer.last_name}",
            country=customer.country
        )
        
        return {
            'status': 'found' if media_result['articles'] else 'clear',
            'articles': media_result['articles'][:5],  # Top 5
            'sentiment': media_result.get('sentiment', 'neutral')
        }
    
    def _check_internal_fraud_db(self, customer: Customer) -> dict:
        """Check internal fraud database"""
        # Implementation
        pass
    
    def _check_external_databases(self, customer: Customer) -> dict:
        """Check external databases"""
        # Implementation
        pass

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta

@dataclass
class Transaction:
    transaction_id: str
    customer_id: str
    amount: float
    currency: str
    timestamp: datetime
    transaction_type: str
    merchant_category: str
    source_account: str
    destination_account: str
    source_country: str
    destination_country: str

class AMLMonitor:
    """
    Anti-Money Laundering transaction monitoring
    """
    
    def __init__(self):
        self.rules = self._load_monitoring_rules()
        self.alert_manager = AlertManager()
    
    def _load_monitoring_rules(self) -> List[dict]:
        """
        Load AML monitoring rules
        """
        
        return [
            {
                'name': 'large_transaction',
                'threshold': 10000,
                'window_days': 1,
                'description': 'Single transaction above threshold'
            },
            {
                'name': 'velocity_rule',
                'transaction_count': 5,
                'window_hours': 1,
                'description': 'Multiple transactions in short period'
            },
            {
                'name': 'round_amount',
                'threshold': 5000,
                'description': 'Round amount transactions (potential structuring)'
            },
            {
                'name': 'high_risk_country',
                'countries': ['IR', 'KP', 'SY', 'CU', 'RU'],
                'description': 'Transaction with high-risk country'
            },
            {
                'name': 'layering_pattern',
                'threshold': 5000,
                'transaction_count': 3,
                'window_hours': 24,
                'description': 'Potential layering (multiple transfers)'
            }
        ]
    
    def monitor_transaction(self, transaction: Transaction) -> Optional[dict]:
        """
        Check transaction against monitoring rules
        """
        
        triggered_rules = []
        
        for rule in self.rules:
            if self._evaluate_rule(rule, transaction):
                triggered_rules.append({
                    'rule_name': rule['name'],
                    'description': rule['description'],
                    'severity': rule.get('severity', 'medium')
                })
        
        if triggered_rules:
            alert = self._create_alert(transaction, triggered_rules)
            self.alert_manager.create_alert(alert)
            return alert
        
        return None
    
    def _evaluate_rule(self, rule: dict, transaction: Transaction) -> bool:
        """
        Evaluate a single monitoring rule
        """
        
        rule_name = rule['name']
        
        if rule_name == 'large_transaction':
            return transaction.amount >= rule['threshold']
        
        elif rule_name == 'round_amount':
            return (
                transaction.amount >= rule['threshold'] and
                transaction.amount % 1000 == 0
            )
        
        elif rule_name == 'high_risk_country':
            return (
                transaction.source_country in rule['countries'] or
                transaction.destination_country in rule['countries']
            )
        
        # For rules requiring historical context, would query database
        elif rule_name == 'velocity_rule':
            return self._check_velocity(transaction, rule)
        
        return False
    
    def _check_velocity(self, transaction: Transaction, rule: dict) -> bool:
        """
        Check transaction velocity (requires database query)
        """
        
        # Would query transaction history
        # count = self.db.get_transaction_count(
        #     customer_id=transaction.customer_id,
        #     window_hours=rule['window_hours']
        # )
        # return count >= rule['transaction_count']
        
        return False
    
    def _create_alert(self, transaction: Transaction, triggered_rules: List[dict]) -> dict:
        """
        Create compliance alert
        """
        
        return {
            'alert_id': f"AML-{transaction.transaction_id}",
            'transaction_id': transaction.transaction_id,
            'customer_id': transaction.customer_id,
            'amount': transaction.amount,
            'currency': transaction.currency,
            'timestamp': transaction.timestamp,
            'triggered_rules': triggered_rules,
            'priority': self._calculate_priority(triggered_rules),
            'status': 'open',
            'created_at': datetime.utcnow()
        }
    
    def _calculate_priority(self, triggered_rules: List[dict]) -> str:
        """
        Calculate alert priority based on triggered rules
        """
        
        severities = [r['severity'] for r in triggered_rules]
        
        if 'critical' in severities:
            return 'critical'
        elif 'high' in severities:
            return 'high'
        elif 'medium' in severities:
            return 'medium'
        return 'low'


class AlertManager:
    """
    Manage compliance alerts
    """
    
    def __init__(self):
        self.db = AlertDatabase()
    
    def create_alert(self, alert: dict):
        """
        Create and route alert
        """
        
        # Store alert
        self.db.save_alert(alert)
        
        # Route to analyst queue
        self._route_to_queue(alert)
        
        # Create case if critical
        if alert['priority'] == 'critical':
            self._create_case(alert)
    
    def _route_to_queue(self, alert: dict):
        """Route alert to appropriate analyst queue"""
        pass
    
    def _create_case(self, alert: dict):
        """Create investigation case"""
        pass

import re
from typing import List, Dict, Optional

class SanctionsScreener:
    """
    Sanctions list screening
    """
    
    def __init__(self):
        self.sanctions_lists = self._load_sanctions_lists()
        self.name_matcher = NameMatcher()
    
    def _load_sanctions_lists(self) -> Dict:
        """
        Load sanctions lists
        """
        
        return {
            'ofac_sdn': self._load_ofac_sdn(),
            'un_sanctions': self._load_un_sanctions(),
            'eu_sanctions': self._load_eu_sanctions(),
            'hmt_sanctions': self._load_hmt_sanctions()
        }
    
    def screen(self, name: str, country: str = None, 
               document_number: str = None) -> dict:
        """
        Screen against all sanctions lists
        """
        
        matches = []
        
        # Clean and normalize name
        normalized_name = self._normalize_name(name)
        
        for list_name, list_data in self.sanctions_lists.items():
            # Check name matches
            name_matches = self.name_matcher.match(
                normalized_name,
                list_data['names']
            )
            
            if name_matches:
                for match in name_matches:
                    matches.append({
                        'list': list_name,
                        'matched_name': match['name'],
                        'entity_type': match.get('type', 'individual'),
                        'program': match.get('program', 'unknown'),
                        'score': match['score']
                    })
            
            # Check country
            if country:
                country_matches = self._check_country_sanctions(
                    country, list_data
                )
                matches.extend(country_matches)
            
            # Check document number
            if document_number:
                doc_matches = self._check_document_sanctions(
                    document_number, list_data
                )
                matches.extend(doc_matches)
        
        return {
            'matches': matches,
            'sources': list(set(m['list'] for m in matches)),
            'overall_status': 'match' if matches else 'clear'
        }
    
    def _normalize_name(self, name: str) -> str:
        """
        Normalize name for matching
        """
        
        # Convert to uppercase
        normalized = name.upper()
        
        # Remove special characters
        normalized = re.sub(r'[^A-Z\s]', '', normalized)
        
        # Remove extra whitespace
        normalized = ' '.join(normalized.split())
        
        return normalized
    
    def _check_country_sanctions(self, country: str, 
                                  list_data: dict) -> List[dict]:
        """
        Check country-based sanctions
        """
        
        matches = []
        
        # Check if country is subject to sanctions
        if 'country_sanctions' in list_data:
            if country in list_data['country_sanctions']:
                matches.append({
                    'type': 'country',
                    'country': country,
                    'list': list_data.get('name', 'unknown')
                })
        
        return matches
    
    def _check_document_sanctions(self, document_number: str,
                                  list_data: dict) -> List[dict]:
        """
        Check document number against sanctions
        """
        
        matches = []
        
        if 'documents' in list_data:
            normalized_doc = document_number.upper()
            
            for sanctioned_doc in list_data['documents']:
                if normalized_doc == sanctioned_doc['number']:
                    matches.append({
                        'type': 'document',
                        'document': document_number,
                        'list': list_data.get('name', 'unknown'),
                        'name': sanctioned_doc['associated_name']
                    })
        
        return matches


class NameMatcher:
    """
    Fuzzy name matching for sanctions screening
    """
    
    def match(self, query: str, reference_names: List[str]) -> List[dict]:
        """
        Find matching names
        """
        
        matches = []
        
        for ref_name in reference_names:
            # Calculate similarity score
            score = self._calculate_similarity(query, ref_name)
            
            # Threshold for match
            if score >= 0.85:
                matches.append({
                    'name': ref_name,
                    'score': score
                })
        
        return matches
    
    def _calculate_similarity(self, name1: str, name2: str) -> float:
        """
        Calculate name similarity score
        """
        
        # Could use various algorithms:
        # - Levenshtein distance
        # - Jaro-Winkler
        # - Token-based matching
        
        # Simple token overlap for example
        tokens1 = set(name1.split())
        tokens2 = set(name2.split())
        
        if not tokens1 or not tokens2:
            return 0.0
        
        intersection = tokens1.intersection(tokens2)
        union = tokens1.union(tokens2)
        
        return len(intersection) / len(union)

from datetime import datetime
from typing import List, Dict

class RegulatoryReporter:
    """
    Automated regulatory reporting
    """
    
    def __init__(self):
        self.reporting_config = self._load_reporting_config()
    
    def _load_reporting_config(self) -> Dict:
        """
        Load reporting requirements by jurisdiction
        """
        
        return {
            'us': {
                'fincen_ctr': {
                    'threshold': 10000,
                    'currency': 'USD',
                    'fields': ['filer', 'transaction', 'account_holder'],
                    'deadline': '15 days'
                },
                'fincen_sar': {
                    'threshold': 5000,
                    'fields': ['subject', 'suspicious_activity', 'Narrative'],
                    'deadline': '30 days'
                }
            },
            'eu': {
                'aml_directive': {
                    'threshold': 15000,
                    'fields': ['beneficiary', 'originator', 'amount']
                }
            },
            'uk': {
                'fca_suspicious_activity': {
                    'fields': ['report_type', 'subject_details', 'narrative']
                }
            }
        }
    
    def generate_ctr(self, transactions: List[dict]) -> dict:
        """
        Generate Currency Transaction Report
        """
        
        # Filter transactions above threshold
        threshold = self.reporting_config['us']['fincen_ctr']['threshold']
        large_transactions = [
            t for t in transactions 
            if t['amount'] >= threshold
        ]
        
        if not large_transactions:
            return None
        
        ctr = {
            'report_type': 'CTR',
            'filing_institution': self._get_institution_info(),
            'report_date': datetime.utcnow(),
            'transactions': [
                {
                    'date': t['timestamp'],
                    'amount': t['amount'],
                    'currency': t['currency'],
                    'type': t['type'],
                    'account': t.get('account_number')
                }
                for t in large_transactions
            ],
            'filer_info': self._get_filer_info()
        }
        
        return ctr
    
    def generate_sar(self, alerts: List[dict]) -> dict:
        """
        Generate Suspicious Activity Report
        """
        
        # Filter critical/high alerts
        critical_alerts = [
            a for a in alerts 
            if a['priority'] in ['critical', 'high']
        ]
        
        if not critical_alerts:
            return None
        
        sar = {
            'report_type': 'SAR',
            'filing_institution': self._get_institution_info(),
            'report_date': datetime.utcnow(),
            'subjects': [],
            'suspicious_activities': [],
            'narrative': self._generate_narrative(critical_alerts)
        }
        
        # Add subject information
        for alert in critical_alerts:
            sar['subjects'].append({
                'subject_type': 'individual',
                'subject_id': alert.get('customer_id'),
                'risk_factors': alert.get('risk_factors', [])
            })
        
        return sar
    
    def _generate_narrative(self, alerts: List[dict]) -> str:
        """
        Generate SAR narrative from alerts
        """
        
        narrative_parts = [
            "This report is filed pursuant to 31 U.S.C. § 5318(g) and 31 CFR § 1020.320.",
            "",
            f"The institution identified {len(alerts)} suspicious transaction pattern(s)."
        ]
        
        for i, alert in enumerate(alerts, 1):
            narrative_parts.append("")
            narrative_parts.append(f"Pattern {i}:")
            narrative_parts.append(f"  - Alert Type: {alert.get('alert_type', 'N/A')}")
            narrative_parts.append(f"  - Amount: ${alert.get('amount', 0):,.2f}")
            narrative_parts.append(f"  - Description: {alert.get('description', 'N/A')}")
        
        return "\n".join(narrative_parts)
    
    def _get_institution_info(self) -> dict:
        """Get institution information"""
        # Implementation
        pass
    
    def _get_filer_info(self) -> dict:
        """Get filer information"""
        # Implementation
        pass

from datetime import datetime
from typing import Any, Dict
import json

class ComplianceAuditLogger:
    """
    Immutable audit trail for compliance
    """
    
    def __init__(self):
        self.audit_log = []
    
    def log_event(self, event_type: str, actor: str, 
                  action: str, details: Dict[str, Any]):
        """
        Log compliance event
        """
        
        event = {
            'timestamp': datetime.utcnow().isoformat(),
            'event_type': event_type,
            'actor': actor,
            'action': action,
            'details': details,
            'hash': None  # Will be calculated
        }
        
        # Calculate hash for integrity
        event['hash'] = self._calculate_hash(event)
        
        # Store (append-only)
        self.audit_log.append(event)
    
    def _calculate_hash(self, event: dict) -> str:
        """
        Calculate hash for event integrity
        """
        
        import hashlib
        
        # Create deterministic string representation
        event_str = json.dumps(event, sort_keys=True, default=str)
        
        # Remove hash from calculation
        event_str = event_str.replace(f'"hash": {event.get("hash")}', '"hash": null')
        
        return hashlib.sha256(event_str.encode()).hexdigest()
    
    def verify_integrity(self) -> bool:
        """
        Verify audit log integrity
        """
        
        for i, event in enumerate(self.audit_log):
            expected_hash = self._calculate_hash(event)
            
            if event['hash'] != expected_hash:
                print(f"Integrity violation at event {i}")
                return False
        
        return True
    
    def query_audit_log(self, start_date: datetime = None, 
                        end_date: datetime = None,
                        actor: str = None,
                        event_type: str = None) -> list:
        """
        Query audit log with filters
        """
        
        results = self.audit_log
        
        if start_date:
            results = [
                r for r in results 
                if datetime.fromisoformat(r['timestamp']) >= start_date
            ]
        
        if end_date:
            results = [
                r for r in results 
                if datetime.fromisoformat(r['timestamp']) <= end_date
            ]
        
        if actor:
            results = [r for r in results if r['actor'] == actor]
        
        if event_type:
            results = [r for r in results if r['event_type'] == event_type]
        
        return results


# Example audit events
audit_logger = ComplianceAuditLogger()

# Log KYC check
audit_logger.log_event(
    event_type='KYC',
    actor='system',
    action='kyc_check_completed',
    details={
        'customer_id': 'CUST-12345',
        'result': 'approved',
        'risk_level': 'medium',
        'checks_performed': ['document', 'sanctions', 'pep', 'media']
    }
)

# Log transaction alert
audit_logger.log_event(
    event_type='AML',
    actor='system',
    action='alert_created',
    details={
        'alert_id': 'AML-001',
        'customer_id': 'CUST-12345',
        'triggered_rules': ['velocity_rule', 'large_transaction'],
        'priority': 'high'
    }
)

# Log analyst action
audit_logger.log_event(
    event_type='ANALYST',
    actor='analyst_001',
    action='alert_resolved',
    details={
        'alert_id': 'AML-001',
        'resolution': 'false_positive',
        'notes': 'Customer authorized transaction, verified with phone call'
    }
)

# Anti-pattern: One-time KYC only
def bad_kyc():
    """
    Anti-pattern: Only check KYC at onboarding
    """
    # Customer passes KYC once
    # Never re-verified
    # Risk profile never updated
    
    return "Stale data, increased risk!"

# Good pattern: Continuous monitoring
def good_kyc():
    """
    Ongoing KYC and risk assessment
    """
    
    # Re-assess periodically (annually or based on risk)
    # Monitor for adverse news
    # Update risk scores
    # Trigger re-verification on risk changes

# Anti-pattern: Accept high false positive rate
def bad_false_positives():
    """
    Anti-pattern: "We'll review everything"
    """
    # 95% of alerts are false positives
    # Analysts overwhelmed
    # Real fraud missed!
    
    return "Inefficient, expensive!"

# Good pattern: Reduce false positives
def good_false_positives():
    """
    Optimize rules to reduce false positives
    """
    
    # Tune thresholds
    # Add exception lists
    # Implement ML-based filtering
    # Use risk-based prioritization