# Cost Optimization

Strategies for optimizing agent system costs: compute, tokens, storage, and operational efficiency.

---

## Cost Analysis Framework

### Cost Components
```
Total Cost = Compute + Tokens + Storage + Network + Operations
```

| Component | Typical % | Optimization Levers |
|---|---|---|
| **LLM Tokens** | 40-60% | Model selection, caching, prompt compression |
| **Compute** | 20-30% | Right-sizing, spot instances, serverless |
| **Storage** | 5-10% | Tiering, compression, lifecycle policies |
| **Network** | 5-10% | CDN, compression, regional deployment |
| **Operations** | 5-10% | Automation, observability efficiency |

---

## Token Cost Optimization

### Model Selection Strategy
```python
class ModelSelector:
    def __init__(self):
        self.models = {
            "gpt-4o": ModelProfile(cost_per_1k_in=0.005, cost_per_1k_out=0.015, quality=1.0, speed=1.0),
            "gpt-4o-mini": ModelProfile(cost_per_1k_in=0.00015, cost_per_1k_out=0.0006, quality=0.85, speed=2.0),
            "gpt-3.5-turbo": ModelProfile(cost_per_1k_in=0.0005, cost_per_1k_out=0.0015, quality=0.7, speed=3.0),
            "claude-3.5-sonnet": ModelProfile(cost_per_1k_in=0.003, cost_per_1k_out=0.015, quality=0.95, speed=1.2),
            "claude-3-haiku": ModelProfile(cost_per_1k_in=0.00025, cost_per_1k_out=0.00125, quality=0.75, speed=4.0),
            "gemini-1.5-flash": ModelProfile(cost_per_1k_in=0.000075, cost_per_1k_out=0.0003, quality=0.8, speed=3.0),
        }
    
    def select_model(self, task: Task, budget_per_request: float = None, 
                     quality_threshold: float = 0.7) -> str:
        """Select cheapest model meeting quality threshold"""
        candidates = [
            m for m in self.models.values() 
            if m.quality >= quality_threshold
        ]
        
        if budget_per_request:
            candidates = [m for m in candidates 
                         if m.estimated_cost_per_request() <= budget_per_request]
        
        return min(candidates, key=lambda m: m.estimated_cost_per_request()).name
    
    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        m = self.models[model]
        return (input_tokens * m.cost_per_1k_in + output_tokens * m.cost_per_1k_out) / 1000
```

### Prompt Compression
```python
class PromptCompressor:
    def __init__(self, tokenizer, target_ratio: float = 0.5):
        self.tokenizer = tokenizer
        self.target_ratio = target_ratio
    
    def compress(self, text: str, preserve_keywords: list[str] = None) -> str:
        """Compress prompt while preserving key information"""
        tokens = self.tokenizer.encode(text)
        target_tokens = int(len(tokens) * self.target_ratio)
        
        if len(tokens) <= target_tokens:
            return text
        
        # Strategy 1: Remove redundant whitespace
        compressed = self._remove_redundancy(text)
        
        # Strategy 2: Summarize with LLM (if still too long)
        if len(self.tokenizer.encode(compressed)) > target_tokens:
            compressed = self._llm_summarize(text, target_tokens)
        
        # Strategy 3: Preserve keywords, truncate rest
        if len(self.tokenizer.encode(compressed)) > target_tokens:
            compressed = self._preserve_keywords(compressed, preserve_keywords, target_tokens)
        
        return compressed
    
    def _preserve_keywords(self, text: str, keywords: list[str], max_tokens: int) -> str:
        # Keep sentences containing keywords
        sentences = sent_tokenize(text)
        scored = []
        for sent in sentences:
            score = sum(1 for kw in keywords if kw.lower() in sent.lower())
            scored.append((score, sent))
        
        # Sort by score, take top until token limit
        scored.sort(key=lambda x: x[0], reverse=True)
        result = []
        tokens = 0
        for score, sent in scored:
            sent_tokens = len(self.tokenizer.encode(sent))
            if tokens + sent_tokens <= max_tokens:
                result.append(sent)
                tokens += sent_tokens
            else:
                break
        return " ".join(result)
```

