Anti-Money Laundering with Ollama: Suspicious Transaction Detection Made Simple

# Install Ollama (if not already installed)
curl -fsSL https://ollama.ai/install.sh | sh

# Install Python packages
pip install pandas numpy scikit-learn requests json

import subprocess
import json

def setup_ollama_model():
    """Download and configure Ollama model for AML detection."""
    
    # Pull the recommended model for text analysis
    try:
        subprocess.run(["ollama", "pull", "llama3.1:8b"], check=True)
        print("✅ Model downloaded successfully")
    except subprocess.CalledProcessError:
        print("❌ Failed to download model")
        return False
    
    return True

# Run setup
setup_ollama_model()

import pandas as pd
import numpy as np
import requests
import json
from datetime import datetime, timedelta

class AMLDetector:
    def __init__(self, ollama_url="http://localhost:11434"):
        """Initialize AML detection system with Ollama integration."""
        self.ollama_url = ollama_url
        self.model = "llama3.1:8b"
        self.risk_thresholds = {
            'high': 0.8,
            'medium': 0.5,
            'low': 0.2
        }
    
    def analyze_transaction_pattern(self, transaction_data):
        """Analyze transaction patterns for suspicious activity."""
        
        # Prepare transaction summary for AI analysis
        prompt = self._create_analysis_prompt(transaction_data)
        
        # Send to Ollama for analysis
        response = self._query_ollama(prompt)
        
        # Parse AI response and calculate risk score
        risk_assessment = self._parse_ai_response(response)
        
        return risk_assessment
    
    def _create_analysis_prompt(self, data):
        """Create detailed prompt for transaction analysis."""
        
        prompt = f"""
        Analyze this financial transaction for money laundering indicators:
        
        Transaction Details:
        - Amount: ${data['amount']:,.2f}
        - Source Account: {data['source_account']}
        - Destination Account: {data['dest_account']}
        - Transaction Type: {data['transaction_type']}
        - Timestamp: {data['timestamp']}
        - Geographic Location: {data['location']}
        
        Account History:
        - Previous 30-day volume: ${data['historical_volume']:,.2f}
        - Average transaction size: ${data['avg_transaction']:,.2f}
        - Frequency pattern: {data['frequency_pattern']}
        
        Analyze for these money laundering indicators:
        1. Structuring (amounts just below reporting thresholds)
        2. Unusual velocity or volume changes
        3. Round dollar amounts indicating possible cash conversion
        4. Geographic risk factors
        5. Time patterns suggesting coordination
        
        Provide risk assessment (LOW/MEDIUM/HIGH) and specific reasoning.
        """
        
        return prompt
    
    def _query_ollama(self, prompt):
        """Send analysis request to Ollama."""
        
        payload = {
            "model": self.model,
            "prompt": prompt,
            "stream": False
        }
        
        try:
            response = requests.post(
                f"{self.ollama_url}/api/generate",
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()['response']
        
        except requests.RequestException as e:
            print(f"Error communicating with Ollama: {e}")
            return None
    
    def _parse_ai_response(self, ai_response):
        """Extract risk assessment from AI analysis."""
        
        if not ai_response:
            return {'risk_level': 'UNKNOWN', 'confidence': 0.0, 'reasoning': 'Analysis failed'}
        
        # Extract risk level from response
        risk_level = 'LOW'
        if 'HIGH' in ai_response.upper():
            risk_level = 'HIGH'
        elif 'MEDIUM' in ai_response.upper():
            risk_level = 'MEDIUM'
        
        # Calculate confidence score based on response detail
        confidence = min(len(ai_response.split()) / 100, 1.0)
        
        return {
            'risk_level': risk_level,
            'confidence': confidence,
            'reasoning': ai_response.strip(),
            'timestamp': datetime.now().isoformat()
        }

class TransactionMonitor:
    def __init__(self, aml_detector):
        """Initialize real-time transaction monitoring."""
        self.detector = aml_detector
        self.suspicious_alerts = []
        self.processed_count = 0
    
    def process_transaction_stream(self, transaction_stream):
        """Monitor and analyze incoming transactions."""
        
        for transaction in transaction_stream:
            # Enrich transaction with historical context
            enriched_transaction = self._enrich_transaction_data(transaction)
            
            # Perform AML analysis
            risk_assessment = self.detector.analyze_transaction_pattern(enriched_transaction)
            
            # Handle suspicious activity
            if risk_assessment['risk_level'] in ['HIGH', 'MEDIUM']:
                self._handle_suspicious_transaction(transaction, risk_assessment)
            
            self.processed_count += 1
            
            # Log progress
            if self.processed_count % 100 == 0:
                print(f"Processed {self.processed_count} transactions")
    
    def _enrich_transaction_data(self, transaction):
        """Add historical context and calculated features."""
        
        # Calculate velocity metrics
        account_history = self._get_account_history(transaction['source_account'])
        
        enriched = transaction.copy()
        enriched.update({
            'historical_volume': account_history['30_day_volume'],
            'avg_transaction': account_history['avg_amount'],
            'frequency_pattern': account_history['frequency_pattern'],
            'velocity_score': self._calculate_velocity_score(transaction, account_history)
        })
        
        return enriched
    
    def _calculate_velocity_score(self, transaction, history):
        """Calculate transaction velocity anomaly score."""
        
        current_amount = transaction['amount']
        avg_amount = history['avg_amount']
        
        if avg_amount > 0:
            velocity_score = abs(current_amount - avg_amount) / avg_amount
        else:
            velocity_score = 1.0  # Maximum anomaly for new accounts
        
        return min(velocity_score, 5.0)  # Cap at 5x normal
    
    def _handle_suspicious_transaction(self, transaction, risk_assessment):
        """Process and log suspicious activity."""
        
        alert = {
            'transaction_id': transaction['id'],
            'timestamp': datetime.now().isoformat(),
            'risk_level': risk_assessment['risk_level'],
            'confidence': risk_assessment['confidence'],
            'reasoning': risk_assessment['reasoning'],
            'transaction_amount': transaction['amount'],
            'accounts_involved': [transaction['source_account'], transaction['dest_account']]
        }
        
        self.suspicious_alerts.append(alert)
        
        # Log immediate alert
        print(f"🚨 SUSPICIOUS ACTIVITY DETECTED:")
        print(f"   Transaction ID: {transaction['id']}")
        print(f"   Risk Level: {risk_assessment['risk_level']}")
        print(f"   Amount: ${transaction['amount']:,.2f}")
        print(f"   Reasoning: {risk_assessment['reasoning'][:100]}...")
        print()
    
    def _get_account_history(self, account_id):
        """Retrieve account transaction history (mock implementation)."""
        
        # In production, query your transaction database
        return {
            '30_day_volume': np.random.uniform(10000, 500000),
            'avg_amount': np.random.uniform(1000, 10000),
            'frequency_pattern': np.random.choice(['regular', 'sporadic', 'burst'])
        }

class AdvancedAMLSystem:
    def __init__(self, ollama_detector):
        """Initialize advanced AML detection with multiple layers."""
        self.ai_detector = ollama_detector
        self.rule_engine = self._initialize_rules()
        self.pattern_analyzer = PatternAnalyzer()
    
    def comprehensive_analysis(self, transaction):
        """Perform multi-layered suspicious transaction analysis."""
        
        results = {
            'rule_based': self._apply_rule_engine(transaction),
            'ai_analysis': self.ai_detector.analyze_transaction_pattern(transaction),
            'pattern_analysis': self.pattern_analyzer.detect_patterns(transaction),
            'network_analysis': self._analyze_transaction_network(transaction)
        }
        
        # Combine all detection methods
        final_assessment = self._combine_assessments(results)
        
        return final_assessment
    
    def _initialize_rules(self):
        """Set up rule-based detection criteria."""
        
        return {
            'structuring': {
                'threshold': 9999.99,  # Just below $10k reporting requirement
                'tolerance': 0.01
            },
            'velocity': {
                'max_daily_transactions': 50,
                'max_hourly_amount': 100000
            },
            'geographic': {
                'high_risk_countries': ['Country1', 'Country2'],  # Use actual FATF list
                'domestic_threshold': 50000
            },
            'round_amounts': {
                'suspicious_patterns': [1000, 5000, 10000, 25000, 50000]
            }
        }
    
    def _apply_rule_engine(self, transaction):
        """Apply traditional AML rules."""
        
        violations = []
        
        # Check for structuring
        amount = transaction['amount']
        if abs(amount - self.rule_engine['structuring']['threshold']) <= self.rule_engine['structuring']['tolerance']:
            violations.append('STRUCTURING_DETECTED')
        
        # Check for round amounts
        if amount in self.rule_engine['round_amounts']['suspicious_patterns']:
            violations.append('ROUND_AMOUNT_PATTERN')
        
        # Check geographic risk
        if transaction.get('dest_country') in self.rule_engine['geographic']['high_risk_countries']:
            violations.append('HIGH_RISK_JURISDICTION')
        
        return {
            'violations': violations,
            'risk_score': len(violations) / 3.0  # Normalize to 0-1 scale
        }
    
    def _combine_assessments(self, results):
        """Merge results from all detection methods."""
        
        # Weight different detection methods
        weights = {
            'rule_based': 0.3,
            'ai_analysis': 0.4,
            'pattern_analysis': 0.2,
            'network_analysis': 0.1
        }
        
        # Calculate weighted risk score
        total_score = 0
        for method, weight in weights.items():
            if method in results and 'confidence' in results[method]:
                total_score += results[method]['confidence'] * weight
        
        # Determine final risk level
        if total_score >= 0.8:
            final_risk = 'HIGH'
        elif total_score >= 0.5:
            final_risk = 'MEDIUM'
        else:
            final_risk = 'LOW'
        
        return {
            'final_risk_level': final_risk,
            'composite_score': total_score,
            'contributing_factors': results,
            'recommendation': self._generate_recommendation(final_risk, results)
        }

import random
from datetime import datetime, timedelta

class TransactionGenerator:
    def __init__(self):
        """Initialize transaction generator for testing."""
        self.account_types = ['checking', 'savings', 'business', 'investment']
        self.transaction_types = ['wire', 'ach', 'check', 'card', 'cash']
        self.countries = ['US', 'UK', 'CA', 'DE', 'FR', 'suspicious_country']
    
    def generate_normal_transactions(self, count=100):
        """Generate normal transaction patterns."""
        
        transactions = []
        base_time = datetime.now() - timedelta(days=30)
        
        for i in range(count):
            transaction = {
                'id': f'TXN_{i:06d}',
                'amount': random.uniform(100, 5000),  # Normal range
                'source_account': f'ACC_{random.randint(1000, 9999)}',
                'dest_account': f'ACC_{random.randint(1000, 9999)}',
                'transaction_type': random.choice(self.transaction_types),
                'timestamp': (base_time + timedelta(hours=random.randint(0, 720))).isoformat(),
                'location': random.choice(self.countries[:-1]),  # Exclude suspicious country
                'dest_country': random.choice(self.countries[:-1])
            }
            transactions.append(transaction)
        
        return transactions
    
    def generate_suspicious_transactions(self, count=20):
        """Generate transactions with money laundering indicators."""
        
        suspicious_transactions = []
        base_time = datetime.now() - timedelta(days=7)
        
        for i in range(count):
            # Create different types of suspicious patterns
            pattern_type = random.choice(['structuring', 'round_amounts', 'velocity', 'geographic'])
            
            if pattern_type == 'structuring':
                amount = 9999.99  # Just below reporting threshold
            elif pattern_type == 'round_amounts':
                amount = random.choice([10000, 25000, 50000])
            elif pattern_type == 'velocity':
                amount = random.uniform(100000, 500000)  # Unusually large
            else:  # geographic
                amount = random.uniform(10000, 100000)
            
            transaction = {
                'id': f'SUS_{i:06d}',
                'amount': amount,
                'source_account': f'ACC_{random.randint(1000, 9999)}',
                'dest_account': f'ACC_{random.randint(1000, 9999)}',
                'transaction_type': random.choice(self.transaction_types),
                'timestamp': (base_time + timedelta(hours=random.randint(0, 168))).isoformat(),
                'location': 'US' if pattern_type != 'geographic' else 'suspicious_country',
                'dest_country': 'suspicious_country' if pattern_type == 'geographic' else 'US'
            }
            suspicious_transactions.append(transaction)
        
        return suspicious_transactions

# Test the system
def run_aml_test():
    """Execute comprehensive AML system test."""
    
    print("🔍 Starting AML Detection System Test")
    print("=" * 50)
    
    # Initialize system components
    detector = AMLDetector()
    monitor = TransactionMonitor(detector)
    generator = TransactionGenerator()
    
    # Generate test data
    normal_transactions = generator.generate_normal_transactions(100)
    suspicious_transactions = generator.generate_suspicious_transactions(20)
    
    # Combine and shuffle transactions
    all_transactions = normal_transactions + suspicious_transactions
    random.shuffle(all_transactions)
    
    print(f"Processing {len(all_transactions)} transactions...")
    print(f"Expected suspicious: {len(suspicious_transactions)}")
    print()
    
    # Process all transactions
    monitor.process_transaction_stream(all_transactions)
    
    # Display results
    print("\n📊 DETECTION RESULTS:")
    print(f"Total Processed: {monitor.processed_count}")
    print(f"Suspicious Detected: {len(monitor.suspicious_alerts)}")
    print(f"Detection Rate: {len(monitor.suspicious_alerts)/len(suspicious_transactions)*100:.1f}%")
    
    return monitor.suspicious_alerts

# Run the test
test_results = run_aml_test()

class ProductionAMLSystem:
    def __init__(self, config):
        """Initialize production-ready AML system."""
        self.config = config
        self.cache = {}
        self.batch_size = config.get('batch_size', 10)
        self.performance_metrics = {
            'processed': 0,
            'detected': 0,
            'false_positives': 0,
            'processing_time': []
        }
    
    def optimize_ollama_performance(self):
        """Configure Ollama for optimal AML performance."""
        
        # Set model parameters for faster inference
        optimization_settings = {
            'num_ctx': 4096,  # Context window size
            'num_predict': 256,  # Max tokens to generate
            'temperature': 0.1,  # Lower temperature for consistent results
            'top_p': 0.9,
            'repeat_penalty': 1.1
        }
        
        return optimization_settings
    
    def batch_process_transactions(self, transactions):
        """Process transactions in optimized batches."""
        
        results = []
        
        for i in range(0, len(transactions), self.batch_size):
            batch = transactions[i:i + self.batch_size]
            
            start_time = datetime.now()
            batch_results = self._process_batch(batch)
            processing_time = (datetime.now() - start_time).total_seconds()
            
            results.extend(batch_results)
            self.performance_metrics['processing_time'].append(processing_time)
        
        return results
    
    def _process_batch(self, batch):
        """Process a batch of transactions efficiently."""
        
        # Prepare batch prompt for multiple transactions
        batch_prompt = self._create_batch_prompt(batch)
        
        # Single API call for entire batch
        response = self._query_ollama_batch(batch_prompt)
        
        # Parse batch response
        return self._parse_batch_response(response, batch)
    
    def generate_compliance_report(self, period_start, period_end):
        """Generate regulatory compliance report."""
        
        report = {
            'report_period': {
                'start': period_start.isoformat(),
                'end': period_end.isoformat()
            },
            'summary_statistics': {
                'total_transactions_processed': self.performance_metrics['processed'],
                'suspicious_activities_detected': self.performance_metrics['detected'],
                'false_positive_rate': self._calculate_false_positive_rate(),
                'system_uptime': self._calculate_uptime(),
                'average_processing_time': np.mean(self.performance_metrics['processing_time'])
            },
            'detection_categories': self._categorize_detections(),
            'risk_trend_analysis': self._analyze_risk_trends(),
            'recommendations': self._generate_compliance_recommendations()
        }
        
        return report

from typing import Dict, List, Optional
import asyncio
import aiohttp

class BankingSystemIntegration:
    def __init__(self, bank_api_config):
        """Initialize integration with banking systems."""
        self.api_config = bank_api_config
        self.aml_detector = AMLDetector()
        self.alert_queue = asyncio.Queue()
    
    async def real_time_transaction_monitoring(self):
        """Monitor transactions in real-time from banking system."""
        
        while True:
            try:
                # Fetch new transactions from banking API
                new_transactions = await self._fetch_pending_transactions()
                
                # Process each transaction
                for transaction in new_transactions:
                    risk_assessment = await self._async_analyze_transaction(transaction)
                    
                    if risk_assessment['risk_level'] in ['HIGH', 'MEDIUM']:
                        await self._send_alert_to_compliance_team(transaction, risk_assessment)
                
                # Wait before next polling cycle
                await asyncio.sleep(self.api_config['polling_interval'])
                
            except Exception as e:
                print(f"Error in monitoring loop: {e}")
                await asyncio.sleep(60)  # Wait before retrying
    
    async def _fetch_pending_transactions(self) -> List[Dict]:
        """Fetch new transactions from banking system API."""
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.api_config['base_url']}/transactions/pending",
                headers={'Authorization': f"Bearer {self.api_config['api_key']}"}
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get('transactions', [])
                else:
                    print(f"API error: {response.status}")
                    return []
    
    async def _send_alert_to_compliance_team(self, transaction, risk_assessment):
        """Send suspicious activity alert to compliance team."""
        
        alert_payload = {
            'alert_type': 'SUSPICIOUS_TRANSACTION',
            'transaction_id': transaction['id'],
            'risk_level': risk_assessment['risk_level'],
            'confidence_score': risk_assessment['confidence'],
            'detected_patterns': risk_assessment['reasoning'],
            'recommended_actions': self._get_recommended_actions(risk_assessment),
            'timestamp': datetime.now().isoformat()
        }
        
        # Send to compliance management system
        async with aiohttp.ClientSession() as session:
            await session.post(
                f"{self.api_config['compliance_url']}/alerts",
                json=alert_payload,
                headers={'Authorization': f"Bearer {self.api_config['api_key']}"}
            )
        
        print(f"🚨 Alert sent for transaction {transaction['id']}")

