Cost Management

# cost_management/analytics.py
import boto3
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class CostMetric:
    date: str
    service: str
    amount: float
    currency: str = "USD"

class CostAnalytics:
    """
    AWS Cost Explorer integration for detailed
    cost analytics and forecasting.
    """
    
    def __init__(self):
        self.ce = boto3.client('ce')
        self.org = boto3.client('organizations')
    
    def get_monthly_costs(
        self,
        months: int = 6,
        granularity: str = "MONTHLY"
    ) -> List[Dict]:
        """Get cost breakdown by month."""
        
        end_date = datetime.now()
        start_date = end_date - timedelta(days=months * 30)
        
        response = self.ce.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.strftime('%Y-%m-%d'),
                'End': end_date.strftime('%Y-%m-%d')
            },
            Granularity=granularity,
            Metrics=['UnblendedCost', 'UsageQuantity'],
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'SERVICE'}
            ]
        )
        
        return self._parse_cost_response(response)
    
    def get_cost_forecast(
        self,
        days_ahead: int = 30
    ) -> Dict:
        """Forecast future costs based on trends."""
        
        end_date = datetime.now() + timedelta(days=days_ahead)
        
        response = self.ce.get_cost_forecast(
            TimePeriod={
                'Start': datetime.now().strftime('%Y-%m-%d'),
                'End': end_date.strftime('%Y-%m-%d')
            },
            Metric='UNBLENDED_COST',
            Granularity='MONTHLY',
            PredictionIntervalLevel=80
        )
        
        return {
            "forecast_amount": float(response['Total']['Amount']),
            "lower_bound": float(response['ForecastResultsByTime'][0]['MeanValue']) * 0.9,
            "upper_bound": float(response['ForecastResultsByTime'][0]['MeanValue']) * 1.1,
            "confidence": 80
        }
    
    def get_cost_anomalies(
        self,
        threshold_percentage: float = 20
    ) -> List[Dict]:
        """Detect unusual cost spikes."""
        
        response = self.ce.get_anomalies(
            DateInterval={
                'StartDate': (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d'),
                'EndDate': datetime.now().strftime('%Y-%m-%d')
            },
            TotalImpact={
                'NumericOperator': 'GREATER_THAN',
                'StartValue': 100
            }
        )
        
        anomalies = []
        for anomaly in response.get('Anomalies', []):
            impact = float(anomaly['Impact']['TotalImpact'])
            anomalies.append({
                "id": anomaly['AnomalyId'],
                "date": anomaly['AnomalyStartDate'],
                "service": anomaly['RootCauses'][0]['Service'] if anomaly['RootCauses'] else "Unknown",
                "impact": impact,
                "status": anomaly['AnomalyScore']['CurrentScore']
            })
        
        return anomalies