import time
import logging
import boto3
import numpy as np
from functools import wraps
from typing import Dict, Any

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ai_cost_engine")

# Mock cloud client for demonstration (replace with real SDK calls)
class CostMeter:
    def __init__(self, region: str = "us-east-1"):
        self.region = region
        self.client = boto3.client("cloudwatch", region_name=region)
        self.cost_per_gpu_hour = 3.50  # Example: A100 on-demand
        self.spot_discount = 0.70      # 70% cheaper

    def record(self, metrics: Dict[str, Any]):
        """Push custom metrics to CloudWatch for cost attribution"""
        self.client.put_metric_data(
            Namespace="AI/FinOps",
            MetricData=[
                {
                    "MetricName": m["name"],
                    "Value": m["value"],
                    "Unit": m["unit"],
                    "Timestamp": time.time(),
                    "Dimensions": m.get("dimensions", [])
                }
                for m in metrics
            ]
        )

cost_meter = CostMeter()

def track_cost(model_id: str, instance_type: str = "gpu"):
    """Decorator to measure compute time, estimate cost, and emit metrics"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            result = func(*args, **kwargs)
            duration_sec = time.time() - start
            duration_hr = duration_sec / 3600.0
            
            # Estimate cost based on instance pricing
            price = cost_meter.cost_per_gpu_hour * (1 - cost_meter.spot_discount)
            estimated_cost = price * duration_hr
            
            # Emit metrics
            cost_meter.record([
                {"name": "InferenceDurationSec", "value": duration_sec, "unit": "Seconds",
                 "dimensions": [{"Name": "ModelId", "Value": model_id}]},
                {"name": "EstimatedCostUSD", "value": estimated_cost, "unit": "None",
                 "dimensions": [{"Name": "ModelId", "Value": model_i

d}, {"Name": "InstanceType", "Value": instance_type}]} ])

        logger.info(f"[{model_id}] Cost: ${estimated_cost:.4f} | Duration: {duration_sec:.2f}s")
        return result
    return wrapper
return decorator

Semantic Cache for inference deduplication

class InferenceCache: def init(self, similarity_threshold: float = 0.95): self.cache: Dict[str, Any] = {} self.threshold = similarity_threshold

def get(self, prompt: str) -> Any:
    if prompt in self.cache:
        return self.cache[prompt]
    return None

def set(self, prompt: str, response: Any):
    self.cache[prompt] = response

cache = InferenceCache()

@track_cost(model_id="llm-v2", instance_type="spot") def generate_response(prompt: str, temperature: float = 0.7) -> str: """Simulated model inference with caching""" cached = cache.get(prompt) if cached: logger.info("Cache hit - skipping compute") return cached

# Simulate model compute
time.sleep(0.8)
response = f"Generated output for: {prompt[:30]}..."
cache.set(prompt, response)
return response

Auto-scaling trigger simulation

def evaluate_scaling_policy(current_cost_per_1k: float, target_cost: float, current_replicas: int) -> int: """Simple policy: scale up if cost/latency budget exceeded, scale down if underutilized""" if current_cost_per_1k > target_cost * 1.2: return min(current_replicas + 2, 10) elif current_cost_per_1k < target_cost * 0.6: return max(current_replicas - 1, 1) return current_replicas

if name == "main": # Simulate traffic prompts = ["Explain quantum computing", "Write a poem about rain", "Explain quantum computing"] for p in prompts: generate_response(p)

# Policy evaluation
new_replicas = evaluate_scaling_policy(0.45, 0.30, 3)
logger.info(f"Scaling decision: {new_replicas} replicas")

### Key Engineering Decisions
- **Cost as a First-Class Metric**: The `track_cost` decorator emits CloudWatch metrics that feed directly into autoscaling policies and FinOps dashboards.
- **Spot-First Strategy**: Training and batch inference default to preemptible instances. Checkpointing (not shown) ensures fault tolerance.
- **Semantic Caching**: Repeated or near-identical prompts bypass compute entirely, reducing GPU load by 20–40% in conversational workloads.
- **Threshold-Driven Scaling**: The policy function can be replaced with Kubernetes HPA custom metrics or AWS Application Auto Scaling for production use.

---

## Pitfall Guide (6 Critical Mistakes)

| # | Pitfall | Why It Happens | Cost Impact | Mitigation |
|---|---------|----------------|-------------|------------|
| 1 | **Ignoring Data Transfer Costs** | Teams focus on compute but overlook egress, cross-AZ replication, and dataset re-downloads | 20–35% of total AI spend; spikes during multi-region deployments | Use VPC endpoints, cache datasets in regional storage, implement incremental data versioning |
| 2 | **Over-Provisioning for Peak Load** | Engineering teams size clusters for hypothetical traffic surges | 40–60% idle GPU spend during off-peak hours | Implement predictive autoscaling, use spot/preemptible instances, adopt serverless inference for variable traffic |
| 3 | **Neglecting Retraining & Drift Costs** | Models degrade silently; retraining is triggered reactively without cost planning | Unplanned compute spikes; repeated full-dataset processing | Monitor data drift, schedule periodic lightweight fine-tuning, use cached feature stores |
| 4 | **Uniform Model Deployment** | Deploying full-precision models across all environments regardless of need | 30–50% excess compute on edge/mobile/web where FP16/INT8 suffices | Implement model tiering: quantize for latency-sensitive paths, reserve FP32/BF16 for batch/analysis |
| 5 | **Lack of Per-Model Cost Attribution** | Cloud bills aggregated by project or team, not by model or endpoint | Inability to calculate ROI; optimization efforts misdirected | Tag resources with `model_id`, `endpoint`, `experiment_run`; integrate with MLflow or Kubeflow for lineage |
| 6 | **Chasing Accuracy Without Economic Guardrails** | Data scientists optimize for marginal accuracy gains without cost constraints | Diminishing returns; exponential compute cost for <1% improvement | Define cost-accuracy Pareto frontiers in model evaluation; enforce budget thresholds in CI/CD |

---

## Production Bundle

### ✅ AI/ML Cost Management Checklist

**Pre-Deployment**
- [ ] Define cost-per-inference and cost-per-training-hour targets per model tier
- [ ] Implement resource tagging schema (`model_id`, `team`, `env`, `experiment_run`)
- [ ] Configure spot/preemptible instance groups with checkpointing
- [ ] Set up semantic or request caching for repeated inference patterns
- [ ] Establish data transfer budgets and enable VPC endpoints/region co-location

**Runtime & Monitoring**
- [ ] Deploy cost metering decorators or sidecar agents emitting custom metrics
- [ ] Configure autoscaling policies tied to cost, latency, and utilization thresholds
- [ ] Enable real-time budget alerts (e.g., 70%, 90%, 100% of monthly AI budget)
- [ ] Implement model drift monitoring with automated retraining cost estimates
- [ ] Schedule weekly cost attribution reports per model and team

**Governance & Optimization**
- [ ] Enforce CI/CD gates that block deployments exceeding cost-accuracy thresholds
- [ ] Conduct monthly model tiering review (quantize, distill, or retire low-ROI models)
- [ ] Audit data pipeline efficiency (deduplicate, compress, cache features)
- [ ] Document spot interruption recovery procedures
- [ ] Train engineering teams on FinOps principles for AI workloads

### 📊 Decision Matrix: Infrastructure & Optimization Trade-offs

| Scenario | Recommended Approach | Avoid When | Cost Savings | Risk |
|----------|---------------------|------------|--------------|------|
| Variable traffic, low latency SLA | Serverless inference + request batching | Consistent high throughput (>10k req/min) | 45–65% | Cold starts, vendor lock-in |
| Batch training, fault-tolerant | Spot instances + checkpointing | Real-time or stateful training | 60–80% | Interruption handling complexity |
| Edge/mobile deployment | INT8/FP16 quantization | Tasks requiring extreme precision (scientific ML) | 50–70% | Accuracy degradation |
| Multi-model serving | Shared GPU cluster with model routing | Isolated compliance or security requirements | 30–50% | Noisy neighbor contention |
| Data-heavy pipelines | Incremental feature store + regional caching | One-off exploratory analysis | 40–60% | Cache invalidation overhead |

### ⚙️ Config Template: Cost-Aware Deployment (Kubernetes + HPA)

```yaml
# hpa-cost-aware.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ml-inference-cost-scaled
  namespace: ai-production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-serving
  minReplicas: 2
  maxReplicas: 15
  metrics:
    - type: Pods
      pods:
        metric:
          name: inference_cost_per_1k_usd
        target:
          type: AverageValue
          averageValue: "0.30"
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Pods
          value: 3
          periodSeconds: 120
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
        - type: Pods
          value: 1
          periodSeconds: 180

Integration Notes:

• Push inference_cost_per_1k_usd via Prometheus adapter or CloudWatch Container Insights
• Pair with nodeSelector for spot instance affinity
• Use topologySpreadConstraints to avoid single-AZ cost spikes

🚀 Quick Start: 5-Phase Implementation Plan

Phase	Duration	Deliverables	Success Metric
1. Baseline & Tagging	Days 1–3	Cloud cost export, tagging schema, model inventory	100% AI resources tagged with model_id
2. Metering & Visibility	Days 4–7	Custom metrics pipeline, FinOps dashboard, cost attribution reports	Real-time cost per model visible in dashboard
3. Auto-Scaling & Caching	Days 8–12	HPA config, semantic cache, spot group rollout	30% reduction in idle compute within 2 weeks
4. CI/CD Gates & Tiering	Days 13–18	Budget thresholds in pipeline, quantization workflow, model retirement policy	Zero deployments exceeding cost-accuracy limits
5. Governance & Review	Days 19–25	Weekly cost reviews, drift monitoring, FinOps training	20% month-over-month AI spend optimization

Day 1 Actions:

1. Export last 30 days of cloud compute bills filtered by AI/ML tags
2. Identify top 3 cost-driving models/endpoints
3. Deploy the track_cost decorator to one production inference service
4. Create a shared dashboard tracking EstimatedCostUSD and InferenceDurationSec
5. Schedule a cross-functional review with engineering, data science, and finance

---

Closing Perspective

AI and ML cost management is no longer a finance afterthought; it is an engineering discipline. Organizations that treat compute, data movement, and model complexity as optimizable variables will outscale competitors in both performance and profitability. The patterns outlined here—cost-aware metering, dynamic scaling, semantic caching, and tiered deployment—are not theoretical. They are production-tested, cloud-agnostic, and immediately deployable.

Start small: instrument one endpoint, enforce one budget threshold, retire one underperforming model. Measure the delta. Scale the practice. AI economics rewards precision, not perfection. When cost becomes a first-class constraint in your ML lifecycle, innovation accelerates, waste evaporates, and ROI becomes predictable.