### Prompt Caching
```python
class PromptCache:
    def __init__(self, redis: Redis, ttl: int = 3600):
        self.redis = redis
        self.ttl = ttl
    
    def cache_key(self, prompt: str, model: str, params: dict) -> str:
        content = f"{model}:{json.dumps(prompt, sort_keys=True)}:{json.dumps(params, sort_keys=True)}"
        return f"prompt_cache:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"
    
    async def get(self, prompt: str, model: str, params: dict) -> Optional[str]:
        key = self.cache_key(prompt, model, params)
        return await self.redis.get(key)
    
    async def set(self, prompt: str, model: str, params: dict, response: str):
        key = self.cache_key(prompt, model, params)
        await self.redis.setex(f"prompt_cache:{key}", self.ttl, response)
    
    async def get_or_compute(self, prompt: str, model: str, params: dict, 
                             compute_fn: Callable) -> str:
        cached = await self.get(prompt, model, {})
        if cached:
            return cached
        
        response = await compute_fn(prompt, model)
        await self.set(prompt, model, {}, response)
        return response
```

### Semantic Caching (Embedding-based)
```python
class SemanticPromptCache:
    def __init__(self, redis: Redis, embedder: EmbeddingModel, 
                 similarity_threshold: float = 0.95, ttl: int = 3600):
        self.redis = redis
        self.embedder = embedder
        self.threshold = similarity_threshold
        self.ttl = ttl
    
    async def get(self, prompt: str) -> Optional[str]:
        embedding = await self.embedder.embed(prompt)
        
        # Search for similar prompts
        similar = await self.search_similar(prompt, top_k=5)
        
        for cached_prompt, cached_response, similarity in similar:
            if similarity >= self.threshold:
                return cached_response
        return None
    
    async def set(self, prompt: str, response: str):
        embedding = await self.embedder.embed(prompt)
        key = f"semantic_cache:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"
        
        await self.redis.hset(f"semantic_cache:{prompt[:50]}", mapping={
            "prompt": prompt,
            "response": json.dumps(response),
            "embedding": json.dumps(embedding.tolist())
        })
        await self.redis.expire(f"semantic_cache:{prompt[:50]}", self.ttl)
```

---

## Compute Cost Optimization

### Right-Sizing Instances
```python
class InstanceRightSizer:
    def __init__(self, metrics_client: MetricsClient):
        self.metrics = metrics_client
    
    async def analyze(self, deployment: str, window: str = "7d") -> RightSizingRecommendation:
        metrics = await self.metrics.get_container_metrics(deployment, window)
        
        cpu_util = statistics.mean(m.cpu_utilization for m in metrics)
        memory_util = statistics.mean(m.memory_utilization for m in metrics)
        
        # Current instance
        current = self.get_current_instance_type()
        
        # Find optimal
        optimal = self.find_optimal_instance(cpu_util, memory_util)
        
        savings = self.calculate_savings(current, optimal)
        
        return RightSizingRecommendation(
            current_instance=current,
            recommended_instance=optimal,
            cpu_utilization=cpu_util,
            memory_utilization=memory_util,
            estimated_monthly_savings=savings
        )
    
    def find_optimal_instance(self, cpu_util: float, mem_util: float) -> InstanceType:
        # Target 70% utilization
        target_cpu = 0.7
        target_mem = 0.7
        
        cpu_factor = cpu_util / target_cpu if cpu_util > 0 else 1
        mem_factor = memory_util / target_mem if memory_util > 0 else 1
        
        scale_factor = max(cpu_factor, mem_factor)
        
        # Find instance with appropriate capacity
        for instance in INSTANCE_TYPES:
            if (instance.cpu >= current.cpu / scale_factor and 
                instance.memory >= current.memory / scale_factor):
                return instance
        
        return current  # Already optimal
```

