1import os
2import boto3
3from datetime import datetime
4import torch
5import torch.distributed as dist
6
7class CheckpointManager:
8    def __init__(self, s3_bucket, checkpoint_interval=3600):  # Checkpoint every hour
9        self.s3_bucket = s3_bucket
10        self.checkpoint_interval = checkpoint_interval
11        self.last_checkpoint = datetime.now()
12        self.s3_client = boto3.client('s3')
13        
14    def save_checkpoint(self, model, optimizer, epoch, loss, path='./checkpoints'):
15        """Save checkpoint locally and to S3."""
16        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
17        checkpoint_path = os.path.join(path, f'checkpoint_{timestamp}.pt')
18        
19        # Save locally
20        torch.save({
21            'epoch': epoch,
22            'model_state_dict': model.state_dict(),
23            'optimizer_state_dict': optimizer.state_dict(),
24            'loss': loss,
25            'timestamp': timestamp
26        }, checkpoint_path)
27        
28        # Upload to S3 for durability
29        s3_key = f'checkpoints/{timestamp}.pt'
30        self.s3_client.upload_file(checkpoint_path, self.s3_bucket, s3_key)
31        
32        print(f"Checkpoint saved to {checkpoint_path} and S3://{self.s3_bucket}/{s3_key}")
33        self.last_checkpoint = datetime.now()
34        
35    def should_checkpoint(self):
36        """Determine if it's time for a checkpoint."""
37        elapsed = (datetime.now() - self.last_checkpoint).total_seconds()
38        return elapsed >= self.checkpoint_interval
39        
40    def load_latest_checkpoint(self, path='./checkpoints'):
41        """Load latest checkpoint from S3 or local."""
42        # First check S3 for most recent
43        response = self.s3_client.list_objects_v2(
44            Bucket=self.s3_bucket,
45            Prefix='checkpoints/'
46        )
47        
48        if 'Contents' in response:
49            # Get most recent checkpoint
50            checkpoints = sorted(response['Contents'], key=lambda x: x['Key'], reverse=True)
51            latest_s3_key = checkpoints[0]['Key']
52            
53            # Download from S3
54            local_path = os.path.join(path, os.path.basename(latest_s3_key))
55            self.s3_client.download_file(self.s3_bucket, latest_s3_key, local_path)
56            
57            checkpoint = torch.load(local_path)
58            print(f"Resuming from checkpoint: {latest_s3_key}")
59            return checkpoint
60        
61        return None
62
63# Integration with training loop
64def training_loop_with_checkpoints(model, optimizer, dataloader, checkpoint_manager, epochs=100):
65    """Training loop with automatic checkpointing."""
66    # Try to resume from checkpoint
67    checkpoint = checkpoint_manager.load_latest_checkpoint()
68    start_epoch = 0
69    
70    if checkpoint:
71        model.load_state_dict(checkpoint['model_state_dict'])
72        optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
73        start_epoch = checkpoint['epoch'] + 1
74        print(f"Resuming from epoch {start_epoch}")
75    
76    for epoch in range(start_epoch, epochs):
77        for batch_idx, (data, target) in enumerate(dataloader):
78            # Training step here...
79            loss = train_step(model, optimizer, data, target)
80            
81            # Check if we should checkpoint
82            if checkpoint_manager.should_checkpoint():
83                checkpoint_manager.save_checkpoint(model, optimizer, epoch, loss)
84                
85        # Also checkpoint at end of each epoch
86        checkpoint_manager.save_checkpoint(model, optimizer, epoch, loss)
87        
88        # Handle spot interruption gracefully
89        if is_spot_interruption_imminent():
90            print("Spot interruption detected - saving final checkpoint")
91            checkpoint_manager.save_checkpoint(model, optimizer, epoch, loss)
92            break
93

