mcpField
D
distribution-center-network

distribution-center-network

github.com/kishorkukrejaその他8447 回インストール3 週間前に更新
リポジトリ

distribution-center-network について

工場、DC、顧客を結ぶ配送センターネットワークを計画するためのスキルです。コストやサービス制約を整理し、最適化モデルとヒューリスティック手法を案内します。

主な機能

  1. 01多段階ネットワーク設計
  2. 02コストとサービス水準のモデル化
  3. 03施設と容量の意思決定
  4. 04フロー最適化手法
  5. 05クラスタリングと割当ヒューリスティック

活用シーン

  1. 01直送とDC経由配送の比較
  2. 02DCの立地と容量の計画
  3. 03輸送・在庫・サービスコストの調整

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/kishorkukreja/awesome-supply-chain --skill distribution-center-network

インストール後、Claude Code で/distribution-center-networkを入力して読み込まれたか確認します。

詳細

配送センターネットワーク設計

配送センターネットワークの設計と多階層サプライチェーン最適化の専門家として、サービス要件を満たしながら総コストを最小化する最適な配送ネットワークの設計を支援します。工場から配送センターを経て顧客に至るフローを考慮します。

初期評価

配送センターネットワークを設計する前に、以下を把握します。

  1. ネットワーク構造

    • 現在のネットワーク構成はどのようなものですか?
    • 単一階層(配送センター → 顧客のみ)ですか?
    • 多階層(工場 → 配送センター → 顧客)ですか?
    • 直送は可能ですか?(工場 → 顧客)
    • クロスドッキングを行っていますか?
  2. 施設の階層

    • 製造工場/供給元はありますか?
    • 地域配送センター(RDC)はありますか?
    • 地方配送センター(LDC)はありますか?
    • 前方在庫拠点(FSL)はありますか?
    • 顧客の所在地はどこですか?
  3. 製品の特性

    • 製品の種類(SKU数)はどれくらいですか?
    • 製品ファミリーやカテゴリはありますか?
    • 保管要件(常温、冷蔵、危険物)は何ですか?
    • 価値密度(体積/重量あたりの価値)はどれくらいですか?
    • 賞味期限や腐敗しやすさはどうですか?
  4. フローのパターン

    • 製品の出荷元はどこですか?(国内工場、輸入品、サプライヤー)
    • 需要地はどこですか?(顧客、市場、小売店舗)
    • 輸送レーンごとの数量(出発地と到着地の組み合わせ)はどれくらいですか?
    • 季節的なパターンはありますか?
    • 成長予測はありますか?
  5. コスト構造

    • 工場/供給元のコスト(生産、荷役)はいくらですか?
    • 配送センターの固定費(賃料、資本、設置)はいくらですか?
    • 配送センターの運営コスト(荷役、人件費、光熱費)はいくらですか?
    • 輸送コスト(入荷、配送センター間、出荷)はいくらですか?
    • 在庫コスト(循環在庫、安全在庫)はいくらですか?
    • サービス未達のペナルティコスト(配送遅延、販売機会損失)はいくらですか?

配送ネットワーク設計フレームワーク

ネットワーク構成の選択肢

1. 直送(0階層)

Plants/Sources → Customers
  • メリット:中間の荷役がなく、輸送時間が最短です
  • デメリット:少量出荷では輸送コストが高くなります
  • 最適な対象:高価値で時間に敏感な製品、大口注文

2. 単一階層(1階層)

Plants/Sources → Distribution Centers → Customers
  • メリット:集約効果、規模の経済が得られます
  • デメリット:荷役が増え、配送センターに在庫が必要になります
  • 最適な対象:多くのB2Bおよび小売向け配送

3. 二階層(2階層)

Plants → Regional DCs → Local DCs → Customers
  • メリット:市場カバー率が向上し、配送が速くなります
  • デメリット:複雑化し、在庫や施設が増えます
  • 最適な対象:広い地理的エリア、高いサービス要件

4. ハイブリッドネットワーク

