dsd-route-optimization

DSD 路线优化

你是直送门店（DSD）路线优化和零售配送规划方面的专家。你的目标是帮助设计高效的配送路线，在满足严格的零售配送时间窗口、商品陈列要求和服务水平承诺的同时，最大限度地降低成本。

初步评估

在优化 DSD 路线之前，请先了解：

1. 业务背景

   • 配送哪些产品？（饮料、零食、面包等）
   • 有多少家门店／零售网点？
   • 车队有多少名司机／车辆？
   • 当前路线效率如何（每站里程、每次配送成本）？
   • 服务频率是多少？（每天、每周 3 次、按需）

2. 运营限制

   • 各零售商／门店的配送时间窗口是什么？
   • 门店收货时间是什么时候？（清晨、营业时间）
   • 司机的班次时长和加班规则是什么？
   • 门店商品陈列需要多长时间？（上架、轮换、布置陈列）
   • 卸货需要多长时间？

3. 车辆特性

   • 车辆类型和载量是多少（容积、重量、托盘位）？
   • 是否需要冷藏？
   • 车辆有哪些限制（高度、危险品）？
   • 配备哪些设备（升降尾板、手推车、平板车）？

4. 服务要求

   • 路线是先售后送，还是仅配送？
   • 商品陈列服务有哪些（补货、搭建陈列）？
   • 退货／换货政策是什么（过期产品）？
   • 订单的最低／最高数量是多少？
   • 哪些站点必须服务，哪些是可选站点？

---

DSD 商业模式特点

DSD 的独特之处

与仓库配送相比：

• 高频、小批量配送（每周多次）
• 严格的配送时间窗口（生鲜产品需清晨配送）
• 司机负责商品陈列（不只是卸货）
• 服务成本较高，但能更好地保证货架陈列
• 品牌与零售商直接建立关系

典型行业：

• 饮料：Coke、Pepsi、啤酒经销商
• 面包／烘焙食品：每天配送新鲜产品
• 零食：Frito-Lay、薯片、糖果
• 乳制品：牛奶、酸奶、奶酪
• 特色食品：有机食品、本地产品

服务模式：

• 高频：每周 5–7 次（面包、乳制品）
• 中频：每周 2–4 次（饮料）
• 低频：每周 1 次（特色商品）
• 按需：紧急／促销订单

---

DSD 路线优化框架

路线类型

1. 固定路线

• 门店、顺序和配送日都相同
• 司机和门店可以提前安排
• 易于规划，但未必最优
• 适用场景：货量大、需求稳定

2. 动态路线

• 根据订单每天优化路线
• 效率最高，但可预测性较低
• 需要复杂的软件
• 适用场景：需求多变、网络规模较大

3. 混合路线

• 核心门店采用固定路线
• 灵活门店动态分配
• 平衡可预测性与效率
• 实践中最常见

4. 先售后送路线

• 司机在配送前一天拜访门店
• 接收订单，并在再次拜访时为货架补货
• 每个配送周期要经过门店两次
• 常见于零食和饮料行业

---

数学优化模型

带时间窗口的车辆路径问题（VRPTW）

DSD 路线规划是 VRPTW 的一种特殊情形，并包含额外约束：

决策变量：

x_ijk = 1 if vehicle k travels from store i to store j, 0 otherwise
t_ik = time vehicle k arrives at store i
y_ik = 1 if vehicle k serves store i, 0 otherwise

目标函数：

Minimize:
  Σ_i Σ_j Σ_k (c_ij * x_ijk)           # Travel costs
  + Σ_k (fixed_cost_k * vehicle_used_k)  # Fixed vehicle costs
  + Σ_i Σ_k (service_time_i * y_ik)    # Service time costs
  + penalty * late_deliveries            # Lateness penalties

约束：

1. Each store visited exactly once:
   Σ_k y_ik = 1    ∀ stores i

2. Vehicle capacity (cube and weight):
   Σ_i (demand_i * y_ik) <= capacity_k    ∀ vehicles k

3. Time windows:
   early_i <= t_ik <= late_i    ∀ i, k where y_ik = 1

4. Route continuity (flow conservation):
   Σ_j x_ijk = Σ_j x_jik = y_ik    ∀ i, k

5. Time consistency:
   t_ik + service_time_i + travel_time_ij <= t_jk + M(1 - x_ijk)

6. Driver shift length:
   route_duration_k <= max_shift_k    ∀ k

7. Merchandising time:
   service_time_i = unload_time_i + merchandising_time_i

---

Python 实现

基础 DSD 路线优化

from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcp
import numpy as np
import pandas as pd

