COST05-BP05 - Perform cost analysis for different usage over time

import boto3
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
import matplotlib.pyplot as plt
from scipy import stats

@dataclass
class UsageScenario:
    name: str
    description: str
    base_usage: float
    growth_rate: float  # Annual growth rate
    seasonality_factor: float  # Seasonal variation (0-1)
    burst_factor: float  # Burst usage multiplier
    burst_frequency: float  # Frequency of bursts (0-1)

@dataclass
class CostModel:
    service_name: str
    fixed_cost_monthly: float
    variable_cost_per_unit: float
    commitment_discount: float  # Discount for commitments
    scaling_efficiency: float  # Cost efficiency as scale increases
    setup_cost: float
    migration_cost: float

class TimeBasedCostAnalyzer:
    def __init__(self):
        self.ce_client = boto3.client('ce')
        self.cloudwatch = boto3.client('cloudwatch')
        self.pricing = boto3.client('pricing', region_name='us-east-1')
        
    def analyze_cost_over_time(self, scenarios: List[UsageScenario], 
                              cost_models: List[CostModel], 
                              analysis_months: int = 36) -> Dict:
        """Perform comprehensive cost analysis over time for different scenarios"""
        
        analysis_results = {
            'scenarios': {},
            'service_comparisons': {},
            'recommendations': [],
            'break_even_analysis': {},
            'total_cost_projections': {}
        }
        
        # Analyze each scenario
        for scenario in scenarios:
            scenario_results = {}
            
            # Generate usage projections for this scenario
            usage_projections = self.generate_usage_projections(scenario, analysis_months)
            
            # Calculate costs for each service model
            for cost_model in cost_models:
                service_costs = self.calculate_service_costs_over_time(
                    usage_projections, cost_model, analysis_months
                )
                scenario_results[cost_model.service_name] = service_costs
            
            analysis_results['scenarios'][scenario.name] = scenario_results
        
        # Perform cross-scenario and cross-service analysis
        analysis_results['service_comparisons'] = self.compare_services_across_scenarios(
            analysis_results['scenarios'], scenarios, cost_models
        )
        
        analysis_results['break_even_analysis'] = self.perform_break_even_analysis(
            scenarios, cost_models, analysis_months
        )
        
        analysis_results['recommendations'] = self.generate_time_based_recommendations(
            analysis_results['scenarios'], analysis_results['break_even_analysis']
        )
        
        return analysis_results
    
    def generate_usage_projections(self, scenario: UsageScenario, months: int) -> List[float]:
        """Generate usage projections based on scenario parameters"""
        
        projections = []
        
        for month in range(months):
            # Base growth calculation
            growth_multiplier = (1 + scenario.growth_rate / 12) ** month
            base_usage = scenario.base_usage * growth_multiplier
            
            # Apply seasonality (sine wave with 12-month period)
            seasonal_adjustment = 1 + scenario.seasonality_factor * np.sin(2 * np.pi * month / 12)
            seasonal_usage = base_usage * seasonal_adjustment
            
            # Apply burst patterns
            if np.random.random() < scenario.burst_frequency:
                burst_usage = seasonal_usage * scenario.burst_factor
            else:
                burst_usage = seasonal_usage
            
            projections.append(max(0, burst_usage))
        
        return projections
    
    def calculate_service_costs_over_time(self, usage_projections: List[float], 
                                        cost_model: CostModel, months: int) -> Dict:
        """Calculate costs for a specific service over time"""
        
        monthly_costs = []
        cumulative_costs = []
        total_cost = cost_model.setup_cost  # Start with setup cost
        
        for month, usage in enumerate(usage_projections):
            # Calculate monthly cost
            fixed_cost = cost_model.fixed_cost_monthly
            
            # Variable cost with scaling efficiency
            scaling_factor = max(0.5, 1 - (usage / 10000) * cost_model.scaling_efficiency)
            variable_cost = usage * cost_model.variable_cost_per_unit * scaling_factor
            
            # Apply commitment discounts (assume commitment after 3 months)
            if month >= 3:
                commitment_discount = cost_model.commitment_discount
            else:
                commitment_discount = 0
            
            monthly_cost = (fixed_cost + variable_cost) * (1 - commitment_discount)
            monthly_costs.append(monthly_cost)
            
            total_cost += monthly_cost
            cumulative_costs.append(total_cost)
        
        return {
            'monthly_costs': monthly_costs,
            'cumulative_costs': cumulative_costs,
            'total_cost': total_cost,
            'average_monthly_cost': np.mean(monthly_costs),
            'cost_trend': self.calculate_cost_trend(monthly_costs),
            'cost_volatility': np.std(monthly_costs)
        }
    
    def compare_services_across_scenarios(self, scenario_results: Dict, 
                                        scenarios: List[UsageScenario], 
                                        cost_models: List[CostModel]) -> Dict:
        """Compare services across different scenarios"""
        
        comparisons = {}
        
        for scenario in scenarios:
            scenario_name = scenario.name
            scenario_data = scenario_results[scenario_name]
            
            # Find the most cost-effective service for this scenario
            service_costs = {
                service: data['total_cost'] 
                for service, data in scenario_data.items()
            }
            
            best_service = min(service_costs.keys(), key=lambda x: service_costs[x])
            worst_service = max(service_costs.keys(), key=lambda x: service_costs[x])
            
            cost_savings = service_costs[worst_service] - service_costs[best_service]
            savings_percentage = (cost_savings / service_costs[worst_service]) * 100
            
            comparisons[scenario_name] = {
                'best_service': best_service,
                'worst_service': worst_service,
                'cost_savings': cost_savings,
                'savings_percentage': savings_percentage,
                'service_rankings': sorted(service_costs.items(), key=lambda x: x[1]),
                'cost_differences': self.calculate_cost_differences(service_costs)
            }
        
        return comparisons
    
    def perform_break_even_analysis(self, scenarios: List[UsageScenario], 
                                   cost_models: List[CostModel], months: int) -> Dict:
        """Perform break-even analysis between different services"""
        
        break_even_results = {}
        
        # Compare each pair of services
        for i, model1 in enumerate(cost_models):
            for j, model2 in enumerate(cost_models[i+1:], i+1):
                comparison_key = f"{model1.service_name}_vs_{model2.service_name}"
                
                # Find break-even points for different scenarios
                break_even_points = {}
                
                for scenario in scenarios:
                    break_even_usage = self.find_break_even_usage(model1, model2, scenario)
                    break_even_time = self.find_break_even_time(model1, model2, scenario, months)
                    
                    break_even_points[scenario.name] = {
                        'break_even_usage': break_even_usage,
                        'break_even_time_months': break_even_time,
                        'recommendation': self.generate_break_even_recommendation(
                            model1, model2, break_even_usage, break_even_time
                        )
                    }
                
                break_even_results[comparison_key] = break_even_points
        
        return break_even_results
    
    def find_break_even_usage(self, model1: CostModel, model2: CostModel, 
                             scenario: UsageScenario) -> Optional[float]:
        """Find the usage level where two services have equal cost"""
        
        # Simplified break-even calculation
        # In practice, this would be more complex considering all factors
        
        if model1.variable_cost_per_unit == model2.variable_cost_per_unit:
            return None  # No break-even point if variable costs are equal
        
        fixed_diff = model2.fixed_cost_monthly - model1.fixed_cost_monthly
        variable_diff = model1.variable_cost_per_unit - model2.variable_cost_per_unit
        
        if variable_diff == 0:
            return None
        
        break_even_usage = fixed_diff / variable_diff
        
        return max(0, break_even_usage) if break_even_usage > 0 else None
    
    def find_break_even_time(self, model1: CostModel, model2: CostModel, 
                            scenario: UsageScenario, max_months: int) -> Optional[int]:
        """Find the time when cumulative costs of two services are equal"""
        
        usage_projections = self.generate_usage_projections(scenario, max_months)
        
        costs1 = self.calculate_service_costs_over_time(usage_projections, model1, max_months)
        costs2 = self.calculate_service_costs_over_time(usage_projections, model2, max_months)
        
        cumulative1 = costs1['cumulative_costs']
        cumulative2 = costs2['cumulative_costs']
        
        # Find crossover point
        for month in range(1, max_months):
            if cumulative1[month-1] <= cumulative2[month-1] and cumulative1[month] > cumulative2[month]:
                return month
            elif cumulative2[month-1] <= cumulative1[month-1] and cumulative2[month] > cumulative1[month]:
                return month
        
        return None
    
    def generate_time_based_recommendations(self, scenario_results: Dict, 
                                          break_even_analysis: Dict) -> List[Dict]:
        """Generate recommendations based on time-based cost analysis"""
        
        recommendations = []
        
        # Analyze each scenario for recommendations
        for scenario_name, scenario_data in scenario_results.items():
            service_costs = {
                service: data['total_cost'] 
                for service, data in scenario_data.items()
            }
            
            best_service = min(service_costs.keys(), key=lambda x: service_costs[x])
            
            # Generate scenario-specific recommendation
            recommendation = {
                'scenario': scenario_name,
                'recommended_service': best_service,
                'rationale': f"Lowest total cost over analysis period: ${service_costs[best_service]:,.2f}",
                'cost_savings': max(service_costs.values()) - service_costs[best_service],
                'confidence_level': self.calculate_recommendation_confidence(scenario_data),
                'considerations': self.generate_considerations(scenario_name, scenario_data, break_even_analysis)
            }
            
            recommendations.append(recommendation)
        
        # Generate cross-scenario recommendations
        cross_scenario_rec = self.generate_cross_scenario_recommendations(scenario_results)
        recommendations.extend(cross_scenario_rec)
        
        return recommendations
    
    def calculate_cost_trend(self, monthly_costs: List[float]) -> str:
        """Calculate the trend in monthly costs"""
        
        if len(monthly_costs) < 2:
            return "insufficient_data"
        
        # Calculate linear regression slope
        x = np.arange(len(monthly_costs))
        slope, _, r_value, _, _ = stats.linregress(x, monthly_costs)
        
        if abs(r_value) < 0.3:
            return "stable"
        elif slope > 0:
            return "increasing"
        else:
            return "decreasing"
    
    def create_cost_visualization(self, analysis_results: Dict, output_path: str):
        """Create visualizations for cost analysis results"""
        
        fig, axes = plt.subplots(2, 2, figsize=(15, 12))
        
        # Plot 1: Cost comparison across scenarios
        scenarios = list(analysis_results['scenarios'].keys())
        services = list(analysis_results['scenarios'][scenarios[0]].keys())
        
        scenario_costs = {}
        for scenario in scenarios:
            scenario_costs[scenario] = [
                analysis_results['scenarios'][scenario][service]['total_cost']
                for service in services
            ]
        
        x = np.arange(len(services))
        width = 0.35
        
        for i, scenario in enumerate(scenarios):
            axes[0, 0].bar(x + i * width, scenario_costs[scenario], width, label=scenario)
        
        axes[0, 0].set_xlabel('Services')
        axes[0, 0].set_ylabel('Total Cost ($)')
        axes[0, 0].set_title('Total Cost Comparison Across Scenarios')
        axes[0, 0].set_xticks(x + width / 2)
        axes[0, 0].set_xticklabels(services)
        axes[0, 0].legend()
        
        # Plot 2: Monthly cost trends
        for scenario in scenarios:
            for service in services:
                monthly_costs = analysis_results['scenarios'][scenario][service]['monthly_costs']
                axes[0, 1].plot(monthly_costs, label=f"{scenario}-{service}")
        
        axes[0, 1].set_xlabel('Month')
        axes[0, 1].set_ylabel('Monthly Cost ($)')
        axes[0, 1].set_title('Monthly Cost Trends')
        axes[0, 1].legend()
        
        # Plot 3: Cumulative cost comparison
        for scenario in scenarios:
            for service in services:
                cumulative_costs = analysis_results['scenarios'][scenario][service]['cumulative_costs']
                axes[1, 0].plot(cumulative_costs, label=f"{scenario}-{service}")
        
        axes[1, 0].set_xlabel('Month')
        axes[1, 0].set_ylabel('Cumulative Cost ($)')
        axes[1, 0].set_title('Cumulative Cost Comparison')
        axes[1, 0].legend()
        
        # Plot 4: Cost savings potential
        savings_data = []
        labels = []
        
        for scenario, comparison in analysis_results['service_comparisons'].items():
            savings_data.append(comparison['savings_percentage'])
            labels.append(scenario)
        
        axes[1, 1].bar(labels, savings_data)
        axes[1, 1].set_xlabel('Scenario')
        axes[1, 1].set_ylabel('Potential Savings (%)')
        axes[1, 1].set_title('Cost Savings Potential by Scenario')
        
        plt.tight_layout()
        plt.savefig(output_path)
        plt.close()

