pip install anthropic boto3 python-dotenv

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ce:GetCostAndUsage",
        "ce:GetCostForecast",
        "ce:GetAnomalies"
      ],
      "Resource": "*"
    }
  ]
}

import boto3
from datetime import datetime, timedelta
from typing import dict
import json
 
 
def get_cost_by_service(days: int = 30) -> dict:
    """
    Pull daily AWS costs broken down by service for the past N days.
    Returns structured data ready for analysis.
    """
    client = boto3.client("ce", region_name="us-east-1")
 
    end_date = datetime.now().strftime("%Y-%m-%d")
    start_date = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
 
    response = client.get_cost_and_usage(
        TimePeriod={
            "Start": start_date,
            "End": end_date
        },
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[
            {"Type": "DIMENSION", "Key": "SERVICE"}
        ]
    )
 
    # Process into a clean structure
    daily_costs = {}
    service_totals = {}
 
    for result in response["ResultsByTime"]:
        date = result["TimePeriod"]["Start"]
        daily_costs[date] = {}
 
        for group in result["Groups"]:
            service = group["Keys"][0]
            cost = float(group["Metrics"]["UnblendedCost"]["Amount"])
 
            if cost > 0.01:  # Filter out negligible costs
                daily_costs[date][service] = round(cost, 4)
                service_totals[service] = service_totals.get(service, 0) + cost
 
    # Sort services by total cost
    top_services = sorted(
        service_totals.items(),
        key=lambda x: x[1],
        reverse=True
    )[:15]  # Top 15 services
 
    return {
        "period": {"start": start_date, "end": end_date},
        "daily_costs": daily_costs,
        "service_totals": dict(top_services),
        "total_spend": sum(service_totals.values())
    }
 
 
def calculate_anomaly_signals(cost_data: dict) -> dict:
    """
    Calculate basic anomaly signals before sending to Claude.
    This gives Claude better context for its analysis.
    """
    daily = cost_data["daily_costs"]
    dates = sorted(daily.keys())
 
    if len(dates) < 7:
        return {"error": "Need at least 7 days of data"}
 
    # Calculate per-service daily averages and recent vs baseline
    service_signals = {}
 
    # Get all services
    all_services = set()
    for day_costs in daily.values():
        all_services.update(day_costs.keys())
 
    for service in all_services:
        # Get cost per day for this service
        costs_by_date = [daily.get(d, {}).get(service, 0) for d in dates]
 
        if max(costs_by_date) < 0.1:
            continue  # Skip negligible services
 
        # Baseline: first 60% of period
        baseline_end = int(len(costs_by_date) * 0.6)
        baseline_costs = costs_by_date[:baseline_end]
        recent_costs = costs_by_date[baseline_end:]
 
        baseline_avg = sum(baseline_costs) / len(baseline_costs) if baseline_costs else 0
        recent_avg = sum(recent_costs) / len(recent_costs) if recent_costs else 0
 
        if baseline_avg > 0:
            change_pct = ((recent_avg - baseline_avg) / baseline_avg) * 100
        else:
            change_pct = 100 if recent_avg > 0 else 0
 
        service_signals[service] = {
            "baseline_daily_avg": round(baseline_avg, 4),
            "recent_daily_avg": round(recent_avg, 4),
            "change_pct": round(change_pct, 1),
            "max_single_day": round(max(costs_by_date), 4),
            "is_anomalous": abs(change_pct) > 30 and recent_avg > 1.0
        }
 
    return service_signals

import anthropic
import os
from dotenv import load_dotenv
 
load_dotenv()
 
 
def analyze_costs_with_claude(cost_data: dict, anomaly_signals: dict) -> str:
    """
    Send cost data to Claude for intelligent anomaly analysis.
    """
    client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
    # Build summary of anomalous services
    anomalous = {
        svc: signals for svc, signals in anomaly_signals.items()
        if signals.get("is_anomalous")
    }
 
    # Build the analysis prompt
    prompt = f"""You are an AWS FinOps expert analyzing cost anomalies.
 
## AWS Cost Summary
- Analysis period: {cost_data["period"]["start"]} to {cost_data["period"]["end"]}
- Total spend this period: ${cost_data["total_spend"]:.2f}
 
## Top Services by Total Cost
{json.dumps(cost_data["service_totals"], indent=2)}
 
## Anomaly Signals (services with >30% change from baseline)
{json.dumps(anomalous, indent=2) if anomalous else "No significant anomalies detected."}
 
## All Service Signals
{json.dumps(anomaly_signals, indent=2)}
 
Based on this data, provide:
 
1. **Anomaly Summary**: Which services show unusual cost patterns? Be specific about the dollar amounts and percentage changes.
 
2. **Likely Causes**: For each anomalous service, what are the most common reasons for this type of cost increase? Reference AWS-specific causes (e.g., "EC2: forgotten instance, wrong instance type, missing auto-scaling schedule").
 
3. **Investigation Commands**: Exact AWS CLI commands to investigate each anomaly. For example:
   - For EC2 spikes: how to list running instances by cost
   - For S3 spikes: how to find buckets with unexpected storage
   - For Data Transfer spikes: how to identify the source
 
4. **Remediation Steps**: Specific actions to reduce costs, with estimated savings.
 
5. **Cost Forecast Risk**: Based on the recent trend, what is the projected monthly spend if no action is taken?
 
Keep the response practical and actionable. Skip services with no anomalies."""
 
    message = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return message.content[0].text
 
 
def generate_cost_report():
    """Main entry point — pull data, analyze, print report."""
    print("Fetching AWS cost data from Cost Explorer...")
    cost_data = get_cost_by_service(days=30)
 
    print(f"Total spend (30 days): ${cost_data['total_spend']:.2f}")
    print("Calculating anomaly signals...")
 
    anomaly_signals = calculate_anomaly_signals(cost_data)
 
    anomalous_count = sum(
        1 for s in anomaly_signals.values()
        if s.get("is_anomalous")
    )
    print(f"Found {anomalous_count} potentially anomalous services")
 
    print("Sending to Claude API for analysis...")
    analysis = analyze_costs_with_claude(cost_data, anomaly_signals)
 
    print("\n" + "="*60)
    print("AWS COST ANOMALY REPORT")
    print("="*60)
    print(analysis)
    print("="*60)
 
    return analysis
 
 
if __name__ == "__main__":
    generate_cost_report()

# Check S3 request metrics (if enabled)
aws cloudwatch get-metric-statistics \
  --namespace AWS/S3 \
  --metric-name AllRequests \
  --dimensions Name=BucketName,Value=YOUR-BUCKET \
  --start-time 2026-06-15T00:00:00Z \
  --end-time 2026-07-04T00:00:00Z \
  --period 86400 \
  --statistics Sum
 
# Check data transfer out
aws s3api list-buckets --query 'Buckets[].Name' --output text | \
  xargs -I {} aws cloudwatch get-metric-statistics \
    --namespace AWS/S3 --metric-name BytesDownloaded \
    --dimensions Name=BucketName,Value={}

# .github/workflows/cost-check.yml
on:
  schedule:
    - cron: "0 9 * * MON"  # Every Monday at 9 AM
 
jobs:
  cost-anomaly-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_COST_ROLE_ARN }}
          aws-region: us-east-1
      - run: pip install anthropic boto3
      - run: python cost_analyzer.py
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}