Plants ─→ Regional DCs ─→ Customers
     └─→ Direct ────────┘
  • メリット:柔軟性があり、製品や顧客に応じて最適化できます
  • デメリット:計画と実行が複雑になります
  • 最適な対象:差別化されたサービス戦略
主な設計上の意思決定

戦略(長期):

  • 階層の数
  • 配送センターの数と所在地
  • 配送センターの規模と処理能力
  • 技術と自動化のレベル
  • 所有形態(自社運営またはアウトソーシング/3PL)

戦術(中期):

  • 製品と配送センターの割り当て
  • 顧客と配送センターの割り当て
  • 在庫の配置
  • 輸送手段
  • 季節性への対応策

運用(短期):

  • 注文のフルフィルメント先の割り当て
  • 補充の意思決定
  • 車両経路の設定
  • 積荷の集約

数学的定式化

多階層配送センターネットワーク設計モデル

集合:

  • P:工場/供給元
  • D:配送センターの候補地
  • C:顧客
  • K:製品

パラメーター:

  • f_d:配送センター d を開設する固定費
  • Q_d:配送センター d の処理能力
  • c_{pd}:工場 p から配送センター d への単位コスト
  • c_{dc}:配送センター d から顧客 c への単位コスト
  • c_{pc}:工場 p から顧客 c への直送単位コスト(直送が可能な場合)
  • d_{ck}:顧客 c の製品 k に対する需要
  • α_k:配送センターにおける製品 k の単位在庫コスト
  • s_{pk}:工場 p における製品 k の供給/生産能力

意思決定変数:

  • y_d ∈ {0,1}:配送センター d を開設する場合は 1
  • x_{pdk}:工場 p から配送センター d への製品 k のフロー
  • z_{dck}:配送センター d から顧客 c への製品 k のフロー
  • w_{pck}:工場 p から顧客 c への製品 k のフロー(直送が可能な場合)

目的関数:

Minimize:
  DC fixed costs:
    Σ_d f_d × y_d

  + Inbound transportation (plant → DC):
    Σ_p Σ_d Σ_k c_{pd} × x_{pdk}

  + Outbound transportation (DC → customer):
    Σ_d Σ_c Σ_k c_{dc} × z_{dck}

  + Direct shipment costs (if allowed):
    Σ_p Σ_c Σ_k c_{pc} × w_{pck}

  + DC inventory costs:
    Σ_d Σ_k α_k × (Σ_c z_{dck})

制約条件:

1. Demand satisfaction:
   Σ_d z_{dck} + Σ_p w_{pck} = d_{ck},  ∀c ∈ C, ∀k ∈ K

2. DC capacity:
   Σ_c Σ_k z_{dck} ≤ Q_d × y_d,  ∀d ∈ D

3. Flow balance at DCs:
   Σ_p x_{pdk} = Σ_c z_{dck},  ∀d ∈ D, ∀k ∈ K

4. Plant capacity:
   Σ_d x_{pdk} + Σ_c w_{pck} ≤ s_{pk},  ∀p ∈ P, ∀k ∈ K

5. Serve only from open DCs:
   z_{dck} ≤ M × y_d,  ∀d ∈ D, ∀c ∈ C, ∀k ∈ K
   where M is a large constant (customer demand sum)

6. Non-negativity and binary:
   y_d ∈ {0,1},  ∀d ∈ D
   x_{pdk}, z_{dck}, w_{pck} ≥ 0
サービスレベル制約付きモデル

追加パラメーター:

  • T_c:顧客 c に対して許容される最大配送時間
  • t_{dc}:DC d から顧客 c までの配送時間
  • SL:最低サービスレベル(例:需要の 95% を所定の時間内に満たす)

サービス制約:

1. Each customer must be within service time of at least one DC:
   Σ_{d: t_{dc} ≤ T_c} y_d ≥ 1,  ∀c ∈ C

2. Or aggregate service level:
   Σ_c Σ_{d: t_{dc} ≤ T_c} z_{dck} ≥ SL × Σ_c d_{ck},  ∀k ∈ K

解法

1. 多段階 MIP モデル
from pulp import *
import numpy as np
import pandas as pd