Cost_Analysis_Configuration:
  analysis_id: "COST-ANALYSIS-2024-Q1"
  analysis_period_months: 36
  
  usage_scenarios:
    steady_growth:
      name: "Steady Growth"
      description: "Consistent 20% annual growth"
      base_usage: 1000
      growth_rate: 0.20
      seasonality_factor: 0.1
      burst_factor: 1.2
      burst_frequency: 0.1
      
    seasonal_business:
      name: "Seasonal Business"
      description: "High seasonality with holiday peaks"
      base_usage: 800
      growth_rate: 0.15
      seasonality_factor: 0.4
      burst_factor: 2.0
      burst_frequency: 0.2
      
    startup_growth:
      name: "Startup Growth"
      description: "Rapid initial growth then stabilization"
      base_usage: 100
      growth_rate: 1.0  # 100% first year, then decreasing
      seasonality_factor: 0.05
      burst_factor: 3.0
      burst_frequency: 0.3
      
  service_models:
    ec2_on_demand:
      service_name: "EC2 On-Demand"
      fixed_cost_monthly: 0
      variable_cost_per_unit: 0.10
      commitment_discount: 0
      scaling_efficiency: 0.1
      setup_cost: 500
      migration_cost: 0
      
    ec2_reserved:
      service_name: "EC2 Reserved"
      fixed_cost_monthly: 50
      variable_cost_per_unit: 0.06
      commitment_discount: 0.4
      scaling_efficiency: 0.1
      setup_cost: 500
      migration_cost: 200
      
    lambda_serverless:
      service_name: "Lambda Serverless"
      fixed_cost_monthly: 0
      variable_cost_per_unit: 0.000016
      commitment_discount: 0
      scaling_efficiency: 0.3
      setup_cost: 100
      migration_cost: 1000
      
  analysis_objectives:
    - "Identify most cost-effective service for each scenario"
    - "Determine break-even points between services"
    - "Optimize for 3-year total cost of ownership"
    - "Account for migration and setup costs"
    - "Consider operational complexity and risk"

