┌─────────────┐     ┌──────────────┐     ┌─────────────┐     ┌─────────────┐
│   Customer  │────▶│   Usage      │────▶│   Pricing   │────▶│   Invoice   │
│   Activity  │     │   Metering   │     │   Engine    │     │   Generator│
└─────────────┘     └──────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │                   │
       ▼                   ▼                   ▼                   ▼
┌─────────────┐     ┌──────────────┐     ┌─────────────┐     ┌─────────────┐
│   Events    │     │   Usage      │     │   Price     │     │   Payment   │
│   Stream    │     │   Aggregation│     │   Tiers     │     │   Processing│
└─────────────┘     └──────────────┘     └─────────────┘     └─────────────┘

# Usage-based pricing configuration
pricing_model:
  type: "tiered"  # pure, tiered, hybrid, freemium
  
  # Usage metrics
  metrics:
    - name: "api_calls"
      unit: "count"
      aggregation: "sum"
      description: "Number of API calls"
    - name: "storage_gb"
      unit: "GB"
      aggregation: "max"
      description: "Storage usage in GB"
    - name: "compute_hours"
      unit: "hours"
      aggregation: "sum"
      description: "Compute usage in hours"
  
  # Pricing tiers
  tiers:
    - name: "starter"
      min_usage: 0
      max_usage: 10000
      price_per_unit: 0.001
      features: ["basic_support"]
    
    - name: "growth"
      min_usage: 10001
      max_usage: 100000
      price_per_unit: 0.0008
      features: ["priority_support", "analytics"]
    
    - name: "enterprise"
      min_usage: 100001
      max_usage: null
      price_per_unit: 0.0005
      features: ["dedicated_support", "sla", "custom_integrations"]
  
  # Volume discounts
  volume_discounts:
    - threshold: 1000000
      discount_percent: 10
    - threshold: 10000000
      discount_percent: 20
  
  # Billing settings
  billing:
    frequency: "monthly"
    currency: "USD"
    minimum_charge: 10.0
    free_tier:
      enabled: true
      free_units: 1000

# ❌ Bad - Fixed pricing, no usage tracking
def calculate_price(customer_id):
    # All customers pay the same regardless of usage
    return 100.0  # $100 per month

# ✅ Good - Usage-based pricing with tiers
def calculate_usage_price(customer_id, usage_metrics):
    # Calculate price based on actual usage
    api_calls = usage_metrics['api_calls']
    storage_gb = usage_metrics['storage_gb']
    
    # Tiered pricing for API calls
    api_price = calculate_tiered_price(api_calls, API_PRICE_TIERS)
    
    # Storage pricing
    storage_price = storage_gb * STORAGE_PRICE_PER_GB
    
    total_price = api_price + storage_price
    return total_price

# ❌ Bad - No volume discounts
def calculate_price(quantity, unit_price):
    return quantity * unit_price

# ✅ Good - Volume discounts
def calculate_price_with_discounts(quantity, unit_price, discount_tiers):
    price = quantity * unit_price
    
    # Apply volume discounts
    for threshold, discount in discount_tiers:
        if quantity >= threshold:
            price *= (1 - discount)
    
    return price

"""
Production-ready Usage-Based Pricing Engine
"""
from typing import Dict, List, Optional, Any, Tuple
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class PricingModel(Enum):
    """Types of pricing models."""
    PURE = "pure"
    TIERED = "tiered"
    HYBRID = "hybrid"
    FREEMIUM = "freemium"


class AggregationMethod(Enum):
    """Usage aggregation methods."""
    SUM = "sum"
    MAX = "max"
    MIN = "min"
    AVG = "avg"
    COUNT = "count"


@dataclass
class UsageMetric:
    """Usage metric definition."""
    name: str
    unit: str
    aggregation: AggregationMethod
    description: str


@dataclass
class PricingTier:
    """Pricing tier definition."""
    name: str
    min_usage: float
    max_usage: Optional[float]
    price_per_unit: float
    features: List[str] = field(default_factory=list)


