Performance Optimization

Advanced Caching System

import redis
import hashlib
import pickle
import time
import threading
from typing import Any, Optional, Dict, List
from dataclasses import dataclass
from abc import ABC, abstractmethod

@dataclass
class CacheEntry:
    value: Any
    timestamp: float
    ttl: float
    access_count: int = 0
    hit_score: float = 0.0

class CacheLayer(ABC):
    @abstractmethod
    def get(self, key: str) -> Optional[Any]:
        pass
    
    @abstractmethod
    def set(self, key: str, value: Any, ttl: float = 3600) -> bool:
        pass
    
    @abstractmethod
    def delete(self, key: str) -> bool:
        pass

class MemoryCache(CacheLayer):
    def __init__(self, max_size: int = 10000):
        self.cache: Dict[str, CacheEntry] = {}
        self.max_size = max_size
        self.lock = threading.RLock()
    
    def get(self, key: str) -> Optional[Any]:
        with self.lock:
            entry = self.cache.get(key)
            if entry is None:
                return None
            
            # Check TTL
            if time.time() - entry.timestamp > entry.ttl:
                del self.cache[key]
                return None
            
            # Update access statistics
            entry.access_count += 1
            entry.hit_score = self._calculate_hit_score(entry)
            
            return entry.value
    
    def set(self, key: str, value: Any, ttl: float = 3600) -> bool:
        with self.lock:
            # Evict if at capacity
            if len(self.cache) >= self.max_size:
                self._evict_lru()
            
            self.cache[key] = CacheEntry(
                value=value,
                timestamp=time.time(),
                ttl=ttl,
                access_count=1,
                hit_score=1.0
            )
            return True
    
    def _evict_lru(self):
        """Evict least recently used items"""
        if not self.cache:
            return
        
        # Find item with lowest hit score
        lru_key = min(self.cache.keys(), 
                     key=lambda k: self.cache[k].hit_score)
        del self.cache[lru_key]
    
    def _calculate_hit_score(self, entry: CacheEntry) -> float:
        """Calculate hit score based on recency and frequency"""
        age = time.time() - entry.timestamp
        frequency_score = entry.access_count / (age + 1)
        recency_score = 1 / (age + 1)
        return 0.7 * frequency_score + 0.3 * recency_score

class RedisCache(CacheLayer):
    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
        self.redis_client = redis.Redis(host=host, port=port, db=db)
    
    def get(self, key: str) -> Optional[Any]:
        try:
            data = self.redis_client.get(key)
            if data is None:
                return None
            return pickle.loads(data)
        except Exception:
            return None
    
    def set(self, key: str, value: Any, ttl: float = 3600) -> bool:
        try:
            serialized = pickle.dumps(value)
            return self.redis_client.setex(key, int(ttl), serialized)
        except Exception:
            return False
    
    def delete(self, key: str) -> bool:
        try:
            return self.redis_client.delete(key) > 0
        except Exception:
            return False