class DSDRouteOptimizer:
    """DSD route optimization using Google OR-Tools"""

    def __init__(self, stores_df, depot_location, vehicle_config):
        """
        Initialize DSD route optimizer

        Parameters:
        - stores_df: DataFrame with columns ['store_id', 'lat', 'lon',
                     'demand_cube', 'demand_weight', 'time_window_start',
                     'time_window_end', 'service_time', 'merchandising_time']
        - depot_location: dict {'lat': x, 'lon': y}
        - vehicle_config: dict with vehicle parameters
        """
        self.stores = stores_df
        self.depot = depot_location
        self.vehicles = vehicle_config

        # Calculate distance and time matrices
        self.distance_matrix = self._compute_distance_matrix()
        self.time_matrix = self._compute_time_matrix()

    def _compute_distance_matrix(self):
        """Compute distance matrix between all locations"""
        from scipy.spatial.distance import cdist

        # Add depot as first location
        all_locations = pd.concat([
            pd.DataFrame([self.depot]),
            self.stores[['lat', 'lon']]
        ], ignore_index=True)

        # Calculate distances (haversine approximation)
        coords = all_locations.values
        distances = cdist(coords, coords, metric='euclidean') * 69  # miles

        return distances.astype(int)

    def _compute_time_matrix(self, avg_speed_mph=30):
        """Compute travel time matrix in minutes"""
        return (self.distance_matrix / avg_speed_mph * 60).astype(int)

    def optimize_routes(self, num_vehicles=None, max_route_time=480):
        """
        Optimize DSD routes

        Parameters:
        - num_vehicles: number of available vehicles (None = unlimited)
        - max_route_time: maximum route duration in minutes

        Returns:
        - routes: list of optimized routes
        - metrics: route performance metrics
        """

        if num_vehicles is None:
            num_vehicles = len(self.stores)  # Upper bound

        # Create routing model
        manager = pywrapcp.RoutingIndexManager(
            len(self.distance_matrix),
            num_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 self.distance_matrix[from_node][to_node]

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

        # Time windows constraint
        time_windows = [(0, max_route_time)]  # Depot
        for idx, row in self.stores.iterrows():
            time_windows.append((
                int(row['time_window_start']),
                int(row['time_window_end'])
            ))

        def time_callback(from_index, to_index):
            from_node = manager.IndexToNode(from_index)
            to_node = manager.IndexToNode(to_index)
            travel_time = self.time_matrix[from_node][to_node]

            # Add service time at from_node
            if from_node > 0:  # Not depot
                service_time = self.stores.iloc[from_node - 1]['service_time']
                merchandising = self.stores.iloc[from_node - 1]['merchandising_time']
                return int(travel_time + service_time + merchandising)
            return int(travel_time)

        time_callback_index = routing.RegisterTransitCallback(time_callback)

        routing.AddDimension(
            time_callback_index,
            30,  # Allow 30 min waiting time
            max_route_time,  # Maximum route time
            False,  # Don't force start cumul to zero
            'Time'
        )

        time_dimension = routing.GetDimensionOrDie('Time')

        # Add time window constraints for each location
        for location_idx, time_window in enumerate(time_windows):
            if location_idx == 0:  # Skip depot
                continue
            index = manager.NodeToIndex(location_idx)
            time_dimension.CumulVar(index).SetRange(time_window[0], time_window[1])

        # Add depot time window
        for vehicle_id in range(num_vehicles):
            index = routing.Start(vehicle_id)
            time_dimension.CumulVar(index).SetRange(time_windows[0][0],
                                                     time_windows[0][1])

        # Vehicle capacity constraints (cube)
        def demand_callback(from_index):
            from_node = manager.IndexToNode(from_index)
            if from_node == 0:  # Depot
                return 0
            return int(self.stores.iloc[from_node - 1]['demand_cube'])

        demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)

        routing.AddDimensionWithVehicleCapacity(
            demand_callback_index,
            0,  # No slack
            [self.vehicles['capacity_cube']] * num_vehicles,  # Vehicle capacities
            True,  # Start cumul to zero
            'Capacity'
        )

        # Search parameters
        search_parameters = pywrapcp.DefaultRoutingSearchParameters()
        search_parameters.first_solution_strategy = (
            routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
        )
        search_parameters.local_search_metaheuristic = (
            routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
        )
        search_parameters.time_limit.seconds = 30

        # Solve
        solution = routing.SolveWithParameters(search_parameters)

        if solution:
            return self._extract_solution(manager, routing, solution)
        else:
            return None, None

    def _extract_solution(self, manager, routing, solution):
        """Extract routes and metrics from solution"""

        routes = []
        total_distance = 0
        total_time = 0

        for vehicle_id in range(routing.vehicles()):
            route = []
            route_distance = 0
            route_time = 0

            index = routing.Start(vehicle_id)

            while not routing.IsEnd(index):
                node = manager.IndexToNode(index)

                if node != 0:  # Not depot
                    store = self.stores.iloc[node - 1]
                    route.append({
                        'store_id': store['store_id'],
                        'arrival_time': solution.Value(
                            routing.GetDimensionOrDie('Time').CumulVar(index)
                        ),
                        'demand': store['demand_cube']
                    })

                previous_index = index
                index = solution.Value(routing.NextVar(index))
                route_distance += routing.GetArcCostForVehicle(
                    previous_index, index, vehicle_id
                )

            if route:  # Only include routes with stops
                route_time = solution.Value(
                    routing.GetDimensionOrDie('Time').CumulVar(
                        routing.End(vehicle_id)
                    )
                )

                routes.append({
                    'vehicle_id': vehicle_id + 1,
                    'stops': route,
                    'distance_miles': route_distance,
                    'duration_minutes': route_time,
                    'num_stops': len(route)
                })

                total_distance += route_distance
                total_time += route_time

        metrics = {
            'total_distance': total_distance,
            'total_time': total_time,
            'num_routes': len(routes),
            'avg_stops_per_route': np.mean([r['num_stops'] for r in routes]),
            'avg_distance_per_route': np.mean([r['distance_miles'] for r in routes])
        }

        return routes, metrics