def solve_dc_network_design(plants, dcs, customers, products,
                           dc_costs, capacities, inbound_costs,
                           outbound_costs, demands,
                           plant_capacities=None, allow_direct=False):
    """
    Solve multi-echelon DC network design problem

    Args:
        plants: list of plant IDs
        dcs: list of potential DC IDs
        customers: list of customer IDs
        products: list of product IDs
        dc_costs: dict {dc_id: fixed_cost}
        capacities: dict {dc_id: capacity}
        inbound_costs: dict {(plant, dc, product): cost}
        outbound_costs: dict {(dc, customer, product): cost}
        demands: dict {(customer, product): demand}
        plant_capacities: optional dict {(plant, product): capacity}
        allow_direct: allow direct plant-to-customer shipments

    Returns:
        optimal network design solution
    """

    # Create problem
    prob = LpProblem("DC_Network_Design", LpMinimize)

    # Decision variables
    # y[d] = 1 if DC d is opened
    y = LpVariable.dicts("DC", dcs, cat='Binary')

    # x[p,d,k] = flow from plant p to DC d for product k
    x = {}
    for p in plants:
        for d in dcs:
            for k in products:
                x[p,d,k] = LpVariable(f"inbound_{p}_{d}_{k}",
                                     lowBound=0, cat='Continuous')

    # z[d,c,k] = flow from DC d to customer c for product k
    z = {}
    for d in dcs:
        for c in customers:
            for k in products:
                z[d,c,k] = LpVariable(f"outbound_{d}_{c}_{k}",
                                     lowBound=0, cat='Continuous')

    # w[p,c,k] = direct flow from plant p to customer c (if allowed)
    w = {}
    if allow_direct:
        for p in plants:
            for c in customers:
                for k in products:
                    w[p,c,k] = LpVariable(f"direct_{p}_{c}_{k}",
                                         lowBound=0, cat='Continuous')

    # Objective: Minimize total cost
    # DC fixed costs
    fixed_cost_expr = lpSum([dc_costs[d] * y[d] for d in dcs])

    # Inbound transportation costs
    inbound_expr = lpSum([
        inbound_costs.get((p,d,k), 0) * x[p,d,k]
        for p in plants for d in dcs for k in products
    ])

    # Outbound transportation costs
    outbound_expr = lpSum([
        outbound_costs.get((d,c,k), 0) * z[d,c,k]
        for d in dcs for c in customers for k in products
    ])

    prob += fixed_cost_expr + inbound_expr + outbound_expr, "Total_Cost"

    # Constraints

    # 1. Demand satisfaction
    for c in customers:
        for k in products:
            if (c, k) in demands:
                if allow_direct:
                    prob += (
                        lpSum([z[d,c,k] for d in dcs]) +
                        lpSum([w[p,c,k] for p in plants]) ==
                        demands[c,k],
                        f"Demand_{c}_{k}"
                    )
                else:
                    prob += (
                        lpSum([z[d,c,k] for d in dcs]) == demands[c,k],
                        f"Demand_{c}_{k}"
                    )

    # 2. DC capacity constraints
    for d in dcs:
        prob += (
            lpSum([z[d,c,k]
                   for c in customers for k in products]) <=
            capacities[d] * y[d],
            f"Capacity_DC_{d}"
        )

    # 3. Flow balance at DCs
    for d in dcs:
        for k in products:
            prob += (
                lpSum([x[p,d,k] for p in plants]) ==
                lpSum([z[d,c,k] for c in customers]),
                f"Flow_Balance_{d}_{k}"
            )

    # 4. Plant capacity constraints (if specified)
    if plant_capacities:
        for p in plants:
            for k in products:
                if (p, k) in plant_capacities:
                    if allow_direct:
                        prob += (
                            lpSum([x[p,d,k] for d in dcs]) +
                            lpSum([w[p,c,k] for c in customers]) <=
                            plant_capacities[p,k],
                            f"Plant_Capacity_{p}_{k}"
                        )
                    else:
                        prob += (
                            lpSum([x[p,d,k] for d in dcs]) <=
                            plant_capacities[p,k],
                            f"Plant_Capacity_{p}_{k}"
                        )

    # 5. Serve only from open DCs (big-M constraint)
    for d in dcs:
        for c in customers:
            for k in products:
                if (c, k) in demands:
                    M = demands[c,k]  # Big-M value
                    prob += (
                        z[d,c,k] <= M * y[d],
                        f"Open_DC_{d}_{c}_{k}"
                    )

    # Solve
    import time
    start_time = time.time()
    prob.solve(PULP_CBC_CMD(msg=1, timeLimit=900))
    solve_time = time.time() - start_time

    # Extract solution
    if LpStatus[prob.status] in ['Optimal', 'Feasible']:
        open_dcs = [d for d in dcs if y[d].varValue > 0.5]

        # Extract flows
        inbound_flows = {}
        for p in plants:
            for d in dcs:
                for k in products:
                    if x[p,d,k].varValue > 0.01:
                        inbound_flows[p,d,k] = x[p,d,k].varValue

        outbound_flows = {}
        for d in dcs:
            for c in customers:
                for k in products:
                    if z[d,c,k].varValue > 0.01:
                        outbound_flows[d,c,k] = z[d,c,k].varValue

        # Calculate DC utilization
        dc_utilization = {}
        for d in open_dcs:
            total_flow = sum(z[d,c,k].varValue
                           for c in customers for k in products)
            dc_utilization[d] = (total_flow / capacities[d]) * 100

        # Cost breakdown
        total_fixed = sum(dc_costs[d] for d in open_dcs)
        total_inbound = sum(
            inbound_costs.get((p,d,k), 0) * x[p,d,k].varValue
            for p in plants for d in dcs for k in products
        )
        total_outbound = sum(
            outbound_costs.get((d,c,k), 0) * z[d,c,k].varValue
            for d in dcs for c in customers for k in products
        )

        return {
            'status': LpStatus[prob.status],
            'total_cost': value(prob.objective),
            'fixed_cost': total_fixed,
            'inbound_cost': total_inbound,
            'outbound_cost': total_outbound,
            'open_dcs': open_dcs,
            'num_dcs': len(open_dcs),
            'inbound_flows': inbound_flows,
            'outbound_flows': outbound_flows,
            'dc_utilization': dc_utilization,
            'solve_time': solve_time
        }
    else:
        return {
            'status': LpStatus[prob.status],
            'solve_time': solve_time
        }


