Interview Question

"Design a dynamic pricing system like Amazon or Uber that can optimize prices in real-time based on demand, competition, inventory, and user behavior, while maintaining fairness and maximizing revenue."

Difficulty: Hard | Frequently asked at Amazon, Uber, Airbnb, airlines, e-commerce companies

1. Requirements Gathering

Functional Requirements

1. Real-time Pricing: Update prices in real-time based on conditions
2. Demand Forecasting: Predict demand at different price points
3. Competitor Monitoring: Track competitor prices
4. Inventory Awareness: Consider inventory levels in pricing
5. User Segmentation: Different pricing for different user segments
6. Price Rules: Business rules for minimum/maximum prices
7. A/B Testing: Test different pricing strategies

Non-Functional Requirements

1. Latency: < 100ms for price calculation
2. Throughput: 100,000+ price calculations per second
3. Accuracy: Price predictions within 5% of actual demand
4. Fairness: No discriminatory pricing
5. Compliance: Follow pricing regulations
6. Scale: Millions of products, billions of price points
7. Auditability: All price changes must be logged

2. High-Level Architecture Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                         DATA SOURCES                                        │
│  Sales Data │ Competitor Prices │ Inventory │ User Behavior │ Market Trends │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         DATA PIPELINE                                       │
│  Real-time Streaming │ Feature Computation │ Demand Forecasting │ Elasticity│
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┼───────────────┐
                    ▼               ▼               ▼
┌────────────────────────┐ ┌───────────────┐ ┌──────────────────────┐
│  DEMAND FORECASTING    │ │ PRICE         │ │ COMPETITOR           │
│  (Time Series)         │ │ ELASTICITY    │ │ MONITORING           │
│  (< 1s)                │ │ (< 100ms)     │ │ (Real-time)          │
└────────────────────────┘ └───────────────┘ └──────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                        PRICE OPTIMIZATION ENGINE                            │
│  Objective Function │ Constraints │ Business Rules │ Fairness Constraints   │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┼───────────────┐
                    ▼               ▼               ▼
┌────────────────────────┐ ┌───────────────┐ ┌──────────────────────┐
│  PRICE OUTPUT          │ │ A/B TESTING   │ │ MONITORING           │
│  (Product Prices)      │ │ FRAMEWORK     │ │ DASHBOARD            │
└────────────────────────┘ └───────────────┘ └──────────────────────┘

3. Data Pipeline Design

3.1 Pricing Data Model

from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
from decimal import Decimal

@dataclass
class Product:
    product_id: str
    category: str
    current_price: Decimal
    cost: Decimal
    inventory_count: int
    min_price: Decimal
    max_price: Decimal
    
@dataclass
class PricePoint:
    product_id: str
    price: Decimal
    timestamp: datetime
    demand: int
    revenue: Decimal
    conversion_rate: float

@dataclass
class CompetitorPrice:
    product_id: str
    competitor: str
    price: Decimal
    timestamp: datetime
    in_stock: bool

3.2 Demand Forecasting

class DemandForecaster:
    def __init__(self):
        self.model = self.build_model()
    
    def build_model(self):
        model = tf.keras.Sequential([
            tf.keras.layers.LSTM(128, return_sequences=True),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.LSTM(64),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.Dense(32, activation='relu'),
            tf.keras.layers.Dense(1)
        ])
        return model
    
    async def forecast_demand(self, product_id, price, horizon_days=7):
        # Get historical data
        history = await self.get_historical_data(product_id)
        
        # Get features
        features = await self.extract_features(product_id, price, history)
        
        # Predict demand
        demand = self.model.predict(features)
        
        return {
            'predicted_demand': demand,
            'confidence_interval': self.compute_confidence(demand),
            'forecast_horizon': horizon_days
        }

class PriceElasticityEstimator:
    def __init__(self):
        self.elasticity_cache = {}
    
    async def estimate_elasticity(self, product_id):
        # Get historical price-demand data
        data = await self.get_price_demand_data(product_id)
        
        # Calculate elasticity
        # Elasticity = % change in demand / % change in price
        elasticity = self.calculate_elasticity(data)
        
        # Cache result
        self.elasticity_cache[product_id] = elasticity
        
        return elasticity
    
    def calculate_elasticity(self, data):
        prices = [d['price'] for d in data]
        demands = [d['demand'] for d in data]
        
        # Log-log regression
        log_prices = np.log(prices)
        log_demands = np.log(demands)
        
        slope, intercept, r_value, p_value, std_err = stats.linregress(
            log_prices, log_demands
        )
        
        return slope  # Elasticity coefficient