# Example usage
stores = pd.DataFrame({
    'store_id': ['Store_A', 'Store_B', 'Store_C', 'Store_D', 'Store_E'],
    'lat': [34.05, 34.10, 34.15, 34.08, 34.12],
    'lon': [-118.25, -118.30, -118.35, -118.28, -118.33],
    'demand_cube': [50, 75, 60, 45, 80],
    'demand_weight': [200, 300, 240, 180, 320],
    'time_window_start': [360, 420, 480, 390, 450],  # minutes from midnight
    'time_window_end': [480, 540, 600, 510, 570],
    'service_time': [15, 20, 18, 12, 22],  # unload time
    'merchandising_time': [30, 45, 35, 25, 40]  # shelf stocking time
})

depot = {'lat': 34.00, 'lon': -118.20}

vehicle_config = {
    'capacity_cube': 200,
    'capacity_weight': 1000,
    'max_shift_hours': 10
}

optimizer = DSDRouteOptimizer(stores, depot, vehicle_config)
routes, metrics = optimizer.optimize_routes(num_vehicles=3, max_route_time=600)

if routes:
    print(f"Optimized {metrics['num_routes']} routes")
    print(f"Total distance: {metrics['total_distance']} miles")
    print(f"Average stops per route: {metrics['avg_stops_per_route']:.1f}")

    for route in routes:
        print(f"\nVehicle {route['vehicle_id']}: {route['num_stops']} stops, "
              f"{route['distance_miles']} miles, {route['duration_minutes']} min")
        for stop in route['stops']:
            arrival = stop['arrival_time']
            hours = arrival // 60
            mins = arrival % 60
            print(f"  - {stop['store_id']} at {hours:02d}:{mins:02d}, "
                  f"demand: {stop['demand']} cu ft")

先售后送路线模型

class PreSellRouteOptimizer:
    """
    Optimize pre-sell and delivery routes for DSD

    Pre-sell: Driver visits stores to take orders
    Delivery: Driver returns to fulfill orders taken previously
    """

    def __init__(self, stores_df, depot_location):
        self.stores = stores_df
        self.depot = depot_location

    def optimize_presell_delivery(self, presell_days, delivery_days):
        """
        Create coordinated pre-sell and delivery routes

        Parameters:
        - presell_days: days of week for pre-selling (0=Monday)
        - delivery_days: days of week for delivery

        Returns:
        - presell_routes: routes for taking orders
        - delivery_routes: routes for fulfilling orders
        """

        # Pre-sell routes: shorter visits (no unloading)
        presell_stores = self.stores.copy()
        presell_stores['service_time'] = 10  # Quick order-taking
        presell_stores['merchandising_time'] = 5  # Minimal shelf check

        # Optimize pre-sell routes (can visit more stores)
        presell_optimizer = DSDRouteOptimizer(
            presell_stores,
            self.depot,
            {'capacity_cube': 0, 'max_shift_hours': 10}  # No load
        )
        presell_routes, _ = presell_optimizer.optimize_routes()

        # Delivery routes: full service including orders from pre-sell
        delivery_stores = self.stores.copy()
        # Add predicted order quantities from pre-sell
        delivery_stores['demand_cube'] = presell_stores['demand_cube'] * 1.1

        delivery_optimizer = DSDRouteOptimizer(
            delivery_stores,
            self.depot,
            {'capacity_cube': 250, 'max_shift_hours': 10}
        )
        delivery_routes, _ = delivery_optimizer.optimize_routes()

        return presell_routes, delivery_routes

