AI-Native Platforms 2026: Build for the $8.4B API Economy

from dataclasses import dataclass
from typing import List, Optional
import asyncio
from enum import Enum

class GPUType(Enum):
    H100 = "NVIDIA H100 80GB"
    A100 = "NVIDIA A100 80GB"
    L4 = "NVIDIA L4 24GB"
    T4 = "NVIDIA T4 16GB"

@dataclass
class GPUResource:
    gpu_id: str
    gpu_type: GPUType
    memory_total: int  # GB
    memory_available: int  # GB
    utilization: float  # 0.0 to 1.0
    current_tenant: Optional[str] = None

class GPUResourceManager:
    """Dynamic GPU allocation for multi-tenant AI workloads"""

    def __init__(self):
        self.gpu_pool: List[GPUResource] = []
        self.allocation_lock = asyncio.Lock()
        self.cost_per_hour = {
            GPUType.H100: 4.50,
            GPUType.A100: 3.00,
            GPUType.L4: 1.00,
            GPUType.T4: 0.50
        }

    async def allocate_gpu(
        self,
        tenant_id: str,
        memory_required: int,
        prefer_type: Optional[GPUType] = None
    ) -> Optional[GPUResource]:
        """Allocate GPU with intelligent selection"""
        async with self.allocation_lock:
            # Find best-fit GPU
            candidates = [
                gpu for gpu in self.gpu_pool
                if gpu.current_tenant is None
                and gpu.memory_available >= memory_required
            ]

            if not candidates:
                return None

            # Prefer specified type, otherwise choose most cost-effective
            if prefer_type:
                candidates = [g for g in candidates if g.gpu_type == prefer_type]

            if not candidates:
                return None

            # Select lowest cost GPU that meets requirements
            best_gpu = min(
                candidates,
                key=lambda g: self.cost_per_hour[g.gpu_type]
            )

            best_gpu.current_tenant = tenant_id
            best_gpu.memory_available -= memory_required

            return best_gpu

    async def release_gpu(self, gpu_id: str, memory_freed: int):
        """Release GPU resources back to pool"""
        async with self.allocation_lock:
            for gpu in self.gpu_pool:
                if gpu.gpu_id == gpu_id:
                    gpu.memory_available += memory_freed
                    if gpu.memory_available >= gpu.memory_total * 0.95:
                        gpu.current_tenant = None
                    break

    def get_cost_estimate(
        self,
        gpu_type: GPUType,
        hours: float
    ) -> float:
        """Calculate cost estimate for GPU usage"""
        return self.cost_per_hour[gpu_type] * hours

class MultiTenantGPUScheduler:
    """Hybrid spatial-temporal GPU scheduling"""

    def __init__(self, gpu_memory_gb: int):
        self.total_memory = gpu_memory_gb
        # Reserve 20% for high-priority spatial partitions
        self.spatial_memory = gpu_memory_gb * 0.2
        # 80% for temporal multiplexing
        self.temporal_memory = gpu_memory_gb * 0.8
        self.spatial_tenants = {}
        self.temporal_queue = asyncio.Queue()

    async def allocate_spatial(
        self,
        tenant_id: str,
        memory_gb: int,
        priority: str = "high"
    ) -> bool:
        """Allocate fixed GPU partition for latency-sensitive workloads"""
        if memory_gb > self.spatial_memory:
            return False

        self.spatial_tenants[tenant_id] = {
            "memory": memory_gb,
            "priority": priority,
            "allocated_at": asyncio.get_event_loop().time()
        }
        self.spatial_memory -= memory_gb
        return True

    async def enqueue_temporal(
        self,
        tenant_id: str,
        inference_request: dict
    ):
        """Queue request for temporal multiplexing"""
        await self.temporal_queue.put({
            "tenant_id": tenant_id,
            "request": inference_request,
            "queued_at": asyncio.get_event_loop().time()
        })

from typing import Tuple
from dataclasses import dataclass

@dataclass
class WorkloadProfile:
    model_size_gb: float
    latency_p95_ms: int
    throughput_rps: int
    batch_size: int
    daily_budget_usd: float