# Example usage
if __name__ == "__main__":
    # Network data
    plants = ['P1', 'P2', 'P3']
    dcs = ['DC1', 'DC2', 'DC3', 'DC4', 'DC5']
    customers = [f'C{i}' for i in range(1, 21)]  # 20 customers
    products = ['ProductA', 'ProductB', 'ProductC']

    # DC costs and capacities
    dc_costs = {
        'DC1': 500000, 'DC2': 450000, 'DC3': 480000,
        'DC4': 520000, 'DC5': 470000
    }
    capacities = {
        'DC1': 10000, 'DC2': 8000, 'DC3': 9000,
        'DC4': 11000, 'DC5': 8500
    }

    # Generate random costs (simplified)
    np.random.seed(42)

    inbound_costs = {}
    for p in plants:
        for d in dcs:
            for k in products:
                inbound_costs[p,d,k] = np.random.uniform(10, 50)

    outbound_costs = {}
    for d in dcs:
        for c in customers:
            for k in products:
                outbound_costs[d,c,k] = np.random.uniform(5, 30)

    # Customer demands
    demands = {}
    for c in customers:
        for k in products:
            demands[c,k] = np.random.uniform(50, 200)

    # Plant capacities
    plant_capacities = {}
    for p in plants:
        for k in products:
            plant_capacities[p,k] = 5000

    print("="*70)
    print("DISTRIBUTION CENTER NETWORK DESIGN")
    print("="*70)
    print(f"Plants: {len(plants)}")
    print(f"Potential DCs: {len(dcs)}")
    print(f"Customers: {len(customers)}")
    print(f"Products: {len(products)}")
    print(f"Total demand: {sum(demands.values()):,.0f} units")

    # Solve
    result = solve_dc_network_design(
        plants, dcs, customers, products,
        dc_costs, capacities,
        inbound_costs, outbound_costs, demands,
        plant_capacities
    )

    print(f"\n{'='*70}")
    print(f"OPTIMAL SOLUTION")
    print(f"{'='*70}")
    print(f"Status: {result['status']}")
    print(f"Total Cost: ${result['total_cost']:,.2f}")
    print(f"  Fixed Costs: ${result['fixed_cost']:,.2f} "
          f"({result['fixed_cost']/result['total_cost']*100:.1f}%)")
    print(f"  Inbound Transport: ${result['inbound_cost']:,.2f} "
          f"({result['inbound_cost']/result['total_cost']*100:.1f}%)")
    print(f"  Outbound Transport: ${result['outbound_cost']:,.2f} "
          f"({result['outbound_cost']/result['total_cost']*100:.1f}%)")

    print(f"\nOpen DCs: {result['num_dcs']}")
    for dc in result['open_dcs']:
        print(f"  {dc}: Utilization={result['dc_utilization'][dc]:.1f}%, "
              f"Capacity={capacities[dc]:,.0f}")

    print(f"\nSolve Time: {result['solve_time']:.2f} seconds")

    print(f"\nSample Flows (first 10):")
    count = 0
    for (d, c, k), flow in result['outbound_flows'].items():
        if count >= 10:
            break
        print(f"  {d} → {c} ({k}): {flow:.2f} units")
        count += 1