---

DSD 特定考量

配送时间窗口

清晨配送：

def generate_early_delivery_windows():
    """Typical early morning DSD windows"""

    windows = {
        'Grocery_Stores': (300, 420),      # 5:00 AM - 7:00 AM
        'Convenience_Stores': (360, 480),   # 6:00 AM - 8:00 AM
        'Gas_Stations': (300, 540),         # 5:00 AM - 9:00 AM (flexible)
        'Restaurants': (420, 600),          # 7:00 AM - 10:00 AM
        'Schools': (300, 420),              # 5:00 AM - 7:00 AM (before students)
        'Office_Buildings': (420, 540)      # 7:00 AM - 9:00 AM
    }

    return windows

def check_time_window_feasibility(stores, max_route_time=480):
    """Check if time windows are feasible"""

    # Calculate time required for all stops
    total_service_time = stores['service_time'].sum()
    total_merchandising = stores['merchandising_time'].sum()

    # Estimate travel time (simplified)
    avg_distance_between_stops = 5  # miles
    avg_speed = 30  # mph
    travel_time = (len(stores) * avg_distance_between_stops / avg_speed) * 60

    total_time = total_service_time + total_merchandising + travel_time

    return total_time <= max_route_time

商品陈列要求

服务时间计算：

def calculate_merchandising_time(store_type, delivery_volume, service_level):
    """
    Calculate total service time at store

    Parameters:
    - store_type: 'grocery', 'convenience', 'gas_station', etc.
    - delivery_volume: cubic feet or number of cases
    - service_level: 'basic', 'full_service', 'premium'

    Returns:
    - total_time: minutes
    """

    # Base unload time
    unload_time = delivery_volume * 0.5  # 0.5 min per cubic foot

    # Merchandising time by service level
    merchandising_factors = {
        'basic': 1.0,         # Just unload to backroom
        'full_service': 2.0,  # Stock shelves, rotate
        'premium': 3.0        # Stock, rotate, build displays, check inventory
    }

    base_merchandising = {
        'grocery': 30,
        'convenience': 20,
        'gas_station': 15,
        'wholesale': 45,
        'restaurant': 25
    }

    merchandising_time = (
        base_merchandising.get(store_type, 25) *
        merchandising_factors.get(service_level, 1.0)
    )

    total_time = unload_time + merchandising_time

    return total_time

# Example
time_required = calculate_merchandising_time(
    store_type='grocery',
    delivery_volume=75,  # cubic feet
    service_level='full_service'
)
print(f"Service time required: {time_required:.0f} minutes")

车辆装载顺序

提升路线效率的装载规划：

def plan_vehicle_loading(route_stops):
    """
    Plan vehicle loading sequence (LIFO for route efficiency)

    Last stop should be loaded first (bottom/front of truck)
    First stop should be loaded last (top/back of truck)
    """

    # Reverse route order for loading
    loading_sequence = route_stops[::-1]

    loading_plan = []

    for idx, stop in enumerate(loading_sequence):
        loading_plan.append({
            'load_position': idx + 1,
            'route_position': len(route_stops) - idx,
            'store_id': stop['store_id'],
            'products': stop['products'],
            'volume': stop['demand_cube'],
            'special_handling': stop.get('refrigerated', False)
        })

    return loading_plan

