AWS Cost Explorer: Complete Guide to Usage Analysis, Cost Optimization & Cloud FinOps 2024

2026-07-28

Master AWS Cost Explorer with this comprehensive 3500+ word guide covering dashboard navigation, Cost and Usage Reports setup, cost allocation tags, forecasting, boto3 API automation, rightsizing recommendations, Reserved Instance analysis, Savings Plans optimization, and Lambda-based automated reporting.

#!/bin/bash
# Enable Cost Explorer using AWS CLI
# Note: This must be run from the management account

# Check if Cost Explorer is already enabled
aws ce get-cost-and-usage \
    --time-period Start=2024-01-01,End=2024-01-02 \
    --granularity DAILY \
    --metrics "UnblendedCost" \
    2>&1

# If you receive an error, Cost Explorer needs to be enabled via the console
# Navigate to: AWS Console > Billing > Cost Explorer > Enable Cost Explorer

# After enabling, wait 24 hours for historical data to become available
# You can then verify access with:
aws ce get-cost-and-usage \
    --time-period Start=2024-01-01,End=2024-01-31 \
    --granularity MONTHLY \
    --metrics "BlendedCost" "UnblendedCost" "UsageQuantity" \
    --group-by Type=DIMENSION,Key=SERVICE

# Get cost by linked account for multi-account organizations
aws ce get-cost-and-usage \
    --time-period Start=2024-01-01,End=2024-01-31 \
    --granularity MONTHLY \
    --metrics "UnblendedCost" \
    --group-by Type=DIMENSION,Key=LINKED_ACCOUNT

Example Filter Configurations:

1. Production Environment Analysis:
   Filter: Tag: Environment = "Production"
   Group by: Service
   Time: Last 6 months, Monthly
   Use Case: Understand production infrastructure costs for capacity planning

2. Development Cost Tracking:
   Filter: Tag: Environment = "Development"
   AND Linked Account IN [dev-account-1, dev-account-2]
   Group by: Usage Type
   Time: Last 30 days, Daily
   Use Case: Identify development waste and optimize sandbox environments

3. EC2 Deep Dive:
   Filter: Service = "Amazon Elastic Compute Cloud - Compute"
   Group by: Instance Type
   Time: Last 3 months, Monthly
   Use Case: Identify rightsizing opportunities and instance family optimization

4. Data Transfer Analysis:
   Filter: Usage Type Group = "Data Transfer"
   Group by: Region
   Time: Last 12 months, Monthly
   Use Case: Understand cross-region and internet egress patterns

5. Untagged Resource Identification:
   Filter: Tag: CostCenter = "" (empty/untagged)
   Group by: Service
   Time: Last 30 days
   Use Case: Identify resources missing required cost allocation tags

# Terraform configuration for AWS Cost and Usage Reports with Athena integration

provider "aws" {
  region = "us-east-1"  # CUR must be created in us-east-1
}

# S3 bucket for CUR data
resource "aws_s3_bucket" "cur_bucket" {
  bucket = "company-cost-and-usage-reports-${data.aws_caller_identity.current.account_id}"

  tags = {
    Purpose     = "cost-management"
    ManagedBy   = "terraform"
    Environment = "shared"
  }
}

resource "aws_s3_bucket_versioning" "cur_bucket" {
  bucket = aws_s3_bucket.cur_bucket.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "cur_bucket" {
  bucket = aws_s3_bucket.cur_bucket.id

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = "AES256"
    }
  }
}

resource "aws_s3_bucket_lifecycle_configuration" "cur_lifecycle" {
  bucket = aws_s3_bucket.cur_bucket.id

  rule {
    id     = "transition-to-ia"
    status = "Enabled"

    transition {
      days          = 90
      storage_class = "STANDARD_IA"
    }

    transition {
      days          = 365
      storage_class = "GLACIER"
    }

    expiration {
      days = 2555  # 7 years for compliance
    }
  }
}

resource "aws_s3_bucket_policy" "cur_bucket_policy" {
  bucket = aws_s3_bucket.cur_bucket.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid    = "AllowCURDelivery"
        Effect = "Allow"
        Principal = {
          Service = "billingreports.amazonaws.com"
        }
        Action = [
          "s3:GetBucketAcl",
          "s3:GetBucketPolicy"
        ]
        Resource = aws_s3_bucket.cur_bucket.arn
        Condition = {
          StringEquals = {
            "aws:SourceAccount" = data.aws_caller_identity.current.account_id
            "aws:SourceArn"     = "arn:aws:cur:us-east-1:${data.aws_caller_identity.current.account_id}:definition/*"
          }
        }
      },
      {
        Sid    = "AllowCURWrite"
        Effect = "Allow"
        Principal = {
          Service = "billingreports.amazonaws.com"
        }
        Action   = "s3:PutObject"
        Resource = "${aws_s3_bucket.cur_bucket.arn}/*"
        Condition = {
          StringEquals = {
            "aws:SourceAccount" = data.aws_caller_identity.current.account_id
            "aws:SourceArn"     = "arn:aws:cur:us-east-1:${data.aws_caller_identity.current.account_id}:definition/*"
          }
        }
      }
    ]
  })
}

# Cost and Usage Report Definition
resource "aws_cur_report_definition" "main" {
  report_name                = "comprehensive-cost-report"
  time_unit                  = "HOURLY"
  format                     = "Parquet"
  compression                = "Parquet"
  additional_schema_elements = ["RESOURCES", "SPLIT_COST_ALLOCATION_DATA"]
  s3_bucket                  = aws_s3_bucket.cur_bucket.id
  s3_region                  = "us-east-1"
  s3_prefix                  = "cur"
  report_versioning          = "OVERWRITE_REPORT"
  refresh_closed_reports     = true

  additional_artifacts = ["ATHENA"]

  depends_on = [aws_s3_bucket_policy.cur_bucket_policy]
}

# Athena workgroup for CUR queries
resource "aws_athena_workgroup" "cur_workgroup" {
  name = "cost-analysis-workgroup"

  configuration {
    enforce_workgroup_configuration    = true
    publish_cloudwatch_metrics_enabled = true

    result_configuration {
      output_location = "s3://${aws_s3_bucket.cur_bucket.id}/athena-results/"
      
      encryption_configuration {
        encryption_option = "SSE_S3"
      }
    }

    bytes_scanned_cutoff_per_query = 10737418240  # 10 GB limit per query
  }

  tags = {
    Purpose = "cost-analysis"
  }
}