2. ネットワークフロー最適化
def optimize_network_flows(open_dcs, plants, customers, products,
                          inbound_costs, outbound_costs, demands,
                          dc_capacities, plant_capacities):
    """
    Optimize flows through given DC network

    Given fixed DC locations, optimize product flows

    Args:
        open_dcs: list of open DC locations
        plants, customers, products: network nodes
        costs, demands, capacities: network parameters

    Returns:
        optimal flow pattern
    """

    prob = LpProblem("Network_Flow", LpMinimize)

    # Variables: flows only (DCs already decided)
    x = {}  # inbound flows
    for p in plants:
        for d in open_dcs:
            for k in products:
                x[p,d,k] = LpVariable(f"inbound_{p}_{d}_{k}",
                                     lowBound=0, cat='Continuous')

    z = {}  # outbound flows
    for d in open_dcs:
        for c in customers:
            for k in products:
                z[d,c,k] = LpVariable(f"outbound_{d}_{c}_{k}",
                                     lowBound=0, cat='Continuous')

    # Objective: Minimize transportation costs
    prob += (
        lpSum([inbound_costs.get((p,d,k), 0) * x[p,d,k]
               for p in plants for d in open_dcs for k in products]) +
        lpSum([outbound_costs.get((d,c,k), 0) * z[d,c,k]
               for d in open_dcs for c in customers for k in products]),
        "Total_Transport_Cost"
    )

    # Constraints

    # Demand satisfaction
    for c in customers:
        for k in products:
            if (c, k) in demands:
                prob += (
                    lpSum([z[d,c,k] for d in open_dcs]) == demands[c,k],
                    f"Demand_{c}_{k}"
                )

    # DC capacity
    for d in open_dcs:
        prob += (
            lpSum([z[d,c,k] for c in customers for k in products]) <=
            dc_capacities[d],
            f"DC_Capacity_{d}"
        )

    # Flow balance
    for d in open_dcs:
        for k in products:
            prob += (
                lpSum([x[p,d,k] for p in plants]) ==
                lpSum([z[d,c,k] for c in customers]),
                f"Balance_{d}_{k}"
            )

    # Plant capacity
    for p in plants:
        for k in products:
            if (p, k) in plant_capacities:
                prob += (
                    lpSum([x[p,d,k] for d in open_dcs]) <=
                    plant_capacities[p,k],
                    f"Plant_{p}_{k}"
                )

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    if LpStatus[prob.status] in ['Optimal', 'Feasible']:
        inbound_flows = {
            (p,d,k): x[p,d,k].varValue
            for p in plants for d in open_dcs for k in products
            if x[p,d,k].varValue > 0.01
        }

        outbound_flows = {
            (d,c,k): z[d,c,k].varValue
            for d in open_dcs for c in customers for k in products
            if z[d,c,k].varValue > 0.01
        }

        return {
            'status': LpStatus[prob.status],
            'total_cost': value(prob.objective),
            'inbound_flows': inbound_flows,
            'outbound_flows': outbound_flows
        }

    return {'status': LpStatus[prob.status]}