class HardwareSelector:
    """Select optimal GPU/ASIC for workload"""

    def recommend_hardware(
        self,
        profile: WorkloadProfile
    ) -> Tuple[GPUType, str]:
        """Return (hardware_type, reasoning)"""

        # Memory check
        if profile.model_size_gb > 80:
            return (
                GPUType.H100,
                "Model >80GB requires H100 80GB VRAM"
            )

        # Latency requirements
        if profile.latency_p95_ms < 50:
            if profile.model_size_gb > 40:
                return (
                    GPUType.H100,
                    "Sub-50ms latency + large model needs H100"
                )
            else:
                return (
                    GPUType.A100,
                    "Sub-50ms latency achievable on A100"
                )

        # High throughput, relaxed latency
        if profile.throughput_rps > 500 and profile.latency_p95_ms > 200:
            daily_cost_l4 = (24 * 1.00) # $1/hour
            daily_cost_a100 = (24 * 3.00) # $3/hour

            if daily_cost_l4 < profile.daily_budget_usd:
                return (
                    GPUType.L4,
                    f"L4 meets throughput at ${daily_cost_l4}/day"
                )

        # Default to A100 for balanced performance
        return (
            GPUType.A100,
            "A100 provides balanced performance/cost"
        )

import torch
from typing import Dict, Tuple
import hashlib

class KVCacheOptimizer:
    """Memory-efficient KV caching for transformer models"""

    def __init__(self, max_cache_size_gb: float = 10.0):
        self.max_cache_bytes = int(max_cache_size_gb * 1024**3)
        self.cache: Dict[str, Tuple[torch.Tensor, torch.Tensor]] = {}
        self.cache_stats = {"hits": 0, "misses": 0, "evictions": 0}
        self.current_size_bytes = 0

    def _hash_prompt(self, prompt: str, model_id: str) -> str:
        """Create hash for prompt + model combination"""
        return hashlib.sha256(
            f"{model_id}:{prompt}".encode()
        ).hexdigest()

    def get_kv_cache(
        self,
        prompt: str,
        model_id: str
    ) -> Tuple[torch.Tensor, torch.Tensor, bool]:
        """Retrieve cached KV tensors if available"""
        cache_key = self._hash_prompt(prompt, model_id)

        if cache_key in self.cache:
            self.cache_stats["hits"] += 1
            k_cache, v_cache = self.cache[cache_key]
            # Move to end (LRU)
            self.cache[cache_key] = self.cache.pop(cache_key)
            return k_cache, v_cache, True

        self.cache_stats["misses"] += 1
        return None, None, False

    def store_kv_cache(
        self,
        prompt: str,
        model_id: str,
        k_cache: torch.Tensor,
        v_cache: torch.Tensor
    ):
        """Store KV cache with LRU eviction"""
        cache_key = self._hash_prompt(prompt, model_id)

        # Calculate size
        cache_size = (
            k_cache.element_size() * k_cache.nelement() +
            v_cache.element_size() * v_cache.nelement()
        )

        # Evict if necessary (LRU)
        while (self.current_size_bytes + cache_size > self.max_cache_bytes
               and len(self.cache) > 0):
            # Remove oldest entry
            oldest_key = next(iter(self.cache))
            old_k, old_v = self.cache.pop(oldest_key)
            self.current_size_bytes -= (
                old_k.element_size() * old_k.nelement() +
                old_v.element_size() * old_v.nelement()
            )
            self.cache_stats["evictions"] += 1

        # Store new cache
        self.cache[cache_key] = (k_cache, v_cache)
        self.current_size_bytes += cache_size

    def get_hit_rate(self) -> float:
        """Calculate cache hit rate"""
        total = self.cache_stats["hits"] + self.cache_stats["misses"]
        if total == 0:
            return 0.0
        return self.cache_stats["hits"] / total

    def get_memory_usage_gb(self) -> float:
        """Current cache memory usage in GB"""
        return self.current_size_bytes / (1024**3)

import ray
from typing import List

@ray.remote(num_gpus=1)
class ModelShard:
    """Single shard of distributed model"""

    def __init__(self, shard_id: int, num_shards: int, model_config: dict):
        self.shard_id = shard_id
        self.num_shards = num_shards
        # Load only this shard's layers
        self.layers = self._load_shard_layers(model_config)

    def _load_shard_layers(self, config: dict):
        """Load model layers for this shard"""
        # Implementation: load subset of layers
        pass

    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
        """Forward pass through this shard's layers"""
        for layer in self.layers:
            hidden_states = layer(hidden_states)
        return hidden_states

class DistributedModelServer:
    """Serve large models across multiple GPUs"""

    def __init__(self, num_shards: int = 4, model_config: dict = None):
        ray.init(ignore_reinit_error=True)

        # Initialize model shards
        self.shards = [
            ModelShard.remote(i, num_shards, model_config)
            for i in range(num_shards)
        ]

    async def generate(
        self,
        prompt: str,
        max_tokens: int = 100
    ) -> str:
        """Generate text using distributed model"""
        # Tokenize
        input_ids = self._tokenize(prompt)
        hidden_states = self._embed(input_ids)

        # Pass through shards sequentially
        for shard in self.shards:
            hidden_states = await shard.forward.remote(hidden_states)

        # Decode
        output_text = self._decode(hidden_states)
        return output_text

    def _tokenize(self, text: str):
        # Implementation
        pass

    def _embed(self, input_ids):
        # Implementation
        pass

    def _decode(self, hidden_states):
        # Implementation
        pass