3.3 Price Optimization

class PriceOptimizer:
    def __init__(self):
        self.demand_forecaster = DemandForecaster()
        self.elasticity_estimator = PriceElasticityEstimator()
    
    async def optimize_price(self, product_id):
        # Get product constraints
        product = await self.get_product(product_id)
        
        # Get demand forecast at different prices
        price_range = np.linspace(
            float(product.min_price),
            float(product.max_price),
            20
        )
        
        revenue_predictions = []
        for price in price_range:
            demand = await self.demand_forecaster.forecast_demand(product_id, price)
            revenue = price * demand['predicted_demand']
            revenue_predictions.append(revenue)
        
        # Find optimal price
        optimal_idx = np.argmax(revenue_predictions)
        optimal_price = price_range[optimal_idx]
        
        # Apply business rules
        final_price = self.apply_business_rules(optimal_price, product)
        
        return {
            'optimal_price': final_price,
            'expected_revenue': revenue_predictions[optimal_idx],
            'expected_demand': await self.demand_forecaster.forecast_demand(
                product_id, final_price
            ),
            'confidence': self.compute_confidence(revenue_predictions)
        }
    
    def apply_business_rules(self, price, product):
        # Ensure within bounds
        price = max(float(product.min_price), min(float(product.max_price), price))
        
        # Round to nearest cent
        price = round(price, 2)
        
        # Apply minimum margin
        min_margin = 0.2  # 20% minimum margin
        min_price = float(product.cost) * (1 + min_margin)
        price = max(price, min_price)
        
        return price

4. Model Selection and Training

4.1 Demand Forecasting Models

class DemandForecastingEnsemble:
    def __init__(self):
        self.models = {
            'lstm': LSTMModel(),
            'prophet': ProphetModel(),
            'xgboost': XGBoostModel()
        }
    
    async def forecast(self, product_id, price, horizon):
        predictions = {}
        for name, model in self.models.items():
            pred = await model.predict(product_id, price, horizon)
            predictions[name] = pred
        
        # Weighted average
        weights = {'lstm': 0.4, 'prophet': 0.3, 'xgboost': 0.3}
        final_pred = sum(predictions[name] * weights[name] for name in predictions)
        
        return final_pred

4.2 Reinforcement Learning for Pricing

class PricingRLAgent:
    def __init__(self):
        self.q_network = self.build_q_network()
        self.target_network = self.build_q_network()
        self.replay_buffer = ReplayBuffer()
    
    def build_q_network(self):
        model = tf.keras.Sequential([
            tf.keras.layers.Dense(128, activation='relu'),
            tf.keras.layers.Dense(64, activation='relu'),
            tf.keras.layers.Dense(32, activation='relu'),
            tf.keras.layers.Dense(n_actions)  # Price actions
        ])
        return model
    
    def get_state(self, product_id, context):
        return np.array([
            context['current_price'],
            context['inventory_level'],
            context['demand_forecast'],
            context['competitor_price'],
            context['time_of_day'],
            context['day_of_week']
        ])
    
    def choose_action(self, state, epsilon=0.1):
        if np.random.random() < epsilon:
            return np.random.randint(n_actions)
        q_values = self.q_network.predict(state.reshape(1, -1))
        return np.argmax(q_values[0])

5. Serving Architecture

5.1 Real-time Pricing Pipeline

Price Request → Load Balancer → Feature Computation → Model Inference → Price Calculation → Response
                  (< 5ms)         (< 20ms)            (< 50ms)          (< 20ms)         (< 5ms)

5.2 Caching Strategy

