Logistics Expert

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
from enum import Enum

class StorageType(Enum):
    PALLET = "pallet"
    SHELF = "shelf"
    BULK = "bulk"
    COLD = "cold_storage"

@dataclass
class Location:
    location_id: str
    zone: str
    aisle: str
    rack: str
    level: int
    storage_type: StorageType
    capacity: float
    current_load: float

@dataclass
class Product:
    sku: str
    name: str
    category: str
    weight: float
    volume: float
    storage_requirements: str

@dataclass
class InventoryItem:
    item_id: str
    sku: str
    quantity: int
    location_id: str
    received_date: datetime
    expiry_date: Optional[datetime]
    batch_number: str

class WMS:
    """Warehouse Management System"""

    def __init__(self, db):
        self.db = db

    def receive_shipment(self, shipment):
        """Process incoming shipment"""
        items_received = []

        for item in shipment.items:
            # Find optimal storage location
            location = self.find_optimal_location(item)

            # Create inventory record
            inventory_item = InventoryItem(
                item_id=generate_id(),
                sku=item.sku,
                quantity=item.quantity,
                location_id=location.location_id,
                received_date=datetime.now(),
                expiry_date=item.expiry_date,
                batch_number=item.batch_number
            )

            self.db.save_inventory(inventory_item)
            self.update_location_capacity(location, item)

            items_received.append(inventory_item)

        return {
            'shipment_id': shipment.shipment_id,
            'items_received': len(items_received),
            'status': 'completed'
        }

    def find_optimal_location(self, item):
        """Find best storage location for item"""
        product = self.db.get_product(item.sku)
        available_locations = self.db.get_available_locations(
            storage_type=product.storage_requirements,
            min_capacity=product.volume * item.quantity
        )

        # Prioritize locations
        # 1. Same SKU for efficient picking
        # 2. Closest to shipping area for fast-moving items
        # 3. Maximize space utilization

        same_sku_locations = [
            loc for loc in available_locations
            if self.has_same_sku(loc, item.sku)
        ]

        if same_sku_locations:
            return same_sku_locations[0]

        # Select closest to shipping for fast-moving items
        if product.category == 'fast-moving':
            return min(available_locations, key=lambda l: l.distance_to_shipping)

        # Otherwise, optimize space utilization
        return max(available_locations, key=lambda l: l.utilization_score)

    def pick_order(self, order_id):
        """Generate picking list and route"""
        order = self.db.get_order(order_id)
        picking_list = []

        for line_item in order.line_items:
            inventory = self.db.find_inventory(
                sku=line_item.sku,
                quantity=line_item.quantity
            )

            picking_list.append({
                'sku': line_item.sku,
                'quantity': line_item.quantity,
                'location': inventory.location_id,
                'batch': inventory.batch_number
            })

        # Optimize picking route
        optimized_route = self.optimize_picking_route(picking_list)

        return {
            'order_id': order_id,
            'picking_list': optimized_route,
            'estimated_time': self.estimate_picking_time(optimized_route)
        }

    def optimize_picking_route(self, picking_list):
        """Optimize warehouse picking route"""
        # Sort by zone, aisle, rack for efficient walking path
        sorted_picks = sorted(
            picking_list,
            key=lambda x: (
                self.get_location_zone(x['location']),
                self.get_location_aisle(x['location']),
                self.get_location_rack(x['location'])
            )
        )

        return sorted_picks

    def check_stock_level(self, sku):
        """Check current stock level"""
        total_quantity = self.db.sum_quantity_by_sku(sku)
        product = self.db.get_product(sku)

        status = 'normal'
        if total_quantity <= product.reorder_point:
            status = 'reorder'
        elif total_quantity <= product.safety_stock:
            status = 'critical'

        return {
            'sku': sku,
            'quantity': total_quantity,
            'status': status,
            'reorder_point': product.reorder_point
        }

Route Optimization

import numpy as np
from scipy.spatial.distance import cdist
from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcp

class RouteOptimizer:
    """Vehicle routing optimization"""

    def optimize_delivery_routes(self, depot, deliveries, vehicles):
        """Optimize delivery routes using OR-Tools"""
        # Create distance matrix
        locations = [depot] + deliveries
        distance_matrix = self.create_distance_matrix(locations)

        # Create routing model
        manager = pywrapcp.RoutingIndexManager(
            len(distance_matrix),
            len(vehicles),
            0  # depot index
        )

        routing = pywrapcp.RoutingModel(manager)

        # Distance callback
        def distance_callback(from_index, to_index):
            from_node = manager.IndexToNode(from_index)
            to_node = manager.IndexToNode(to_index)
            return distance_matrix[from_node][to_node]

        transit_callback_index = routing.RegisterTransitCallback(distance_callback)
        routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)

        # Add capacity constraints
        def demand_callback(from_index):
            from_node = manager.IndexToNode(from_index)
            return deliveries[from_node - 1].weight if from_node > 0 else 0

        demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)

        routing.AddDimensionWithVehicleCapacity(
            demand_callback_index,
            0,  # null capacity slack
            [v.capacity for v in vehicles],
            True,  # start cumul to zero
            'Capacity'
        )

        # Solve
        search_parameters = pywrapcp.DefaultRoutingSearchParameters()
        search_parameters.first_solution_strategy = (
            routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
        )

        solution = routing.SolveWithParameters(search_parameters)

        if solution:
            return self.extract_routes(manager, routing, solution, locations)

        return None

    def create_distance_matrix(self, locations):
        """Create distance matrix from coordinates"""
        coords = np.array([(loc.lat, loc.lon) for loc in locations])
        return cdist(coords, coords, metric='euclidean')

    def extract_routes(self, manager, routing, solution, locations):
        """Extract optimized routes from solution"""
        routes = []

        for vehicle_id in range(routing.vehicles()):
            route = []
            index = routing.Start(vehicle_id)

            while not routing.IsEnd(index):
                node = manager.IndexToNode(index)
                route.append(locations[node])
                index = solution.Value(routing.NextVar(index))

            routes.append({
                'vehicle_id': vehicle_id,
                'stops': route,
                'total_distance': solution.ObjectiveValue()
            })

        return routes