from enum import Enum
import torch

class MemoryTier(Enum):
    VRAM = "GPU VRAM"  # Fastest, most expensive
    RAM = "System RAM"  # Medium speed/cost
    DISK = "NVMe SSD"  # Slowest, cheapest

class HybridMemoryManager:
    """Coordinate RAM, VRAM, and disk for large models"""

    def __init__(
        self,
        vram_budget_gb: float = 40.0,
        ram_budget_gb: float = 128.0,
        disk_budget_gb: float = 512.0
    ):
        self.vram_budget = vram_budget_gb * 1024**3
        self.ram_budget = ram_budget_gb * 1024**3
        self.disk_budget = disk_budget_gb * 1024**3

        self.vram_used = 0
        self.ram_used = 0
        self.disk_used = 0

        self.layer_locations = {}  # layer_id -> MemoryTier

    def place_layer(
        self,
        layer_id: str,
        layer_size_bytes: int,
        access_frequency: float  # 0.0 to 1.0
    ) -> MemoryTier:
        """Intelligently place layer in memory hierarchy"""

        # Hot layers (frequently accessed) -> VRAM
        if access_frequency > 0.7 and self.vram_used + layer_size_bytes <= self.vram_budget:
            self.vram_used += layer_size_bytes
            tier = MemoryTier.VRAM

        # Warm layers -> RAM
        elif access_frequency > 0.3 and self.ram_used + layer_size_bytes <= self.ram_budget:
            self.ram_used += layer_size_bytes
            tier = MemoryTier.RAM

        # Cold layers -> Disk
        else:
            if self.disk_used + layer_size_bytes <= self.disk_budget:
                self.disk_used += layer_size_bytes
                tier = MemoryTier.DISK
            else:
                raise MemoryError("Insufficient storage across all tiers")

        self.layer_locations[layer_id] = tier
        return tier

    def get_layer_latency_ms(self, tier: MemoryTier) -> float:
        """Expected latency for layer access"""
        latencies = {
            MemoryTier.VRAM: 0.1,  # 100 microseconds
            MemoryTier.RAM: 2.0,   # 2ms
            MemoryTier.DISK: 15.0  # 15ms
        }
        return latencies[tier]

    def optimize_placement(self, access_stats: dict):
        """Re-optimize layer placement based on access patterns"""
        # Collect (layer_id, access_frequency) pairs
        layers = []
        for layer_id, freq in access_stats.items():
            if layer_id in self.layer_locations:
                layers.append((layer_id, freq))

        # Sort by access frequency (descending)
        layers.sort(key=lambda x: x[1], reverse=True)

        # Reset allocations
        self.vram_used = 0
        self.ram_used = 0
        self.disk_used = 0
        self.layer_locations.clear()

        # Re-place layers
        for layer_id, freq in layers:
            # Get layer size (would be stored separately)
            layer_size = self._get_layer_size(layer_id)
            self.place_layer(layer_id, layer_size, freq)

    def _get_layer_size(self, layer_id: str) -> int:
        # Implementation: retrieve layer size from metadata
        return 1024**3  # Placeholder: 1GB

from typing import Optional, Dict, Any
import asyncio
import httpx
from enum import Enum
import time
import os

class AIProvider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    TOGETHER = "together"
    REPLICATE = "replicate"