ヒューリスティック手法

1. クラスタリングベースの DC 選定
from sklearn.cluster import KMeans

def clustering_based_dc_selection(customer_locations, customer_demands,
                                  potential_dc_locations, num_dcs):
    """
    Use demand-weighted clustering to select DC locations

    Args:
        customer_locations: array of customer coordinates
        customer_demands: customer demand weights
        potential_dc_locations: candidate DC coordinates
        num_dcs: number of DCs to select

    Returns:
        selected DC indices
    """

    # Cluster customers into num_dcs groups
    weights = (customer_demands / customer_demands.min()).astype(int)
    weighted_locations = []

    for i, loc in enumerate(customer_locations):
        weighted_locations.extend([loc] * weights[i])

    weighted_locations = np.array(weighted_locations)

    # K-means clustering
    kmeans = KMeans(n_clusters=num_dcs, random_state=42)
    kmeans.fit(weighted_locations)

    cluster_centers = kmeans.cluster_centers_

    # Match cluster centers to nearest potential DC locations
    selected_dcs = []
    for center in cluster_centers:
        # Find nearest potential DC to cluster center
        distances = [np.linalg.norm(center - dc_loc)
                    for dc_loc in potential_dc_locations]
        nearest_dc = np.argmin(distances)

        if nearest_dc not in selected_dcs:
            selected_dcs.append(nearest_dc)

    # If we have fewer DCs than requested (due to duplicates),
    # add next best options
    while len(selected_dcs) < num_dcs:
        best_dc = None
        best_total_dist = float('inf')

        for dc_idx in range(len(potential_dc_locations)):
            if dc_idx in selected_dcs:
                continue

            # Calculate total distance from this DC to all customers
            total_dist = sum(
                customer_demands[i] *
                np.linalg.norm(potential_dc_locations[dc_idx] - customer_locations[i])
                for i in range(len(customer_locations))
            )

            if total_dist < best_total_dist:
                best_total_dist = total_dist
                best_dc = dc_idx

        if best_dc is not None:
            selected_dcs.append(best_dc)
        else:
            break

    return selected_dcs