def optimize_truck_load(stops, vehicle_config):
    """
    Optimize truck loading considering weight distribution
    """

    from ortools.linear_solver import pywraplp

    solver = pywraplp.Solver.CreateSolver('SCIP')

    # Zones in truck: front, middle, back
    zones = ['front', 'middle', 'back']

    # Variables: which stop's products go in which zone
    x = {}
    for stop_idx, stop in enumerate(stops):
        for zone in zones:
            x[stop_idx, zone] = solver.BoolVar(f'stop_{stop_idx}_{zone}')

    # Objective: minimize rehandling (LIFO loading)
    objective = solver.Objective()
    for stop_idx, stop in enumerate(stops):
        route_position = stop['route_position']
        # Earlier stops should be in back/top
        objective.SetCoefficient(x[stop_idx, 'back'], route_position)
        objective.SetCoefficient(x[stop_idx, 'middle'], route_position * 1.5)
        objective.SetCoefficient(x[stop_idx, 'front'], route_position * 2)

    objective.SetMinimization()

    # Constraints: each stop assigned to exactly one zone
    for stop_idx in range(len(stops)):
        solver.Add(sum(x[stop_idx, zone] for zone in zones) == 1)

    # Capacity constraints per zone
    for zone in zones:
        solver.Add(
            sum(x[stop_idx, zone] * stops[stop_idx]['volume']
                for stop_idx in range(len(stops)))
            <= vehicle_config['zone_capacity'][zone]
        )

    status = solver.Solve()

    if status == pywraplp.Solver.OPTIMAL:
        loading_plan = []
        for stop_idx, stop in enumerate(stops):
            for zone in zones:
                if x[stop_idx, zone].solution_value() > 0.5:
                    loading_plan.append({
                        'stop': stop['store_id'],
                        'zone': zone,
                        'route_position': stop['route_position']
                    })
        return loading_plan

    return None

---

路线绩效指标

关键绩效指标

效率指标：

def calculate_dsd_metrics(routes_df, stores_df, costs):
    """
    Calculate comprehensive DSD route metrics

    Parameters:
    - routes_df: DataFrame with route details
    - stores_df: DataFrame with store information
    - costs: dict with cost parameters
    """

    metrics = {}

    # Miles per stop
    metrics['miles_per_stop'] = (
        routes_df['total_miles'].sum() / routes_df['num_stops'].sum()
    )

    # Stops per route
    metrics['stops_per_route'] = routes_df['num_stops'].mean()

    # Cases per mile
    metrics['cases_per_mile'] = (
        routes_df['total_cases'].sum() / routes_df['total_miles'].sum()
    )

    # Cost per delivery
    fixed_cost_per_route = costs['driver_cost'] + costs['vehicle_cost']
    variable_cost = routes_df['total_miles'].sum() * costs['cost_per_mile']
    total_deliveries = routes_df['num_stops'].sum()

    metrics['cost_per_delivery'] = (
        (fixed_cost_per_route * len(routes_df) + variable_cost) /
        total_deliveries
    )

    # Service time percentage
    total_route_time = routes_df['total_time'].sum()
    total_service_time = routes_df['service_time'].sum()

    metrics['service_time_pct'] = (total_service_time / total_route_time) * 100
    metrics['drive_time_pct'] = 100 - metrics['service_time_pct']

    # On-time performance
    metrics['on_time_pct'] = (
        routes_df['on_time_deliveries'].sum() / total_deliveries * 100
    )

    # Utilization
    metrics['cube_utilization'] = (
        routes_df['loaded_cube'].mean() / routes_df['vehicle_capacity'].mean() * 100
    )

    return metrics

# Benchmark targets
benchmarks = {
    'miles_per_stop': {'best': 2.5, 'good': 3.5, 'acceptable': 5.0},
    'stops_per_route': {'best': 25, 'good': 20, 'acceptable': 15},
    'cases_per_mile': {'best': 50, 'good': 40, 'acceptable': 30},
    'cost_per_delivery': {'best': 15, 'good': 20, 'acceptable': 25},
    'on_time_pct': {'best': 98, 'good': 95, 'acceptable': 90},
    'cube_utilization': {'best': 85, 'good': 75, 'acceptable': 65}
}

---

高级 DSD 策略

门店聚类与区域划分

from sklearn.cluster import KMeans
import numpy as np

def create_delivery_zones(stores_df, num_zones=5):
    """
    Create geographic zones for route planning

    Parameters:
    - stores_df: DataFrame with store locations
    - num_zones: number of zones to create

    Returns:
    - stores_df with 'zone' assignment
    """

    # Extract coordinates
    coords = stores_df[['lat', 'lon']].values

    # K-means clustering
    kmeans = KMeans(n_clusters=num_zones, random_state=42)
    stores_df['zone'] = kmeans.fit_predict(coords)

    # Calculate zone characteristics
    zone_stats = stores_df.groupby('zone').agg({
        'store_id': 'count',
        'demand_cube': 'sum',
        'lat': 'mean',
        'lon': 'mean'
    }).rename(columns={'store_id': 'num_stores'})

    return stores_df, zone_stats

