# Install AWS CLI and configure credentials
pip install awscli boto3==1.34.44
aws configure

# Required IAM permissions:
# - lambda:CreateFunction, lambda:InvokeFunction
# - sagemaker:CreateModel, sagemaker:CreateEndpoint
# - s3:GetObject, s3:PutObject
# - states:CreateStateMachine, states:StartExecution
# - ecr:CreateRepository (for custom containers)

pip install boto3==1.34.44
pip install sagemaker==2.200.0
pip install scikit-learn==1.3.2
pip install pandas==2.1.4
pip install numpy==1.26.3

# lambda_inference.py - Deploy scikit-learn model in AWS Lambda
import json
import boto3
import pickle
import numpy as np
from typing import Dict, Any

# S3 client for model loading
s3_client = boto3.client('s3')

# Global variable for model caching (survives across warm starts)
MODEL_CACHE = {}

def load_model_from_s3(bucket: str, key: str):
    """
    Load pickled model from S3 with caching.

    AWS Lambda keeps global variables in memory between invocations
    (warm starts), so we cache the model to avoid repeated S3 fetches.
    """
    cache_key = f"{bucket}/{key}"

    if cache_key in MODEL_CACHE:
        print(f"Using cached model: {cache_key}")
        return MODEL_CACHE[cache_key]

    print(f"Loading model from S3: s3://{bucket}/{key}")

    # Download model from S3
    response = s3_client.get_object(Bucket=bucket, Key=key)
    model_bytes = response['Body'].read()

    # Unpickle model
    model = pickle.loads(model_bytes)

    # Cache for future invocations
    MODEL_CACHE[cache_key] = model

    print(f"Model loaded and cached. Memory usage: {len(model_bytes) / 1024 / 1024:.2f} MB")

    return model

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    AWS Lambda handler for ML inference.

    Event structure:
    {
        "model_bucket": "my-ml-models",
        "model_key": "fraud-detection/model.pkl",
        "features": [1.2, 3.4, 5.6, ...],
        "return_probabilities": true
    }

    Returns:
    {
        "statusCode": 200,
        "body": {
            "prediction": 1,
            "probability": 0.87,
            "model_version": "v1.2.3",
            "inference_time_ms": 45
        }
    }
    """
    import time
    start_time = time.time()

    try:
        # Parse input
        model_bucket = event.get('model_bucket', 'my-ml-models')
        model_key = event.get('model_key', 'models/fraud-detection.pkl')
        features = event['features']  # Required field
        return_proba = event.get('return_probabilities', False)

        # Validate input
        if not isinstance(features, list):
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'Features must be a list'})
            }

        # Load model (cached on warm starts)
        model = load_model_from_s3(model_bucket, model_key)

        # Prepare features
        X = np.array(features).reshape(1, -1)

        # Make prediction
        prediction = int(model.predict(X)[0])

        response_body = {
            'prediction': prediction,
            'model_version': model_key.split('/')[-1].replace('.pkl', ''),
            'inference_time_ms': int((time.time() - start_time) * 1000)
        }

        # Add probabilities if requested
        if return_proba and hasattr(model, 'predict_proba'):
            probabilities = model.predict_proba(X)[0].tolist()
            response_body['probabilities'] = probabilities
            response_body['confidence'] = max(probabilities)

        return {
            'statusCode': 200,
            'body': json.dumps(response_body),
            'headers': {
                'Content-Type': 'application/json',
                'X-Model-Version': model_key
            }
        }

    except KeyError as e:
        return {
            'statusCode': 400,
            'body': json.dumps({
                'error': f'Missing required field: {str(e)}'
            })
        }

    except Exception as e:
        print(f"Error during inference: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': 'Internal server error',
                'message': str(e)
            })
        }

# lambda_deployment.py - Deploy Lambda function using AWS CDK
from aws_cdk import (
    Stack,
    Duration,
    aws_lambda as lambda_,
    aws_s3 as s3,
    aws_iam as iam,
    aws_apigateway as apigw
)
from constructs import Construct

class MLInferenceLambdaStack(Stack):
    """
    CDK Stack for serverless ML inference.

    Deploys:
    - Lambda function with scikit-learn layer
    - S3 bucket for models
    - API Gateway REST API
    - IAM roles with least privilege
    """

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # S3 bucket for ML models
        model_bucket = s3.Bucket(
            self, "ModelBucket",
            bucket_name="ml-models-serverless",
            versioned=True,  # Enable versioning for model rollback
            encryption=s3.BucketEncryption.S3_MANAGED,
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL
        )

        # Lambda layer with scikit-learn
        # Pre-built layer: https://github.com/model-zoo/scikit-learn-lambda-layer
        sklearn_layer = lambda_.LayerVersion(
            self, "SklearnLayer",
            code=lambda_.Code.from_asset("layers/sklearn-layer.zip"),
            compatible_runtimes=[lambda_.Runtime.PYTHON_3_11],
            description="scikit-learn 1.3.2 + numpy + pandas"
        )

        # Lambda function
        inference_lambda = lambda_.Function(
            self, "InferenceFunction",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="lambda_inference.lambda_handler",
            code=lambda_.Code.from_asset("lambda"),
            timeout=Duration.seconds(30),
            memory_size=512,  # Adjust based on model size
            layers=[sklearn_layer],
            environment={
                "MODEL_BUCKET": model_bucket.bucket_name,
                "DEFAULT_MODEL_KEY": "models/fraud-detection.pkl"
            },
            reserved_concurrent_executions=100  # Limit concurrent invocations
        )

        # Grant Lambda read access to S3 models
        model_bucket.grant_read(inference_lambda)

        # API Gateway REST API
        api = apigw.RestApi(
            self, "InferenceAPI",
            rest_api_name="ML Inference API",
            description="Serverless ML inference endpoint",
            deploy_options=apigw.StageOptions(
                stage_name="prod",
                throttling_rate_limit=1000,  # Requests per second
                throttling_burst_limit=2000,
                logging_level=apigw.MethodLoggingLevel.INFO,
                metrics_enabled=True
            )
        )

        # /predict endpoint
        predict_resource = api.root.add_resource("predict")
        predict_integration = apigw.LambdaIntegration(
            inference_lambda,
            proxy=False,
            integration_responses=[
                apigw.IntegrationResponse(
                    status_code="200",
                    response_templates={
                        "application/json": "$input.json('$.body')"
                    }
                )
            ]
        )

        predict_resource.add_method(
            "POST",
            predict_integration,
            method_responses=[
                apigw.MethodResponse(
                    status_code="200",
                    response_models={
                        "application/json": apigw.Model.EMPTY_MODEL
                    }
                )
            ]
        )

        # Output API endpoint URL
        from aws_cdk import CfnOutput
        CfnOutput(
            self, "APIEndpoint",
            value=api.url,
            description="ML Inference API endpoint"
        )

# Install AWS CDK
npm install -g aws-cdk

# Bootstrap CDK (first time only)
cdk bootstrap

# Deploy stack
cdk deploy MLInferenceLambdaStack

# Output:
# MLInferenceLambdaStack.APIEndpoint = https://xxxxx.execute-api.us-east-1.amazonaws.com/prod/

# test_lambda_inference.py - Test deployed Lambda function
import requests
import json

def test_inference(api_endpoint: str, features: list):
    """Test the deployed Lambda inference API."""

    # Prepare request
    payload = {
        "features": features,
        "return_probabilities": True
    }

    # Send POST request
    response = requests.post(
        f"{api_endpoint}/predict",
        json=payload,
        headers={"Content-Type": "application/json"}
    )

    # Parse response
    if response.status_code == 200:
        result = response.json()
        print("✓ Prediction successful:")
        print(f"  Prediction: {result['prediction']}")
        print(f"  Confidence: {result.get('confidence', 'N/A')}")
        print(f"  Inference Time: {result['inference_time_ms']}ms")
        return result
    else:
        print(f"✗ Prediction failed: {response.status_code}")
        print(f"  Error: {response.text}")
        return None

# Example: Fraud detection
api_endpoint = "https://xxxxx.execute-api.us-east-1.amazonaws.com/prod"

# Transaction features: [amount, merchant_category, hour_of_day, ...]
transaction_features = [250.0, 5411, 14, 1, 0, 1, 0.8, 3]

result = test_inference(api_endpoint, transaction_features)

# sagemaker_serverless_deployment.py - Deploy PyTorch model to SageMaker Serverless
import boto3
import sagemaker
from sagemaker.pytorch import PyTorchModel
from sagemaker.serverless import ServerlessInferenceConfig
import tarfile
import os

class SageMakerServerlessDeployment:
    """Deploy ML models to SageMaker Serverless Inference."""

    def __init__(self, role_arn: str, bucket_name: str):
        self.role = role_arn
        self.bucket = bucket_name
        self.sagemaker_session = sagemaker.Session()
        self.sagemaker_client = boto3.client('sagemaker')

    def package_model(self, model_path: str, output_tar: str = "model.tar.gz"):
        """
        Package model files into tarball for SageMaker.

        Required structure:
        model.tar.gz
        ├── code/
        │   └── inference.py  # Must define model_fn, input_fn, predict_fn, output_fn
        └── model.pth         # PyTorch model weights
        """
        with tarfile.open(output_tar, "w:gz") as tar:
            tar.add(model_path, arcname=".")

        print(f"✓ Model packaged: {output_tar}")
        return output_tar

    def upload_model_to_s3(self, local_path: str, s3_prefix: str = "models"):
        """Upload packaged model to S3."""
        s3_client = boto3.client('s3')

        s3_key = f"{s3_prefix}/{os.path.basename(local_path)}"
        s3_client.upload_file(local_path, self.bucket, s3_key)

        s3_uri = f"s3://{self.bucket}/{s3_key}"
        print(f"✓ Model uploaded: {s3_uri}")

        return s3_uri

    def deploy_serverless_endpoint(
        self,
        model_data_url: str,
        endpoint_name: str,
        framework_version: str = "2.1.0",
        python_version: str = "py310",
        memory_size_mb: int = 4096,
        max_concurrency: int = 20
    ) -> str:
        """
        Deploy model to SageMaker Serverless Inference endpoint.

        Args:
            model_data_url: S3 URI to model.tar.gz
            endpoint_name: Unique endpoint name
            memory_size_mb: Memory allocation (1024, 2048, 3072, 4096, 5120, 6144MB)
            max_concurrency: Max concurrent invocations (1-200)

        Returns: Endpoint name
        """
        # Create PyTorchModel
        pytorch_model = PyTorchModel(
            model_data=model_data_url,
            role=self.role,
            framework_version=framework_version,
            py_version=python_version,
            entry_point="inference.py",  # Must be in code/ directory
            sagemaker_session=self.sagemaker_session
        )

        # Serverless configuration
        serverless_config = ServerlessInferenceConfig(
            memory_size_in_mb=memory_size_mb,
            max_concurrency=max_concurrency
        )

        # Deploy endpoint
        print(f"Deploying serverless endpoint: {endpoint_name}")
        print("This may take 5-10 minutes...")

        predictor = pytorch_model.deploy(
            endpoint_name=endpoint_name,
            serverless_inference_config=serverless_config
        )

        print(f"✓ Endpoint deployed: {endpoint_name}")
        print(f"  Memory: {memory_size_mb}MB")
        print(f"  Max Concurrency: {max_concurrency}")

        return endpoint_name

    def invoke_endpoint(self, endpoint_name: str, payload: dict):
        """
        Invoke SageMaker serverless endpoint.

        Args:
            endpoint_name: Deployed endpoint name
            payload: Input data as dictionary

        Returns: Model predictions
        """
        import json

        runtime_client = boto3.client('sagemaker-runtime')

        response = runtime_client.invoke_endpoint(
            EndpointName=endpoint_name,
            ContentType='application/json',
            Body=json.dumps(payload)
        )

        result = json.loads(response['Body'].read().decode())

        print(f"✓ Inference completed")
        print(f"  Cold start: {'Yes' if response['ResponseMetadata']['HTTPHeaders'].get('x-amzn-sagemaker-cold-start') == 'true' else 'No'}")

        return result

    def delete_endpoint(self, endpoint_name: str):
        """Delete serverless endpoint to stop billing."""
        self.sagemaker_client.delete_endpoint(EndpointName=endpoint_name)
        self.sagemaker_client.delete_endpoint_config(EndpointConfigName=endpoint_name)
        print(f"✓ Endpoint deleted: {endpoint_name}")


# Example: Deploy sentiment analysis model
if __name__ == "__main__":
    # Setup
    ROLE_ARN = "arn:aws:iam::123456789012:role/SageMakerExecutionRole"
    BUCKET_NAME = "my-ml-models"

    deployer = SageMakerServerlessDeployment(
        role_arn=ROLE_ARN,
        bucket_name=BUCKET_NAME
    )

    # Package model
    model_tar = deployer.package_model(model_path="pytorch_model/")

    # Upload to S3
    model_s3_uri = deployer.upload_model_to_s3(model_tar)

    # Deploy serverless endpoint
    endpoint_name = "sentiment-analysis-serverless"
    deployer.deploy_serverless_endpoint(
        model_data_url=model_s3_uri,
        endpoint_name=endpoint_name,
        memory_size_mb=2048,
        max_concurrency=10
    )

    # Test inference
    test_payload = {
        "inputs": "This product exceeded my expectations! Highly recommend."
    }

    result = deployer.invoke_endpoint(endpoint_name, test_payload)
    print(f"Sentiment: {result}")

# code/inference.py - Custom inference handler for SageMaker
import torch
import json
import numpy as np
from transformers import AutoTokenizer, AutoModelForSequenceClassification

# Global variables for model caching
model = None
tokenizer = None
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

def model_fn(model_dir: str):
    """
    Load model from model_dir.

    This function is called once when the endpoint is created.
    The model is kept in memory for all subsequent predictions.

    Args:
        model_dir: Directory containing model artifacts

    Returns: Loaded model
    """
    global model, tokenizer

    print(f"Loading model from: {model_dir}")

    # Load tokenizer
    tokenizer = AutoTokenizer.from_pretrained(model_dir)

    # Load model
    model = AutoModelForSequenceClassification.from_pretrained(model_dir)
    model.to(device)
    model.eval()

    print(f"✓ Model loaded on device: {device}")

    return model

def input_fn(request_body: str, content_type: str = "application/json"):
    """
    Deserialize input data.

    Args:
        request_body: Raw request body
        content_type: Content type header

    Returns: Deserialized input
    """
    if content_type == "application/json":
        data = json.loads(request_body)
        return data
    else:
        raise ValueError(f"Unsupported content type: {content_type}")

def predict_fn(input_data: dict, model):
    """
    Make prediction.

    Args:
        input_data: Deserialized input
        model: Loaded model from model_fn

    Returns: Model predictions
    """
    texts = input_data.get("inputs")

    if isinstance(texts, str):
        texts = [texts]

    # Tokenize inputs
    inputs = tokenizer(
        texts,
        padding=True,
        truncation=True,
        max_length=512,
        return_tensors="pt"
    )

    inputs = {k: v.to(device) for k, v in inputs.items()}

    # Run inference
    with torch.no_grad():
        outputs = model(**inputs)
        logits = outputs.logits

    # Get predictions
    probabilities = torch.nn.functional.softmax(logits, dim=-1)
    predictions = torch.argmax(probabilities, dim=-1)

    # Format results
    results = []
    for i, text in enumerate(texts):
        results.append({
            "text": text,
            "label": model.config.id2label[predictions[i].item()],
            "score": probabilities[i, predictions[i]].item()
        })

    return results

def output_fn(prediction: list, accept: str = "application/json"):
    """
    Serialize prediction output.

    Args:
        prediction: Model predictions
        accept: Accept header

    Returns: Serialized response
    """
    if accept == "application/json":
        return json.dumps(prediction), accept
    else:
        raise ValueError(f"Unsupported accept type: {accept}")

# step_functions_ml_pipeline.py - ML pipeline orchestration
import boto3
import json
from typing import Dict

class MLPipelineOrchestrator:
    """Orchestrate ML workflows with AWS Step Functions."""

    def __init__(self, role_arn: str):
        self.sfn_client = boto3.client('stepfunctions')
        self.role_arn = role_arn

    def create_ml_pipeline_state_machine(self, state_machine_name: str) -> str:
        """
        Create Step Functions state machine for ML pipeline.

        Pipeline stages:
        1. Data validation
        2. Feature engineering
        3. Model training (SageMaker)
        4. Model evaluation
        5. Model deployment (conditional)
        6. Notification
        """
        definition = {
            "Comment": "End-to-end ML pipeline with conditional deployment",
            "StartAt": "ValidateData",
            "States": {
                "ValidateData": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::lambda:invoke",
                    "Parameters": {
                        "FunctionName": "data-validation-function",
                        "Payload": {
                            "data_path.$": "$.data_path",
                            "validation_rules.$": "$.validation_rules"
                        }
                    },
                    "ResultPath": "$.validation_result",
                    "Next": "CheckDataQuality",
                    "Catch": [
                        {
                            "ErrorEquals": ["States.ALL"],
                            "Next": "NotifyFailure"
                        }
                    ]
                },
                "CheckDataQuality": {
                    "Type": "Choice",
                    "Choices": [
                        {
                            "Variable": "$.validation_result.Payload.is_valid",
                            "BooleanEquals": True,
                            "Next": "FeatureEngineering"
                        }
                    ],
                    "Default": "NotifyFailure"
                },
                "FeatureEngineering": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::lambda:invoke",
                    "Parameters": {
                        "FunctionName": "feature-engineering-function",
                        "Payload": {
                            "data_path.$": "$.data_path",
                            "feature_config.$": "$.feature_config"
                        }
                    },
                    "ResultPath": "$.features_result",
                    "Next": "TrainModel"
                },
                "TrainModel": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::sagemaker:createTrainingJob.sync",
                    "Parameters": {
                        "TrainingJobName.$": "$.training_job_name",
                        "RoleArn": self.role_arn,
                        "AlgorithmSpecification": {
                            "TrainingImage.$": "$.training_image",
                            "TrainingInputMode": "File"
                        },
                        "InputDataConfig": [
                            {
                                "ChannelName": "training",
                                "DataSource": {
                                    "S3DataSource": {
                                        "S3DataType": "S3Prefix",
                                        "S3Uri.$": "$.features_result.Payload.output_path",
                                        "S3DataDistributionType": "FullyReplicated"
                                    }
                                },
                                "ContentType": "text/csv"
                            }
                        ],
                        "OutputDataConfig": {
                            "S3OutputPath.$": "$.model_output_path"
                        },
                        "ResourceConfig": {
                            "InstanceType": "ml.m5.xlarge",
                            "InstanceCount": 1,
                            "VolumeSizeInGB": 10
                        },
                        "StoppingCondition": {
                            "MaxRuntimeInSeconds": 3600
                        }
                    },
                    "ResultPath": "$.training_result",
                    "Next": "EvaluateModel"
                },
                "EvaluateModel": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::lambda:invoke",
                    "Parameters": {
                        "FunctionName": "model-evaluation-function",
                        "Payload": {
                            "model_path.$": "$.training_result.ModelArtifacts.S3ModelArtifacts",
                            "test_data_path.$": "$.test_data_path",
                            "evaluation_metrics": ["accuracy", "precision", "recall", "f1"]
                        }
                    },
                    "ResultPath": "$.evaluation_result",
                    "Next": "CheckModelQuality"
                },
                "CheckModelQuality": {
                    "Type": "Choice",
                    "Choices": [
                        {
                            "Variable": "$.evaluation_result.Payload.metrics.accuracy",
                            "NumericGreaterThanEquals": 0.90,
                            "Next": "DeployModel"
                        }
                    ],
                    "Default": "NotifyPoorPerformance"
                },
                "DeployModel": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::sagemaker:createEndpoint",
                    "Parameters": {
                        "EndpointName.$": "$.endpoint_name",
                        "EndpointConfigName.$": "$.endpoint_config_name"
                    },
                    "ResultPath": "$.deployment_result",
                    "Next": "NotifySuccess"
                },
                "NotifySuccess": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::sns:publish",
                    "Parameters": {
                        "TopicArn": "arn:aws:sns:us-east-1:123456789012:ml-pipeline-notifications",
                        "Subject": "ML Pipeline Success",
                        "Message.$": "States.Format('Pipeline completed successfully. Model accuracy: {}', $.evaluation_result.Payload.metrics.accuracy)"
                    },
                    "End": True
                },
                "NotifyFailure": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::sns:publish",
                    "Parameters": {
                        "TopicArn": "arn:aws:sns:us-east-1:123456789012:ml-pipeline-notifications",
                        "Subject": "ML Pipeline Failure",
                        "Message": "Pipeline failed during data validation"
                    },
                    "End": True
                },
                "NotifyPoorPerformance": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::sns:publish",
                    "Parameters": {
                        "TopicArn": "arn:aws:sns:us-east-1:123456789012:ml-pipeline-notifications",
                        "Subject": "ML Pipeline - Model Below Threshold",
                        "Message.$": "States.Format('Model accuracy {} below threshold 0.90', $.evaluation_result.Payload.metrics.accuracy)"
                    },
                    "End": True
                }
            }
        }

        # Create state machine
        response = self.sfn_client.create_state_machine(
            name=state_machine_name,
            definition=json.dumps(definition),
            roleArn=self.role_arn,
            type='STANDARD',  # Use 'EXPRESS' for high-volume, short-duration workflows
            loggingConfiguration={
                'level': 'ALL',
                'includeExecutionData': True,
                'destinations': [
                    {
                        'cloudWatchLogsLogGroup': {
                            'logGroupArn': f'arn:aws:logs:us-east-1:123456789012:log-group:/aws/vendedlogs/states/{state_machine_name}'
                        }
                    }
                ]
            }
        )

        state_machine_arn = response['stateMachineArn']
        print(f"✓ State machine created: {state_machine_arn}")

        return state_machine_arn

    def start_pipeline_execution(self, state_machine_arn: str, input_params: Dict) -> str:
        """
        Start ML pipeline execution.

        Args:
            state_machine_arn: ARN of state machine
            input_params: Pipeline input parameters

        Returns: Execution ARN
        """
        response = self.sfn_client.start_execution(
            stateMachineArn=state_machine_arn,
            input=json.dumps(input_params)
        )

        execution_arn = response['executionArn']
        print(f"✓ Pipeline execution started: {execution_arn}")

        return execution_arn

    def get_execution_status(self, execution_arn: str) -> Dict:
        """Get pipeline execution status and output."""
        response = self.sfn_client.describe_execution(executionArn=execution_arn)

        status = response['status']
        print(f"Pipeline status: {status}")

        if status == 'SUCCEEDED':
            output = json.loads(response.get('output', '{}'))
            print(f"Pipeline output: {output}")
            return output
        elif status == 'FAILED':
            print(f"Failure cause: {response.get('cause', 'Unknown')}")

        return response


# Example usage
if __name__ == "__main__":
    ROLE_ARN = "arn:aws:iam::123456789012:role/StepFunctionsExecutionRole"

    orchestrator = MLPipelineOrchestrator(role_arn=ROLE_ARN)

    # Create state machine
    state_machine_arn = orchestrator.create_ml_pipeline_state_machine(
        state_machine_name="fraud-detection-pipeline"
    )

    # Start pipeline execution
    pipeline_input = {
        "data_path": "s3://my-bucket/raw-data/transactions.csv",
        "validation_rules": {
            "required_columns": ["amount", "merchant_id", "timestamp"],
            "max_null_percentage": 0.05
        },
        "feature_config": {
            "aggregation_window": "7d",
            "categorical_features": ["merchant_id", "category"]
        },
        "training_job_name": "fraud-detection-training-20250122",
        "training_image": "382416733822.dkr.ecr.us-east-1.amazonaws.com/xgboost:latest",
        "model_output_path": "s3://my-bucket/models/fraud-detection/",
        "test_data_path": "s3://my-bucket/test-data/transactions-test.csv",
        "endpoint_name": "fraud-detection-endpoint",
        "endpoint_config_name": "fraud-detection-config"
    }

    execution_arn = orchestrator.start_pipeline_execution(
        state_machine_arn, pipeline_input
    )

    # Monitor execution
    import time
    while True:
        result = orchestrator.get_execution_status(execution_arn)
        if result['status'] in ['SUCCEEDED', 'FAILED', 'TIMED_OUT', 'ABORTED']:
            break
        time.sleep(10)

# cost_analysis.py - Analyze serverless ML costs
import boto3
from datetime import datetime, timedelta
import pandas as pd

class ServerlessMLCostAnalyzer:
    """Analyze costs for serverless ML infrastructure."""

    def __init__(self):
        self.ce_client = boto3.client('ce')  # Cost Explorer
        self.cloudwatch = boto3.client('cloudwatch')

    def get_lambda_costs(self, days: int = 30) -> pd.DataFrame:
        """Get Lambda function costs for last N days."""
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=days)

        response = self.ce_client.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.strftime('%Y-%m-%d'),
                'End': end_date.strftime('%Y-%m-%d')
            },
            Granularity='DAILY',
            Filter={
                'Dimensions': {
                    'Key': 'SERVICE',
                    'Values': ['AWS Lambda']
                }
            },
            Metrics=['UnblendedCost'],
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'USAGE_TYPE'}
            ]
        )

        # Parse response into DataFrame
        data = []
        for result in response['ResultsByTime']:
            date = result['TimePeriod']['Start']
            for group in result['Groups']:
                data.append({
                    'Date': date,
                    'UsageType': group['Keys'][0],
                    'Cost': float(group['Metrics']['UnblendedCost']['Amount'])
                })

        df = pd.DataFrame(data)
        print(f"\nLambda costs (last {days} days):")
        print(df.groupby('UsageType')['Cost'].sum().sort_values(ascending=False))

        return df

    def get_sagemaker_serverless_metrics(self, endpoint_name: str) -> Dict:
        """Get SageMaker Serverless endpoint metrics."""
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(days=1)

        metrics_to_fetch = [
            'ModelInvocations',
            'ModelSetupTime',
            'Invocation4XXErrors',
            'Invocation5XXErrors'
        ]

        results = {}

        for metric_name in metrics_to_fetch:
            response = self.cloudwatch.get_metric_statistics(
                Namespace='AWS/SageMaker',
                MetricName=metric_name,
                Dimensions=[
                    {'Name': 'EndpointName', 'Value': endpoint_name}
                ],
                StartTime=start_time,
                EndTime=end_time,
                Period=3600,  # 1 hour
                Statistics=['Sum', 'Average']
            )

            if response['Datapoints']:
                results[metric_name] = {
                    'Sum': sum(dp['Sum'] for dp in response['Datapoints']),
                    'Average': sum(dp['Average'] for dp in response['Datapoints']) / len(response['Datapoints'])
                }

        print(f"\nSageMaker Serverless metrics ({endpoint_name}):")
        for metric, values in results.items():
            print(f"  {metric}: {values}")

        return results

    def estimate_monthly_cost(
        self,
        lambda_invocations: int,
        lambda_duration_ms: float,
        lambda_memory_mb: int,
        sagemaker_invocations: int = 0,
        sagemaker_memory_gb: int = 4
    ) -> Dict:
        """
        Estimate monthly costs for serverless ML workload.

        Pricing (as of 2025):
        - Lambda: $0.20 per 1M requests + $0.0000166667 per GB-second
        - SageMaker Serverless: $0.000025 per inference + $0.000011 per MB-sec
        """
        # Lambda costs
        lambda_request_cost = (lambda_invocations / 1_000_000) * 0.20
        lambda_gb_seconds = (lambda_invocations * (lambda_duration_ms / 1000) * (lambda_memory_mb / 1024))
        lambda_compute_cost = lambda_gb_seconds * 0.0000166667

        lambda_total = lambda_request_cost + lambda_compute_cost

        # SageMaker Serverless costs
        sagemaker_inference_cost = sagemaker_invocations * 0.000025
        # Assume 2 second average inference time
        sagemaker_compute_cost = (sagemaker_invocations * 2 * sagemaker_memory_gb * 1024) * 0.000011

        sagemaker_total = sagemaker_inference_cost + sagemaker_compute_cost

        return {
            'lambda': {
                'request_cost': lambda_request_cost,
                'compute_cost': lambda_compute_cost,
                'total': lambda_total
            },
            'sagemaker': {
                'inference_cost': sagemaker_inference_cost,
                'compute_cost': sagemaker_compute_cost,
                'total': sagemaker_total
            },
            'total_monthly_cost': lambda_total + sagemaker_total
        }


# Example usage
if __name__ == "__main__":
    analyzer = ServerlessMLCostAnalyzer()

    # Get actual Lambda costs
    lambda_costs_df = analyzer.get_lambda_costs(days=30)

    # Get SageMaker metrics
    sagemaker_metrics = analyzer.get_sagemaker_serverless_metrics(
        endpoint_name="sentiment-analysis-serverless"
    )

    # Estimate future costs
    cost_estimate = analyzer.estimate_monthly_cost(
        lambda_invocations=100_000,  # 100k predictions/month
        lambda_duration_ms=250,      # 250ms average
        lambda_memory_mb=512,
        sagemaker_invocations=50_000,  # 50k SageMaker predictions/month
        sagemaker_memory_gb=2
    )

    print("\nMonthly Cost Estimate:")
    print(f"  Lambda: ${cost_estimate['lambda']['total']:.2f}")
    print(f"  SageMaker: ${cost_estimate['sagemaker']['total']:.2f}")
    print(f"  TOTAL: ${cost_estimate['total_monthly_cost']:.2f}")

# Use Lambda provisioned concurrency (keeps N instances warm)
lambda_client = boto3.client('lambda')

lambda_client.put_provisioned_concurrency_config(
    FunctionName='ml-inference-function',
    Qualifier='$LATEST',
    ProvisionedConcurrentExecutions=5  # Keep 5 warm instances
)

# Cost: $0.000004167 per GB-second (vs cold start delays)
# Recommended for production endpoints with consistent traffic

# Use Lambda aliases for A/B testing
lambda_client.create_alias(
    FunctionName='ml-inference-function',
    Name='prod',
    FunctionVersion='2',
    RoutingConfig={
        'AdditionalVersionWeights': {
            '3': 0.10  # Route 10% traffic to new version
        }
    }
)

# Create CloudWatch alarms for ML endpoints
cloudwatch = boto3.client('cloudwatch')

cloudwatch.put_metric_alarm(
    AlarmName='ml-lambda-errors-high',
    ComparisonOperator='GreaterThanThreshold',
    EvaluationPeriods=2,
    MetricName='Errors',
    Namespace='AWS/Lambda',
    Period=300,
    Statistic='Sum',
    Threshold=10.0,
    ActionsEnabled=True,
    AlarmActions=['arn:aws:sns:us-east-1:123456789012:ml-alerts'],
    AlarmDescription='Alert when Lambda errors exceed threshold',
    Dimensions=[
        {'Name': 'FunctionName', 'Value': 'ml-inference-function'}
    ]
)

# Test import locally
python3 -c "import sklearn; print(sklearn.__version__)"

# Check model size
import os
model_size_mb = os.path.getsize('model.tar.gz') / 1024 / 1024
print(f"Model size: {model_size_mb:.2f} MB")