EC2 Cost Analysis & Savings Plan Recommendations with Cost Explorer

import boto3
from datetime import datetime, timedelta
import json


def analyze_and_recommend_savings(lookback_days: int = 30) -> dict:
    """
    Analyses EC2 on-demand costs and surfaces AWS-generated
    Savings Plan and Reserved Instance recommendations.

    Cost Explorer is a GLOBAL service — the client always uses
    region "us-east-1" regardless of where your resources are.

    TimePeriod dates must be ISO 8601 format: "YYYY-MM-DD".
    End date is exclusive (same as Python range() semantics).
    """
    # Cost Explorer is only available in us-east-1
    ce = boto3.client("ce", region_name="us-east-1")

    end   = datetime.utcnow().strftime("%Y-%m-%d")
    start = (datetime.utcnow() - timedelta(days=lookback_days)).strftime("%Y-%m-%d")

    print(f"Analysing EC2 costs from {start} to {end}...\n")

    # ── Step 1: Current on-demand EC2 spend by instance type ──────
    # get_cost_and_usage() is the primary Cost Explorer API.
    #
    # Filter: We want only EC2 compute on-demand charges.
    # The "And" operator requires ALL nested conditions to match.
    # SERVICE dimension filters by the AWS service name.
    # PURCHASE_TYPE dimension filters by pricing model.
    #
    # Metrics: "BlendedCost" is the dollar amount.
    #          "UsageQuantity" is hours of usage.
    #
    # GroupBy: Breaks the cost by INSTANCE_TYPE dimension so we can
    # see which instance types are driving the spend.
    cost_response = ce.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="MONTHLY",
        Filter={
            "And": [
                {
                    "Dimensions": {
                        "Key":    "SERVICE",
                        "Values": ["Amazon Elastic Compute Cloud - Compute"],
                    }
                },
                {
                    "Dimensions": {
                        "Key":    "PURCHASE_TYPE",
                        "Values": ["On-Demand"],
                    }
                },
            ]
        },
        Metrics=["BlendedCost", "UsageQuantity"],
        GroupBy=[{"Type": "DIMENSION", "Key": "INSTANCE_TYPE"}],
    )

    # Aggregate costs across months (in case lookback spans multiple months)
    instance_costs: dict[str, float] = {}
    for result in cost_response["ResultsByTime"]:
        for group in result["Groups"]:
            instance_type = group["Keys"][0]
            cost = float(group["Metrics"]["BlendedCost"]["Amount"])
            instance_costs[instance_type] = instance_costs.get(instance_type, 0) + cost

    # ── Step 2: Savings Plans recommendation ──────────────────────
    # get_savings_plans_purchase_recommendation() calls AWS's recommendation
    # engine which analyses your historical usage and computes the optimal
    # hourly commitment to maximise savings.
    #
    # SavingsPlansType options:
    #   "COMPUTE_SP" — most flexible, applies to any EC2, Lambda, Fargate
    #   "EC2_INSTANCE_SP" — higher savings, locked to a region/instance family
    #
    # TermInYears: "ONE_YEAR" or "THREE_YEARS"
    # PaymentOption: "NO_UPFRONT", "PARTIAL_UPFRONT", "ALL_UPFRONT"
    sp_response = ce.get_savings_plans_purchase_recommendation(
        SavingsPlansType="COMPUTE_SP",
        TermInYears="ONE_YEAR",
        PaymentOption="NO_UPFRONT",
        LookbackPeriodInDays="THIRTY_DAYS",
    )
    sp_summary = sp_response.get(
        "SavingsPlansPurchaseRecommendation", {}
    ).get("SavingsPlansPurchaseRecommendationSummary", {})

    # ── Step 3: Reserved Instance recommendation ──────────────────
    # get_reservation_purchase_recommendation() gives per-instance-type
    # RI recommendations — more specific than Savings Plans but higher savings.
    ri_response = ce.get_reservation_purchase_recommendation(
        Service="Amazon EC2",
        TermInYears="ONE_YEAR",
        PaymentOption="NO_UPFRONT",
        LookbackPeriodInDays="THIRTY_DAYS",
    )
    ri_recommendations = ri_response.get("Recommendations", [])

    # ── Print report ──────────────────────────────────────────────
    total_cost = sum(instance_costs.values())

    print("=" * 60)
    print("EC2 ON-DEMAND SPEND BY INSTANCE TYPE (Last 30 Days)")
    print("=" * 60)
    for itype, cost in sorted(instance_costs.items(), key=lambda x: -x[1]):
        bar = "█" * min(int(cost / total_cost * 30), 30)
        print(f"  {itype:<20} ${cost:>10.2f}  {bar}")
    print(f"\n  {'TOTAL ON-DEMAND':<20} ${total_cost:>10.2f}")

    print("\n" + "=" * 60)
    print("SAVINGS PLAN RECOMMENDATION (Compute SP, 1-Year, No Upfront)")
    print("=" * 60)
    if sp_summary:
        monthly_savings  = float(sp_summary.get("EstimatedMonthlySavingsAmount", 0))
        savings_pct      = sp_summary.get("EstimatedSavingsPercentage", "N/A")
        hourly_commit    = float(sp_summary.get("HourlyCommitmentToPurchase", 0))
        current_spend    = float(sp_summary.get("CurrentOnDemandSpend", 0))

        print(f"  Current On-Demand Spend:   ${current_spend:>10,.2f}")
        print(f"  Recommended Hourly Commit: ${hourly_commit:>10.4f}/hr")
        print(f"  Estimated Monthly Savings: ${monthly_savings:>10,.2f}")
        print(f"  Savings Percentage:        {savings_pct}%")
        print(f"  Annual Savings Estimate:   ${monthly_savings * 12:>10,.2f}")
    else:
        print("  No Savings Plan recommendation available (need 30+ days of data)")

    print("\n" + "=" * 60)
    print("RESERVED INSTANCE RECOMMENDATIONS (Top 5, 1-Year, No Upfront)")
    print("=" * 60)
    for i, rec in enumerate(ri_recommendations[:5], 1):
        details = rec.get("RecommendationDetails", [{}])[0]
        spec    = details.get("InstanceDetails", {}).get("EC2InstanceDetails", {})

        print(f"\n  [{i}] {spec.get('InstanceType', 'N/A')} in {spec.get('Region', 'N/A')}")
        print(f"      Platform:     {spec.get('Platform', 'Linux')}")
        print(f"      Recommended:  {details.get('RecommendedNumberOfInstancesToPurchase', 'N/A')} instances")
        print(f"      Monthly Save: ${float(details.get('EstimatedMonthlySavings', 0)):,.2f}")
        print(f"      Upfront Cost: ${float(details.get('UpfrontCost', 0)):,.2f}")

    result = {
        "period":              {"start": start, "end": end},
        "total_on_demand_usd": round(total_cost, 2),
        "instance_breakdown":  {k: round(v, 2) for k, v in instance_costs.items()},
        "savings_plan": {
            "monthly_savings_usd": round(float(sp_summary.get("EstimatedMonthlySavingsAmount", 0)), 2),
            "savings_pct":         sp_summary.get("EstimatedSavingsPercentage", "N/A"),
            "hourly_commitment":   round(float(sp_summary.get("HourlyCommitmentToPurchase", 0)), 4),
        } if sp_summary else {},
    }

    # Save JSON report
    with open("cost_report.json", "w") as f:
        json.dump(result, f, indent=2)
    print("\nFull report saved to cost_report.json")

    return result