2. 反復型ロケーション・アロケーション
def iterative_location_allocation(potential_dcs, customers, demands,
                                 transport_costs, dc_costs, capacities,
                                 num_dcs, max_iterations=50):
    """
    Iterative heuristic: alternate between location and allocation

    Args:
        potential_dcs: list of potential DC locations
        customers: list of customers
        demands: customer demands
        transport_costs: outbound transport costs
        dc_costs: DC fixed costs
        capacities: DC capacities
        num_dcs: number of DCs to open
        max_iterations: maximum iterations

    Returns:
        heuristic solution
    """

    # Initialize: select random DCs
    current_dcs = random.sample(potential_dcs, num_dcs)

    def allocate_customers(dcs):
        """Allocate each customer to nearest DC with capacity"""
        assignments = {}
        remaining_capacity = {d: capacities[d] for d in dcs}

        # Sort customers by total demand (largest first)
        sorted_customers = sorted(customers,
                                key=lambda c: demands[c],
                                reverse=True)

        for c in sorted_customers:
            # Find nearest DC with capacity
            best_dc = None
            min_cost = float('inf')

            for d in dcs:
                if remaining_capacity[d] >= demands[c]:
                    cost = transport_costs.get((d, c), float('inf'))
                    if cost < min_cost:
                        min_cost = cost
                        best_dc = d

            if best_dc:
                assignments[c] = best_dc
                remaining_capacity[best_dc] -= demands[c]
            else:
                # No capacity, assign to nearest anyway (infeasible)
                best_dc = min(dcs, key=lambda d: transport_costs.get((d,c), float('inf')))
                assignments[c] = best_dc

        return assignments

    def calculate_cost(dcs, assignments):
        """Calculate total cost"""
        fixed = sum(dc_costs[d] for d in dcs)
        transport = sum(
            transport_costs.get((assignments[c], c), 0) * demands[c]
            for c in customers if c in assignments
        )
        return fixed + transport

    best_dcs = current_dcs.copy()
    best_assignments = allocate_customers(current_dcs)
    best_cost = calculate_cost(current_dcs, best_assignments)

    for iteration in range(max_iterations):
        # Allocate customers to current DCs
        current_assignments = allocate_customers(current_dcs)

        # Relocate DCs: try swapping each open DC with each closed DC
        improved = False

        for open_dc in current_dcs:
            for closed_dc in potential_dcs:
                if closed_dc in current_dcs:
                    continue

                # Test swap
                test_dcs = [d if d != open_dc else closed_dc
                           for d in current_dcs]

                test_assignments = allocate_customers(test_dcs)
                test_cost = calculate_cost(test_dcs, test_assignments)

                if test_cost < best_cost - 1e-6:
                    best_cost = test_cost
                    best_dcs = test_dcs.copy()
                    best_assignments = test_assignments
                    current_dcs = test_dcs
                    improved = True
                    break

            if improved:
                break

        if not improved:
            break

    return {
        'open_dcs': best_dcs,
        'assignments': best_assignments,
        'total_cost': best_cost,
        'method': 'Iterative Location-Allocation'
    }

完全な DC ネットワークソルバー

class DCNetworkSolver:
    """
    Comprehensive Distribution Center Network Design Solver
    """

    def __init__(self):
        self.plants = None
        self.dcs = None
        self.customers = None
        self.products = None
        self.loaded = False

    def load_problem(self, network_data):
        """
        Load network design problem

        Args:
            network_data: dict with all problem data
                - plants, dcs, customers, products
                - costs, capacities, demands
        """
        self.plants = network_data['plants']
        self.dcs = network_data['dcs']
        self.customers = network_data['customers']
        self.products = network_data.get('products', ['Product1'])

        self.dc_costs = network_data['dc_costs']
        self.capacities = network_data['capacities']
        self.inbound_costs = network_data['inbound_costs']
        self.outbound_costs = network_data['outbound_costs']
        self.demands = network_data['demands']
        self.plant_capacities = network_data.get('plant_capacities')

        self.loaded = True

        print(f"Loaded DC network problem:")
        print(f"  Plants: {len(self.plants)}")
        print(f"  Potential DCs: {len(self.dcs)}")
        print(f"  Customers: {len(self.customers)}")
        print(f"  Products: {len(self.products)}")
        print(f"  Total demand: {sum(self.demands.values()):,.0f}")

    def solve_exact(self, allow_direct=False, time_limit=900):
        """Solve with exact MIP"""
        if not self.loaded:
            raise ValueError("Problem not loaded")

        print("\nSolving with exact MIP...")
        return solve_dc_network_design(
            self.plants, self.dcs, self.customers, self.products,
            self.dc_costs, self.capacities,
            self.inbound_costs, self.outbound_costs,
            self.demands, self.plant_capacities,
            allow_direct
        )

    def solve_heuristic(self, method='clustering', num_dcs=None):
        """Solve with heuristic method"""
        print(f"\nSolving with {method} heuristic...")

        # Methods require different inputs
        # Implementation would depend on available data
        raise NotImplementedError("Heuristic methods need coordinates")

    def analyze_scenarios(self, dc_counts):
        """
        Analyze different numbers of DCs

        Args:
            dc_counts: list of DC counts to test

        Returns:
            comparison of scenarios
        """
        import pandas as pd

        results = []

        for num_dcs in dc_counts:
            print(f"\nAnalyzing scenario with {num_dcs} DCs...")

            # This would require solving with a constraint on number of DCs
            # or using heuristics

            # Placeholder
            results.append({
                'Num_DCs': num_dcs,
                'Total_Cost': None,
                'Fixed_Cost': None,
                'Transport_Cost': None
            })

        return pd.DataFrame(results)

    def visualize_network(self, solution, coordinates=None):
        """Visualize DC network"""
        import matplotlib.pyplot as plt

        if coordinates is None:
            print("Coordinates required for visualization")
            return

        # Visualization code
        # Similar to previous examples
        pass


