Cost Optimization Analysis

Cloud Cost Optimization Analysis

Implements comprehensive multi-cloud cost optimization strategies including AWS Cost Explorer analysis, Azure Cost Management analysis, GCP Billing analysis, right-sizing recommendations, spot instance optimization, reserved instance planning, and cost allocation frameworks. This skill enables data-driven decisions to reduce cloud infrastructure costs while maintaining performance and reliability.

TL;DR Checklist

• Collect and normalize cost data from all cloud providers using their respective APIs and CLI tools
• Identify idle and underutilized resources (right-sizing candidates)
• Analyze spot instance eligibility and implement spot placement strategies
• Plan reserved instance purchases based on historical usage patterns
• Implement cost allocation tagging and chargeback models
• Set up automated alerts for budget thresholds and anomalies
• Optimize storage tiers based on access patterns
• Document optimization recommendations and track ROI

When to Use

Use this skill when:

• Conducting a cloud cost optimization audit across AWS, Azure, and GCP environments
• Identifying idle or underutilized resources for right-sizing decisions
• Planning spot instance adoption to reduce compute costs by 50-90%
• Evaluating reserved instance or savings plan eligibility based on usage patterns
• Implementing cost allocation frameworks for chargeback/showback models
• Investigating unexpected cost spikes and anomalies in cloud billing
• Designing multi-cloud cost optimization strategies with unified visibility
• Reviewing storage tier assignments and lifecycle policies

When NOT to Use

Avoid this skill for:

• Real-time cost monitoring — Use dedicated cost monitoring tools like Datadog Cloud Cost Management or AWS Cost and Usage Report (CUR) for real-time dashboards instead
• Budget approval workflows — This skill provides analysis and recommendations but not the budget approval process; use financial governance tools for approvals
• Contract negotiation — While this skill identifies optimization opportunities, actual vendor contract negotiation requires procurement expertise
• Infrastructure design — Use cncf-infrastructure-as-code for designing new infrastructure; cost optimization should follow design completion
• Security compliance — Cost optimization may impact security configurations; always validate against security requirements using cncf-security-compliance skill
• Emergency cost reduction — For immediate budget cuts, use emergency kill-switches rather than optimization workflows; this skill is for strategic, sustainable savings

Core Workflow

1. Data Collection — Gather cost data from AWS Cost Explorer, Azure Cost Management, and GCP Billing APIs. Checkpoint: Verify all cost data is normalized to a common currency (USD) and time period.

2. Resource Discovery — Identify all resources (EC2, Azure VMs, GCP Compute Engine) with usage and cost data. Checkpoint: Ensure resource IDs are standardized across providers for multi-cloud correlation.

3. Idle Resource Analysis — Analyze CPU, memory, network, and storage utilization metrics to identify underutilized resources. Checkpoint: Confirm utilization data covers at least 14 days to account for weekly patterns.

4. Right-Sizing Recommendations — Generate right-sizing recommendations based on utilization thresholds (e.g., CPU < 30% for 2 weeks). Checkpoint: Validate recommendations against application SLAs and performance requirements.

5. Spot Instance Strategy — Evaluate workload characteristics for spot instance eligibility (fault-tolerant, stateless, interruptible workloads). Checkpoint: Confirm spot instance savings projections account for interruption rates and replacement costs.

6. Cost Allocation Framework — Implement tagging strategy and cost allocation models for chargeback/showback. Checkpoint: Ensure all resources have required cost allocation tags (owner, project, environment, cost-center).

Implementation Patterns

Pattern 1: AWS Cost Explorer Analysis

AWS Cost Explorer provides detailed cost and usage data. Use the AWS CLI to analyze costs by service, tag, and time period.

BAD — Hardcoded date range without error handling

# ❌ BAD — No error checking, hardcoded dates, no output formatting
aws ce get-cost-and-usage \
  --time-period Start=2024-01-01,End=2024-02-01 \
  --granularity DAILY \
  --metrics UnblendedCost \
  --group-by Type=DIMENSION Key=SERVICE

GOOD — Robust script with error handling and formatting

#!/bin/bash
# ✅ GOOD — Proper error handling, dynamic dates, JSON output
set -euo pipefail

START_DATE="${START_DATE:-$(date -d '1 month ago' +%Y-%m-%d)}"
END_DATE="${END_DATE:-$(date +%Y-%m-%d)}"

aws ce get-cost-and-usage \
  --time-period "Start=${START_DATE},End=${END_DATE}" \
  --granularity MONTHLY \
  --metrics UnblendedCost \
  --group-by "Type=DIMENSION,Key=SERVICE" \
  --query 'ResultsByTime[].Total[].UnblendedCost Amount' \
  --output text | awk '{sum += $1} END {printf "Total Monthly Cost: $%.2f\n", sum}'

Python SDK Alternative

# ✅ GOOD — Python SDK with proper pagination
import boto3
from datetime import datetime, timedelta