if __name__ == "__main__":
    analyze_and_recommend_savings(lookback_days=30)

ce = boto3.client("ce", region_name="us-east-1")

end   = datetime.utcnow().strftime("%Y-%m-%d")
start = (datetime.utcnow() - timedelta(days=lookback_days)).strftime("%Y-%m-%d")

cost_response = ce.get_cost_and_usage(
    TimePeriod={"Start": start, "End": end},
    Granularity="MONTHLY",
    Filter={
        "And": [
            {"Dimensions": {"Key": "SERVICE",       "Values": ["Amazon Elastic Compute Cloud - Compute"]}},
            {"Dimensions": {"Key": "PURCHASE_TYPE",  "Values": ["On-Demand"]}},
        ]
    },
    Metrics=["BlendedCost", "UsageQuantity"],
    GroupBy=[{"Type": "DIMENSION", "Key": "INSTANCE_TYPE"}],
)

for result in cost_response["ResultsByTime"]:
    for group in result["Groups"]:
        instance_type = group["Keys"][0]
        cost = float(group["Metrics"]["BlendedCost"]["Amount"])
        instance_costs[instance_type] = instance_costs.get(instance_type, 0) + cost

sp_response = ce.get_savings_plans_purchase_recommendation(
    SavingsPlansType="COMPUTE_SP",
    TermInYears="ONE_YEAR",
    PaymentOption="NO_UPFRONT",
    LookbackPeriodInDays="THIRTY_DAYS",
)

sp_summary = sp_response.get(
    "SavingsPlansPurchaseRecommendation", {}
).get("SavingsPlansPurchaseRecommendationSummary", {})

for itype, cost in sorted(instance_costs.items(), key=lambda x: -x[1]):
    bar = "█" * min(int(cost / total_cost * 30), 30)
    print(f"  {itype:<20} ${cost:>10.2f}  {bar}")

ri_response = ce.get_reservation_purchase_recommendation(
    Service="Amazon EC2",
    TermInYears="ONE_YEAR",
    PaymentOption="NO_UPFRONT",
    LookbackPeriodInDays="THIRTY_DAYS",
)