class AIProviderGateway:
    """Route requests across OpenAI, Anthropic, Together, etc."""

    def __init__(self):
        self.providers = {
            AIProvider.OPENAI: {
                "endpoint": "https://api.openai.com/v1/chat/completions",
                "api_key": os.getenv("OPENAI_API_KEY"),
                "cost_per_1k_tokens": {"input": 0.01, "output": 0.03}
            },
            AIProvider.ANTHROPIC: {
                "endpoint": "https://api.anthropic.com/v1/messages",
                "api_key": os.getenv("ANTHROPIC_API_KEY"),
                "cost_per_1k_tokens": {"input": 0.008, "output": 0.024}
            },
            AIProvider.TOGETHER: {
                "endpoint": "https://api.together.xyz/v1/chat/completions",
                "api_key": os.getenv("TOGETHER_API_KEY"),
                "cost_per_1k_tokens": {"input": 0.002, "output": 0.006}
            }
        }

        self.provider_health = {
            provider: {"available": True, "latency_ms": 0, "error_rate": 0.0}
            for provider in AIProvider
        }

        self.circuit_breakers = {
            provider: CircuitBreaker(failure_threshold=5, timeout_seconds=60)
            for provider in AIProvider
        }

    async def chat_completion(
        self,
        messages: list,
        model: str = "gpt-4",
        temperature: float = 0.7,
        max_tokens: int = 1000,
        preferred_provider: Optional[AIProvider] = None
    ) -> Dict[str, Any]:
        """Route chat completion to optimal provider"""

        # Select provider
        if preferred_provider and self.provider_health[preferred_provider]["available"]:
            provider = preferred_provider
        else:
            provider = self._select_best_provider(model)

        # Check circuit breaker
        if not self.circuit_breakers[provider].is_available():
            # Fallback to next best provider
            provider = self._select_fallback_provider(provider, model)

        try:
            start_time = time.time()

            # Make request
            response = await self._make_request(
                provider,
                messages,
                model,
                temperature,
                max_tokens
            )

            # Update health metrics
            latency_ms = (time.time() - start_time) * 1000
            self._update_health(provider, success=True, latency_ms=latency_ms)

            return response

        except Exception as e:
            self._update_health(provider, success=False)
            self.circuit_breakers[provider].record_failure()

            # Retry with fallback provider
            fallback = self._select_fallback_provider(provider, model)
            return await self.chat_completion(
                messages, model, temperature, max_tokens, fallback
            )

    def _select_best_provider(self, model: str) -> AIProvider:
        """Select provider based on cost and latency"""
        # Simple selection: lowest cost
        costs = {
            AIProvider.TOGETHER: 0.002,
            AIProvider.ANTHROPIC: 0.008,
            AIProvider.OPENAI: 0.01
        }

        available_providers = [
            p for p, health in self.provider_health.items()
            if health["available"]
        ]

        if not available_providers:
            raise Exception("No providers available")

        return min(available_providers, key=lambda p: costs.get(p, float('inf')))

    def _select_fallback_provider(
        self,
        failed_provider: AIProvider,
        model: str
    ) -> AIProvider:
        """Select fallback when primary fails"""
        available = [
            p for p in AIProvider
            if p != failed_provider
            and self.provider_health[p]["available"]
        ]

        if not available:
            raise Exception("No fallback providers available")

        # Select provider with lowest error rate
        return min(
            available,
            key=lambda p: self.provider_health[p]["error_rate"]
        )

    async def _make_request(
        self,
        provider: AIProvider,
        messages: list,
        model: str,
        temperature: float,
        max_tokens: int
    ) -> dict:
        """Make API request to provider"""
        config = self.providers[provider]

        async with httpx.AsyncClient() as client:
            response = await client.post(
                config["endpoint"],
                headers={
                    "Authorization": f"Bearer {config['api_key']}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model,
                    "messages": messages,
                    "temperature": temperature,
                    "max_tokens": max_tokens
                },
                timeout=30.0
            )
            response.raise_for_status()
            return response.json()

    def _update_health(
        self,
        provider: AIProvider,
        success: bool,
        latency_ms: float = 0
    ):
        """Update provider health metrics"""
        health = self.provider_health[provider]

        if success:
            # Exponential moving average for latency
            alpha = 0.3
            health["latency_ms"] = (
                alpha * latency_ms +
                (1 - alpha) * health["latency_ms"]
            )
            # Decrease error rate
            health["error_rate"] *= 0.95
        else:
            # Increase error rate
            health["error_rate"] = min(1.0, health["error_rate"] + 0.1)

        # Mark unavailable if error rate too high
        health["available"] = health["error_rate"] < 0.5

