이 글은 누구를 위한 것인가
- 서울, 부산, 인천 등 여러 창고에서 주문을 처리하는 이커머스 팀
- 어떤 창고에서 출고할지 자동화하고 싶은 물류 담당자
- 분할 배송과 단일 배송의 기준을 코드로 구현해야 하는 개발자
들어가며
창고가 하나면 단순하다. 창고가 여러 개가 되는 순간 "어디서 보낼까?"가 복잡한 최적화 문제가 된다. 배송비, 재고, 속도, 고객 위치 — 이 모든 변수를 고려해야 한다.
이 글은 bluefoxdev.kr의 이커머스 물류 최적화 가이드 를 참고하여 작성했습니다.
1. 창고 라우팅 결정 요소
[주문 라우팅 의사결정 트리]
입력: 주문 (상품 목록, 배송지)
Step 1: 재고 가용성 확인
→ 각 창고별 재고 보유 여부
Step 2: 단일 창고 배송 가능?
→ 모든 상품이 한 창고에 있음
→ 비용 최적 창고 선택
Step 3: 단일 창고 불가 → 분할 배송
→ 최소 창고 수로 커버 가능한 조합
→ 분할 배송 비용 vs 단일 재입고 후 발송 비교
Step 4: 우선순위 적용
Priority 1: 재고 가용성
Priority 2: 배송 가능 일자 (SLA)
Priority 3: 배송비 최소화
Priority 4: 창고 부하 균형
2. 라우팅 엔진 구현
from dataclasses import dataclass
from typing import Optional
import itertools
@dataclass
class Warehouse:
id: str
name: str
lat: float
lng: float
@dataclass
class WarehouseInventory:
warehouse_id: str
sku: str
available_qty: int
@dataclass
class ShippingRate:
from_warehouse_id: str
to_region: str
base_cost: float # 기본 배송비
per_kg_cost: float # kg당 추가 비용
estimated_days: int # 예상 배송일
def find_optimal_warehouse(
order_items: list[dict], # [{"sku": "A", "qty": 2}, ...]
delivery_address: dict,
warehouses: list[Warehouse],
inventory: dict[str, dict[str, int]], # {warehouse_id: {sku: qty}}
shipping_rates: dict[str, ShippingRate],
) -> dict:
"""
최적 창고 조합 찾기
Returns: {"assignments": [{warehouse_id, items}], "total_cost": N, "estimated_days": N}
"""
# 각 창고의 충족 가능 SKU 계산
warehouse_coverage = {}
for wh in warehouses:
wh_inv = inventory.get(wh.id, {})
covered = []
for item in order_items:
sku = item["sku"]
needed = item["qty"]
available = wh_inv.get(sku, 0)
if available >= needed:
covered.append(sku)
warehouse_coverage[wh.id] = set(covered)
all_skus = {item["sku"] for item in order_items}
# 단일 창고로 처리 가능한지 확인
for wh in warehouses:
if warehouse_coverage[wh.id] >= all_skus:
rate = shipping_rates.get(wh.id)
return {
"type": "single",
"assignments": [{"warehouse_id": wh.id, "items": order_items}],
"total_cost": rate.base_cost if rate else 0,
"estimated_days": rate.estimated_days if rate else 3,
}
# 최소 창고 수 분할 배송 찾기
best_solution = None
min_cost = float('inf')
for r in range(2, len(warehouses) + 1):
for combo in itertools.combinations(warehouses, r):
combined_coverage = set()
for wh in combo:
combined_coverage |= warehouse_coverage[wh.id]
if combined_coverage >= all_skus:
# 이 조합의 배송비 계산
total_cost = sum(
shipping_rates[wh.id].base_cost
for wh in combo
if wh.id in shipping_rates
)
max_days = max(
shipping_rates[wh.id].estimated_days
for wh in combo
if wh.id in shipping_rates
)
if total_cost < min_cost:
min_cost = total_cost
best_solution = {
"type": "split",
"assignments": _assign_items_to_warehouses(
order_items, combo, warehouse_coverage
),
"total_cost": total_cost,
"estimated_days": max_days,
}
if best_solution:
break
return best_solution or {"type": "unavailable", "assignments": [], "total_cost": 0}
def _assign_items_to_warehouses(order_items, warehouses, warehouse_coverage):
"""아이템을 창고에 배분"""
assignments = {wh.id: [] for wh in warehouses}
assigned_skus = set()
for item in order_items:
sku = item["sku"]
for wh in warehouses:
if sku in warehouse_coverage[wh.id] and sku not in assigned_skus:
assignments[wh.id].append(item)
assigned_skus.add(sku)
break
return [
{"warehouse_id": wh_id, "items": items}
for wh_id, items in assignments.items()
if items
]
3. 창고 부하 균형
def balance_warehouse_load(routing_result: dict, warehouse_loads: dict[str, int]) -> dict:
"""
동일 비용의 창고 중 부하가 적은 곳 선택
warehouse_loads: {warehouse_id: 현재_처리중_주문수}
"""
if routing_result["type"] == "single":
# 같은 비용의 창고들 중 가장 한가한 곳 선택
# (실제로는 같은 지역 창고들 중 선택)
assignments = routing_result["assignments"]
candidates = [a["warehouse_id"] for a in assignments]
least_loaded = min(candidates, key=lambda wh_id: warehouse_loads.get(wh_id, 0))
routing_result["assignments"][0]["warehouse_id"] = least_loaded
return routing_result
마무리
다중 창고 라우팅은 단순히 가까운 창고에서 보내는 것이 아니다. 재고 가용성, 배송비, 배송 속도, 창고 부하를 동시에 고려해야 한다. 처음엔 단순한 규칙 기반(가장 가까운 창고)으로 시작하고, 데이터를 보면서 최적화 알고리즘으로 발전시키는 것이 현실적이다.