@dataclass
class VolumeDiscount:
    """Volume discount definition."""
    threshold: float
    discount_percent: float


@dataclass
class UsageRecord:
    """Individual usage record."""
    customer_id: str
    timestamp: datetime
    metric_name: str
    value: float
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Invoice:
    """Invoice for a customer."""
    customer_id: str
    billing_period_start: datetime
    billing_period_end: datetime
    usage_breakdown: Dict[str, float]
    charges: Dict[str, float]
    total_amount: float
    currency: str


class UsageMeter:
    """
    Enterprise-grade usage metering system.
    """
    
    def __init__(self, metrics: List[UsageMetric]):
        """
        Initialize usage meter.
        
        Args:
            metrics: List of usage metrics to track
        """
        self.metrics = {m.name: m for m in metrics}
        self.usage_records: List[UsageRecord] = []
        
        logger.info(f"Usage meter initialized with {len(metrics)} metrics")
    
    def record_usage(
        self,
        customer_id: str,
        metric_name: str,
        value: float,
        timestamp: Optional[datetime] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> None:
        """
        Record usage for a customer.
        
        Args:
            customer_id: Customer ID
            metric_name: Name of the metric
            value: Usage value
            timestamp: Timestamp of usage (default: now)
            metadata: Additional metadata
        """
        if metric_name not in self.metrics:
            raise ValueError(f"Unknown metric: {metric_name}")
        
        if timestamp is None:
            timestamp = datetime.utcnow()
        
        record = UsageRecord(
            customer_id=customer_id,
            timestamp=timestamp,
            metric_name=metric_name,
            value=value,
            metadata=metadata or {}
        )
        
        self.usage_records.append(record)
        logger.debug(f"Recorded usage: {customer_id} {metric_name}={value}")
    
    def aggregate_usage(
        self,
        customer_id: str,
        metric_name: str,
        start_date: datetime,
        end_date: datetime
    ) -> float:
        """
        Aggregate usage for a customer over a period.
        
        Args:
            customer_id: Customer ID
            metric_name: Name of the metric
            start_date: Start of period
            end_date: End of period
            
        Returns:
            Aggregated usage value
        """
        if metric_name not in self.metrics:
            raise ValueError(f"Unknown metric: {metric_name}")
        
        metric = self.metrics[metric_name]
        
        # Filter records for customer and period
        records = [
            r for r in self.usage_records
            if r.customer_id == customer_id
            and r.metric_name == metric_name
            and start_date <= r.timestamp <= end_date
        ]
        
        if not records:
            return 0.0
        
        # Aggregate based on method
        values = [r.value for r in records]
        
        if metric.aggregation == AggregationMethod.SUM:
            return sum(values)
        elif metric.aggregation == AggregationMethod.MAX:
            return max(values)
        elif metric.aggregation == AggregationMethod.MIN:
            return min(values)
        elif metric.aggregation == AggregationMethod.AVG:
            return sum(values) / len(values)
        elif metric.aggregation == AggregationMethod.COUNT:
            return len(values)
        else:
            raise ValueError(f"Unknown aggregation method: {metric.aggregation}")
    
    def get_customer_usage(
        self,
        customer_id: str,
        start_date: datetime,
        end_date: datetime
    ) -> Dict[str, float]:
        """
        Get all usage for a customer over a period.
        
        Args:
            customer_id: Customer ID
            start_date: Start of period
            end_date: End of period
            
        Returns:
            Dictionary of metric names to aggregated values
        """
        usage = {}
        
        for metric_name in self.metrics:
            usage[metric_name] = self.aggregate_usage(
                customer_id, metric_name, start_date, end_date
            )
        
        return usage


class PricingEngine:
    """
    Enterprise-grade usage-based pricing engine.
    """
    
    def __init__(
        self,
        pricing_model: PricingModel,
        metrics: List[UsageMetric],
        tiers: List[PricingTier],
        volume_discounts: List[VolumeDiscount] = None,
        base_price: float = 0.0,
        free_tier_units: float = 0.0
    ):
        """
        Initialize pricing engine.
        
        Args:
            pricing_model: Type of pricing model
            metrics: List of usage metrics
            tiers: Pricing tiers
            volume_discounts: Volume discount tiers
            base_price: Base subscription price (for hybrid model)
            free_tier_units: Free units (for freemium model)
        """
        self.pricing_model = pricing_model
        self.metrics = {m.name: m for m in metrics}
        self.tiers = sorted(tiers, key=lambda t: t.min_usage)
        self.volume_discounts = sorted(volume_discounts or [], key=lambda d: d.threshold)
        self.base_price = base_price
        self.free_tier_units = free_tier_units
        
        self.usage_meter = UsageMeter(metrics)
        
        logger.info(f"Pricing engine initialized: {pricing_model.value} model")
    
    def calculate_tiered_price(
        self,
        usage: float,
        metric_name: str
    ) -> Tuple[float, PricingTier]:
        """
        Calculate price using tiered pricing.
        
        Args:
            usage: Usage amount
            metric_name: Name of the metric
            
        Returns:
            Tuple of (price, tier)
        """
        # Find applicable tier
        tier = None
        for t in self.tiers:
            if t.min_usage <= usage and (t.max_usage is None or usage <= t.max_usage):
                tier = t
                break
        
        if tier is None:
            # Default to highest tier if usage exceeds all tiers
            tier = self.tiers[-1]
        
        # Calculate price
        price = usage * tier.price_per_unit
        
        return price, tier
    
    def apply_volume_discounts(
        self,
        price: float,
        total_usage: float
    ) -> float:
        """
        Apply volume discounts to price.
        
        Args:
            price: Original price
            total_usage: Total usage across all metrics
            
        Returns:
            Discounted price
        """
        for discount in self.volume_discounts:
            if total_usage >= discount.threshold:
                price *= (1 - discount.discount_percent / 100)
        
        return price
    
    def calculate_price(
        self,
        customer_id: str,
        start_date: datetime,
        end_date: datetime
    ) -> Invoice:
        """
        Calculate price for a customer for a billing period.
        
        Args:
            customer_id: Customer ID
            start_date: Start of billing period
            end_date: End of billing period
            
        Returns:
            Invoice object
        """
        # Get usage for the period
        usage = self.usage_meter.get_customer_usage(
            customer_id, start_date, end_date
        )
        
        # Calculate charges
        charges = {}
        total_usage = 0.0
        
        for metric_name, value in usage.items():
            # Apply free tier
            if value > self.free_tier_units:
                billable_usage = value - self.free_tier_units
            else:
                billable_usage = 0.0
            
            # Calculate tiered price
            if billable_usage > 0:
                price, tier = self.calculate_tiered_price(billable_usage, metric_name)
                charges[metric_name] = {
                    'usage': billable_usage,
                    'price': price,
                    'tier': tier.name,
                    'unit_price': tier.price_per_unit
                }
                total_usage += billable_usage
        
        # Calculate total before discounts
        subtotal = sum(c['price'] for c in charges.values())
        
        # Add base price for hybrid model
        if self.pricing_model == PricingModel.HYBRID:
            subtotal += self.base_price
            charges['base_subscription'] = {
                'usage': 1,
                'price': self.base_price,
                'tier': 'base',
                'unit_price': self.base_price
            }
        
        # Apply volume discounts
        total_amount = self.apply_volume_discounts(subtotal, total_usage)
        
        # Create invoice
        invoice = Invoice(
            customer_id=customer_id,
            billing_period_start=start_date,
            billing_period_end=end_date,
            usage_breakdown=usage,
            charges=charges,
            total_amount=total_amount,
            currency="USD"
        )
        
        logger.info(
            f"Invoice generated for {customer_id}: "
            f"${total_amount:.2f} for {start_date} to {end_date}"
        )
        
        return invoice
    
    def estimate_price(
        self,
        projected_usage: Dict[str, float]
    ) -> float:
        """
        Estimate price based on projected usage.
        
        Args:
            projected_usage: Dictionary of metric names to projected usage
            
        Returns:
            Estimated price
        """
        total_price = 0.0
        total_usage = 0.0
        
        for metric_name, value in projected_usage.items():
            # Apply free tier
            if value > self.free_tier_units:
                billable_usage = value - self.free_tier_units
            else:
                billable_usage = 0.0
            
            # Calculate tiered price
            if billable_usage > 0:
                price, _ = self.calculate_tiered_price(billable_usage, metric_name)
                total_price += price
                total_usage += billable_usage
        
        # Add base price for hybrid model
        if self.pricing_model == PricingModel.HYBRID:
            total_price += self.base_price
        
        # Apply volume discounts
        total_price = self.apply_volume_discounts(total_price, total_usage)
        
        return total_price


# Example usage
if __name__ == "__main__":
    # Define usage metrics
    metrics = [
        UsageMetric(
            name="api_calls",
            unit="count",
            aggregation=AggregationMethod.SUM,
            description="Number of API calls"
        ),
        UsageMetric(
            name="storage_gb",
            unit="GB",
            aggregation=AggregationMethod.MAX,
            description="Storage usage in GB"
        )
    ]
    
    # Define pricing tiers
    tiers = [
        PricingTier(
            name="starter",
            min_usage=0,
            max_usage=10000,
            price_per_unit=0.001,
            features=["basic_support"]
        ),
        PricingTier(
            name="growth",
            min_usage=10001,
            max_usage=100000,
            price_per_unit=0.0008,
            features=["priority_support", "analytics"]
        ),
        PricingTier(
            name="enterprise",
            min_usage=100001,
            max_usage=None,
            price_per_unit=0.0005,
            features=["dedicated_support", "sla"]
        )
    ]
    
    # Define volume discounts
    discounts = [
        VolumeDiscount(threshold=1000000, discount_percent=10),
        VolumeDiscount(threshold=10000000, discount_percent=20)
    ]
    
    # Create pricing engine
    engine = PricingEngine(
        pricing_model=PricingModel.TIERED,
        metrics=metrics,
        tiers=tiers,
        volume_discounts=discounts,
        free_tier_units=1000
    )
    
    # Record usage for a customer
    customer_id = "cust_001"
    billing_start = datetime(2025, 1, 1)
    billing_end = datetime(2025, 1, 31)
    
    # Simulate usage records
    for i in range(100):
        engine.usage_meter.record_usage(
            customer_id=customer_id,
            metric_name="api_calls",
            value=1000,
            timestamp=billing_start + timedelta(days=i)
        )
    
    engine.usage_meter.record_usage(
        customer_id=customer_id,
        metric_name="storage_gb",
        value=500,
        timestamp=billing_start + timedelta(days=1)
    )
    
    # Calculate invoice
    invoice = engine.calculate_price(customer_id, billing_start, billing_end)
    
    print(f"\nInvoice for {customer_id}:")
    print(f"  Billing Period: {billing_start.date()} to {billing_end.date()}")
    print(f"  Total Amount: ${invoice.total_amount:.2f}")
    print(f"\n  Usage Breakdown:")
    for metric, value in invoice.usage_breakdown.items():
        print(f"    {metric}: {value}")
    print(f"\n  Charges:")
    for charge_name, charge_info in invoice.charges.items():
        print(f"    {charge_name}:")
        print(f"      Usage: {charge_info['usage']}")
        print(f"      Price: ${charge_info['price']:.2f}")
        print(f"      Tier: {charge_info['tier']}")
    
    # Estimate price for projected usage
    projected = {
        "api_calls": 50000,
        "storage_gb": 1000
    }
    estimated = engine.estimate_price(projected)
    print(f"\nEstimated price for projected usage: ${estimated:.2f}")

Revenue per Customer = Σ(Usage × Unit Price)

Gross Margin = (Revenue - COGS) / Revenue

Customer LTV = Average Monthly Profit × Customer Lifetime

Pricing Elasticity = % Change in Demand / % Change in Price