class CircuitBreaker:
    """Prevent cascade failures"""

    def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout_seconds = timeout_seconds
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    def is_available(self) -> bool:
        """Check if circuit breaker allows requests"""
        if self.state == "CLOSED":
            return True

        if self.state == "OPEN":
            # Check if timeout expired
            if time.time() - self.last_failure_time > self.timeout_seconds:
                self.state = "HALF_OPEN"
                return True
            return False

        if self.state == "HALF_OPEN":
            return True

        return False

    def record_failure(self):
        """Record failure and potentially open circuit"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

    def record_success(self):
        """Record success and potentially close circuit"""
        if self.state == "HALF_OPEN":
            self.state = "CLOSED"
            self.failure_count = 0

from dataclasses import dataclass
from datetime import datetime, timedelta
import asyncio

@dataclass
class CostEvent:
    timestamp: datetime
    provider: AIProvider
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    tenant_id: str

class APICostTracker:
    """Real-time cost tracking and budget enforcement"""

    def __init__(self):
        self.cost_events = []
        self.tenant_budgets = {}  # tenant_id -> budget_usd
        self.tenant_spent = {}  # tenant_id -> spent_usd
        self.alerts = []

    def set_budget(self, tenant_id: str, daily_budget_usd: float):
        """Set daily budget for tenant"""
        self.tenant_budgets[tenant_id] = daily_budget_usd
        if tenant_id not in self.tenant_spent:
            self.tenant_spent[tenant_id] = 0.0

    def track_cost(
        self,
        provider: AIProvider,
        model: str,
        input_tokens: int,
        output_tokens: int,
        tenant_id: str
    ) -> float:
        """Track cost and return total spent"""
        # Calculate cost
        cost_config = self._get_cost_config(provider, model)
        cost_usd = (
            (input_tokens / 1000) * cost_config["input"] +
            (output_tokens / 1000) * cost_config["output"]
        )

        # Record event
        event = CostEvent(
            timestamp=datetime.now(),
            provider=provider,
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost_usd=cost_usd,
            tenant_id=tenant_id
        )
        self.cost_events.append(event)

        # Update tenant spending
        if tenant_id not in self.tenant_spent:
            self.tenant_spent[tenant_id] = 0.0
        self.tenant_spent[tenant_id] += cost_usd

        # Check budget
        if tenant_id in self.tenant_budgets:
            budget = self.tenant_budgets[tenant_id]
            spent = self.tenant_spent[tenant_id]

            if spent > budget * 0.9:
                self._send_alert(
                    tenant_id,
                    f"90% of budget used: ${spent:.2f} / ${budget:.2f}"
                )

            if spent > budget:
                raise BudgetExceededError(
                    f"Tenant {tenant_id} exceeded daily budget: "
                    f"${spent:.2f} > ${budget:.2f}"
                )

        return self.tenant_spent[tenant_id]

    def get_daily_cost(self, tenant_id: str) -> float:
        """Get today's cost for tenant"""
        today = datetime.now().date()
        total = sum(
            event.cost_usd
            for event in self.cost_events
            if event.tenant_id == tenant_id
            and event.timestamp.date() == today
        )
        return total

    def get_cost_breakdown(
        self,
        tenant_id: str,
        days: int = 7
    ) -> Dict[str, float]:
        """Get cost breakdown by provider"""
        cutoff = datetime.now() - timedelta(days=days)

        breakdown = {}
        for event in self.cost_events:
            if event.tenant_id == tenant_id and event.timestamp > cutoff:
                provider_name = event.provider.value
                if provider_name not in breakdown:
                    breakdown[provider_name] = 0.0
                breakdown[provider_name] += event.cost_usd

        return breakdown

    def _get_cost_config(self, provider: AIProvider, model: str) -> dict:
        """Get cost per 1K tokens"""
        # Simplified - would be more comprehensive in production
        configs = {
            AIProvider.OPENAI: {"input": 0.01, "output": 0.03},
            AIProvider.ANTHROPIC: {"input": 0.008, "output": 0.024},
            AIProvider.TOGETHER: {"input": 0.002, "output": 0.006}
        }
        return configs.get(provider, {"input": 0.01, "output": 0.03})

    def _send_alert(self, tenant_id: str, message: str):
        """Send budget alert"""
        self.alerts.append({
            "timestamp": datetime.now(),
            "tenant_id": tenant_id,
            "message": message
        })
        # In production: send email, Slack notification, etc.

class BudgetExceededError(Exception):
    pass

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

@dataclass
class WorkloadMetrics:
    timestamp: datetime
    queue_depth: int
    avg_latency_ms: float
    gpu_utilization: float
    requests_per_second: float

