kishorkukreja/awesome-supply-chain/last-mile-delivery

import numpy as np
import matplotlib.pyplot as plt
from scipy.spatial import ConvexHull
class DeliveryDensityAnalyzer:
    """
    Analyze and optimize delivery density
    Higher density = more efficient routes and lower costs
    """
    def __init__(self, delivery_locations, orders_per_location):
        self.locations = np.array(delivery_locations)
        self.orders = np.array(orders_per_location)
    def calculate_density_metrics(self):
        """
        Calculate delivery density metrics
        Returns deliveries per square mile/km
        """
        # Calculate convex hull (service area)
        if len(self.locations) < 3:
            return {'error': 'Need at least 3 locations'}
        hull = ConvexHull(self.locations)
        service_area = hull.volume  # area in 2D
        total_deliveries = self.orders.sum()
        delivery_density = total_deliveries / service_area if service_area > 0 else 0
        return {
            'total_deliveries': total_deliveries,
            'service_area': service_area,
            'delivery_density': delivery_density,
            'num_locations': len(self.locations)
        }
    def identify_high_density_zones(self, radius=2.0):
        """
        Identify clusters of high delivery density
        These zones justify dedicated resources
        """
        from sklearn.cluster import DBSCAN
        # Cluster deliveries
        clustering = DBSCAN(eps=radius, min_samples=5).fit(self.locations)
        labels = clustering.labels_
        # Analyze clusters
        clusters = []
        for cluster_id in set(labels):
            if cluster_id == -1:  # Noise
                continue
            cluster_mask = labels == cluster_id
            cluster_locations = self.locations[cluster_mask]
            cluster_orders = self.orders[cluster_mask]
            # Calculate cluster metrics
            center = cluster_locations.mean(axis=0)
            total_orders = cluster_orders.sum()
            radius_from_center = np.max(
                np.linalg.norm(cluster_locations - center, axis=1)
            )
            area = np.pi * radius_from_center ** 2
            density = total_orders / area if area > 0 else 0
            clusters.append({
                'cluster_id': cluster_id,
                'center': center,
                'num_locations': cluster_mask.sum(),
                'total_orders': total_orders,
                'radius': radius_from_center,
                'density': density
            })
        return sorted(clusters, key=lambda x: x['density'], reverse=True)
    def recommend_density_improvements(self):
        """
        Recommend strategies to improve density
        """
        metrics = self.calculate_density_metrics()
        recommendations = []
        # Benchmark density (deliveries per sq mile)
        if metrics['delivery_density'] < 20:
            recommendations.append({
                'priority': 'High',
                'strategy': 'Focus marketing on underserved areas',
                'rationale': 'Density below 20 deliveries/sq mi is inefficient',
                'target': 'Increase to 30+ deliveries/sq mi'
            })
        if metrics['delivery_density'] < 50:
            recommendations.append({
                'priority': 'Medium',
                'strategy': 'Implement time-based incentives',
                'rationale': 'Encourage customers to cluster delivery times',
                'target': 'Increase deliveries per route by 20%'
            })
        # Check for sparse outliers
        clusters = self.identify_high_density_zones()
        if len(clusters) > 0:
            recommendations.append({
                'priority': 'High',
                'strategy': 'Focus on high-density clusters first',
                'rationale': f'Found {len(clusters)} high-density zones',
                'target': 'Prioritize service in dense clusters'
            })
        return recommendations
# Example usage
delivery_locations = np.random.rand(100, 2) * 10  # 100 locations in 10x10 grid
orders_per_location = np.random.poisson(3, 100)  # Poisson-distributed orders
analyzer = DeliveryDensityAnalyzer(delivery_locations, orders_per_location)
metrics = analyzer.calculate_density_metrics()
recommendations = analyzer.recommend_density_improvements()
print(f"Delivery density: {metrics['delivery_density']:.1f} deliveries/sq unit")
print(f"Recommendations: {len(recommendations)}")

import pandas as pd
from datetime import datetime, timedelta
class TimeWindowOptimizer:
    """
    Optimize delivery time windows
    Balance customer preference with operational efficiency
    """
    def __init__(self, historical_deliveries):
        """
        Parameters:
        - historical_deliveries: DataFrame with delivery data
          columns: ['customer_id', 'requested_window', 'actual_delivery',
                   'window_width_hours', 'successful']
        """
        self.deliveries = historical_deliveries
    def analyze_window_preferences(self):
        """
        Analyze customer time window preferences
        Returns distribution of preferred windows
        """
        # Extract hour from requested windows
        self.deliveries['preferred_start_hour'] = pd.to_datetime(
            self.deliveries['requested_window']
        ).dt.hour
        # Distribution by hour
        hourly_demand = self.deliveries.groupby('preferred_start_hour').size()
        return {
            'hourly_distribution': hourly_demand,
            'peak_hour': hourly_demand.idxmax(),
            'peak_demand': hourly_demand.max()
        }
    def calculate_window_efficiency(self):
        """
        Calculate efficiency by window width
        Wider windows = easier routing, but lower satisfaction
        """
        efficiency = self.deliveries.groupby('window_width_hours').agg({
            'successful': 'mean',  # Success rate
            'customer_id': 'count'  # Volume
        }).rename(columns={'customer_id': 'count', 'successful': 'success_rate'})
        return efficiency
    def recommend_window_strategy(self):
        """
        Recommend optimal window strategy
        Balance customer preference with operational needs
        """
        window_efficiency = self.calculate_window_efficiency()
        recommendations = []
        # Default: Offer 2-hour windows during peak times
        recommendations.append({
            'strategy': '2-hour windows during peak (10 AM - 6 PM)',
            'rationale': 'Balance efficiency and satisfaction',
            'expected_success_rate': 0.92
        })
        # Premium: 1-hour windows for higher fee
        recommendations.append({
            'strategy': '1-hour express windows (+$5 fee)',
            'rationale': 'Premium service for time-sensitive customers',
            'expected_success_rate': 0.85,
            'upcharge': 5.00
        })
        # Economy: 4-hour or all-day windows for discount
        recommendations.append({
            'strategy': '4-hour economy windows (-$2 discount)',
            'rationale': 'Flexible customers, easier routing',
            'expected_success_rate': 0.97,
            'discount': -2.00
        })
        return recommendations
    def dynamic_window_pricing(self, date, available_capacity):
        """
        Calculate dynamic pricing for time slots
        Surge pricing when capacity is constrained
        """
        # Predict demand for each time slot
        slots = [
            '8-10 AM', '10-12 PM', '12-2 PM',
            '2-4 PM', '4-6 PM', '6-8 PM'
        ]
        pricing = []
        for slot in slots:
            # Estimate demand (simplified - use ML in practice)
            base_demand = 100
            slot_demand = base_demand * np.random.uniform(0.7, 1.3)
            # Calculate capacity utilization
            utilization = slot_demand / available_capacity
            # Dynamic pricing
            if utilization < 0.7:
                price_modifier = 0.8  # 20% discount
                incentive = 'Discount - low demand'
            elif utilization < 0.9:
                price_modifier = 1.0  # Standard price
                incentive = 'Standard pricing'
            else:
                price_modifier = 1.5  # 50% surge
                incentive = 'Surge - high demand'
            pricing.append({
                'time_slot': slot,
                'demand': slot_demand,
                'capacity': available_capacity,
                'utilization': utilization,
                'price_modifier': price_modifier,
                'incentive': incentive
            })
        return pd.DataFrame(pricing)