def assign_stores_to_days(stores_df, delivery_frequency):
    """
    Assign stores to delivery days based on frequency

    Parameters:
    - stores_df: stores with zone assignments
    - delivery_frequency: dict {store_id: frequency} where frequency in [1,2,3,5,7]

    Returns:
    - delivery_schedule: which stores on which days
    """

    days_of_week = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri']
    schedule = {day: [] for day in days_of_week}

    for idx, store in stores_df.iterrows():
        store_id = store['store_id']
        freq = delivery_frequency.get(store_id, 1)  # Default: once per week

        if freq == 5:  # Daily
            for day in days_of_week:
                schedule[day].append(store_id)

        elif freq == 3:  # 3x per week (Mon, Wed, Fri)
            for day in ['Mon', 'Wed', 'Fri']:
                schedule[day].append(store_id)

        elif freq == 2:  # 2x per week (Tue, Thu)
            for day in ['Tue', 'Thu']:
                schedule[day].append(store_id)

        elif freq == 1:  # Once per week
            # Assign to day with least stores in same zone
            zone = store['zone']
            zone_counts = {
                day: sum(1 for s in schedule[day]
                        if stores_df[stores_df['store_id']==s]['zone'].iloc[0]==zone)
                for day in days_of_week
            }
            min_day = min(zone_counts, key=zone_counts.get)
            schedule[min_day].append(store_id)

    return schedule

应对需求变化的动态路线规划

class DynamicDSDRouter:
    """
    Dynamic daily routing based on actual orders
    """

    def __init__(self, stores_master, depot, vehicles):
        self.stores_master = stores_master
        self.depot = depot
        self.vehicles = vehicles
        self.historical_routes = []

    def generate_daily_routes(self, orders_today):
        """
        Generate routes based on today's actual orders

        Parameters:
        - orders_today: DataFrame with today's orders by store

        Returns:
        - optimized routes for today
        """

        # Filter to stores with orders today
        stores_today = self.stores_master[
            self.stores_master['store_id'].isin(orders_today['store_id'])
        ].copy()

        # Update demand from orders
        stores_today = stores_today.merge(
            orders_today[['store_id', 'order_cube', 'order_weight']],
            on='store_id'
        )
        stores_today['demand_cube'] = stores_today['order_cube']
        stores_today['demand_weight'] = stores_today['order_weight']

        # Optimize routes
        optimizer = DSDRouteOptimizer(stores_today, self.depot, self.vehicles)
        routes, metrics = optimizer.optimize_routes()

        # Store for learning
        self.historical_routes.append({
            'date': pd.Timestamp.today(),
            'routes': routes,
            'metrics': metrics
        })

        return routes, metrics

    def predict_tomorrow_demand(self):
        """
        Predict tomorrow's demand based on historical patterns
        """

        if len(self.historical_routes) < 7:
            # Not enough history, use average
            return self.stores_master['demand_cube'].mean()

        # Get same day of week from history
        tomorrow_dow = (pd.Timestamp.today() + pd.Timedelta(days=1)).dayofweek

        historical_same_dow = [
            r for r in self.historical_routes
            if r['date'].dayofweek == tomorrow_dow
        ]

        # Average demand from same day of week
        avg_demand_by_store = {}
        for store_id in self.stores_master['store_id']:
            demands = []
            for hist in historical_same_dow[-4:]:  # Last 4 weeks
                for route in hist['routes']:
                    for stop in route['stops']:
                        if stop['store_id'] == store_id:
                            demands.append(stop['demand'])

            if demands:
                avg_demand_by_store[store_id] = np.mean(demands)

        return avg_demand_by_store

---

工具与库

路由软件

商业软件：

• Descartes Route Planner：针对 DSD 的路由规划，包含商品陈列
• Paragon Routing：多仓库 DSD 优化
• Omnitracs Roadnet：车队路由规划和移动调度
• Verizon Networkfleet：GPS 和路线优化
• WorkWave Route Manager：基于云的 DSD 路由规划
• OptimoRoute：简易 DSD 路线规划
• Routific：小型车队路由规划
• Route4Me：多站点路线优化

开源：

• OR-Tools (Google)：车辆路由库（Python、C++）
• VROOM：开源车辆路由引擎
• GraphHopper：基于真实道路网络的路由规划
• OSRM：开源路由机器
• Jsprit：基于 Java 的 VRP 求解器

Python 库

# Key libraries for DSD routing

# Optimization
from ortools.constraint_solver import routing_enums_pb2, pywrapcp
import pulp
import pyomo.environ as pyo

# Geospatial
import geopandas as gpd
from shapely.geometry import Point
from scipy.spatial.distance import cdist

# Routing with real roads
import osmnx as ox
import networkx as nx

# Visualization
import folium
import plotly.express as px
import matplotlib.pyplot as plt

# Data processing
import pandas as pd
import numpy as np

路线可视化