# Glue database for CUR (created automatically by Athena integration)
resource "aws_glue_catalog_database" "cur_database" {
  name = "cost_and_usage_reports"
}

data "aws_caller_identity" "current" {}

output "cur_bucket_name" {
  value       = aws_s3_bucket.cur_bucket.id
  description = "S3 bucket containing Cost and Usage Reports"
}

output "athena_workgroup" {
  value       = aws_athena_workgroup.cur_workgroup.name
  description = "Athena workgroup for cost queries"
}

-- Top 10 most expensive resources in the last month
SELECT 
    line_item_resource_id,
    line_item_product_code AS service,
    product_product_name AS product_name,
    SUM(line_item_unblended_cost) AS total_cost,
    SUM(line_item_usage_amount) AS total_usage
FROM cost_and_usage_reports.cur_table
WHERE 
    line_item_usage_start_date >= date_add('month', -1, current_date)
    AND line_item_resource_id IS NOT NULL
    AND line_item_resource_id != ''
GROUP BY line_item_resource_id, line_item_product_code, product_product_name
ORDER BY total_cost DESC
LIMIT 10;

-- Daily cost trend by service for the last 30 days
SELECT 
    date_trunc('day', line_item_usage_start_date) AS usage_date,
    line_item_product_code AS service,
    SUM(line_item_unblended_cost) AS daily_cost
FROM cost_and_usage_reports.cur_table
WHERE line_item_usage_start_date >= date_add('day', -30, current_date)
GROUP BY date_trunc('day', line_item_usage_start_date), line_item_product_code
ORDER BY usage_date, daily_cost DESC;

-- EC2 spending by instance type and purchase option
SELECT 
    product_instance_type AS instance_type,
    pricing_term AS purchase_option,
    SUM(line_item_unblended_cost) AS total_cost,
    SUM(line_item_usage_amount) AS total_hours
FROM cost_and_usage_reports.cur_table
WHERE 
    line_item_product_code = 'AmazonEC2'
    AND product_instance_type IS NOT NULL
    AND line_item_usage_start_date >= date_add('month', -1, current_date)
GROUP BY product_instance_type, pricing_term
ORDER BY total_cost DESC;

-- Identify untagged resources and their costs
SELECT 
    line_item_product_code AS service,
    line_item_resource_id AS resource_id,
    SUM(line_item_unblended_cost) AS untagged_cost
FROM cost_and_usage_reports.cur_table
WHERE 
    line_item_usage_start_date >= date_add('month', -1, current_date)
    AND (resource_tags_user_cost_center IS NULL OR resource_tags_user_cost_center = '')
    AND line_item_resource_id IS NOT NULL
GROUP BY line_item_product_code, line_item_resource_id
HAVING SUM(line_item_unblended_cost) > 10
ORDER BY untagged_cost DESC
LIMIT 50;

{
  "tags": {
    "Environment": {
      "tag_key": {
        "@@assign": "Environment"
      },
      "tag_value": {
        "@@assign": [
          "production",
          "staging", 
          "development",
          "sandbox"
        ]
      },
      "enforced_for": {
        "@@assign": [
          "ec2:instance",
          "ec2:volume",
          "rds:db",
          "s3:bucket",
          "lambda:function"
        ]
      }
    },
    "CostCenter": {
      "tag_key": {
        "@@assign": "CostCenter"
      },
      "enforced_for": {
        "@@assign": [
          "ec2:instance",
          "rds:db"
        ]
      }
    }
  }
}

import boto3
from botocore.exceptions import ClientError
from typing import List, Dict

def manage_cost_allocation_tags() -> Dict:
    """
    Manage AWS Cost Allocation Tags programmatically.
    This script lists existing tags, activates recommended tags,
    and provides a summary of tag coverage.
    """
    ce_client = boto3.client('ce')
    results = {
        'active_tags': [],
        'inactive_tags': [],
        'activated_tags': [],
        'errors': []
    }
    
    # Get list of active cost allocation tags
    try:
        response = ce_client.list_cost_allocation_tags(
            Status='Active',
            MaxResults=100
        )
        
        active_tags = response.get('CostAllocationTags', [])
        results['active_tags'] = [tag['TagKey'] for tag in active_tags]
        print(f"Currently active cost allocation tags: {len(active_tags)}")
        for tag in active_tags:
            print(f"  - {tag['TagKey']}: Status={tag['Status']}, Type={tag.get('Type', 'User')}")
    
    except ClientError as e:
        error_msg = f"Error listing active tags: {e}"
        print(error_msg)
        results['errors'].append(error_msg)
        return results
    
    # Get inactive tags
    try:
        inactive_response = ce_client.list_cost_allocation_tags(
            Status='Inactive',
            MaxResults=100
        )
        
        inactive_tags = inactive_response.get('CostAllocationTags', [])
        results['inactive_tags'] = [tag['TagKey'] for tag in inactive_tags]
        print(f"\nInactive tags available for activation: {len(inactive_tags)}")
        for tag in inactive_tags[:10]:  # Show first 10
            print(f"  - {tag['TagKey']}")
        
    except ClientError as e:
        error_msg = f"Error listing inactive tags: {e}"
        print(error_msg)
        results['errors'].append(error_msg)
        return results
    
    # Recommended tags to activate (customize based on your organization)
    recommended_tags = [
        'Environment', 
        'Project', 
        'CostCenter', 
        'Owner', 
        'Application',
        'BusinessUnit',
        'Team'
    ]
    
    # Activate recommended tags that are currently inactive
    tags_to_activate = []
    for tag in inactive_tags:
        if tag['TagKey'] in recommended_tags:
            tags_to_activate.append({
                'TagKey': tag['TagKey'], 
                'Status': 'Active'
            })
    
    if tags_to_activate:
        try:
            print(f"\nActivating {len(tags_to_activate)} cost allocation tags...")
            ce_client.update_cost_allocation_tags_status(
                CostAllocationTagsStatus=tags_to_activate
            )
            results['activated_tags'] = [t['TagKey'] for t in tags_to_activate]
            print("Tags activated successfully!")
            for tag in tags_to_activate:
                print(f"  - Activated: {tag['TagKey']}")
        except ClientError as e:
            error_msg = f"Error activating tags: {e}"
            print(error_msg)
            results['errors'].append(error_msg)
    else:
        print("\nNo recommended tags need activation.")
    
    return results