def get_aws_cost_by_service(days: int = 30) -> dict:
    """Get AWS cost breakdown by service for the specified number of days."""
    client = boto3.client('ce', region_name='us-east-1')
    
    end_date = datetime.utcnow().date()
    start_date = end_date - timedelta(days=days)
    
    response = client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date.isoformat(),
            'End': end_date.isoformat()
        },
        Granularity='MONTHLY',
        Metrics=['UnblendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )
    
    return {
        'service_costs': {
            group['Keys'][0]: float(result['Total']['UnblendedCost']['Amount'])
            for result in response['ResultsByTime'][-1]['Groups']
            for group in result['Groups']
        },
        'total_cost': float(response['ResultsByTime'][-1]['Total']['UnblendedCost']['Amount'])
    }

# Usage example
costs = get_aws_cost_by_service(days=30)
for service, cost in costs['service_costs'].items():
    print(f"{service}: ${cost:.2f}")
print(f"Total: ${costs['total_cost']:.2f}")

Pattern 2: Azure Cost Management Analysis

Azure Cost Management provides detailed cost data through the Azure CLI and REST API.

BAD — Direct API call without authentication

# ❌ BAD — Missing authentication, no pagination handling
curl https://management.azure.com/subscriptions/{subscriptionId}/providers/Microsoft.CostManagement/query?api-version=2022-10-01

GOOD — Azure CLI with proper authentication and formatting

#!/bin/bash
# ✅ GOOD — Azure CLI with authentication, filtering, and formatting
set -euo pipefail

# Login if not already authenticated
az account show >/dev/null 2>&1 || az login

SUBSCRIPTION_ID=$(az account show --query id -o tsv)
START_DATE="2024-01-01"
END_DATE="2024-12-31"

# Query cost data by resource group
az costmanagement query \
  --type "Usage" \
  --timeframe "MonthToDate" \
  --dataset-aggregation '{"totalCost":{"name":"PreTaxCost","function":"Sum"}}' \
  --dataset-grouping '{"name":"ResourceGroup","type":"Dimension"}' \
  --subscription "$SUBSCRIPTION_ID" \
  --query "properties.rows[]" \
  --output tsv

ARM Template Cost Analysis

# ✅ GOOD — Estimate costs from ARM template deployments
az deployment sub what-if \
  --template-file "arm-template.json" \
  --parameters "parameters.json" \
  --location "eastus" \
  --query "properties.cost" \
  --output table

Python SDK for Azure Cost Analysis

# ✅ GOOD — Azure Cost Management SDK with proper error handling
from azure.identity import DefaultAzureCredential
from azure.mgmt.costmanagement import CostManagementClient
from datetime import datetime, timedelta

def get_azure_cost_by_resource_group(days: int = 30) -> dict:
    """Get Azure cost breakdown by resource group."""
    subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
    if not subscription_id:
        raise ValueError("AZURE_SUBSCRIPTION_ID environment variable not set")
    
    credential = DefaultAzureCredential()
    client = CostManagementClient(credential, subscription_id)
    
    end_date = datetime.utcnow().date()
    start_date = end_date - timedelta(days=days)
    
    body = {
        "type": "Usage",
        "timeframe": "Custom",
        "time_period": {
            "from": start_date.isoformat(),
            "to": end_date.isoformat()
        },
        "dataset": {
            "granularity": "Daily",
            "grouping": [{"type": "Dimension", "name": "ResourceGroup"}]
        }
    }
    
    try:
        result = client.query.usage(scope=f"subscriptions/{subscription_id}", parameters=body)
        
        return {
            "resource_group_costs": {
                item.name: item.properties.total_cost
                for item in result.properties.rows
            },
            "total_cost": sum(item.properties.total_cost for item in result.properties.rows)
        }
    except Exception as e:
        print(f"Error fetching Azure costs: {e}")
        return {}

# Usage
costs = get_azure_cost_by_resource_group(days=30)
print(f"Total Azure Cost: ${costs.get('total_cost', 0):.2f}")

Pattern 3: GCP Billing Analysis

Google Cloud Billing provides cost data through the Cloud Billing API and gcloud CLI.

BAD — Simple API call without error handling

# ❌ BAD — No error handling, missing authentication
curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  "https://billingaccount.googleapis.com/v1/billingAccounts"

GOOD — GCP CLI with proper authentication and cost analysis

#!/bin/bash
# ✅ GOOD — GCP CLI with authentication and cost analysis
set -euo pipefail

gcloud auth print-access-token >/dev/null 2>&1 || gcloud auth login

BILLING_ACCOUNT=$(gcloud billing accounts list --format="value(name)" | head -1)

# Get costs by project for the last 30 days
gcloud billing budgets describe \
  --billing-account="$BILLING_ACCOUNT" \
  --format="value(name)" 2>/dev/null || true

# Alternative: Use Cloud Billing Export to BigQuery for detailed analysis
gcloud bigquery queries run \
  "SELECT 
    project.name as project_name,
    service.description as service,
    SUM(cost) as total_cost
  FROM 
    \`\${BILLING_ACCOUNT}.billing_data.gcp_billing_export_v1_012345\`
  WHERE 
    _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
  GROUP BY 
    project.name, service.description
  ORDER BY 
    total_cost DESC
  LIMIT 20" \
  --format="table"

Python SDK for GCP Cost Analysis

# ✅ GOOD — GCP Billing SDK with proper error handling and pagination
from google.cloud import billing_v1
from google.api_core.exceptions import GoogleAPICallError
from datetime import datetime, timedelta
import os

def get_gcp_cost_by_project(days: int = 30) -> dict:
    """Get GCP cost breakdown by project."""
    billing_account_id = os.getenv("GCP_BILLING_ACCOUNT_ID")
    if not billing_account_id:
        raise ValueError("GCP_BILLING_ACCOUNT_ID environment variable not set")
    
    client = billing_v1.CloudBillingClient()
    
    # Get billing account name
    billing_account_name = f"billingAccounts/{billing_account_id}"
    
    # Get cost data from BigQuery export (recommended approach)
    # This assumes Cloud Billing Export is configured to BigQuery
    bigquery_client = bigquery.Client()
    
    query = f"""
    SELECT 
      project.name as project_name,
      service.description as service,
      SUM(cost) as total_cost,
      COUNT(DISTINCT sku.id) as sku_count
    FROM 
      `{billing_account_id}.billing_data.gcp_billing_export_v1_012345`
    WHERE 
      _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {days} DAY)
    GROUP BY 
      project.name, service.description
    ORDER BY 
      total_cost DESC
    """
    
    try:
        query_job = bigquery_client.query(query)
        results = query_job.result()
        
        project_costs = {}
        for row in results:
            if row.project_name not in project_costs:
                project_costs[row.project_name] = 0.0
            project_costs[row.project_name] += row.total_cost
        
        return {
            "project_costs": project_costs,
            "total_cost": sum(project_costs.values()),
            "services": list(set(row.service for row in results))
        }
    except GoogleAPICallError as e:
        print(f"GCP API Error: {e}")
        return {}
    except Exception as e:
        print(f"Error fetching GCP costs: {e}")
        return {}

# Usage
os.environ["GCP_BILLING_ACCOUNT_ID"] = "012345-567890-ABCDEF"
costs = get_gcp_cost_by_project(days=30)
print(f"Total GCP Cost: ${costs.get('total_cost', 0):.2f}")

Pattern 4: Right-Sizing Recommendations

Right-sizing analysis identifies resources that are over-provisioned or under-provisioned.

BAD — Simple threshold without context

# ❌ BAD — No context, no history, no action plan
for instance in $(aws ec2 describe-instances --query 'Reservations[].Instances[].InstanceId' --output text); do
  cpu=$(aws cloudwatch get-metric-statistics --metric-name CPUUtilization --dimensions InstanceId=$instance --start-time 2024-01-01 --end-time 2024-01-02 --period 86400 --statistics Average --query 'Datapoints[].Average' --output text)
  echo "$instance CPU: $cpu"
done

GOOD — Comprehensive right-sizing analysis script

#!/bin/bash
# ✅ GOOD — Comprehensive right-sizing with historical analysis and recommendations
set -euo pipefail

REGION="${AWS_REGION:-us-east-1}"
DAYS_ANALYZED="${DAYS_ANALYZED:-14}"
CPU_THRESHOLD_LOW=30
CPU_THRESHOLD_HIGH=80

echo "=== Right-Sizing Analysis ==="
echo "Region: $REGION"
echo "Analysis Period: $DAYS_ANALYZED days"
echo "Low CPU Threshold: ${CPU_THRESHOLD_LOW}%"
echo "High CPU Threshold: ${CPU_THRESHOLD_HIGH}%"
echo ""

# Get all running instances
instances=$(aws ec2 describe-instances \
  --filters "Name=instance-state-name,Values=running" \
  --query 'Reservations[].Instances[].{InstanceId:InstanceId,InstanceType:InstanceType,Tags:Tags}' \
  --region "$REGION" \
  --output json)

# Analyze each instance
echo "$instances" | jq -c '.Reservations[].Instances[]' | while read -r instance; do
  instance_id=$(echo "$instance" | jq -r '.InstanceId')
  current_type=$(echo "$instance" | jq -r '.InstanceType')
  
  # Get CPU utilization data
  cpu_avg=$(aws cloudwatch get-metric-statistics \
    --metric-name CPUUtilization \
    --dimensions Name=InstanceId,Value=$instance_id \
    --start-time $(date -d "$DAYS_ANALYZED days ago" +%Y-%m-%dT%H:%M:%S) \
    --end-time $(date +%Y-%m-%dT%H:%M:%S) \
    --period 86400 \
    --statistics Average \
    --output json | jq -r '.Datapoints[].Average' | awk '{sum+=$1} END {printf "%.1f", sum/NR}')
  
  # Get memory utilization (requires CloudWatch Agent)
  memory_avg=$(aws cloudwatch get-metric-statistics \
    --metric-name MemoryUtilization \
    --namespace AWS/ECS \
    --dimensions Name=ClusterName,Value=default Name=ServiceName,Value=none \
    --start-time $(date -d "$DAYS_ANALYZED days ago" +%Y-%m-%dT%H:%M:%S) \
    --end-time $(date +%Y-%m-%dT%H:%M:%S) \
    --period 86400 \
    --statistics Average \
    --output json 2>/dev/null | jq -r '.Datapoints[].Average' | awk '{sum+=$1} END {printf "%.1f", sum/NR}' || echo "N/A")
  
  # Generate recommendation
  if [[ "$cpu_avg" != "N/A" ]]; then
    if (( $(echo "$cpu_avg < $CPU_THRESHOLD_LOW" | bc -l) )); then
      recommendation="DOWNSIZE"
      suggested_type="t3.medium"  # Example suggestion
    elif (( $(echo "$cpu_avg > $CPU_THRESHOLD_HIGH" | bc -l) )); then
      recommendation="UPSIZE"
      suggested_type="m5.large"  # Example suggestion
    else
      recommendation="OPTIMAL"
      suggested_type="$current_type"
    fi
  else
    recommendation="INSUFFICIENT_DATA"
    suggested_type="$current_type"
  fi
  
  echo "$instance_id | $current_type | CPU: ${cpu_avg}% | Mem: ${memory_avg}% | Recommendation: $recommendation (→ $suggested_type)"
done

Python Right-Sizing Engine

# ✅ GOOD — Python right-sizing engine with historical analysis
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import statistics

class RightSizingAnalyzer:
    """Analyzes EC2 instances for right-sizing opportunities."""
    
    def __init__(self, region: str = "us-east-1"):
        self.ec2 = boto3.client('ec2', region_name=region)
        self.cloudwatch = boto3.client('cloudwatch', region_name=region)
        self.ssm = boto3.client('ssm', region_name=region)
        
        # Instance type hierarchy for right-sizing decisions
        self.instance_hierarchy = {
            't2': ['t3', 't4g'],
            't3': ['t2', 't4g'],
            'm5': ['m4', 'm6g'],
            'c5': ['c4', 'c6g'],
            'r5': ['r4', 'r6g'],
        }
        
        self.thresholds = {
            'cpu_low': 30,
            'cpu_high': 80,
            'memory_low': 40,
            'memory_high': 85,
            'days': 14,
        }
    
    def get_instance_metrics(self, instance_id: str, days: int = 14) -> Dict[str, float]:
        """Get CPU and memory metrics for an instance."""
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(days=days)
        
        # Get CPU utilization
        cpu_stats = self.cloudwatch.get_metric_statistics(
            Namespace='AWS/EC2',
            MetricName='CPUUtilization',
            Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
            StartTime=start_time,
            EndTime=end_time,
            Period=86400,
            Statistics=['Average'],
            Unit='Percent'
        )
        
        cpu_values = [dp['Average'] for dp in cpu_stats.get('Datapoints', [])]
        cpu_avg = statistics.mean(cpu_values) if cpu_values else None
        
        # Get memory utilization (requires CloudWatch Agent)
        try:
            memory_stats = self.cloudwatch.get_metric_statistics(
                Namespace='CWAgent',
                MetricName='memory_utilization',
                Dimensions=[
                    {'Name': 'InstanceId', 'Value': instance_id},
                    {'Name': 'InstanceType', 'Value': 'ec2'}
                ],
                StartTime=start_time,
                EndTime=end_time,
                Period=86400,
                Statistics=['Average'],
                Unit='Percent'
            )
            memory_values = [dp['Average'] for dp in memory_stats.get('Datapoints', [])]
            memory_avg = statistics.mean(memory_values) if memory_values else None
        except Exception:
            memory_avg = None
        
        return {
            'cpu_avg': round(cpu_avg, 2) if cpu_avg else None,
            'memory_avg': round(memory_avg, 2) if memory_avg else None,
            'cpu_data_points': len(cpu_values),
            'memory_data_points': len(memory_values) if memory_avg else 0,
        }
    
    def suggest_right_size(self, instance_type: str, metrics: Dict[str, float]) -> Tuple[str, str]:
        """Suggest a right-sized instance type based on metrics."""
        cpu = metrics.get('cpu_avg')
        memory = metrics.get('memory_avg')
        
        # If insufficient data, return original type
        if not cpu and not memory:
            return instance_type, "INSUFFICIENT_DATA"
        
        # If high CPU, consider up-sizing
        if cpu and cpu > self.thresholds['cpu_high']:
            return self._upsize(instance_type), "HIGH_CPU"
        
        # If low CPU, consider down-sizing
        if cpu and cpu < self.thresholds['cpu_low']:
            return self._downsize(instance_type), "LOW_CPU"
        
        # If high memory, consider memory-optimized
        if memory and memory > self.thresholds['memory_high']:
            return self._memory_optimize(instance_type), "HIGH_MEMORY"
        
        # If low memory, consider general-purpose
        if memory and memory < self.thresholds['memory_low']:
            return self._general_purpose(instance_type), "LOW_MEMORY"
        
        return instance_type, "OPTIMAL"
    
    def _downsize(self, instance_type: str) -> str:
        """Downsize instance type if possible."""
        base_type = next((k for k in self.instance_hierarchy.keys() if instance_type.startswith(k)), None)
        if base_type:
            smaller_types = self.instance_hierarchy[base_type]
            if smaller_types:
                return f"{smaller_types[0]}.large"
        return instance_type
    
    def _upsize(self, instance_type: str) -> str:
        """Upsize instance type if needed."""
        base_type = next((k for k in self.instance_hierarchy.keys() if instance_type.startswith(k)), None)
        if base_type:
            larger_types = [t for t in self.instance_hierarchy.get(base_type, []) if t != instance_type]
            if larger_types:
                return f"{larger_types[0]}.xlarge"
        return instance_type
    
    def _memory_optimize(self, instance_type: str) -> str:
        """Suggest memory-optimized instance type."""
        if instance_type.startswith('t'):
            return 'r5.large'
        elif instance_type.startswith('m'):
            return 'r5.xlarge'
        return f"r{instance_type[1:]}" if instance_type[1] != 'r' else instance_type
    
    def _general_purpose(self, instance_type: str) -> str:
        """Suggest general-purpose instance type."""
        if instance_type.startswith('r'):
            return 'm5.large'
        elif instance_type.startswith('c'):
            return 't3.large'
        return f"m{instance_type[1:]}" if instance_type[1] != 'm' else instance_type
    
    def analyze_instance(self, instance_id: str) -> Dict:
        """Analyze a single instance for right-sizing."""
        # Get instance details
        response = self.ec2.describe_instances(InstanceIds=[instance_id])
        instance = response['Reservations'][0]['Instances'][0]
        instance_type = instance['InstanceType']
        
        # Get metrics
        metrics = self.get_instance_metrics(instance_id, self.thresholds['days'])
        
        # Suggest right size
        suggested_type, reason = self.suggest_right_size(instance_type, metrics)
        
        return {
            'instance_id': instance_id,
            'current_type': instance_type,
            'suggested_type': suggested_type,
            'reason': reason,
            'metrics': metrics,
            'potential_savings': self._estimate_savings(instance_type, suggested_type),
        }
    
    def _estimate_savings(self, current_type: str, suggested_type: str) -> float:
        """Estimate potential monthly savings from right-sizing."""
        # Simplified pricing model (replace with actual pricing data)
        pricing = {
            't3.large': 0.0832,
            't3.xlarge': 0.1664,
            't3.2xlarge': 0.3328,
            'm5.large': 0.096,
            'm5.xlarge': 0.192,
            'm5.2xlarge': 0.384,
            'r5.large': 0.126,
            'r5.xlarge': 0.252,
            'r5.2xlarge': 0.504,
        }
        
        current_price = pricing.get(current_type, 0.1)
        suggested_price = pricing.get(suggested_type, 0.1)
        
        monthly_usage_hours = 730  # Average hours per month
        monthly_savings = (current_price - suggested_price) * monthly_usage_hours
        
        return round(max(0, monthly_savings), 2)
    
    def analyze_all_instances(self) -> List[Dict]:
        """Analyze all instances in the region."""
        instances = self.ec2.describe_instances(
            Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
        )
        
        results = []
        for reservation in instances['Reservations']:
            for instance in reservation['Instances']:
                result = self.analyze_instance(instance['InstanceId'])
                results.append(result)
        
        return results

# Usage example
if __name__ == "__main__":
    analyzer = RightSizingAnalyzer(region="us-east-1")
    results = analyzer.analyze_all_instances()
    
    print("Right-Sizing Analysis Results:")
    print("-" * 80)
    
    total_monthly_savings = 0
    for result in results:
        if result['reason'] != 'OPTIMAL':
            print(f"Instance: {result['instance_id']}")
            print(f"  Current: {result['current_type']}")
            print(f"  Suggested: {result['suggested_type']} ({result['reason']})")
            print(f"  Monthly Savings: ${result['potential_savings']}")
            print()
            total_monthly_savings += result['potential_savings']
    
    print(f"Total Potential Monthly Savings: ${total_monthly_savings:.2f}")

Pattern 5: Spot Instance Strategy

Spot instance optimization reduces compute costs by up to 90% for fault-tolerant workloads.

BAD — Simple spot request without termination handling

# ❌ BAD — No termination handling, no replacement strategy
aws ec2 request-spot-instances \
  --spot-price "0.05" \
  --instance-count 1 \
  --type "one-time" \
  --launch-specification file://launch-spec.json

GOOD — Comprehensive spot instance management script

#!/bin/bash
# ✅ GOOD — Spot instance with termination handling, replacement strategy, and scaling
set -euo pipefail

SPOT_PRICE="${SPOT_PRICE:-0.05}"
INSTANCE_TYPE="${INSTANCE_TYPE:-c5.large}"
REGION="${AWS_REGION:-us-east-1}"
MIN_INSTANCES="${MIN_INSTANCES:-2}"
MAX_INSTANCES="${MAX_INSTANCES:-10}"

# Spot request with replacement strategy
create_spot_fleet() {
  local fleet_config=$(cat <<EOF
{
  "IamFleetRole": "arn:aws:iam::123456789012:role/spot-fleet-role",
  "TargetCapacity": $MIN_INSTANCES,
  "SpotPrice": "$SPOT_PRICE",
  "AllocationStrategy": "lowestPrice",
  "LaunchSpecifications": [
    {
      "ImageId": "ami-0123456789abcdef0",
      "InstanceType": "$INSTANCE_TYPE",
      "KeyName": "my-key",
      "SubnetId": "subnet-0123456789abcdef0",
      "IamInstanceProfile": {
        "Arn": "arn:aws:iam::123456789012:instance-profile/spot-instance-profile"
      },
      "SpotPrice": "$SPOT_PRICE"
    }
  ],
  "TerminateInstancesWithExpiration": true,
  "ReplaceUnhealthyInstances": true
}
EOF
)
  
  aws ec2 request-spot-fleet \
    --spot-fleet-request-data "$fleet_config" \
    --region "$REGION"
}

# Check spot instance health
check_spot_health() {
  local fleet_id="$1"
  
  # Get spot instance requests
  aws ec2 describe-spot-instance-requests \
    --filters "Name=spot-fleet-request-id,Values=$fleet_id" \
    --query 'SpotInstanceRequests[].{InstanceId:InstanceId,State:State,Status:Status}' \
    --output table
  
  # Check for terminations
  aws ec2 describe-spot-instance-requests \
    --filters "Name=spot-fleet-request-id,Values=$fleet_id" \
    --query 'SpotInstanceRequests[?State==`closed`]' \
    --output text
}

# Scale spot fleet based on demand
scale_spot_fleet() {
  local fleet_id="$1"
  local new_capacity="$2"
  
  aws ec2 modify-spot-fleet-request \
    --spot-fleet-request-id "$fleet_id" \
    --target-capacity "$new_capacity" \
    --terminate-instances-with-expiration
}

# Main execution
case "${1:-create}" in
  create)
    echo "Creating spot fleet with $MIN_INSTANCES instances at $SPOT_PRICE"
    create_spot_fleet
    ;;
  check)
    check_spot_health "$2"
    ;;
  scale)
    scale_spot_fleet "$2" "$3"
    ;;
  *)
    echo "Usage: $0 {create|check|scale}"
    exit 1
    ;;
esac

Python Spot Instance Optimizer

# ✅ GOOD — Python spot instance optimizer with termination handling
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import statistics
import time

class SpotInstanceOptimizer:
    """Optimizes spot instance usage for cost savings."""
    
    def __init__(self, region: str = "us-east-1"):
        self.ec2 = boto3.client('ec2', region_name=region)
        self.cloudwatch = boto3.client('cloudwatch', region_name=region)
        
        self.spot_pricing_history = {}
        self.recommendation_cache = {}
    
    def get_spot_price_history(self, instance_types: List[str], days: int = 7) -> Dict[str, List[float]]:
        """Get historical spot prices for given instance types."""
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(days=days)
        
        prices = {}
        
        for instance_type in instance_types:
            try:
                response = self.ec2.describe_spot_price_history(
                    InstanceTypes=[instance_type],
                    StartTime=start_time,
                    EndTime=end_time,
                    MaxResults=1000
                )
                
                price_list = [float(dp['SpotPrice']) for dp in response['SpotPriceHistory']]
                prices[instance_type] = price_list
                
                # Cache the data
                self.spot_pricing_history[instance_type] = {
                    'prices': price_list,
                    'avg': statistics.mean(price_list) if price_list else 0,
                    'min': min(price_list) if price_list else 0,
                    'max': max(price_list) if price_list else 0,
                    'std': statistics.stdev(price_list) if len(price_list) > 1 else 0,
                }
            except Exception as e:
                print(f"Error fetching spot price history for {instance_type}: {e}")
        
        return prices
    
    def recommend_spot_instance(self, instance_type: str, current_on_demand_price: float) -> Dict:
        """Recommend whether to use spot for given instance type."""
        history = self.spot_pricing_history.get(instance_type, {})
        
        if not history:
            return {
                'instance_type': instance_type,
                'recommendation': 'UNKNOWN',
                'reason': 'Insufficient historical data',
            }
        
        avg_price = history['avg']
        min_price = history['min']
        max_price = history['max']
        std = history['std']
        
        # Calculate savings
        savings_percentage = ((current_on_demand_price - avg_price) / current_on_demand_price) * 100
        
        # Determine recommendation
        if savings_percentage > 50 and std < 0.05:
            recommendation = 'STRONG_RECOMMENDATION'
            reason = f'High savings ({savings_percentage:.1f}%) with low volatility'
        elif savings_percentage > 30:
            recommendation = 'RECOMMENDATION'
            reason = f'Moderate savings ({savings_percentage:.1f}%)'
        elif savings_percentage > 10:
            recommendation = 'CONSIDER'
            reason = f'Some savings ({savings_percentage:.1f}%), monitor volatility'
        else:
            recommendation = 'NOT_RECOMMENDED'
            reason = f'Low savings ({savings_percentage:.1f}%)'
        
        return {
            'instance_type': instance_type,
            'recommendation': recommendation,
            'reason': reason,
            'spot_avg_price': round(avg_price, 4),
            'spot_min_price': round(min_price, 4),
            'spot_max_price': round(max_price, 4),
            'on_demand_price': current_on_demand_price,
            'savings_percentage': round(savings_percentage, 1),
            'price_volatility': round(std, 4),
        }
    
    def check_spot_capacity(self, instance_types: List[str], availability_zone: str) -> Dict[str, bool]:
        """Check spot capacity availability."""
        capacity = {}
        
        for instance_type in instance_types:
            try:
                response = self.ec2.describe_spot_instance_requests(
                    InstanceTypes=[instance_type],
                    AvailabilityZone=availability_zone,
                    MaxResults=1
                )
                capacity[instance_type] = True
            except Exception:
                capacity[instance_type] = False
        
        return capacity
    
    def estimate_savings(self, instance_type: str, hours_per_month: int = 730) -> Dict:
        """Estimate monthly savings from spot instances."""
        history = self.spot_pricing_history.get(instance_type, {})
        
        if not history:
            return {'error': 'No historical data available'}
        
        avg_spot_price = history['avg']
        
        # Get on-demand price (simplified - use actual pricing API in production)
       
            'c5.large': 0.17,
            'c5.xlarge': 0.34,
            'c5.2xlarge': 0.68,
            'm5.large': 0.096,
            'm5.xlarge': 0.192,
            'r5.large': 0.126,
            'r5.xlarge': 0.252,
        }
        
        avg_spot_price * 2)
        
        * hours_per_month
        spot_cost = avg_spot_price * hours_per_month
        savings = on_demand_cost - spot_cost
        savings_percentage = (savings / on_demand_cost) * 100
        
        return {
            'instance_type': instance_type,
            'on_demand_cost': round(on_demand_cost, 2),
            'spot_cost': round(spot_cost, 2),
            'monthly_savings': round(savings, 2),
            'savings_percentage': round(savings_percentage, 1),
            'break_even_price': round(spot_cost / hours_per_month, 4),
        }
    
    def get_all_recommendations(self) -> List[Dict]:
        """Get recommendations for all common instance types."""
        instance_types = ['c5.large', 'c5.xlarge', 'c5.2xlarge', 'm5.large', 'm5.xlarge', 'r5.large', 'r5.xlarge']
       
            'c5.large': 0.17, 'c5.xlarge': 0.34, 'c5.2xlarge': 0.68,
            'm5.large': 0.096, 'm5.xlarge': 0.192,
            'r5.large': 0.126, 'r5.xlarge': 0.252,
        }
        
        self.get_spot_price_history(instance_types)
        
        recommendations = []
        for instance_type in instance_types:
            recommendation = self.recommend_spot_instance(
                instance_type, 
                on_demand_prices.get(instance_type, 0)
            )
            savings = self.estimate_savings(instance_type)
            recommendation.update(savings)
            recommendations.append(recommendation)
        
        return recommendations

# Usage example
if __name__ == "__main__":
    optimizer = SpotInstanceOptimizer(region="us-east-1")
    recommendations = optimizer.get_all_recommendations()
    
    print("Spot Instance Optimization Recommendations")
    print("=" * 80)
    
    total_monthly_savings = 0
    
    for rec in recommendations:
        if rec['recommendation'] in ['STRONG_RECOMMENDATION', 'RECOMMENDATION']:
            print(f"\nInstance: {rec['instance_type']}")
            print(f"  Recommendation: {rec['recommendation']}")
            print(f"  Reason: {rec['reason']}")
            print(f"  On-demand: ${rec['on_demand_price']}/hr")
            print(f"  Spot avg: ${rec['spot_avg_price']}/hr")
            print(f"  Monthly Savings: ${rec['monthly_savings']}")
            
            total_monthly_savings += rec['monthly_savings']
    
    print(f"\n" + "=" * 80)
    print(f"Total Potential Monthly Savings: ${total_monthly_savings:.2f}")

Pattern 6: Reserved Instance Optimization

Reserved instance optimization analyzes usage patterns to recommend RI purchases.

BAD — Manual analysis without historical data

# ❌ BAD — No historical analysis, no ROI calculation
# Just buying RIs without data-driven decisions
aws ec2 purchase-reserved-instances-offering \
  --reserved-instances-offering-id "ri-offering-123" \
  --instance-count 5

GOOD — Data-driven RI optimization script

#!/bin/bash
# ✅ GOOD — Historical analysis with ROI calculation
set -euo pipefail

REGION="${AWS_REGION:-us-east-1}"
ANALYSIS_DAYS="${ANALYSIS_DAYS:-90}"

# Get historical instance usage
get_instance_usage() {
  local instance_type="$1"
  
  # Get average CPU utilization over analysis period
  aws cloudwatch get-metric-statistics \
    --metric-name CPUUtilization \
    --dimensions Name=InstanceId,Value=unknown \
    --namespace AWS/EC2 \
    --start-time $(date -d "$ANALYSIS_DAYS days ago" +%Y-%m-%dT%H:%M:%S) \
    --end-time $(date +%Y-%m-%dT%H:%M:%S) \
    --period 86400 \
    --statistics Average \
    --query 'Datapoints[].Average' \
    --output text | awk '{sum+=$1; count++} END {printf "%.1f", sum/count}'
}

# Calculate RI ROI
calculate_ri_roi() {
  local
  local ri_upfront="$2"
  local ri_annual="$3"
  local usage_hours="$4"
  
  # Calculate total cost with on-demand
  local "$on_demand_hourly * $usage_hours" | bc -l)
  
  # Calculate total cost with RI (upfront + annual)
  local ri_total=$(echo "$ri_upfront + $ri_annual" | bc -l)
  
  # Calculate savings
  local savings=$(echo "$on_demand_total - $ri_total" | bc -l)
  local savings_percentage=$(echo "($savings / $on_demand_total) * 100" | bc -l)
  
  echo "On-Demand Total: \$$on_demand_total"
  echo "RI Total: \$$ri_total"
  echo "Savings: \$$savings ($savings_percentage%)"
}

# Analyze instance usage and recommend RIs
analyze_ri_opportunity() {
  local instance_type="$1"
  local current_count="$2"
  
  echo "=== Reserved Instance Analysis ==="
  echo "Instance Type: $instance_type"
  echo "Current Count: $current_count"
  echo "Analysis Period: $ANALYSIS_DAYS days"
  echo ""
  
  # Get on-demand pricing
  local ec2 describe-spot-price-history \
    --instance-types "$instance_type" \
    --product-description "Linux/UNIX" \
    --query 'SpotPriceHistory[0].SpotPrice' \
    --output text)
  
  # Get RI pricing (simplified - use actual RI pricing API)
  local ri_pricing={
    "c5.large": {"upfront": 450, "annual": 600, "hourly": 0.17},
    "c5.xlarge": {"upfront": 900, "annual": 1200, "hourly": 0.34},
    "m5.large": {"upfront": 300, "annual": 400, "hourly": 0.096},
    "m5.xlarge": {"upfront": 600, "annual": 800, "hourly": 0.192},
  }
  
  if [[ -v ri_pricing["$instance_type"] ]]; then
    local upfront="${ri_pricing[$instance_type, upfront]}"
    local annual="${ri_pricing[$instance_type, annual]}"
    local hourly="${ri_pricing[$instance_type, hourly]}"
    
    # Assume 24/7 usage
    local usage_hours=$((ANALYSIS_DAYS * 24))
    
    echo "Pricing:"
    echo "  On-Demand: \$${hourly}/hour"
    echo "  RI Upfront: \$${upfront}"
    echo "  RI Annual: \$${annual}"
    echo ""
    
    echo "Projected Usage: $usage_hours hours"
    echo ""
    
    calculate_ri_roi "$hourly" "$upfront" "$annual" "$usage_hours"
  else
    echo "RI pricing not available for $instance_type"
  fi
}

# Main execution
instance_type="${1:-c5.large}"
count="${2:-5}"

analyze_ri_opportunity "$instance_type" "$count"

Python RI Optimizer

# ✅ GOOD — Python RI optimizer with ROI calculations
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import statistics

@dataclass
class InstanceUsage:
    instance_id: str
    instance_type: str
    avg_cpu: float
    hours_used: int
    days_used: int

@dataclass
class RIRecommendation:
    instance_type: str
    recommended_count: int
    upfront_cost: float
    annual_cost: float
    on_demand_cost: float
    roi_percentage: float
    break_even_months: float
    recommendation: str

class ReservedInstanceOptimizer:
    """Optimizes reserved instance purchases for cost savings."""
    
    def __init__(self, region: str = "us-east-1"):
        self.ec2 = boto3.client('ec2', region_name=region)
        s

…(truncated)