1import boto3
2
3def create_spot_fleet_configuration():
4    """Create diversified spot fleet for training."""
5    ec2 = boto3.client('ec2')
6    
7    # Diversified across instance families and AZs
8    launch_specifications = [
9        {
10            'InstanceType': 'p4d.24xlarge',  # 8× A100
11            'WeightedCapacity': 1,
12            'AvailabilityZone': 'us-east-1a'
13        },
14        {
15            'InstanceType': 'g5.48xlarge',   # 8× A10G
16            'WeightedCapacity': ...
17
181,
19            'AvailabilityZone': 'us-east0, 1b'
20        },
21        {
22            'InstanceType': 'p3.16xlarge',   # 8× V100
23            'WeightedCapacity': 1,
24            'AvailabilityZone': 'us-east-1c'
25        }
26    ]
27    
28    response = ec2.request_spot_fleet(
29        SpotFleetRequestConfig={
30            'IamFleetRole': 'arn:aws:iam::123456789012:role/SpotFleetRole',
31            'TargetCapacity': 8,  # Total capacity needed
32            'AllocationStrategy': 'capacity-optimized',  # AWS chooses best availability
33            'LaunchSpecifications': launch_specifications,
34            'Type': 'maintain',  # Maintain target capacity
35            'ValidUntil': datetime(2026, 6, 27),  # Max duration
36            'SpotMaintenanceStrategies': {
37                'CapacityRebalance': {
38                    'ReplacementStrategy': 'launch-before-terminate'
39                }
40            }
41        }
42    )
43    
44    return response['SpotFleetRequestId']
45

1import boto3
2import json
3from datetime import datetime, time
4
5class InferenceAutoscaler:
6    def __init__(self, endpoint_name, min_capacity=0, max_capacity=10):
7        self.sagemaker = boto3.client('sagemaker')
8        self.cloudwatch = boto3.client('cloudwatch')
9        self.endpoint_name = endpoint_name
10        self.min_capacity = min_capacity
11        self.max_capacity = max_capacity
12        
13    def create_scaling_policies(self):
14        """Create target tracking and scheduled scaling policies."""
15        
16        # Target tracking based on CPU utilization
17        self.sagemaker.put_scaling_policy(
18            EndpointName=self.endpoint_name,
19            PolicyName='TargetTrackingScaling',
20            PolicyType='TargetTrackingScaling',
21            TargetTrackingScalingPolicyConfiguration={
22                'TargetValue': 70.0,  # 70% CPU utilization target
23                'ScaleInCooldown': 300,  # 5 minutes before scaling in
24                'ScaleOutCooldown': 60,   # 1 minute before scaling out
25                'PredefinedMetricSpecification': {
26                    'PredefinedMetricType': 'SageMakerVariantInvocationsPerInstance'
27                }
28            }
29        )
30        
31        # Scheduled scaling for predictable patterns
32        self.sagemaker.put_scaling_policy(
33            EndpointName=self.endpoint_name,
34            PolicyName='ScheduledScaling',
35            PolicyType='StepScaling',
36            StepScalingPolicyConfiguration={
37                'AdjustmentType': 'ChangeInCapacity',
38                'StepAdjustments': [
39                    {
40                        'MetricIntervalLowerBound': 0,
41                        'MetricIntervalUpperBound': 8,  # 8 AM
42                        'ScalingAdjustment': 2  # Add 2 instances at 8 AM
43                    },
44                    {
45                        'MetricIntervalLowerBound': 18,  # 6 PM
46                        'MetricIntervalUpperBound': 24,
47                        'ScalingAdjustment': -2  # Remove 2 instances at 6 PM
48                    }
49                ],
50                'Cooldown': 300
51            }
52        )
53        
54    def scale_to_zero_during_off_hours(self):
55        """Scale to zero during predictable low-traffic periods."""
56        # Define off-hours (e.g., 10 PM to noi 6 AM)
57        off_hours_start = time(22, 0)  # 10 PM
58        off_hours_end = time(6, 0)     # 6 AM
59        
60        current_time = datetime.now().time()
61        
62        if off_hours_start <= current_time or current_time <= off_hours_end:
63            print("Off-hours detected - scaling to zero")
64            self.sagemaker.update_endpoint(
65                EndpointName=self.endpoint_name,
66                DesiredInferenceUnits=0
67            )
68        else:
69            print("Business hours - maintaining minimum capacity")
70            self.sagemaker.update_endpoint(
71                EndpointName=self.endpoint_name,
72                DesiredInferenceUnits=self.min_capacity
73            )
74    
75    def predictive_scaling_based_on_history(self):
76        """Use CloudWatch metrics to predict future load."""
77        # Get historical invocations
78        response = self.cloudwatch.get_metric_data(
79            MetricDataQueries=[
80                {
81                    'Id': 'invocations',
82                    'MetricStat': {
83                        'Metric': {
84                            'Namespace': 'AWS/SageMaker',
85                            'MetricName': 'Invocations',
86                            'Dimensions': [
87                                {
88                                    'Name': 'EndpointName',
89                                    'Value': self.endpoint_name
90                                }
91                            ]
92                        },
93                        'Period': 3600,  # 1 hour
94                        'Stat': 'Sum',
95                        'Unit': 'Count'
96                    },
97                    'ReturnData': True
98                }
99            ],
100            StartTime=datetime.now() - timedelta(days=7),
101            EndTime=datetime.now()
102        )
103        
104        # Simple prediction: if similar day/hour had high traffic, pre-scale
105        historical_data = response['MetricDataResults'][0]['Values']
106        # Implement your prediction logic here...
107        
108        return predicted_capacity
109