def get_tag_coverage_report() -> None:
    """
    Generate a report showing tag coverage across resources.
    Helps identify resources missing required cost allocation tags.
    """
    ce_client = boto3.client('ce')
    
    from datetime import datetime, timedelta
    end_date = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
    
    # Get costs grouped by a required tag to see coverage
    for tag_key in ['Environment', 'CostCenter', 'Project']:
        try:
            response = ce_client.get_cost_and_usage(
                TimePeriod={'Start': start_date, 'End': end_date},
                Granularity='MONTHLY',
                Metrics=['UnblendedCost'],
                GroupBy=[{'Type': 'TAG', 'Key': tag_key}]
            )
            
            tagged_cost = 0
            untagged_cost = 0
            
            for result in response['ResultsByTime']:
                for group in result['Groups']:
                    tag_value = group['Keys'][0]
                    cost = float(group['Metrics']['UnblendedCost']['Amount'])
                    if tag_value and tag_value != f'{tag_key}

Cost Forecasting and Budget Management

AWS Cost Explorer's forecasting capabilities use machine learning to predict future spending based on historical patterns. Combined with AWS Budgets, this enables proactive cost management and helps prevent budget overruns before they occur.

Understanding Cost Forecasting Accuracy

AWS Cost Explorer forecasts are based on:

• Historical usage patterns from the past 12 months
• Seasonal trends and recurring patterns
• Growth trajectories based on recent changes
• Confidence intervals showing the range of likely outcomes

Forecasts are most accurate when:

• You have at least 3 months of historical data
• Usage patterns are relatively consistent
• No major infrastructure changes are planned

Accessing Forecasts via the Cost Explorer API

import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional

def get_cost_forecast(
    months_ahead: int = 3,
    granularity: str = 'MONTHLY',
    filter_expression: Optional[Dict] = None
) -> Dict:
    """
    Retrieve cost forecasts using Cost Explorer API.
    
    Args:
        months_ahead: Number of months to forecast
        granularity: MONTHLY or DAILY
        filter_expression: Optional filter to scope the forecast
    
    Returns:
        Dictionary containing forecast data and summary
    """
    ce_client = boto3.client('ce')
    
    today = datetime.now()
    # Forecast must start from today or tomorrow
    start_date = (today + timedelta(days=1)).strftime('%Y-%m-%d')
    end_date = (today + timedelta(days=months_ahead * 30)).strftime('%Y-%m-%d')
    
    try:
        request_params = {
            'TimePeriod': {'Start': start_date, 'End': end_date},
            'Metric': 'UNBLENDED_COST',
            'Granularity': granularity,
            'PredictionIntervalLevel': 95  # 95% confidence interval
        }
        
        if filter_expression:
            request_params['Filter'] = filter_expression
        
        response = ce_client.get_cost_forecast(**request_params)
        
        total_forecast = float(response['Total']['Amount'])
        print(f"\nCost Forecast for Next {months_ahead} Months:")
        print(f"{'='*50}")
        print(f"Total Forecast: ${total_forecast:,.2f}")
        print(f"\nMonthly Breakdown:")
        
        forecasts = []
        for forecast in response['ForecastResultsByTime']:
            period_start = forecast['TimePeriod']['Start']
            period_end = forecast['TimePeriod']['End']
            mean_value = float(forecast['MeanValue'])
            lower_bound = float(forecast['PredictionIntervalLowerBound'])
            upper_bound = float(forecast['PredictionIntervalUpperBound'])
            
            forecasts.append({
                'period': period_start,
                'mean': mean_value,
                'lower_bound': lower_bound,
                'upper_bound': upper_bound
            })
            
            print(f"\n  {period_start} to {period_end}:")
            print(f"    Predicted: ${mean_value:,.2f}")
            print(f"    Range (95% CI): ${lower_bound:,.2f} - ${upper_bound:,.2f}")
            variance = upper_bound - lower_bound
            print(f"    Uncertainty: ±${variance/2:,.2f}")
        
        return {
            'total': total_forecast,
            'forecasts': forecasts,
            'confidence_level': 95
        }
        
    except Exception as e:
        print(f"Error getting forecast: {e}")
        return None

def forecast_by_service() -> None:
    """Get forecasts for top spending services."""
    ce_client = boto3.client('ce')
    
    today = datetime.now()
    start_date = (today - timedelta(days=30)).strftime('%Y-%m-%d')
    end_date = today.strftime('%Y-%m-%d')
    
    # First, identify top services
    response = ce_client.get_cost_and_usage(
        TimePeriod={'Start': start_date, 'End': end_date},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )
    
    services = []
    for result in response['ResultsByTime']:
        for group in result['Groups']:
            service = group['Keys'][0]
            cost = float(group['Metrics']['UnblendedCost']['Amount'])
            services.append((service, cost))
    
    top_services = sorted(services, key=lambda x: x[1], reverse=True)[:5]
    
    print("\nForecasts by Top Services:")
    print("="*60)
    
    for service, current_cost in top_services:
        filter_expr = {
            'Dimensions': {
                'Key': 'SERVICE',
                'Values': [service]
            }
        }
        forecast = get_cost_forecast(months_ahead=1, filter_expression=filter_expr)
        if forecast:
            print(f"\n{service[:50]}:")
            print(f"  Current (30d): ${current_cost:,.2f}")
            print(f"  Next Month Forecast: ${forecast['total']:,.2f}")

if __name__ == "__main__":
    get_cost_forecast(months_ahead=3)
    forecast_by_service()

Creating AWS Budgets with Terraform

For infrastructure-as-code management of budgets, use Terraform:

# Terraform configuration for AWS Budgets