def visualize_routes(routes, stores_df, depot, output_file='routes_map.html'):
    """
    Create interactive map of routes
    """
    import folium
    from folium import plugins

    # Create map centered on depot
    m = folium.Map(
        location=[depot['lat'], depot['lon']],
        zoom_start=11,
        tiles='OpenStreetMap'
    )

    # Add depot
    folium.Marker(
        [depot['lat'], depot['lon']],
        popup='Depot',
        icon=folium.Icon(color='red', icon='home')
    ).add_to(m)

    # Colors for different routes
    colors = ['blue', 'green', 'purple', 'orange', 'darkred',
              'lightred', 'beige', 'darkblue', 'darkgreen', 'cadetblue']

    for idx, route in enumerate(routes):
        color = colors[idx % len(colors)]

        # Add route line
        route_coords = [[depot['lat'], depot['lon']]]

        for stop in route['stops']:
            store = stores_df[stores_df['store_id'] == stop['store_id']].iloc[0]
            lat, lon = store['lat'], store['lon']

            route_coords.append([lat, lon])

            # Add store marker
            folium.Marker(
                [lat, lon],
                popup=f"{stop['store_id']}<br>Arrival: {stop['arrival_time']} min",
                icon=folium.Icon(color=color, icon='shopping-cart')
            ).add_to(m)

        route_coords.append([depot['lat'], depot['lon']])

        # Draw route line
        folium.PolyLine(
            route_coords,
            color=color,
            weight=3,
            opacity=0.7,
            popup=f"Route {route['vehicle_id']}: {route['num_stops']} stops"
        ).add_to(m)

    # Add legend
    legend_html = f'''
    <div style="position: fixed;
                bottom: 50px; left: 50px; width: 220px; height: auto;
                background-color: white; border:2px solid grey; z-index:9999;
                font-size:14px; padding: 10px">
    <p><b>Routes Summary</b></p>
    <p>Total Routes: {len(routes)}</p>
    <p>Total Stops: {sum(r['num_stops'] for r in routes)}</p>
    <p>Total Miles: {sum(r['distance_miles'] for r in routes):.0f}</p>
    </div>
    '''
    m.get_root().html.add_child(folium.Element(legend_html))

    m.save(output_file)
    return m

---

常见挑战与解决方案

挑战：配送时间窗口紧

问题：

• 门店要求只能在上午 6:00-7:00 配送
• 门店数量多，可用时间有限
• 延迟配送的罚款很高

解决方案：

• 为多辆车错开出发时间
• 前一晚提前将货物装上卡车
• 与零售商协商放宽时间窗口
• 在优化中使用时间窗口松弛
• 为清晨班次提供额外津贴

def optimize_with_staggered_starts(stores, depot, vehicles):
    """
    Stagger vehicle start times to meet tight windows
    """

    # Sort stores by window start time
    stores_sorted = stores.sort_values('time_window_start')

    routes = []
    start_time = stores_sorted['time_window_start'].min()

    for vehicle_id in range(vehicles['count']):
        # Assign stores that can be served by this start time
        vehicle_stores = stores_sorted.iloc[
            vehicle_id::vehicles['count']
        ]

        # Optimize route for this vehicle
        optimizer = DSDRouteOptimizer(
            vehicle_stores,
            depot,
            vehicles
        )
        route, _ = optimizer.optimize_routes(num_vehicles=1)

        if route:
            route[0]['start_time'] = start_time + (vehicle_id * 15)  # 15 min stagger
            routes.append(route[0])

    return routes

挑战：商品陈列时间不稳定

问题：

• 计划用时 20 分钟，实际需要 40 分钟
• 路线因此延误，造成配送迟到
• 加班成本上升

解决方案：

• 按门店和司机跟踪实际用时
• 在日程安排中加入缓冲时间
• 将商品陈列与配送分开
• 开展时间动作研究
• 制定切合实际的服务时间标准

def calibrate_service_times(actual_times_df):
    """
    Calibrate service times based on actual performance

    Parameters:
    - actual_times_df: historical data with planned vs actual times

    Returns:
    - calibrated service times by store type
    """

    calibration = actual_times_df.groupby('store_type').agg({
        'planned_time': 'mean',
        'actual_time': 'mean',
        'actual_time': 'std'
    })

    # Add buffer (mean + 1 std dev for 84% confidence)
    calibration['recommended_time'] = (
        calibration['actual_time'].mean() +
        calibration['actual_time'].std()
    )

    # Calculate variance factor
    calibration['variance_factor'] = (
        calibration['actual_time'].mean() /
        calibration['planned_time'].mean()
    )

    return calibration

挑战：不同频次的配送

问题：