class AMLModelTrainer:
    def __init__(self, aml_system):
        """Initialize model training and improvement system."""
        self.aml_system = aml_system
        self.feedback_data = []
        self.model_performance = {
            'accuracy': 0.0,
            'precision': 0.0,
            'recall': 0.0,
            'f1_score': 0.0
        }
    
    def collect_feedback(self, transaction_id, predicted_risk, actual_outcome, analyst_notes):
        """Collect feedback from compliance analysts."""
        
        feedback = {
            'transaction_id': transaction_id,
            'predicted_risk': predicted_risk,
            'actual_outcome': actual_outcome,  # 'confirmed_suspicious', 'false_positive', 'needs_review'
            'analyst_notes': analyst_notes,
            'timestamp': datetime.now().isoformat()
        }
        
        self.feedback_data.append(feedback)
        
        # Trigger model update if enough feedback collected
        if len(self.feedback_data) >= 100:
            self._update_detection_parameters()
    
    def _update_detection_parameters(self):
        """Update detection thresholds based on feedback."""
        
        # Analyze feedback patterns
        false_positives = [f for f in self.feedback_data if f['actual_outcome'] == 'false_positive']
        confirmed_suspicious = [f for f in self.feedback_data if f['actual_outcome'] == 'confirmed_suspicious']
        
        # Adjust detection sensitivity
        if len(false_positives) > len(confirmed_suspicious):
            # Too many false positives - increase threshold
            self.aml_system.risk_thresholds['medium'] += 0.05
            self.aml_system.risk_thresholds['high'] += 0.05
            print("📊 Adjusted thresholds to reduce false positives")
        
        elif len(confirmed_suspicious) > len(false_positives) * 2:
            # Missing suspicious transactions - decrease threshold
            self.aml_system.risk_thresholds['medium'] -= 0.05
            self.aml_system.risk_thresholds['high'] -= 0.05
            print("📊 Adjusted thresholds to improve detection sensitivity")
        
        # Clear feedback buffer
        self.feedback_data = []