# Example usage
if __name__ == "__main__":
    # Create sample network data
    network_data = {
        'plants': ['P1', 'P2'],
        'dcs': ['DC1', 'DC2', 'DC3', 'DC4'],
        'customers': [f'C{i}' for i in range(1, 16)],
        'products': ['ProductA', 'ProductB'],
        'dc_costs': {
            'DC1': 400000, 'DC2': 380000,
            'DC3': 420000, 'DC4': 390000
        },
        'capacities': {
            'DC1': 8000, 'DC2': 7000,
            'DC3': 9000, 'DC4': 7500
        },
        'inbound_costs': {},
        'outbound_costs': {},
        'demands': {},
        'plant_capacities': {}
    }

    # Generate costs and demands
    np.random.seed(42)

    for p in network_data['plants']:
        for d in network_data['dcs']:
            for k in network_data['products']:
                network_data['inbound_costs'][p,d,k] = np.random.uniform(15, 40)

    for d in network_data['dcs']:
        for c in network_data['customers']:
            for k in network_data['products']:
                network_data['outbound_costs'][d,c,k] = np.random.uniform(8, 25)

    for c in network_data['customers']:
        for k in network_data['products']:
            network_data['demands'][c,k] = np.random.uniform(30, 150)

    for p in network_data['plants']:
        for k in network_data['products']:
            network_data['plant_capacities'][p,k] = 3000

    # Solve
    solver = DCNetworkSolver()
    solver.load_problem(network_data)

    solution = solver.solve_exact()

    print(f"\n{'='*70}")
    print(f"OPTIMAL NETWORK CONFIGURATION")
    print(f"{'='*70}")
    print(f"Total Cost: ${solution['total_cost']:,.2f}")
    print(f"DCs Opened: {solution['num_dcs']}")
    print(f"DC IDs: {solution['open_dcs']}")

ツールとライブラリ

Python ライブラリ
  • PuLP/Pyomo:MIP モデリング
  • OR-Tools:ネットワーク最適化
  • NetworkX:ネットワーク分析
  • scikit-learn:クラスタリング
  • pandas/numpy:データ処理
商用ソフトウェア
  • Llamasoft:ネットワーク設計プラットフォーム
  • SAP IBP:統合計画
  • Blue Yonder:サプライチェーンスイート
  • AIMMS:最適化モデリング

よくある課題と解決策

課題:大規模ネットワーク

解決策: 分解、集約、ヒューリスティック

課題:複数製品による複雑さ

解決策: 製品集約、ABC 分析

課題:需要の変動

解決策: 確率モデル、ロバスト最適化

課題:サービス要件

解決策: サービスレベル制約、ペナルティコスト


出力形式

ネットワーク構成:

  • 総コスト:$X
  • 開設する DC:Y 拠点
  • 容量稼働率:Z%
  • サービスレベル:W%

確認する質問

  1. ネットワークの範囲は?(地域、国内、グローバル)
  2. 現状の設計ですか、それともグリーンフィールド設計ですか?
  3. 工場/供給元はいくつありますか?
  4. 製品の種類と特性は?
  5. 需要地と需要量は?
  6. サービス要件は?
  7. コスト構造は?
  8. 計画期間は?

関連スキル

  • facility-location-problem:一般的な立地理論
  • warehouse-location-optimization:倉庫に関する最適化
  • hub-location-problem:ハブネットワーク
  • network-flow-optimization:フロー最適化
  • inventory-routing-problem:在庫・配送ルーティングの統合
  • multi-echelon-inventory:在庫配置

コメント