class PricingCache:
    def __init__(self):
        self.price_cache = RedisCluster()
        self.feature_cache = LRUCache()
    
    async def get_price(self, product_id, context):
        cache_key = f"price:{product_id}:{hash(str(context))}"
        
        # Check cache
        cached = await self.price_cache.get(cache_key)
        if cached:
            return cached
        
        # Calculate fresh price
        price = await self.calculate_price(product_id, context)
        
        # Cache result (short TTL for dynamic pricing)
        await self.price_cache.set(cache_key, price, ttl=60)
        
        return price

6. Monitoring and Observability

6.1 Key Metrics

class PricingMetrics:
    BUSINESS_METRICS = ['revenue', 'profit_margin', 'conversion_rate', 'inventory_turnover']
    MODEL_METRICS = ['demand_forecast_accuracy', 'price_elasticity_estimation_error']
    OPERATIONAL_METRICS = ['price_calculation_latency', 'throughput', 'error_rate']
    FAIRNESS_METRICS = ['price_variance_across_segments', 'discrimination_score']

7. Scale Considerations and Trade-offs

7.1 Horizontal Scaling

Product Catalog: Shard by product category
Price Calculations: Horizontal scaling with load balancing
Demand Forecasting: Batch predictions with caching
Competitor Monitoring: Distributed crawling

7.2 Cost vs Performance Trade-offs

8. Advanced Topics

8.1 Competitive Pricing

class CompetitivePricer:
    def __init__(self):
        self.competitor_monitor = CompetitorMonitor()
    
    async def get_competitive_price(self, product_id):
        competitors = await self.competitor_monitor.get_competitor_prices(product_id)
        
        # Calculate competitive position
        avg_competitor_price = np.mean([c['price'] for c in competitors])
        
        # Decide strategy
        strategy = self.determine_strategy(competitors)
        
        if strategy == 'undercut':
            return avg_competitor_price * 0.98  # 2% below average
        elif strategy == 'premium':
            return avg_competitor_price * 1.05  # 5% above average
        else:
            return avg_competitor_price  # Match average

8.2 Personalized Pricing

class PersonalizedPricer:
    def __init__(self):
        self.user_segmenter = UserSegmenter()
    
    async def get_personalized_price(self, product_id, user_id):
        # Get user segment
        segment = await self.user_segmenter.get_segment(user_id)
        
        # Get base price
        base_price = await self.get_base_price(product_id)
        
        # Apply segment-based adjustment
        adjustment = self.get_segment_adjustment(segment)
        
        personalized_price = base_price * adjustment
        
        # Ensure within bounds
        personalized_price = self.enforce_bounds(personalized_price, product_id)
        
        return personalized_price
    
    def get_segment_adjustment(self, segment):
        adjustments = {
            'price_sensitive': 0.95,  # 5% discount
            'loyal_customer': 0.90,   # 10% discount
            'premium_user': 1.05,     # 5% premium
            'new_user': 0.90          # 10% discount
        }
        return adjustments.get(segment, 1.0)

9. Implementation Roadmap

Phase 1: Basic Pricing (Weeks 1-4)

• Cost-plus pricing
• Basic demand forecasting
• Simple rule engine

Phase 2: ML-Based Pricing (Weeks 5-8)

• Demand forecasting models
• Price elasticity estimation
• A/B testing framework

Phase 3: Advanced Optimization (Weeks 9-12)

• Reinforcement learning
• Competitive pricing
• Personalized pricing

Phase 4: Optimization (Weeks 13-16)

• Latency optimization
• Cost optimization
• Compliance monitoring

10. Summary and Key Takeaways

Architecture Recap

1. Demand forecasting: Predict demand at different prices
2. Price optimization: Find optimal price given constraints
3. Real-time updates: Update prices based on conditions
4. Monitoring: Track performance and fairness

Key Metrics

• Revenue: Maximize revenue while maintaining fairness
• Accuracy: Demand predictions within 5%
• Latency: < 100ms for price calculation

Common Interview Mistakes

1. Not discussing fairness constraints
2. Ignoring competitor response
3. Forgetting about regulations
4. Not considering inventory constraints

Further Reading

• "Dynamic Pricing in E-commerce" (Amazon Science)
• "Reinforcement Learning for Pricing" (Uber Research)
• "Price Elasticity Estimation" (Stanford)
• "Fairness in Dynamic Pricing" (ACM)