### Spot Instance Strategy
```python
class SpotInstanceManager:
    def __init__(self, cloud_provider: CloudProvider):
        self.provider = cloud_provider
    
    async def create_spot_fleet(self, spec: SpotFleetSpec) -> SpotFleet:
        # Diversify across instance types and AZs
        instance_types = self.get_diverse_instance_types(spec.requirements)
        
        fleet = await self.provider.create_spot_fleet(
            instance_types=instance_types,
            target_capacity=spec.target_capacity,
            allocation_strategy="diversified",
            max_price=spec.max_price,
            allocation_strategy="lowest-price",
            instance_interruption_behavior="terminate"
        )
        
        # Add fallback to on-demand
        await self.add_on_demand_fallback(fleet, spec.fallback_percentage)
        
        return fleet
    
    async def handle_interruption(self, instance_id: str):
        # Graceful draining
        await self.drain_node(instance_id)
        
        # Replace with new spot or on-demand
        await self.replace_instance(instance_id)
```

---

## Serverless Cost Optimization

### Right-Sizing Lambda
```python
class LambdaOptimizer:
    def __init__(self, cloudwatch: CloudWatchClient):
        self.cw = cloudwatch
    
    async def optimize_function(self, function_name: str) -> OptimizationResult:
        # Get metrics
        metrics = await self.get_metrics(function_name, days=14)
        
        # Analyze
        avg_duration = statistics.mean(m.duration for m in metrics.durations)
        max_memory_used = max(m.max_memory_used for m in metrics.memory)
        avg_memory_used = statistics.mean(m.avg_memory_used for m in metrics.memory)
        
        # Current config
        current_memory = self.get_function_memory(function_name)
        
        # Optimal memory (power of 2, min 128MB)
        optimal_memory = self.calculate_optimal_memory(
            avg_memory_used, max_memory_used
        )
        
        # Estimate savings
        current_cost = self.estimate_monthly_cost(current_memory, metrics)
        optimal_cost = self.estimate_monthly_cost(optimal_memory, metrics)
        
        return OptimizationResult(
            current_memory=current_memory,
            recommended_memory=optimal_memory,
            current_monthly_cost=current_cost,
            projected_monthly_cost=optimal_cost,
            savings_pct=(current_cost - optimal_cost) / current_cost * 100
        )
```

### Provisioned Concurrency Optimization
```python
class ProvisionedConcurrencyOptimizer:
    def __init__(self, lambda_client: LambdaClient):
        self.client = lambda_client
    
    async def optimize(self, function_name: str) -> ConcurrencyConfig:
        # Get utilization metrics
        metrics = await self.get_concurrency_metrics(function_name, days=7)
        
        # Calculate optimal provisioned concurrency
        p90_concurrent = np.percentile(metrics.concurrent_executions, 90)
        p99_concurrent = np.percentile(metrics.concurrent_executions, 99)
        
        # Target: cover P99 with 20% buffer
        recommended = int(p99_concurrent * 1.2)
        
        # Cost analysis
        provisioned_cost = self.calculate_provisioned_cost(recommended)
        on_demand_cost = self.estimate_on_demand_cost(metrics)
        
        return ConcurrencyConfig(
            recommended_provisioned=recommended,
            current_provisioned=self.get_current_provisioned(function_name),
            estimated_savings=on_demand_cost - provisioned_cost
        )
```

---

## Storage Cost Optimization