1import boto3
2from datetime import datetime, timedelta
3import pandas as pd
4
5class AITrainingBudgetManager:
6    def __init__(self, budget_name='ai-training-budget'):
7        self.budgets = boto3.client('budgets')
8        self.cost_explorer = boto3.client('ce')
9        self.budget_name = budget_name
10        
11    def create_ai_training_budget(self, monthly_limit=10000):
12        """Create budget with alerts and actions."""
13        self.budgets.create_budget(
14            AccountId='123456789012',
15            Budget={
16                'BudgetName': self.budget_name,
17                'BudgetLimit': {
18                    'Amount': str(monthly_limit),
19                    'Unit': 'USD'
20                },
21                'TimeUnit': 'MONTHLY',
22                'BudgetType': 'COST',
23                'CostTypes': {
24                    'IncludeTax': True,
25                    'IncludeSubscription': True,
26                    'UseBlended': False,
27                    'IncludeRefund': False,
28                    'IncludeCredit': False,
29                    'IncludeUpfront': False,
30                    'IncludeRecurring': False,
31                    'IncludeOtherSubscription': False,
32                    'IncludeSupport': False,
33                    'IncludeDiscount': False,
34                    'UseAmortized': False
35                }
36            },
37            NotificationsWithSubscribers=[
38                {
39                    'Notification': {
40                        'NotificationType': 'ACTUAL',
41                        'ComparisonOperator': 'GREATER_THAN',
42                        'Threshold': 50,  # Alert at 50% of budget
43                        'ThresholdType': 'PERCENTAGE'
44                    },
45                    'Subscribers': [
46                        {
47                            'SubscriptionType': 'EMAIL',
48                            'Address': 'ai-team@company.com'
49                        }
50                    ]
51                },
52                {
53                    'Notification': {
54                        'NotificationType': 'ACTUAL',
55                        'ComparisonOperator': 'GREATER_THAN',
56                        'Threshold': 80,  # Alert at 80% of budget
57                        'ThresholdType': 'PERCENTAGE'
58                    },
59                    'Subscribers': [
60                        {
61                            'SubscriptionType': 'EMAIL',
62                            'Address': 'ai-director@company.com'
63                        }
64                    ]
65                },
66                {
67                    'Notification': {
68                        'NotificationType': 'ACTUAL',
69                        'ComparisonOperator': 'GREATER_THAN',
70                        'Threshold': 100,  # Alert at 100% of budget
71                        'ThresholdType': 'PERCENTAGE'
72                    },
73                    'Subscribers': [
74                        {
75                            'SubscriptionType': 'EMAIL',
76                            'Address': 'finance@company.com'
77                        }
78                    ]
79                }
80            ]
81        )
82        
83    def analyze_training_costs(self):
84        """Break down AI training costs by project and user."""
85        response = self.cost_explorer.get_cost_and_usage(
86            TimePeriod={
87                'Start': (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d'),
88                'End': datetime.now().strftime('%Y-%m-%d')
89            },
90            Granularity='MONTHLY',
91            Metrics=['UnblendedCost'],
92            GroupBy=[
93                {'Type': 'DIMENSION', 'Key': 'SERVICE'},
94                {'Type': 'TAG', 'Key': 'Project'},
95                {'Type': 'TAG', 'Key': 'User'}
96            ]
97        )
98        
99        # Transform to DataFrame for analysis
100        cost_data = []
101        for group in response['ResultsByTime'][0]['Groups']:
102            cost_data.append({
103                'service': group['Keys'][0],
104                'project': group['Keys'][1] if len(group['Keys']) > 1 else 'untagged',
105                'user': group['Keys'][2] if len(group['Keys']) > 2 else 'untagged',
106                'cost': float(group['Metrics']['UnblendedCost']['Amount'])
107            })
108        
109        df = pd.DataFrame(cost_data)
110        
111        # Identify cost outliers
112        outliers = df[df['cost'] > df['cost'].quantile(0.95)]
113        
114        return df, outliers
115    
116    def enforce_cost_policies(self):
117        """Automatically enforce cost policies."""
118        df, outliers = self.analyze_training_costs()
119        
120        for _, row in outliers.iterrows():
121            print(f"Cost alert: {row['user']} spent ${row['cost']:.2f} on {row['project']}")
122            
123            # Example policy: if cost > $1000 and not approved, stop resources
124            if row['cost'] > 1000 and row['project'] not in self.get_approved_projects():
125                self.stop_training_resources(row['user'], row['project'])
126                
127    def get_approved_projects(self):
128        """Get list of pre-approved high-cost projects."""
129        return ['llm-research', 'production-model-training', 'approved-experiment']
130

1def select_optimal_gpu_instance(model_size, batch_size, dataset_size):
2    """
3    Select most cost-effective GPU instance for given workload.
4    """
5    # Memory requirements calculation
6    # Rule of thumb: 4× model parameters in bytes for training
7    memory_needed_gb = model_size * 4 / 1e9  # Convert parameters to GB
8    
9    # Performance requirements
10    # Estimated training time = dataset_size × iterations / GPU_speed
11    
12    instances = [
13        {'type': 'g5.xlarge', 'memory_gb': 24, 'cost': 1.212, 'speed': 70},
14        {'type': 'g5.12xlarge', 'memory_gb': 96, 'cost': 4.088, 'speed': 280},
15        {'type': 'p4d.24xlarge', 'memory_gb': 320, 'cost': 32.77, 'speed': 624},
16        {'type': 'p5.48xlarge', 'memory_gb': 1128, 'cost': 98.32, 'speed': 1979}
17    ]
18    
19    # Filter by memory requirements
20    suitable_instances = [i for i in instances if i['memory_gb'] >= memory_needed_gb]
21    
22    if not suitable_instances:
23        print(f"Error: Need {memory_needed_gb:.1f}GB, max available is {instances[-1]['memory_gb']}GB")
24        return None
25    
26    # Calculate cost-effectiveness
27    for instance in suitable_instances:
28        # Estimated training hours
29        estimated_hours = dataset_size * 1000 / instance['speed']  # Simplified
30        
31        # Total cost
32        total_cost = estimated_hours * instance['cost']
33        
34        instance['estimated_hours'] = estimated_hours
35        instance['total_cost'] = total_cost
36        instance['cost_per_tflops'] = instance['cost'] / instance['speed']
37    
38    # Choose most cost-effective
39    most_cost_effective = min(suitable_instances, key=lambda x: x['total_cost'])
40    
41    print(f"Recommended: {most_cost_effective['type']}")
42    print(f"Estimated: {most_cost_effective['estimated_hours']:.1f} hours, ${most_cost_effective['total_cost']:.2f}")
43    
44    return most_cost_effective
45

1class CostOptimizedTrainingPipeline:
2    def __init__(self, project_name, budget):
3        self.project_name = project_name
4        self.budget = budget
5        self.checkpoint_manager = CheckpointManager(s3_bucket='training-checkpoints')
6        self.budget_manager = AITrainingBudgetManager()
7        
8    def run_training(self, model_config, dataset):
9        """Run training with cost optimization."""
10        
11        # 1. Right-size instance selection
12        instance = select_optimal_gpu_instance(
13            model_config['size'],
14            model_config['batch_size'],
15            len(dataset)
16        )
17        
18        # 2. Check budget availability
19        if not self.budget_manager.check_budget_available(self.budget):
20            raise Exception(f"Budget exceeded for {self.project_name}")
21            
22        # 3. Launch spot instance fleet
23        fleet_id = create_spot_fleet_configuration()
24        
25        # 4. Train with checkpointing
26        training_loop_with_checkpoints(
27            model=model_config['model'],
28            optimizer=model_config['optimizer'],
29            dataloader=dataset.loader(),
30            checkpoint_manager=self.checkpoint_manager,
31            epochs=model_config['epochs']
32        )
33        
34        # 5. Clean up resources
35        self.cleanup_resources(fleet_id)
36        
37        # 6. Report costs
38        self.report_costs()
39        
40    def report_costs(self):
41        """Generate cost optimization report."""
42        df, outliers = self.budget_manager.analyze_training_costs()
43        
44        print(f"=== Cost Report for {self.project_name} ===")
45        print(f"Total cost: ${df['cost'].sum():.2f}")
46        print(f"Spot instance savings: ${self.calculate_spot_savings():.2f}")
47        print(f"Auto-scaling savings: ${self.calculate_scaling_savings():.2f}")
48        print(f"Right-sizing savings: ${self.calculate_rightsize_savings():.2f}")
49        
50        savings_pct = (self.calculate_total_savings() / self.calculate_ondemand_cost()) * 100
51        print(f"Total savings: {savings_pct:.1f}%")
52