resource "aws_budgets_budget" "monthly_cost" {
  name              = "monthly-aws-budget"
  budget_type       = "COST"
  limit_amount      = "10000"
  limit_unit        = "USD"
  time_unit         = "MONTHLY"
  time_period_start = "2024-01-01_00:00"

  cost_types {
    include_credit             = false
    include_discount           = true
    include_other_subscription = true
    include_recurring          = true
    include_refund             = false
    include_subscription       = true
    include_support            = true
    include_tax                = true
    include_upfront            = true
    use_blended                = false
  }

  notification {
    comparison_operator        = "GREATER_THAN"
    threshold                  = 80
    threshold_type             = "PERCENTAGE"
    notification_type          = "ACTUAL"
    subscriber_email_addresses = ["finops@company.com", "cloud-team@company.com"]
  }

  notification {
    comparison_operator        = "GREATER_THAN"
    threshold                  = 100
    threshold_type             = "PERCENTAGE"
    notification_type          = "FORECASTED"
    subscriber_email_addresses = ["finops@company.com"]
  }

  notification {
    comparison_operator        = "GREATER_THAN"
    threshold                  = 100
    threshold_type             = "PERCENTAGE"
    notification_type          = "ACTUAL"
    subscriber_email_addresses = ["finops@company.com", "engineering-leads@company.com"]
  }
}

# Budget for specific service (EC2)
resource "aws_budgets_budget" "ec2_budget" {
  name         = "ec2-monthly-budget"
  budget_type  = "COST"
  limit_amount = "5000"
  limit_unit   = "USD"
  time_unit    = "MONTHLY"

  cost_filter {
    name   = "Service"
    values = ["Amazon Elastic Compute Cloud - Compute"]
  }

  notification {
    comparison_operator        = "GREATER_THAN"
    threshold                  = 90
    threshold_type             = "PERCENTAGE"
    notification_type          = "ACTUAL"
    subscriber_email_addresses = ["platform-team@company.com"]
  }
}

# Budget with auto-adjustment based on historical spending
resource "aws_budgets_budget" "auto_adjusting" {
  name         = "auto-adjusting-budget"
  budget_type  = "COST"
  limit_amount = "0"  # Will be auto-calculated
  limit_unit   = "USD"
  time_unit    = "MONTHLY"

  auto_adjust_data {
    auto_adjust_type = "HISTORICAL"
    
    historical_options {
      budget_adjustment_period = 3  # Look back 3 periods
    }
  }

  notification {
    comparison_operator        = "GREATER_THAN"
    threshold                  = 110
    threshold_type             = "PERCENTAGE"
    notification_type          = "ACTUAL"
    subscriber_email_addresses = ["finops@company.com"]
  }
}

Creating AWS Budgets Programmatically with boto3

import boto3
from datetime import datetime
from typing import List

def create_monthly_budget(
    budget_name: str, 
    limit_amount: float, 
    email_subscribers: List[str],
    cost_filters: dict = None
) -> bool:
    """
    Create a monthly cost budget with multiple alert thresholds.
    
    Args:
        budget_name: Name for the budget
        limit_amount: Monthly budget limit in USD
        email_subscribers: List of email addresses for notifications
        cost_filters: Optional filters (e.g., by service, tag, account)
    
    Returns:
        True if budget created successfully
    """
    budgets_client = boto3.client('budgets')
    account_id = boto3.client('sts').get_caller_identity()['Account']
    
    budget_definition = {
        'BudgetName': budget_name,
        'BudgetLimit': {
            'Amount': str(limit_amount), 
            'Unit': 'USD'
        },
        'CostTypes': {
            'IncludeTax': True,
            'IncludeSubscription': True,
            'UseBlended': False,
            'IncludeRefund': False,
            'IncludeCredit': False,
            'IncludeUpfront': True,
            'IncludeRecurring': True,
            'IncludeOtherSubscription': True,
            'IncludeSupport': True,
            'IncludeDiscount': True
        },
        'TimeUnit': 'MONTHLY',
        'BudgetType': 'COST'
    }
    
    # Add cost filters if provided
    if cost_filters:
        budget_definition['CostFilters'] = cost_filters
    
    # Define notification thresholds
    notifications_with_subscribers = [
        {
            'Notification': {
                'NotificationType': 'ACTUAL',
                'ComparisonOperator': 'GREATER_THAN',
                'Threshold': 50.0,
                'ThresholdType': 'PERCENTAGE'
            },
            'Subscribers': [
                {'SubscriptionType': 'EMAIL', 'Address': email} 
                for email in email_subscribers
            ]
        },
        {
            'Notification': {
                'NotificationType': 'ACTUAL',
                'ComparisonOperator': 'GREATER_THAN',
                'Threshold': 80.0,
                'ThresholdType': 'PERCENTAGE'
            },
            'Subscribers': [
                {'SubscriptionType': 'EMAIL', 'Address': email} 
                for email in email_subscribers
            ]
        },
        {
            'Notification': {
                'NotificationType': 'ACTUAL',
                'ComparisonOperator': 'GREATER_THAN',
                'Threshold': 100.0,
                'ThresholdType': 'PERCENTAGE'
            },
            'Subscribers': [
                {'SubscriptionType': 'EMAIL', 'Address': email} 
                for email in email_subscribers
            ]
        },
        {
            'Notification': {
                'NotificationType': 'FORECASTED',
                'ComparisonOperator': 'GREATER_THAN',
                'Threshold': 100.0,
                'ThresholdType': 'PERCENTAGE'
            },
            'Subscribers': [
                {'SubscriptionType': 'EMAIL', 'Address': email} 
                for email in email_subscribers
            ]
        }
    ]
    
    try:
        budgets_client.create_budget(
            AccountId=account_id,
            Budget=budget_definition,
            NotificationsWithSubscribers=notifications_with_subscribers
        )
        print(f"Budget '{budget_name}' created successfully!")
        print(f"  Limit: ${limit_amount:,.2f}/month")
        print(f"  Subscribers: {', '.join(email_subscribers)}")
        print(f"  Alerts at: 50%, 80%, 100% actual, 100% forecasted")
        return True
        
    except Exception as e:
        print(f"Error creating budget: {e}")
        return False

