AWS SageMaker AutoML + Lambda Real-time Pricing

Solution for the use case Inventory Demand Forecasting with Dynamic Pricing.

AI tools

AWS SageMaker, AWS Lambda, Amazon S3, CloudWatch, AutoML

Language

Python

Accuracy

85%

Published

2025-07-05

Code (Python)

import boto3
import json
import pandas as pd
from datetime import datetime, timedelta

# SageMaker AutoML Job Creation
def create_automl_job():
    sagemaker = boto3.client("sagemaker")
    
    automl_job_config = {
        "AutoMLJobName": f"demand-forecast-{datetime.now().strftime('%Y%m%d')}",
        "InputDataConfig": [
            {
                "DataSource": {
                    "S3DataSource": {
                        "S3DataType": "S3Prefix",
                        "S3Uri": "s3://retail-analytics/training-data/",
                        "S3DataDistributionType": "FullyReplicated"
                    }
                },
                "TargetAttributeName": "daily_sales",
                "ContentType": "text/csv",
                "CompressionType": "None"
            }
        ],
        "OutputDataConfig": {
            "S3OutputPath": "s3://retail-analytics/automl-output/"
        },
        "ProblemType": "Regression",
        "AutoMLJobObjective": {
            "MetricName": "MSE"
        },
        "RoleArn": "arn:aws:iam::123456789012:role/SageMakerRole",
        "AutoMLJobConfig": {
            "CompletionCriteria": {
                "MaxCandidates": 250,
                "MaxRuntimePerTrainingJobInSeconds": 3600,
                "MaxAutoMLJobRuntimeInSeconds": 86400
            },
            "SecurityConfig": {
                "EnableInterContainerTrafficEncryption": True
            }
        }
    }
    
    response = sagemaker.create_auto_ml_job(**automl_job_config)
    return response["AutoMLJobArn"]

# Lambda function for real-time pricing
def lambda_handler(event, context):
    """AWS Lambda function for real-time pricing decisions"""
    
    # Parse input
    sku = event["sku"]
    store_id = event["store_id"]
    current_inventory = event["current_inventory"]
    competitor_prices = event["competitor_prices"]
    
    # Initialize AWS clients
    sagemaker_runtime = boto3.client("sagemaker-runtime")
    s3 = boto3.client("s3")
    
    try:
        # Get demand forecast from SageMaker endpoint
        forecast_response = sagemaker_runtime.invoke_endpoint(
            EndpointName="demand-forecast-endpoint",
            ContentType="text/csv",
            Body=prepare_forecast_input(sku, store_id, event)
        )
        
        forecast_data = json.loads(forecast_response["Body"].read().decode())
        predicted_demand = forecast_data["predictions"][0]
        
        # Load historical price elasticity
        elasticity_data = load_price_elasticity(s3, sku, store_id)
        
        # Optimize pricing
        optimal_price = optimize_price(
            predicted_demand=predicted_demand,
            current_inventory=current_inventory,
            competitor_prices=competitor_prices,
            price_elasticity=elasticity_data["elasticity"],
            cost=elasticity_data["unit_cost"]
        )
        
        # Log decision for monitoring
        log_pricing_decision(sku, store_id, optimal_price, predicted_demand)
        
        return {
            "statusCode": 200,
            "body": json.dumps({
                "sku": sku,
                "store_id": store_id,
                "recommended_price": round(optimal_price, 2),
                "predicted_demand": round(predicted_demand, 1),
                "confidence": forecast_data.get("confidence", 0.8),
                "timestamp": datetime.now().isoformat()
            })
        }
        
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }

def optimize_price(predicted_demand, current_inventory, competitor_prices, price_elasticity, cost):
    """Price optimization algorithm"""
    import numpy as np
    
    # Price bounds
    min_price = cost * 1.1  # 10% markup minimum
    max_price = np.mean(competitor_prices) * 1.3  # Max 30% above competition
    
    # Generate price candidates
    price_candidates = np.linspace(min_price, max_price, 100)
…

Prompts

AUTOML_CONFIG = """
Configure AutoML job for demand forecasting:

Training Data Location: s3://retail-data/sales-history/
Target Column: daily_sales
Problem Type: Regression
Metrics: RMSE, MAPE, R²

Feature Engineering:
- Automatic time series decomposition
- Holiday and event detection
- Weather correlation analysis
- Price elasticity estimation

Model Selection:
- Linear Models (Ridge, Lasso, ElasticNet)
- Tree-based (XGBoost, Random Forest, Extra Trees)
- Neural Networks (MLP, LSTM for time series)
- Ensemble methods with automatic blending

Hyperparameter Optimization:
- Bayesian optimization for efficient search
- Early stopping to prevent overfitting
- Cross-validation with time series splits
- Multi-objective optimization (accuracy vs speed)
"""

LAMBDA_PRICING = """
Lambda function for real-time pricing optimization:

Input: {
  "sku": "PROD123",
  "store_id": "ST001", 
  "current_inventory": 150,
  "competitor_prices": [19.99, 21.50, 18.75],
  "weather_forecast": "sunny",
  "upcoming_events": ["weekend", "local_festival"]
}

Processing Steps:
1. Fetch latest demand forecast from SageMaker endpoint
2. Calculate price elasticity for this SKU/store combination
3. Run optimization algorithm considering:
   - Revenue maximization
   - Inventory turnover goals
   - Competitive positioning
   - Brand guidelines (min/max price bounds)

Output: {
  "recommended_price": 20.49,
  "confidence": 0.85,
  "expected_sales": 45,
  "revenue_impact": "+12%",
  "reasoning": "High demand predicted due to local festival"
}
"""