class IntelligentCacheManager:
    def __init__(self):
        self.layers = [
            MemoryCache(max_size=1000),    # L1: In-memory
            RedisCache(),                   # L2: Distributed cache
        ]
        
        self.prediction_cache = {}
        self.access_patterns = {}
        self.cache_stats = {
            'hits': 0,
            'misses': 0,
            'predictions': 0,
            'prediction_accuracy': 0.0
        }
    
    def get(self, key: str) -> Optional[Any]:
        """Get value from cache with intelligent prediction"""
        
        # Try each cache layer
        for i, layer in enumerate(self.layers):
            value = layer.get(key)
            if value is not None:
                self.cache_stats['hits'] += 1
                
                # Promote to higher cache levels
                self._promote_to_higher_levels(key, value, i)
                
                # Update access patterns for prediction
                self._update_access_patterns(key)
                
                return value
        
        self.cache_stats['misses'] += 1
        return None
    
    def set(self, key: str, value: Any, ttl: float = 3600) -> bool:
        """Set value in all cache layers"""
        success = True
        
        for layer in self.layers:
            if not layer.set(key, value, ttl):
                success = False
        
        # Trigger predictive caching
        self._predict_related_keys(key)
        
        return success
    
    def intelligent_preload(self, base_key: str, related_keys: List[str]):
        """Preload related keys based on access patterns"""
        
        if base_key in self.access_patterns:
            pattern = self.access_patterns[base_key]
            
            # Predict which keys will be needed
            for related_key in related_keys:
                prediction_score = self._calculate_prediction_score(
                    base_key, related_key, pattern
                )
                
                if prediction_score > 0.7:  # High confidence prediction
                    # Preload the key if not already cached
                    if self.get(related_key) is None:
                        self._preload_key(related_key)
                        self.cache_stats['predictions'] += 1
    
    def batch_invalidate(self, pattern: str):
        """Intelligently invalidate cache entries matching pattern"""
        
        for layer in self.layers:
            if hasattr(layer, 'redis_client'):
                # Redis pattern matching
                keys = layer.redis_client.keys(pattern)
                if keys:
                    layer.redis_client.delete(*keys)
            else:
                # Memory cache pattern matching
                keys_to_delete = [k for k in layer.cache.keys() 
                                if self._matches_pattern(k, pattern)]
                for key in keys_to_delete:
                    layer.delete(key)
    
    def optimize_cache_sizes(self):
        """Dynamically optimize cache sizes based on hit rates"""
        
        total_hits = self.cache_stats['hits']
        total_requests = total_hits + self.cache_stats['misses']
        
        if total_requests > 1000:  # Enough data for optimization
            hit_rate = total_hits / total_requests
            
            # Adjust L1 cache size based on hit rate
            if hit_rate < 0.8:
                # Increase L1 cache size
                if hasattr(self.layers[0], 'max_size'):
                    self.layers[0].max_size = min(
                        self.layers[0].max_size * 1.2, 50000
                    )
            elif hit_rate > 0.95:
                # Decrease L1 cache size to save memory
                if hasattr(self.layers[0], 'max_size'):
                    self.layers[0].max_size = max(
                        self.layers[0].max_size * 0.8, 1000
                    )
    
    def get_cache_statistics(self) -> Dict[str, Any]:
        """Get comprehensive cache statistics"""
        
        total_requests = self.cache_stats['hits'] + self.cache_stats['misses']
        hit_rate = (self.cache_stats['hits'] / total_requests 
                   if total_requests > 0 else 0)
        
        return {
            'hit_rate': hit_rate,
            'total_requests': total_requests,
            'predictions_made': self.cache_stats['predictions'],
            'prediction_accuracy': self.cache_stats['prediction_accuracy'],
            'l1_size': len(self.layers[0].cache) if hasattr(self.layers[0], 'cache') else 0,
            'memory_usage': self._calculate_memory_usage(),
            'performance_gain': self._calculate_performance_gain()
        }
    
    def _promote_to_higher_levels(self, key: str, value: Any, found_level: int):
        """Promote frequently accessed items to higher cache levels"""
        for i in range(found_level):
            self.layers[i].set(key, value)
    
    def _update_access_patterns(self, key: str):
        """Update access patterns for intelligent prediction"""
        if key not in self.access_patterns:
            self.access_patterns[key] = {
                'access_times': [],
                'related_keys': set(),
                'frequency': 0
            }
        
        pattern = self.access_patterns[key]
        pattern['access_times'].append(time.time())
        pattern['frequency'] += 1
        
        # Keep only recent access times (last 1000)
        if len(pattern['access_times']) > 1000:
            pattern['access_times'] = pattern['access_times'][-1000:]

# Example usage for AI security caching
class AISecurityCache:
    def __init__(self):
        self.cache_manager = IntelligentCacheManager()
    
    def cache_scan_result(self, input_hash: str, scan_result: Dict[str, Any], 
                         ttl: float = 1800):  # 30 minutes
        """Cache AI security scan results"""
        cache_key = f"scan_result:{input_hash}"
        return self.cache_manager.set(cache_key, scan_result, ttl)
    
    def get_cached_scan(self, input_hash: str) -> Optional[Dict[str, Any]]:
        """Get cached scan result"""
        cache_key = f"scan_result:{input_hash}"
        return self.cache_manager.get(cache_key)
    
    def cache_model_prediction(self, model_id: str, input_hash: str, 
                              prediction: Any, confidence: float):
        """Cache model predictions with confidence-based TTL"""
        cache_key = f"prediction:{model_id}:{input_hash}"
        
        # Higher confidence predictions cached longer
        ttl = 3600 * confidence  # 1 hour at 100% confidence
        
        self.cache_manager.set(cache_key, {
            'prediction': prediction,
            'confidence': confidence,
            'timestamp': time.time()
        }, ttl)
    
    def preload_user_patterns(self, user_id: str):
        """Preload cache based on user access patterns"""
        base_key = f"user_patterns:{user_id}"
        
        # Get related keys that this user typically accesses
        related_keys = self._get_user_related_keys(user_id)
        
        self.cache_manager.intelligent_preload(base_key, related_keys)

# Usage example
cache = AISecurityCache()

# Cache a scan result
input_text = "Some user input to analyze"
input_hash = hashlib.sha256(input_text.encode()).hexdigest()

scan_result = {
    'risk_score': 0.75,
    'threats': ['prompt_injection'],
    'confidence': 0.92
}

cache.cache_scan_result(input_hash, scan_result)

# Retrieve cached result
cached_result = cache.get_cached_scan(input_hash)
if cached_result:
    print(f"Cache hit! Risk score: {cached_result['risk_score']}")
else:
    print("Cache miss - performing new scan")