Fleet Management

@dataclass
class Vehicle:
    vehicle_id: str
    type: str
    capacity_weight: float
    capacity_volume: float
    fuel_efficiency: float
    status: str
    current_location: dict
    maintenance_due: datetime

class FleetManagement:
    """Fleet tracking and management"""

    def __init__(self, db):
        self.db = db

    def assign_vehicle(self, delivery):
        """Assign optimal vehicle for delivery"""
        available_vehicles = self.db.get_available_vehicles(
            location=delivery.origin,
            min_capacity=delivery.total_weight
        )

        # Score vehicles based on:
        # - Distance from origin
        # - Capacity utilization
        # - Fuel efficiency
        # - Maintenance status

        best_vehicle = min(
            available_vehicles,
            key=lambda v: self.calculate_vehicle_score(v, delivery)
        )

        return best_vehicle

    def track_vehicle(self, vehicle_id):
        """Real-time vehicle tracking"""
        vehicle = self.db.get_vehicle(vehicle_id)
        current_route = self.db.get_current_route(vehicle_id)

        return {
            'vehicle_id': vehicle_id,
            'location': vehicle.current_location,
            'status': vehicle.status,
            'current_delivery': current_route.delivery_id if current_route else None,
            'eta': self.calculate_eta(vehicle, current_route) if current_route else None,
            'fuel_level': vehicle.fuel_level,
            'distance_traveled_today': vehicle.daily_distance
        }

    def schedule_maintenance(self, vehicle_id):
        """Schedule vehicle maintenance"""
        vehicle = self.db.get_vehicle(vehicle_id)
        maintenance_history = self.db.get_maintenance_history(vehicle_id)

        # Predictive maintenance based on:
        # - Mileage
        # - Hours of operation
        # - Previous issues
        # - Manufacturer schedule

        next_maintenance = self.predict_maintenance_date(
            vehicle,
            maintenance_history
        )

        return {
            'vehicle_id': vehicle_id,
            'next_maintenance': next_maintenance,
            'type': 'preventive',
            'estimated_cost': self.estimate_maintenance_cost(vehicle)
        }

Inventory Forecasting

from sklearn.ensemble import RandomForestRegressor
import pandas as pd

class DemandForecasting:
    """Demand forecasting and inventory optimization"""

    def forecast_demand(self, sku, forecast_period_days=30):
        """Forecast product demand"""
        # Get historical sales data
        historical_data = self.db.get_sales_history(sku, days=365)

        # Features: day of week, month, seasonality, promotions
        df = pd.DataFrame(historical_data)
        df['day_of_week'] = pd.to_datetime(df['date']).dt.dayofweek
        df['month'] = pd.to_datetime(df['date']).dt.month
        df['is_promotion'] = df['promotion_id'].notna().astype(int)

        X = df[['day_of_week', 'month', 'is_promotion']]
        y = df['quantity_sold']

        # Train model
        model = RandomForestRegressor(n_estimators=100)
        model.fit(X, y)

        # Generate forecast
        future_dates = pd.date_range(
            start=pd.Timestamp.now(),
            periods=forecast_period_days,
            freq='D'
        )

        forecast_features = pd.DataFrame({
            'day_of_week': future_dates.dayofweek,
            'month': future_dates.month,
            'is_promotion': 0  # Assume no promotions unless specified
        })

        predicted_demand = model.predict(forecast_features)

        return {
            'sku': sku,
            'forecast_period': forecast_period_days,
            'predicted_demand': predicted_demand.tolist(),
            'total_demand': sum(predicted_demand),
            'recommended_order_quantity': self.calculate_order_quantity(
                predicted_demand,
                sku
            )
        }

    def calculate_order_quantity(self, forecast, sku):
        """Calculate economic order quantity"""
        product = self.db.get_product(sku)
        total_demand = sum(forecast)

        # EOQ formula
        eoq = np.sqrt(
            (2 * total_demand * product.order_cost) /
            product.holding_cost
        )

        return int(eoq)

Best Practices

• Implement real-time tracking
• Use predictive analytics
• Optimize inventory levels
• Automate warehouse operations
• Enable multi-modal transportation
• Implement quality control
• Track KPIs (on-time delivery, accuracy)
• Use lean principles
• Enable reverse logistics
• Implement safety protocols

Anti-Patterns

❌ Manual inventory tracking ❌ No route optimization ❌ Overstocking or understocking ❌ Poor warehouse layout ❌ No real-time visibility ❌ Ignoring data analytics ❌ Inefficient picking processes

Resources

• Council of Supply Chain Management Professionals: https://cscmp.org/
• OR-Tools: https://developers.google.com/optimization
• Warehouse Management: https://www.mhi.org/
• Transportation Management: https://www.inboundlogistics.com/