class AIWorkloadScaler:
    """Auto-scale GPU instances based on queue depth and latency"""

    def __init__(
        self,
        min_instances: int = 2,
        max_instances: int = 20,
        target_queue_depth: int = 10,
        target_latency_ms: float = 100.0,
        scale_up_threshold: float = 1.5,
        scale_down_threshold: float = 0.5,
        cooldown_seconds: int = 300
    ):
        self.min_instances = min_instances
        self.max_instances = max_instances
        self.current_instances = min_instances
        self.target_queue_depth = target_queue_depth
        self.target_latency_ms = target_latency_ms
        self.scale_up_threshold = scale_up_threshold
        self.scale_down_threshold = scale_down_threshold
        self.cooldown_seconds = cooldown_seconds
        self.last_scale_time = datetime.now()
        self.metrics_history: List[WorkloadMetrics] = []

    def should_scale(self, metrics: WorkloadMetrics) -> int:
        """
        Determine if scaling is needed
        Returns: positive = scale up, negative = scale down, 0 = no change
        """
        # Cooldown check
        if (datetime.now() - self.last_scale_time).seconds < self.cooldown_seconds:
            return 0

        # Calculate current ratios
        queue_ratio = metrics.queue_depth / self.target_queue_depth
        latency_ratio = metrics.avg_latency_ms / self.target_latency_ms

        # Predictive scaling: look at trend
        trend = self._calculate_trend()

        # Scale up if queue or latency exceeds threshold
        if (queue_ratio > self.scale_up_threshold or
            latency_ratio > self.scale_up_threshold or
            trend > 0.2):  # 20% upward trend

            if self.current_instances < self.max_instances:
                # Calculate desired instances
                desired = min(
                    self.max_instances,
                    int(self.current_instances * 1.5)  # 50% increase
                )
                return desired - self.current_instances

        # Scale down if underutilized
        elif (queue_ratio < self.scale_down_threshold and
              latency_ratio < self.scale_down_threshold and
              metrics.gpu_utilization < 0.3 and
              trend < -0.1):  # 10% downward trend

            if self.current_instances > self.min_instances:
                # Calculate desired instances
                desired = max(
                    self.min_instances,
                    int(self.current_instances * 0.75)  # 25% decrease
                )
                return desired - self.current_instances

        return 0

    def _calculate_trend(self) -> float:
        """Calculate request rate trend over last 5 minutes"""
        if len(self.metrics_history) < 2:
            return 0.0

        # Get metrics from last 5 minutes
        cutoff = datetime.now() - timedelta(minutes=5)
        recent = [
            m for m in self.metrics_history
            if m.timestamp > cutoff
        ]

        if len(recent) < 2:
            return 0.0

        # Simple linear trend
        first_rps = recent[0].requests_per_second
        last_rps = recent[-1].requests_per_second

        if first_rps == 0:
            return 0.0

        return (last_rps - first_rps) / first_rps

    async def scale_instances(self, delta: int):
        """Execute scaling action"""
        new_count = self.current_instances + delta
        new_count = max(self.min_instances, min(self.max_instances, new_count))

        if delta > 0:
            # Scale up
            for i in range(delta):
                await self._launch_instance()
        elif delta < 0:
            # Scale down
            for i in range(abs(delta)):
                await self._terminate_instance()

        self.current_instances = new_count
        self.last_scale_time = datetime.now()

    async def _launch_instance(self):
        """Launch new GPU instance"""
        # Implementation: call cloud provider API
        await asyncio.sleep(0.1)  # Placeholder

    async def _terminate_instance(self):
        """Terminate GPU instance"""
        # Implementation: graceful shutdown + termination
        await asyncio.sleep(0.1)  # Placeholder

import pickle
import os
import json
from functools import lru_cache

class ServerlessAIHandler:
    """Serverless function for AI inference with cold start optimization"""

    # Class-level cache survives across invocations
    _model_cache = {}
    _initialized = False

    def __init__(self):
        if not ServerlessAIHandler._initialized:
            self._warm_start()
            ServerlessAIHandler._initialized = True

    def _warm_start(self):
        """Optimize cold start time"""
        # Pre-load model during container initialization
        model_path = os.getenv("MODEL_PATH", "/opt/model")

        if os.path.exists(f"{model_path}/config.json"):
            # Load lightweight config first
            with open(f"{model_path}/config.json") as f:
                config = json.load(f)
                ServerlessAIHandler._model_cache["config"] = config

            # Lazy-load heavy weights only when needed
            # This reduces cold start from 15s to 2s

    @lru_cache(maxsize=128)
    def _get_model(self, model_id: str):
        """Lazy-load model with caching"""
        if model_id in ServerlessAIHandler._model_cache:
            return ServerlessAIHandler._model_cache[model_id]

        # Load model
        model_path = os.getenv("MODEL_PATH", "/opt/model")
        with open(f"{model_path}/{model_id}.pkl", "rb") as f:
            model = pickle.load(f)

        ServerlessAIHandler._model_cache[model_id] = model
        return model

    async def handle_request(self, event: dict) -> dict:
        """Handle inference request"""
        model_id = event.get("model_id", "default")
        input_text = event.get("input")

        # Get cached model
        model = self._get_model(model_id)

        # Run inference
        output = await self._inference(model, input_text)

        return {
            "statusCode": 200,
            "body": {
                "output": output,
                "model_id": model_id
            }
        }

    async def _inference(self, model, input_text: str) -> str:
        """Run model inference"""
        # Implementation
        return f"Generated response for: {input_text}"

from enum import Enum
import asyncio

class AWSRegion(Enum):
    US_EAST_1 = "us-east-1"
    US_WEST_2 = "us-west-2"
    EU_WEST_1 = "eu-west-1"
    AP_SOUTHEAST_1 = "ap-southeast-1"
    AP_NORTHEAST_1 = "ap-northeast-1"
    SA_EAST_1 = "sa-east-1"