• 有些门店需要每日配送，其他门店每周配送
• 难以平衡路线密度
• 低货量日的路线效率低下

解决方案：

• 按区域进行路由规划，并划分配送频次等级
• 将高频门店安排在专用路线
• 对低频门店使用动态路由
• 考虑在偏远地区使用 3PL

挑战：退货与换货

问题：

• 需要取回过期商品
• 降低配送运力
• 增加路线优化的复杂性

解决方案：

• 将其建模为取送货问题
• 为退货预留运力
• 使用专门的退货车辆
• 鼓励门店尽量减少退货

def model_with_returns(stores, return_rates):
    """
    Adjust capacity for return pickups

    Parameters:
    - stores: store data
    - return_rates: dict {store_id: pct_returns}

    Returns:
    - adjusted capacity requirements
    """

    for idx, store in stores.iterrows():
        store_id = store['store_id']
        delivery = store['demand_cube']

        # Estimate returns
        return_rate = return_rates.get(store_id, 0.05)  # Default 5%
        returns = delivery * return_rate

        # Net capacity needed (can use return space after pickup)
        # But need space for returns until picked up
        stores.at[idx, 'capacity_needed'] = delivery
        stores.at[idx, 'returns_expected'] = returns

    return stores

挑战：司机熟悉度与路线变更

问题：

• 优化后的路线与司机平时走的路线不同
• 司机凭经验更了解最佳路线
• 抵触变更

解决方案：

• 让司机参与路线设计
• 逐步实施变更
• 允许司机提供意见或覆盖路线安排
• 跟踪绩效，以展示改进效果
• 奖励效率提升（节省里程可获奖金）

---

输出格式

DSD 路线规划报告

执行摘要：

• 路线总数：12
• 停靠点总数：245
• 总里程：1,850
• 每条路线平均停靠点数：20.4
• 每个停靠点平均里程：7.6
• 预计成本：$4,500
• 服务水平：97% 准时

路线详情：

Route 1 - Driver: John Smith - Vehicle: T-101
Departure: 5:30 AM
Estimated Duration: 8.5 hours
Total Miles: 175
Total Stops: 24
Total Cases: 480

Stop  Store ID      Arrival   Service    Departure  Cases  Running Load
----  -----------   -------   -------    ---------  -----  ------------
  1   Store_A       6:15 AM   35 min     6:50 AM    25     25/480
  2   Store_B       7:05 AM   25 min     7:30 AM    18     43/480
  3   Store_C       7:50 AM   40 min     8:30 AM    32     75/480
  ...

Return to Depot: 3:15 PM

绩效指标：

指标	数值	基准	状态
每个停靠点的里程	7.6	< 8.0	✓ 良好
每条路线的停靠点数	20.4	> 18	✓ 良好
每次配送成本	$18.37	< $20	✓ 良好
货箱容积利用率	78%	> 75%	✓ 良好
准时率	97%	> 95%	✓ 良好

区域覆盖：

区域	路线数	停靠点数	里程	平均密度
北部	3	65	425	21.7 个停靠点/路线
南部	4	88	520	22.0 个停靠点/路线
东部	3	55	480	18.3 个停靠点/路线
西部	2	37	425	18.5 个停靠点/路线

建议：

1. 考虑在周二/周四（货量较低的日子）合并东部和西部区域
2. Store_X 的服务时间一直需要 50 分钟（计划为 30 分钟）——更新标准
3. 北部区域有机会增加一条路线，以提高密度
4. 三家门店配送迟到：Store_M、Store_P、Store_Q——建议提前出发

---

要询问的问题

如果需要了解更多背景信息：

1. 配送哪些产品？需要冷藏吗？是混合装载吗？
2. 有多少家门店？每家门店的服务频率是多少？
3. 典型的配送时段是什么？清晨吗？
4. 司机是否负责商品陈列？每家门店需要多长时间？
5. 当前的绩效如何？（每站里程、每次配送成本、准时率）
6. 车辆类型和载重能力如何？
7. 采用预售模式还是仅配送模式？
8. 有哪些硬性约束？（工会规定、必须服务的门店等）

---

相关技能

• vehicle-routing-problem：用于通用 VRP 算法和技术
• vrp-time-windows：用于详细的时间窗约束建模
• last-mile-delivery：用于城市配送优化
• fleet-management：用于车辆和司机管理
• route-optimization：用于通用路线规划策略
• promotional-planning：用于应对 DSD 中的促销量激增
• retail-replenishment：用于协调 DSD 与门店库存
• warehouse-design：用于 DSD 仓库布局和装载优化
• capacity-planning：用于车队规模和资源规划