### Tiered Storage
```python
class StorageTierManager:
    def __init__(self, s3_client: S3Client):
        self.s3 = s3_client
    
    async def apply_lifecycle_policy(self, bucket: str, policy: LifecyclePolicy):
        """Apply intelligent tiering"""
        rules = []
        
        # Hot -> Standard (0-30 days)
        rules.append(LifecycleRule(
            id="hot-to-standard",
            status="Enabled",
            filter={"Prefix": ""},
            transitions=[
                Transition(StorageClass="STANDARD_IA", Days=30),
                Transition(StorageClass="GLACIER_IR", Days=90),
                Transition(StorageClass="GLACIER", Days=180),
                Transition(StorageClass="DEEP_ARCHIVE", Days=365)
            ]
        ))
        
        # Delete incomplete multipart uploads
        rules.append(LifecycleRule(
            id="cleanup-multipart",
            status="Enabled",
            abort_incomplete_multipart_upload=AbortIncompleteMultipartUpload(
                DaysAfterInitiation=7
            )
        ))
        
        # Expire old versions
        rules.append(LifecycleRule(
            id="expire-old-versions",
            status="Enabled",
            noncurrent_version_transitions=[
                NoncurrentVersionTransition(StorageClass="STANDARD_IA", NoncurrentDays=30),
                NoncurrentVersionTransition(StorageClass="GLACIER", NoncurrentDays=90)
            ],
            noncurrent_version_expiration=NoncurrentVersionExpiration(NoncurrentDays=365)
        ))
        
        await self.s3.put_bucket_lifecycle_configuration(
            Bucket=bucket,
            LifecycleConfiguration={"Rules": rules}
        )
```

### Intelligent Tiering (S3)
```python
# Enable S3 Intelligent-Tiering
await s3.put_bucket_intelligent_tiering_configuration(
    Bucket=bucket,
    Id="auto-tiering",
    IntelligentTieringConfiguration={
        "Id": "auto-tiering",
        "Status": "Enabled",
        "Tierings": [
            {"Days": 1, "AccessTier": "ARCHIVE_INSTANT_ACCESS"},
            {"Days": 30, "AccessTier": "ARCHIVE"},
            {"Days": 90, "AccessTier": "DEEP_ARCHIVE"}
        ]
    )
)
```

### Database Storage Optimization
```python
class DatabaseStorageOptimizer:
    def __init__(self, pg_pool: AsyncConnectionPool):
        self.pool = pg_pool
    
    async def analyze_table_bloat(self) -> list[BloatReport]:
        query = """
        SELECT 
            schemaname, tablename,
            pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as total_size,
            pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) as table_size,
            pg_size_pretty(pg_total_relation_size - pg_relation_size) as bloat_size,
            round(100 * (pg_total_relation_size - pg_relation_size) / pg_total_relation_size) as bloat_pct
        FROM pg_stat_user_tables
        WHERE pg_total_relation_size > 100 * 1024 * 1024  -- > 100MB
        ORDER BY bloat_pct DESC;
        """
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query)
        
        return [BloatReport(**row) for row in rows]
    
    async def reclaim_space(self, table: str) -> ReclaimResult:
        # pg_repack for online vacuum
        result = await self.execute_pg_repack(table)
        return ReclaimResult(table=table, reclaimed_bytes=result.reclaimed)
```

---

## Network Cost Optimization

### CDN & Caching
```python
class CDNOptimizer:
    def __init__(self, cdn_client: CDNClient):
        self.cdn = cdn_client
    
    async def optimize_cache_policy(self, distribution_id: str) -> CachePolicy:
        # Analyze cache hit ratios
        stats = await self.get_cache_stats(distribution_id)
        
        # Optimal TTL by content type
        policies = {
            "static_assets": CachePolicy(ttl=31536000, vary_headers=["Accept-Encoding"]),
            "api_responses": CachePolicy(ttl=300, vary_headers=["Authorization", "Accept"]),
            "user_content": CachePolicy(ttl=0, no_cache=True),
            "api_static": CachePolicy(ttl=3600, vary_headers=["Accept"])
        }
        
        return CachePolicyConfig(
            behaviors=self.map_paths_to_policies(stats),
            default_ttl=3600
        )
```

### Data Transfer Optimization
```python
class DataTransferOptimizer:
    def __init__(self, cloud_provider: CloudProvider):
        self.provider = cloud_provider
    
    async def optimize_cross_region_transfer(self) -> TransferOptimization:
        # Use VPC endpoints for S3/DynamoDB
        # Enable VPC peering instead of public internet
        # Use S3 Transfer Acceleration for large uploads
        # Enable compression (gzip, brotli)
        # Use multipart upload for large files
        
        recommendations = [
            "Enable VPC endpoints for S3/DynamoDB",
            "Use VPC peering for cross-region",
            "Enable S3 Transfer Acceleration for >100MB",
            "Enable compression (gzip/brotli) for all responses",
            "Use multipart upload for files >100MB",
            "Implement request coalescing for duplicate requests"
        ]
        
        return TransferOptimization(recommendations=recommendations)
```