class MultiRegionAIDeployment:
    """Deploy AI services across multiple regions for low latency"""

    def __init__(self):
        self.regional_endpoints = {
            AWSRegion.US_EAST_1: {"url": "https://api-use1.example.com", "latency_ms": 0},
            AWSRegion.US_WEST_2: {"url": "https://api-usw2.example.com", "latency_ms": 0},
            AWSRegion.EU_WEST_1: {"url": "https://api-euw1.example.com", "latency_ms": 0},
            AWSRegion.AP_SOUTHEAST_1: {"url": "https://api-apse1.example.com", "latency_ms": 0},
            AWSRegion.AP_NORTHEAST_1: {"url": "https://api-apne1.example.com", "latency_ms": 0},
            AWSRegion.SA_EAST_1: {"url": "https://api-sae1.example.com", "latency_ms": 0},
        }

    def select_region(self, client_ip: str) -> AWSRegion:
        """Select optimal region based on client location"""
        # In production: use GeoIP lookup
        # Simplified: parse IP prefix
        if client_ip.startswith("54."):  # US East
            return AWSRegion.US_EAST_1
        elif client_ip.startswith("52."):  # US West
            return AWSRegion.US_WEST_2
        elif client_ip.startswith("3."):  # EU
            return AWSRegion.EU_WEST_1
        elif client_ip.startswith("13."):  # Asia Pacific
            return AWSRegion.AP_SOUTHEAST_1
        else:
            # Default to lowest latency
            return self._get_lowest_latency_region()

    def _get_lowest_latency_region(self) -> AWSRegion:
        """Select region with lowest latency"""
        return min(
            self.regional_endpoints.items(),
            key=lambda x: x[1]["latency_ms"]
        )[0]

    async def route_request(
        self,
        client_ip: str,
        request_data: dict
    ) -> dict:
        """Route request to optimal region"""
        region = self.select_region(client_ip)
        endpoint = self.regional_endpoints[region]["url"]

        # Make request to regional endpoint
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{endpoint}/inference",
                json=request_data,
                timeout=10.0
            )
            return response.json()

class EcommerceRecommendationPlatform:
    """Production recommendation system serving 10M+ users"""

    def __init__(self):
        # GPU pool: 8x NVIDIA L4 instances
        self.gpu_pool = GPUResourceManager()
        for i in range(8):
            self.gpu_pool.gpu_pool.append(
                GPUResource(
                    gpu_id=f"l4-{i}",
                    gpu_type=GPUType.L4,
                    memory_total=24,
                    memory_available=24,
                    utilization=0.0
                )
            )

        # Multi-tier model serving
        self.model_tiers = {
            "small": "recommendation-7b",   # 90% of requests
            "medium": "recommendation-13b", # 9% of requests
            "large": "recommendation-70b"   # 1% of requests (VIP users)
        }

        # Embedding cache for product features
        self.embedding_cache = EmbeddingCache(max_size_gb=20.0)

        # Cost tracker
        self.cost_tracker = APICostTracker()
        self.cost_tracker.set_budget("ecommerce", daily_budget_usd=1667)  # $50K/month

    async def get_recommendations(
        self,
        user_id: str,
        context: dict,
        user_tier: str = "standard"
    ) -> list:
        """Get personalized recommendations"""

        # Select model based on user tier
        if user_tier == "vip":
            model = self.model_tiers["large"]
            memory_required = 35  # GB
        elif user_tier == "premium":
            model = self.model_tiers["medium"]
            memory_required = 13
        else:
            model = self.model_tiers["small"]
            memory_required = 7

        # Check embedding cache
        cache_key = f"user_embed_{user_id}"
        user_embedding, _, cached = self.embedding_cache.get(cache_key)

        if not cached:
            # Compute user embedding
            gpu = await self.gpu_pool.allocate_gpu(
                tenant_id=user_id,
                memory_required=memory_required,
                prefer_type=GPUType.L4
            )

            user_embedding = await self._compute_embedding(user_id, context, gpu)

            # Cache for 1 hour
            self.embedding_cache.store(cache_key, user_embedding, ttl_seconds=3600)

            await self.gpu_pool.release_gpu(gpu.gpu_id, memory_required)

        # Get product recommendations
        recommendations = await self._rank_products(user_embedding, model)

        return recommendations

    async def _compute_embedding(self, user_id: str, context: dict, gpu: GPUResource):
        """Compute user embedding on GPU"""
        # Implementation: run embedding model
        return torch.randn(768)  # Placeholder

    async def _rank_products(self, user_embedding, model: str) -> list:
        """Rank products by relevance"""
        # Implementation: similarity search + ranking
        return ["product_123", "product_456", "product_789"]