if __name__ == "__main__":
    # Create overall monthly budget
    create_monthly_budget(
        "Monthly-AWS-Budget", 
        10000.00, 
        ["finops@company.com", "engineering@company.com"]
    )
    
    # Create EC2-specific budget
    create_monthly_budget(
        "EC2-Monthly-Budget",
        5000.00,
        ["platform-team@company.com"],
        cost_filters={'Service': ['Amazon Elastic Compute Cloud - Compute']}
    )

Rightsizing Recommendations: Optimizing EC2 Instances

AWS Cost Explorer provides rightsizing recommendations that analyze your EC2 usage patterns and suggest optimal instance types. These recommendations can significantly reduce EC2 costs, often by 30-50% for over-provisioned workloads. Rightsizing is one of the highest-impact cost optimization activities you can undertake.

Understanding Rightsizing Recommendations

Cost Explorer analyzes 14 days of CloudWatch metrics including:

• CPU utilization (average and maximum)
• Memory utilization (if CloudWatch Agent is installed)
• Network throughput
• Disk I/O patterns

Based on this analysis, it recommends:

• Modify: Change to a smaller or different instance type
• Terminate: Instance appears unused and should be terminated

Accessing Rightsizing Recommendations with boto3

import boto3
from typing import Dict, List
import json

def get_rightsizing_recommendations(
    service: str = 'AmazonEC2',
    same_family: bool = True
) -> Dict:
    """
    Retrieve and analyze EC2 rightsizing recommendations.
    
    Args:
        service: Service to analyze (currently only AmazonEC2)
        same_family: If True, recommend within same instance family
    
    Returns:
        Dictionary with recommendations summary and details
    """
    ce_client = boto3.client('ce')
    
    recommendation_target = 'SAME_INSTANCE_FAMILY' if same_family else 'CROSS_INSTANCE_FAMILY'
    
    response = ce_client.get_rightsizing_recommendation(
        Service=service,
        Configuration={
            'RecommendationTarget': recommendation_target,
            'BenefitsConsidered': True  # Consider RI/SP in recommendations
        }
    )
    
    recommendations = response.get('RightsizingRecommendations', [])
    summary = response.get('Summary', {})
    
    results = {
        'summary': {
            'total_recommendations': summary.get('TotalRecommendationCount', 0),
            'estimated_monthly_savings': float(summary.get('EstimatedTotalMonthlySavingsAmount', 0)),
            'savings_percentage': float(summary.get('SavingsPercentage', 0))
        },
        'recommendations': [],
        'by_action': {'Modify': [], 'Terminate': []}
    }
    
    print("\nRightsizing Recommendations Summary:")
    print("="*60)
    print(f"Total Recommendations: {results['summary']['total_recommendations']}")
    print(f"Estimated Monthly Savings: ${results['summary']['estimated_monthly_savings']:,.2f}")
    print(f"Potential Savings: {results['summary']['savings_percentage']:.1f}%")
    
    print("\nTop Recommendations:")
    print("-"*60)
    
    for rec in recommendations[:15]:
        action = rec.get('RightsizingType')
        current = rec.get('CurrentInstance', {})
        resource_id = current.get('ResourceId', 'N/A')
        resource_details = current.get('ResourceDetails', {}).get('EC2ResourceDetails', {})
        current_type = resource_details.get('InstanceType', 'Unknown')
        current_cost = float(current.get('MonthlyCost', 0))
        
        rec_details = {
            'resource_id': resource_id,
            'action': action,
            'current_type': current_type,
            'current_monthly_cost': current_cost
        }
        
        if action == 'Modify':
            modify_detail = rec.get('ModifyRecommendationDetail', {})
            target_instances = modify_detail.get('TargetInstances', [])
            
            if target_instances:
                target = target_instances[0]
                target_details = target.get('ResourceDetails', {}).get('EC2ResourceDetails', {})
                target_type = target_details.get('InstanceType', 'Unknown')
                estimated_savings = float(target.get('EstimatedMonthlySavings', 0))
                target_cost = float(target.get('EstimatedMonthlyCost', 0))
                
                rec_details['target_type'] = target_type
                rec_details['estimated_savings'] = estimated_savings
                rec_details['target_monthly_cost'] = target_cost
                
                print(f"  MODIFY: {resource_id[:20]}...")
                print(f"    {current_type} -> {target_type}")
                print(f"    Current: ${current_cost:,.2f}/mo -> Target: ${target_cost:,.2f}/mo")
                print(f"    Savings: ${estimated_savings:,.2f}/mo")
                
        elif action == 'Terminate':
            savings = float(rec.get('TerminateRecommendationDetail', {}).get('EstimatedMonthlySavings', current_cost))
            rec_details['estimated_savings'] = savings
            
            print(f"  TERMINATE: {resource_id[:20]}...")
            print(f"    Instance Type: {current_type}")
            print(f"    Savings: ${savings:,.2f}/mo (appears unused)")
        
        results['recommendations'].append(rec_details)
        results['by_action'][action].append(rec_details)
        print()
    
    # Summary by action type
    print("\nSummary by Action:")
    print(f"  Modify: {len(results['by_action']['Modify'])} instances")
    print(f"  Terminate: {len(results['by_action']['Terminate'])} instances")
    
    return results

if __name__ == "__main__":
    # Get same-family recommendations (safer)
    print("\n=== Same Instance Family Recommendations ===")
    get_rightsizing_recommendations(same_family=True)
    
    # Get cross-family recommendations (more aggressive)
    print("\n=== Cross Instance Family Recommendations ===")
    get_rightsizing_recommendations(same_family=False)

Reserved Instance and Savings Plans Analysis

Reserved Instances and Savings Plans can provide up to 72% discount compared to On-Demand pricing. Cost Explorer provides detailed analysis of coverage, utilization, and purchase recommendations to help you optimize your commitment-based discounts.

Understanding RI vs. Savings Plans

Analyzing RI Coverage and Savings Plans

import boto3
from datetime import datetime, timedelta
from typing import Dict