---

## Monitoring & Alerting for Costs

### Cost Anomaly Detection
```python
class CostAnomalyDetector:
    def __init__(self, cost_explorer: CostExplorerClient):
        self.ce = cost_explorer
    
    async def detect_anomalies(self, account_id: str, days: int = 30) -> list[CostAnomaly]:
        # Get historical costs
        costs = await self.get_daily_costs(account_id, days)
        
        # Statistical anomaly detection
        daily_costs = [c.amount for c in costs]
        mean = statistics.mean(daily_costs)
        stdev = statistics.stdev(daily_costs) if len(daily_costs) > 1 else 0
        
        anomalies = []
        for cost in costs[-7:]:  # Last 7 days
            z_score = abs(cost.amount - mean) / stdev if stdev > 0 else 0
            if z_score > 3:  # 3 sigma
                anomalies.append(CostAnomaly(
                    date=cost.date,
                    amount=cost.amount,
                    expected=mean,
                    z_score=z_score,
                    severity="high" if z_score > 4 else "medium"
                ))
        
        return anomalies
```

### Budget Alerts
```python
class BudgetAlertManager:
    def __init__(self, budgets_client: BudgetsClient):
        self.client = budgets_client
    
    async def create_budget_alert(self, config: BudgetConfig):
        await self.client.create_budget(Budget(
            budget_name=config.name,
            budget_limit=config.limit,
            time_unit=config.time_unit,
            cost_filters=config.filters,
            notifications=[
                BudgetNotification(
                    notification_type="ACTUAL",
                    comparison_operator="GREATER_THAN",
                    threshold=80,
                    notification_type="EMAIL",
                    subscribers=[config.email]
                ),
                BudgetNotification(
                    notification_type="FORECASTED",
                    comparison_operator="GREATER_THAN",
                    threshold=100,
                    notification_type="EMAIL",
                    subscribers=[config.email]
                )
            ]
        ))
```

---

## Cost Allocation & Chargeback

### Tag-Based Cost Allocation
```python
class CostAllocator:
    def __init__(self, cost_explorer: CostExplorerClient):
        self.ce = cost_explorer
    
    async def allocate_costs(self, period: TimeRange, 
                             allocation_tags: list[str]) -> CostAllocation:
        # Get costs with tags
        costs = await self.ce.get_cost_and_usage(
            TimePeriod={"Start": period.start, "End": period.end},
            Granularity="DAILY",
            Metrics=["UnblendedCost"],
            GroupBy=[{"Type": "TAG", "Key": tag} for tag in allocation_tags]
        )
        
        # Aggregate by tag
        allocations = defaultdict(float)
        for result in costs:
            for group in result["Groups"]:
                key = group["Keys"][0] if group["Keys"] else "untagged"
                cost = float(group["Metrics"]["UnblendedCost"]["Amount"])
                allocations[key] += cost
        
        return CostAllocation(
            period=period,
            allocations=dict(allocations),
            total=sum(allocations.values())
        )
```

### Showback/Chargeback Reports
```python
class ChargebackReportGenerator:
    def __init__(self, cost_allocator: CostAllocator):
        self.allocator = cost_allocator
    
    async def generate_monthly_report(self, month: str) -> ChargebackReport:
        allocation = await self.allocator.allocate_costs(
            period=TimeRange.from_month(month),
            allocation_tags=["team", "project", "environment"]
        )
        
        # Generate per-team breakdown
        team_costs = defaultdict(float)
        for tag, cost in allocation.allocations.items():
            if tag.startswith("team:"):
                team_costs[tag[5:]] += cost
        
        return ChargebackReport(
            month=month,
            total_cost=allocation.total,
            by_team=dict(team_costs),
            by_project=self.aggregate_by_tag(allocation, "project"),
            by_environment=self.aggregate_by_tag(allocation, "environment"),
            trends=self.calculate_trends(month),
            recommendations=self.generate_recommendations(allocation)
        )
```