class ComplianceReporting:
    def __init__(self, aml_system):
        """Initialize compliance reporting module."""
        self.aml_system = aml_system
        self.regulatory_requirements = {
            'BSA': {  # Bank Secrecy Act
                'currency_transaction_threshold': 10000,
                'suspicious_activity_threshold': 5000,
                'reporting_timeframe_days': 30
            },
            'FATF': {  # Financial Action Task Force
                'high_risk_jurisdictions': ['Country1', 'Country2'],
                'enhanced_due_diligence_threshold': 25000
            }
        }
    
    def generate_sar_report(self, suspicious_transactions):
        """Generate Suspicious Activity Report (SAR)."""
        
        sar_data = {
            'report_header': {
                'financial_institution': self.aml_system.config.get('institution_name'),
                'report_date': datetime.now().isoformat(),
                'report_type': 'SAR',
                'reporting_period': self._get_reporting_period()
            },
            'suspicious_activities': []
        }
        
        for transaction in suspicious_transactions:
            if self._meets_sar_criteria(transaction):
                sar_entry = {
                    'transaction_id': transaction['id'],
                    'transaction_date': transaction['timestamp'],
                    'amount': transaction['amount'],
                    'parties_involved': {
                        'source': transaction['source_account'],
                        'destination': transaction['dest_account']
                    },
                    'suspicious_activity_description': transaction['risk_assessment']['reasoning'],
                    'supporting_documentation': self._gather_supporting_docs(transaction)
                }
                sar_data['suspicious_activities'].append(sar_entry)
        
        return sar_data
    
    def _meets_sar_criteria(self, transaction):
        """Determine if transaction meets SAR reporting criteria."""
        
        # Check amount threshold
        if transaction['amount'] >= self.regulatory_requirements['BSA']['suspicious_activity_threshold']:
            return True
        
        # Check risk level
        if transaction.get('risk_assessment', {}).get('risk_level') == 'HIGH':
            return True
        
        return False
    
    def audit_trail_generator(self, transaction_id):
        """Generate complete audit trail for regulatory review."""
        
        audit_trail = {
            'transaction_id': transaction_id,
            'detection_timeline': [],
            'analysis_steps': [],
            'system_decisions': [],
            'human_interventions': [],
            'data_sources': []
        }
        
        # Compile comprehensive audit information
        # This would integrate with your logging and monitoring systems
        
        return audit_trail