class HIPAACompliantImagingPlatform:
    """Medical imaging analysis with HIPAA compliance"""

    def __init__(self):
        self.encryption_key = self._load_encryption_key()
        self.audit_logger = ComplianceAuditLogger()
        self.model_registry = MedicalModelRegistry()

    async def analyze_scan(
        self,
        scan_id: str,
        scan_data: bytes,
        patient_id: str,
        urgency: str = "routine"
    ) -> dict:
        """Analyze medical scan with full audit trail"""

        # Audit: log access
        self.audit_logger.log_access(
            resource_type="medical_scan",
            resource_id=scan_id,
            patient_id=patient_id,
            action="analyze",
            timestamp=datetime.now()
        )

        # Encrypt scan data at rest
        encrypted_scan = self._encrypt_phi(scan_data)

        # Select model (production vs. canary for A/B testing)
        model_version = self.model_registry.get_production_model(
            modality="ct_scan",
            use_canary_pct=5.0  # 5% canary traffic
        )

        # Run inference
        results = await self._run_diagnostic_model(
            encrypted_scan,
            model_version,
            priority="high" if urgency == "stat" else "normal"
        )

        # Audit: log results
        self.audit_logger.log_result(
            scan_id=scan_id,
            model_version=model_version,
            confidence=results["confidence"],
            findings=results["findings"],
            timestamp=datetime.now()
        )

        return results

    def _encrypt_phi(self, data: bytes) -> bytes:
        """Encrypt Protected Health Information"""
        # Implementation: AES-256 encryption
        return data  # Placeholder

    async def _run_diagnostic_model(
        self,
        scan_data: bytes,
        model_version: str,
        priority: str
    ) -> dict:
        """Run diagnostic AI model"""
        # Implementation: model inference
        return {
            "confidence": 0.96,
            "findings": ["potential_nodule_upper_left_lobe"],
            "requires_radiologist_review": True
        }

class FutureHardwareStrategy:
    """Plan for hardware transitions"""

    def evaluate_new_hardware(
        self,
        hardware_type: str,
        current_cost_per_inference: float,
        current_latency_ms: float
    ) -> dict:
        """Evaluate if new hardware is worth adopting"""

        # Cost-benefit analysis
        adoption_threshold = {
            "cost_reduction": 0.30,  # 30% cost reduction
            "latency_improvement": 0.40,  # 40% latency improvement
            "or_combination": True
        }

        # Placeholder: would fetch real benchmarks
        new_hardware_cost = current_cost_per_inference * 0.65  # 35% reduction
        new_hardware_latency = current_latency_ms * 0.70  # 30% improvement

        cost_benefit = (current_cost_per_inference - new_hardware_cost) / current_cost_per_inference
        latency_benefit = (current_latency_ms - new_hardware_latency) / current_latency_ms

        should_adopt = (
            cost_benefit >= adoption_threshold["cost_reduction"] or
            latency_benefit >= adoption_threshold["latency_improvement"]
        )

        return {
            "should_adopt": should_adopt,
            "cost_benefit": f"{cost_benefit:.0%}",
            "latency_benefit": f"{latency_benefit:.0%}",
            "estimated_payback_months": 6 if should_adopt else None
        }

class GreenAIOptimizer:
    """Optimize AI infrastructure for energy efficiency"""

    def __init__(self):
        self.carbon_intensity = {
            # gCO2/kWh by region
            AWSRegion.US_EAST_1: 390,
            AWSRegion.US_WEST_2: 90,  # Hydro-powered
            AWSRegion.EU_WEST_1: 250,
            AWSRegion.AP_SOUTHEAST_1: 480,
        }

    def select_green_region(
        self,
        latency_tolerance_ms: int = 200
    ) -> AWSRegion:
        """Select region with lowest carbon intensity"""
        # Filter regions meeting latency requirement
        viable_regions = [
            region for region, intensity in self.carbon_intensity.items()
            if self._estimate_latency(region) <= latency_tolerance_ms
        ]

        # Select lowest carbon
        return min(
            viable_regions,
            key=lambda r: self.carbon_intensity[r]
        )

    def calculate_carbon_footprint(
        self,
        gpu_hours: float,
        region: AWSRegion,
        gpu_type: GPUType = GPUType.A100
    ) -> float:
        """Calculate carbon emissions in kg CO2"""
        # A100 TDP: 400W
        gpu_power_kw = 0.4

        energy_kwh = gpu_hours * gpu_power_kw
        carbon_intensity = self.carbon_intensity[region]

        # Convert gCO2 to kgCO2
        carbon_kg = (energy_kwh * carbon_intensity) / 1000

        return carbon_kg

    def _estimate_latency(self, region: AWSRegion) -> int:
        """Estimate latency to region"""
        # Simplified
        return 100  # ms