---

## Kubernetes Cost Optimization

### Resource Quotas & Limits
```yaml
# Namespace resource quota
apiVersion: v1
kind: ResourceQuota
metadata:
  name: agent-namespace-quota
  namespace: agents
spec:
  hard:
    requests.cpu: "100"
    requests.memory: "200Gi"
    limits.cpu: "200"
    limits.memory: "400Gi"
    requests.nvidia.com/gpu: "4"
    persistentvolumeclaims: "20"
    services.loadbalancers: "5"
---
# LimitRange for default container limits
apiVersion: v1
kind: LimitRange
metadata:
  name: agent-limits
  namespace: agents
spec:
  limits:
  - default:
      cpu: "2000m"
      memory: "2Gi"
    defaultRequest:
      cpu: "250m"
      memory: "512Mi"
    max:
      cpu: "4000m"
      memory: "8Gi"
    type: Container
```

### KubeCost Integration
```yaml
# kubecost-values.yaml
kubecost:
  prometheus:
    server:
      global:
        scrape_interval: 30s
      retention: 15d
  costs:
    defaultClusterId: production-cluster
  networkCosts:
    enabled: true
  kubecostToken: ${KUBECOST_TOKEN}
```

---

## FinOps Practices

### FinOps Framework
```python
class FinOpsManager:
    def __init__(self):
        self.principles = [
            "Team owns cost of their services",
            "Cost is a first-class metric",
            "Decisions driven by business value",
            "Forecasting and planning are essential",
            "Continuous optimization"
        ]
    
    async def generate_finops_report(self) -> FinOpsReport:
        return FinOpsReport(
            maturity=self.assess_maturity(),
            cost_allocation=self.get_cost_allocation(),
            optimization_opportunities=self.get_optimizations(),
            forecasting=self.generate_forecast(),
            governance=self.assess_governance()
        )
    
    def assess_maturity(self) -> MaturityLevel:
        # Crawl, Walk, Run
        checks = {
            "cost_visibility": self.has_cost_visibility(),
            "allocation": self.has_cost_allocation(),
            "optimization": self.has_automated_optimization(),
            "forecasting": self.has_forecasting(),
            "governance": self.has_governance(),
            "culture": self.has_finops_culture()
        }
        score = sum(checks.values()) / len(checks)
        if score >= 0.8: return MaturityLevel.RUN
        if score >= 0.5: return MaturityLevel.WALK
        return MaturityLevel.CRAWL
```

---

## Related Notes

- [[Production/Deployment-Architecture.md]] — Deployment patterns
- [[Production/Scaling-Strategies.md]] — Scaling strategies
- [[Production/Observability.md]] — Cost monitoring
- [[Production/Reliability.md]] — Reliability vs cost
- [[Production/Deployment-Architecture.md]] — Deployment
- [[DevOps/CI-CD-Best-Practices.md]] — CI/CD costs
- [[DevOps/Infrastructure-as-Code.md]] — IaC costs
- [[DevOps/Observability.md]] — Cost observability
- [[DevOps/CI-CD-Best-Practices.md]] — CI/CD costs
- [[Security/Supply-Chain-Security.md]] — Supply chain costs
- [[Frontiers/Research-Directions.md]] — Cost optimization research

---

*Sources: AWS Well-Architected Cost Optimization, Google Cloud Cost Optimization, Azure Cost Management, FinOps Foundation, AWS Compute Optimizer, AWS Cost Explorer, AWS Budgets, Kubecost, CloudHealth, CloudCheckr, Cloudability, Cloudability, FinOps Foundation, AWS Well-Architected Framework, Google Cloud FinOps, Azure Cost Management, CNCF FinOps, CNCF FinOps Working Group.*