Build an AI Infrastructure Cost Forecaster with Claude API and AWS Cost Explorer

pip install anthropic boto3 pandas

import anthropic
import boto3
import json
import pandas as pd
from datetime import datetime, timedelta
 
client = anthropic.Anthropic()
ce = boto3.client("ce", region_name="us-east-1")
 
 
def get_monthly_costs(months_back: int = 6) -> pd.DataFrame:
    """Get monthly cost breakdown for the past N months."""
    end = datetime.today().replace(day=1).strftime("%Y-%m-%d")
    start = (datetime.today() - timedelta(days=months_back * 30)).replace(day=1).strftime("%Y-%m-%d")
 
    response = ce.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="MONTHLY",
        Metrics=["BlendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}]
    )
 
    records = []
    for result in response["ResultsByTime"]:
        period = result["TimePeriod"]["Start"]
        for group in result["Groups"]:
            records.append({
                "month": period,
                "service": group["Keys"][0],
                "cost": float(group["Metrics"]["BlendedCost"]["Amount"])
            })
 
    return pd.DataFrame(records)
 
 
def get_growth_context() -> dict:
    """Get context about usage trends."""
    # Get EC2 running instances trend
    ec2 = boto3.client("ec2", region_name="ap-south-1")
    instances = ec2.describe_instances(
        Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
    )
    instance_count = sum(len(r["Instances"]) for r in instances["Reservations"])
 
    return {
        "current_ec2_count": instance_count,
        "current_month": datetime.today().strftime("%B %Y")
    }
 
 
def forecast_with_claude(cost_df: pd.DataFrame, context: dict, forecast_months: int = 3) -> str:
    """Use Claude to forecast costs based on historical trends."""
 
    # Pivot to monthly totals per service
    monthly_totals = cost_df.groupby("month")["cost"].sum().to_dict()
    top_services = cost_df.groupby("service")["cost"].sum().nlargest(10).to_dict()
 
    prompt = f"""You are a FinOps expert analyzing AWS cloud cost trends.
 
## Historical Monthly Costs (last 6 months)
{json.dumps(monthly_totals, indent=2)}
 
## Top 10 Services by Total Spend
{json.dumps({k: round(v, 2) for k, v in top_services.items()}, indent=2)}
 
## Current Environment Context
- Running EC2 instances: {context['current_ec2_count']}
- Current month: {context['current_month']}
 
Provide a {forecast_months}-month cost forecast with:
 
1. **Monthly Cost Forecast**: Projected total for each of the next {forecast_months} months
2. **Growth Rate**: Calculated month-over-month trend percentage
3. **Key Cost Drivers**: Which services are growing fastest and why
4. **Risk Scenarios**: 
   - Base case (current growth continues)
   - High case (20% above trend — business growth, incidents)
   - Low case (optimization actions taken)
5. **Top 3 Optimization Actions**: Specific changes that would have the biggest cost impact
6. **Break-even Estimate**: When Reserved Instances become cheaper than On-Demand for your top EC2 spend
 
Format as a structured report with exact dollar amounts."""
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return response.content[0].text
 
 
def generate_cost_forecast_report():
    print("Fetching cost history...")
    cost_df = get_monthly_costs(months_back=6)
 
    print("Getting environment context...")
    context = get_growth_context()
 
    print("Generating AI forecast...")
    forecast = forecast_with_claude(cost_df, context)
 
    print("\n" + "="*60)
    print("AWS COST FORECAST REPORT")
    print("="*60)
    print(forecast)
 
    # Save to file
    report_path = f"cost_forecast_{datetime.today().strftime('%Y-%m-%d')}.md"
    with open(report_path, "w") as f:
        f.write(f"# AWS Cost Forecast Report\n")
        f.write(f"Generated: {datetime.today().strftime('%Y-%m-%d %H:%M UTC')}\n\n")
        f.write(forecast)
 
    print(f"\nReport saved: {report_path}")
 
 
if __name__ == "__main__":
    generate_cost_forecast_report()

import resend
 
resend.api_key = "your-resend-api-key"
 
def send_forecast_email(forecast_text: str, recipient: str):
    resend.Emails.send({
        "from": "costs@devopsboys.com",
        "to": recipient,
        "subject": f"AWS Cost Forecast - {datetime.today().strftime('%B %Y')}",
        "html": f"<pre>{forecast_text}</pre>"
    })

# .github/workflows/cost-forecast.yml
name: Monthly Cost Forecast
 
on:
  schedule:
    - cron: '0 9 1 * *'    # 1st of every month at 9 AM
 
jobs:
  forecast:
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ vars.AWS_ROLE_ARN }}
          aws-region: us-east-1
      - name: Run forecast
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          pip install anthropic boto3 pandas
          python cost_forecaster.py