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. 事前販売ルート

• ドライバーが配送の前日に店舗を訪問
• 注文を受け、次回の訪問時に棚へ商品を補充
• 1サイクルにつき店舗を2回訪問
• スナックや飲料で一般的です

---

数理最適化モデル

時間枠付き車両経路問題（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時～7時の納品のみを受け付ける
• 複数の店舗に対応する必要がある一方、時間が限られている
• 納品の遅延に対するペナルティが大きい

解決策：

• 複数の車両の出発時刻をずらす
• 前夜にトラックを事前配置する
• 小売業者と交渉して時間枠を広げる
• 最適化で時間枠の緩和を使用する
• 早朝シフトの割増賃金を設定する

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

課題：配送頻度が混在する納品

問題：

• 毎日納品が必要な店舗もあれば、週1回でよい店舗もある
• ルート密度のバランス調整が難しい
• 取扱量が少ない日のルートが非効率になる

解決策：

• 配送頻度の段階に応じてゾーンベースのルーティングを行う
• 配送頻度の高い店舗を専用ルートにまとめる
• 配送頻度の低い店舗には動的ルーティングを使用する
• 配送先がまばらな地域では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	✓ 良好
1回あたりの納品コスト	$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. 北ゾーンでは、ルートを1つ追加して密度を高められる可能性があります
4. 3店舗で納品が遅れています：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：車両台数とリソースの計画