def analyze_commitment_coverage() -> Dict:
    """
    Comprehensive analysis of Reserved Instance and Savings Plans coverage.
    
    Returns:
        Dictionary with coverage metrics and recommendations
    """
    ce_client = boto3.client('ce')
    
    end_date = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
    
    results = {
        'ri_coverage': {},
        'ri_utilization': {},
        'savings_plans_coverage': {},
        'savings_plans_utilization': {},
        'recommendations': {}
    }
    
    print("\n=== Commitment-Based Discount Analysis ===")
    print("="*60)
    print(f"Analysis Period: {start_date} to {end_date}")
    
    # Get RI Coverage
    print("\n--- Reserved Instance Coverage ---")
    try:
        ri_coverage = ce_client.get_reservation_coverage(
            TimePeriod={'Start': start_date, 'End': end_date},
            Granularity='MONTHLY',
            Metrics=['Hour', 'Cost']
        )
        
        for period in ri_coverage['CoveragesByTime']:
            total = period.get('Total', {})
            coverage_hours = total.get('CoverageHours', {})
            
            on_demand = float(coverage_hours.get('OnDemandHours', 0))
            reserved = float(coverage_hours.get('ReservedHours', 0))
            total_hours = on_demand + reserved
            
            coverage_pct = (reserved / total_hours * 100) if total_hours > 0 else 0
            
            results['ri_coverage'] = {
                'on_demand_hours': on_demand,
                'reserved_hours': reserved,
                'coverage_percentage': coverage_pct
            }
            
            print(f"  RI Coverage: {coverage_pct:.1f}%")
            print(f"  On-Demand Hours: {on_demand:,.0f}")
            print(f"  Reserved Hours: {reserved:,.0f}")
            
    except Exception as e:
        print(f"  Error getting RI coverage: {e}")
    
    # Get RI Utilization
    print("\n--- Reserved Instance Utilization ---")
    try:
        ri_utilization = ce_client.get_reservation_utilization(
            TimePeriod={'Start': start_date, 'End': end_date},
            Granularity='MONTHLY'
        )
        
        for period in ri_utilization['UtilizationsByTime']:
            total = period.get('Total', {})
            utilization_pct = float(total.get('UtilizationPercentage', 0))
            
            results['ri_utilization'] = {
                'utilization_percentage': utilization_pct,
                'purchased_hours': float(total.get('PurchasedHours', 0)),
                'used_hours': float(total.get('TotalActualHours', 0)),
                'unused_hours': float(total.get('UnusedHours', 0))
            }
            
            print(f"  RI Utilization: {utilization_pct:.1f}%")
            if utilization_pct < 80:
                print(f"  WARNING: RI utilization below 80% - review your reservations")
                
    except Exception as e:
        print(f"  Error getting RI utilization: {e}")
    
    # Get Savings Plans Coverage
    print("\n--- Savings Plans Coverage ---")
    try:
        sp_coverage = ce_client.get_savings_plans_coverage(
            TimePeriod={'Start': start_date, 'End': end_date},
            Granularity='MONTHLY'
        )
        
        for period in sp_coverage.get('SavingsPlansCoverages', []):
            coverage = period.get('Coverage', {})
            coverage_pct = float(coverage.get('CoveragePercentage', 0))
            spend_covered = float(coverage.get('SpendCoveredBySavingsPlans', 0))
            on_demand_spend = float(coverage.get('OnDemandCost', 0))
            
            results['savings_plans_coverage'] = {
                'coverage_percentage': coverage_pct,
                'spend_covered': spend_covered,
                'on_demand_cost': on_demand_spend
            }
            
            print(f"  SP Coverage: {coverage_pct:.1f}%")
            print(f"  Spend Covered: ${spend_covered:,.2f}")
            print(f"  On-Demand Spend: ${on_demand_spend:,.2f}")
            
    except Exception as e:
        print(f"  Error getting SP coverage: {e}")
    
    # Get Savings Plans Utilization
    print("\n--- Savings Plans Utilization ---")
    try:
        sp_utilization = ce_client.get_savings_plans_utilization(
            TimePeriod={'Start': start_date, 'End': end_date},
            Granularity='MONTHLY'
        )
        
        total = sp_utilization.get('Total', {})
        utilization_pct = float(total.get('UtilizationPercentage', 0))
        
        results['savings_plans_utilization'] = {
            'utilization_percentage': utilization_pct
        }
        
        print(f"  SP Utilization: {utilization_pct:.1f}%")
        
    except Exception as e:
        print(f"  Error getting SP utilization: {e}")
    
    # Get Savings Plans Purchase Recommendations
    print("\n--- Savings Plans Purchase Recommendations ---")
    for sp_type in ['COMPUTE_SP', 'EC2_INSTANCE_SP']:
        try:
            rec_response = ce_client.get_savings_plans_purchase_recommendation(
                SavingsPlansType=sp_type,
                TermInYears='ONE_YEAR',
                PaymentOption='NO_UPFRONT',
                LookbackPeriodInDays='THIRTY_DAYS'
            )
            
            recommendation = rec_response.get('SavingsPlansPurchaseRecommendation', {})
            summary = recommendation.get('SavingsPlansPurchaseRecommendationSummary', {})
            
            if summary:
                hourly_commitment = float(summary.get('HourlyCommitmentToPurchase', 0))
                monthly_savings = float(summary.get('EstimatedMonthlySavingsAmount', 0))
                savings_pct = float(summary.get('EstimatedSavingsPercentage', 0))
                
                results['recommendations'][sp_type] = {
                    'hourly_commitment': hourly_commitment,
                    'monthly_savings': monthly_savings,
                    'savings_percentage': savings_pct
                }
                
                print(f"\n  {sp_type.replace('_', ' ').title()}:")
                print(f"    Recommended Commitment: ${hourly_commitment:,.3f}/hour")
                print(f"    Estimated Monthly Savings: ${monthly_savings:,.2f}")
                print(f"    Savings Percentage: {savings_pct:.1f}%")
                
        except Exception as e:
            print(f"  Error getting {sp_type} recommendations: {e}")
    
    # Overall recommendation
    print("\n--- Overall Recommendation ---")
    total_coverage = (
        results.get('ri_coverage', {}).get('coverage_percentage', 0) +
        results.get('savings_plans_coverage', {}).get('coverage_percentage', 0)
    )
    
    if total_coverage < 50:
        print("  ACTION NEEDED: Low commitment coverage (<50%)")
        print("  Consider purchasing Savings Plans for stable workloads")
    elif total_coverage < 70:
        print("  OPPORTUNITY: Moderate commitment coverage (50-70%)")
        print("  Review recommendations for additional savings")
    else:
        print("  GOOD: High commitment coverage (>70%)")
        print("  Monitor utilization to ensure efficient use")
    
    return results