def create_time_based_decision_framework():
    """Create framework for time-based service selection decisions"""
    
    framework = {
        'decision_criteria': {
            'short_term': {
                'time_horizon': '0-6 months',
                'primary_factors': ['setup_cost', 'migration_effort', 'immediate_savings'],
                'weight_cost': 0.4,
                'weight_speed': 0.4,
                'weight_risk': 0.2
            },
            'medium_term': {
                'time_horizon': '6-24 months',
                'primary_factors': ['operational_cost', 'scalability', 'flexibility'],
                'weight_cost': 0.5,
                'weight_performance': 0.3,
                'weight_flexibility': 0.2
            },
            'long_term': {
                'time_horizon': '24+ months',
                'primary_factors': ['total_cost_ownership', 'strategic_alignment', 'innovation'],
                'weight_cost': 0.3,
                'weight_strategy': 0.4,
                'weight_innovation': 0.3
            }
        },
        
        'usage_thresholds': {
            'low_usage': {
                'threshold': 'usage < 1000 units/month',
                'recommended_approach': 'pay_as_you_go',
                'considerations': ['minimal_commitment', 'flexibility', 'low_setup_cost']
            },
            'medium_usage': {
                'threshold': '1000 <= usage < 10000 units/month',
                'recommended_approach': 'hybrid_commitment',
                'considerations': ['partial_commitment', 'reserved_capacity', 'cost_optimization']
            },
            'high_usage': {
                'threshold': 'usage >= 10000 units/month',
                'recommended_approach': 'full_commitment',
                'considerations': ['maximum_commitment', 'enterprise_discounts', 'dedicated_support']
            }
        },
        
        'growth_patterns': {
            'stable': {
                'growth_rate': '< 20% annually',
                'recommendation': 'Reserved capacity with some on-demand buffer',
                'commitment_level': 'high'
            },
            'growing': {
                'growth_rate': '20-100% annually',
                'recommendation': 'Mixed approach with increasing reserved capacity',
                'commitment_level': 'medium'
            },
            'rapid_growth': {
                'growth_rate': '> 100% annually',
                'recommendation': 'Flexible, scalable solutions with minimal commitment',
                'commitment_level': 'low'
            }
        }
    }
    
    return framework