if __name__ == "__main__":
    analyze_commitment_coverage()

Automated Cost Reporting with AWS Lambda

Automating cost reporting ensures stakeholders receive regular updates without manual effort. Here's a production-ready Lambda function for automated weekly cost reports with detailed breakdowns:

import boto3
import json
from datetime import datetime, timedelta
from typing import Dict, List

def lambda_handler(event, context):
    """
    AWS Lambda function for automated weekly cost reporting.
    Sends detailed cost breakdown via SNS.
    
    Environment Variables:
        SNS_TOPIC_ARN: ARN of SNS topic for notifications
    """
    import os
    
    ce_client = boto3.client('ce')
    sns_client = boto3.client('sns')
    
    SNS_TOPIC_ARN = os.environ.get('SNS_TOPIC_ARN', 
        'arn:aws:sns:us-east-1:123456789012:cost-reports')
    
    end_date = datetime.now()
    start_date = end_date - timedelta(days=7)
    
    # Previous week for comparison
    prev_end = start_date
    prev_start = prev_end - timedelta(days=7)
    
    # Get current week costs by service
    response = ce_client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date.strftime('%Y-%m-%d'),
            'End': end_date.strftime('%Y-%m-%d')
        },
        Granularity='DAILY',
        Metrics=['UnblendedCost', 'UsageQuantity'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )
    
    # Get previous week for comparison
    prev_response = ce_client.get_cost_and_usage(
        TimePeriod={
            'Start': prev_start.strftime('%Y-%m-%d'),
            'End': prev_end.strftime('%Y-%m-%d')
        },
        Granularity='DAILY',
        Metrics=['UnblendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )
    
    # Process current week
    total_cost = 0
    by_service: Dict[str, float] = {}
    
    for result in response['ResultsByTime']:
        for group in result['Groups']:
            service = group['Keys'][0]
            cost = float(group['Metrics']['UnblendedCost']['Amount'])
            by_service[service] = by_service.get(service, 0) + cost
            total_cost += cost
    
    # Process previous week
    prev_by_service: Dict[str, float] = {}
    prev_total = 0
    
    for result in prev_response['ResultsByTime']:
        for group in result['Groups']:
            service = group['Keys'][0]
            cost = float(group['Metrics']['UnblendedCost']['Amount'])
            prev_by_service[service] = prev_by_service.get(service, 0) + cost
            prev_total += cost
    
    # Calculate week-over-week change
    wow_change = ((total_cost - prev_total) / prev_total * 100) if prev_total > 0 else 0
    change_indicator = "📈" if wow_change > 0 else "📉" if wow_change < 0 else "➡️"
    
    sorted_services = sorted(by_service.items(), key=lambda x: x[1], reverse=True)[:10]
    
    # Build report
    report_lines = [
        "═" * 50,
        "     WEEKLY AWS COST REPORT",
        "═" * 50,
        f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}",
        "",
        f"TOTAL SPEND: ${total_cost:,.2f}",
        f"Previous Week: ${prev_total:,.2f}",
        f"Change: {change_indicator} {wow_change:+.1f}%",
        "",
        "─" * 50,
        "TOP 10 SERVICES BY COST",
        "─" * 50,
    ]
    
    for service, cost in sorted_services:
        prev_cost = prev_by_service.get(service, 0)
        service_change = ((cost - prev_cost) / prev_cost * 100) if prev_cost > 0 else 0
        pct_of_total = (cost / total_cost * 100) if total_cost > 0 else 0
        
        indicator = "↑" if service_change > 5 else "↓" if service_change < -5 else "="
        report_lines.append(
            f"{service[:35]:35} ${cost:>10,.2f} ({pct_of_total:>4.1f}%) {indicator}"
        )
    
    # Add cost anomalies section
    report_lines.extend([
        "",
        "─" * 50,
        "NOTABLE CHANGES (>20% WoW)",
        "─" * 50,
    ])
    
    anomalies_found = False
    for service, cost in sorted_services:
        prev_cost = prev_by_service.get(service, 0)
        if prev_cost > 10:  # Only flag services with meaningful previous spend
            change = ((cost - prev_cost) / prev_cost * 100)
            if abs(change) > 20:
                anomalies_found = True
                report_lines.append(
                    f"  {service[:30]}: {change:+.1f}% (${prev_cost:,.2f} -> ${cost:,.2f})"
                )
    
    if not anomalies_found:
        report_lines.append("  No significant anomalies detected")
    
    report_lines.extend([
        "",
        "═" * 50,
        "Generated by Warqline Cost Analytics",
        "View detailed analysis: https://portal.warqline.com/costs",
    ])
    
    report = "\n".join(report_lines)
    
    # Send via SNS
    sns_client.publish(
        TopicArn=SNS_TOPIC_ARN,
        Subject=f"Weekly AWS Cost Report - ${total_cost:,.2f} ({change_indicator} {wow_change:+.1f}%)",
        Message=report
    )
    
    return {
        'statusCode': 200, 
        'body': json.dumps({
            'total_cost': total_cost,
            'week_over_week_change': wow_change,
            'top_service': sorted_services[0][0] if sorted_services else None
        })
    }

Cost Anomaly Detection and Alerting

AWS Cost Anomaly Detection uses machine learning to continuously monitor your spending patterns and alert you when unusual costs are detected. This proactive approach helps catch unexpected expenses before they become significant issues.

Setting Up Cost Anomaly Detection

import boto3
from typing import List, Optional

def setup_anomaly_detection(
    monitor_name: str = 'AllServicesMonitor',
    subscription_name: str = 'CostAnomalyAlerts',
    threshold: float = 100.0,
    email_addresses: List[str] = None
) -> dict:
    """
    Configure AWS Cost Anomaly Detection monitors and subscriptions.
    
    Args:
        monitor_name: Name for the anomaly monitor
        subscription_name: Name for the alert subscription
        threshold: Dollar threshold for anomaly alerts
        email_addresses: List of email addresses for notifications
    
    Returns:
        Dictionary with monitor and subscription ARNs
    """
    ce_client = boto3.client('ce')
    
    if email_addresses is None:
        email_addresses = ['finops@company.com']
    
    results = {}
    
    # Create a cost monitor for all services
    print("Creating Cost Anomaly Monitor...")
    monitor_response = ce_client.create_anomaly_monitor(
        AnomalyMonitor={
            'MonitorName': monitor_name,
            'MonitorType': 'DIMENSIONAL',
            'MonitorDimension': 'SERVICE'
        }
    )
    monitor_arn = monitor_response['MonitorArn']
    results['monitor_arn'] = monitor_arn
    print(f"  Created monitor: {monitor_arn}")
    
    # Create an alert subscription
    print("\nCreating Alert Subscription...")
    subscribers = [
        {'Type': 'EMAIL', 'Address': email} 
        for email in email_addresses
    ]
    
    subscription_response = ce_client.create_anomaly_subscription(
        AnomalySubscription={
            'SubscriptionName': subscription_name,
            'Threshold': threshold,
            'Frequency': 'IMMEDIATE',  # DAILY or IMMEDIATE
            'MonitorArnList': [monitor_arn],
            'Subscribers': subscribers,
            'ThresholdExpression': {
                'Dimensions': {
                    'Key': 'ANOMALY_TOTAL_IMPACT_ABSOLUTE',
                    'Values': [str(threshold)],
                    'MatchOptions': ['GREATER_THAN_OR_EQUAL']
                }
            }
        }
    )
    
    subscription_arn = subscription_response['SubscriptionArn']
    results['subscription_arn'] = subscription_arn
    print(f"  Created subscription: {subscription_arn}")
    print(f"  Threshold: ${threshold}")
    print(f"  Subscribers: {', '.join(email_addresses)}")
    
    return results

def create_service_specific_monitors() -> List[dict]:
    """
    Create separate monitors for high-spend services for more targeted alerts.
    """
    ce_client = boto3.client('ce')
    
    # Services to create dedicated monitors for
    high_priority_services = [
        ('AmazonEC2', 'EC2ComputeMonitor'),
        ('AmazonRDS', 'RDSDatabaseMonitor'),
        ('AmazonS3', 'S3StorageMonitor'),
    ]
    
    monitors = []
    
    for service, monitor_name in high_priority_services:
        try:
            response = ce_client.create_anomaly_monitor(
                AnomalyMonitor={
                    'MonitorName': monitor_name,
                    'MonitorType': 'CUSTOM',
                    'MonitorSpecification': {
                        'Dimensions': {
                            'Key': 'SERVICE',
                            'Values': [service],
                            'MatchOptions': ['EQUALS']
                        }
                    }
                }
            )
            monitors.append({
                'service': service,
                'monitor_name': monitor_name,
                'monitor_arn': response['MonitorArn']
            })
            print(f"Created monitor for {service}: {response['MonitorArn']}")
            
        except Exception as e:
            print(f"Error creating monitor for {service}: {e}")
    
    return monitors

if __name__ == "__main__":
    # Set up main anomaly detection
    setup_anomaly_detection(
        threshold=50.0,
        email_addresses=['finops@company.com', 'cloud-team@company.com']
    )
    
    # Set up service-specific monitors
    print("\nCreating service-specific monitors...")
    create_service_specific_monitors()

Best Practices for Anomaly Detection

Threshold Configuration: Start with conservative thresholds (higher dollar amounts like $100-500) and gradually lower them as you understand your normal spending patterns. This prevents alert fatigue from false positives while still catching significant anomalies.

Multi-Monitor Strategy: Create separate monitors for different account groups, services, or cost centers. This provides more targeted alerts and helps identify the root cause of anomalies more quickly. Consider monitors for:

• High-spend services (EC2, RDS, S3)
• Production vs. non-production accounts
• Individual business units or projects

Integration with Incident Response: Connect Cost Anomaly Detection alerts to your incident response workflows:

• Create tickets in Jira or ServiceNow automatically
• Send alerts to Slack or Microsoft Teams channels
• Trigger investigation runbooks
• Escalate to on-call personnel for critical anomalies

Cross-Account Cost Analysis Strategies

For organizations using AWS Organizations with multiple accounts, effective cost analysis requires strategies that provide both aggregate and granular visibility across your entire AWS estate.

Consolidated Billing and Cost Allocation

When using AWS Organizations, all member account costs are consolidated into the management account's bill. This enables:

• Aggregate Reporting: View total organizational spend in one place with drill-down capability
• Volume Discounts: Benefit from combined usage for services with tiered pricing
• Centralized Optimization: Purchase Savings Plans or Reserved Instances that apply across accounts
• Simplified Billing: Single invoice for all accounts with detailed line items

Implementing Showback and Chargeback

Many organizations need to allocate cloud costs back to business units or project teams. Here's how to implement effective cost allocation:

import boto3
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, Optional
import csv
import io

def generate_showback_report(
    cost_center_tag: str = 'CostCenter',
    output_format: str = 'console'
) -> Dict:
    """
    Generate a showback report grouped by cost center.
    
    Args:
        cost_center_tag: Tag key to group costs by
        output_format: 'console', 'csv', or 'json'
    
    Returns:
        Dictionary with cost center breakdowns
    """
    ce_client = boto3.client('ce')
    
    end_date = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
    
    response = ce_client.get_cost_and_usage(
        TimePeriod={'Start': start_date, 'End': end_date},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost', 'UsageQuantity'],
        GroupBy=[
            {'Type': 'TAG', 'Key': cost_center_tag},
            {'Type': 'DIMENSION', 'Key': 'SERVICE'}
        ]
    )
    
    cost_centers = defaultdict(lambda: {'total': 0, 'services': {}})
    grand_total = 0
    
    for result in response['ResultsByTime']:
        for group in result['Groups']:
            tag_value = group['Keys'][0]
            # Extract actual tag value (format is "TagKey$TagValue")
            cost_center = tag_value.split('