基于成本优化的多仓库订单动态分配：代码输出异常排查

问题背景 #

现有A、B、C三个仓库，每个仓库设置不同的最低订单量slab，每个slab对应单位订单成本（cps）。需求是将给定总订单量分配至每个仓库至少一个slab，实现总成本最优。当前Python代码输出不符合预期，需排查修正以得到正确的最优分配结果。

示例数据与期望输出 #

示例输入：

• 总订单量：1000
• 仓库slab配置：

  A: [(50, 2), (100, 1.8), (200, 1.5)]  # (最低订单量, cps)
  B: [(30, 2.2), (80, 2), (150, 1.7)]
  C: [(40, 2.1), (90, 1.9), (180, 1.6)]
  

期望最优分配：A:670，B:150，C:180，总成本=670×1.5 + 150×1.7 + 180×1.6 = 1548

现有代码问题分析 #

假设用户提供的代码如下：

def calculate_optimal_allocation(total_order, warehouses):
    allocations = {w: None for w in warehouses}
    remaining = total_order
    # 给每个仓库选最低成本的slab
    for w in warehouses:
        best_slab = min(warehouses[w], key=lambda x: x[1])
        allocations[w] = best_slab[0]
        remaining -= best_slab[0]
    # 增量分配剩余订单
    while remaining > 0:
        best_w = min(warehouses.keys(), key=lambda x: warehouses[x][allocations[x]][1])
        allocations[best_w] += 10
        remaining -=10
    total_cost = sum(allocations[w] * warehouses[w][allocations[x]][1] for w in allocations)
    return allocations, total_cost

代码存在3个核心错误：

1. Slab匹配逻辑错误：用订单量直接作为slab列表的索引，无法正确匹配对应订单量的cps
2. 剩余分配效率低下：按固定增量（10）分配，未直接将剩余订单全部分配给成本最优的仓库
3. 初始分配逻辑缺失弹性：未考虑"总成本最优优先选低cps slab"的核心逻辑，且未处理初始slab总和超过总订单量的情况

修正后的代码 #

def get_slab_cps(order_qty, slabs):
    """根据订单量匹配对应slab的最低cps"""
    # 将slab按最低订单量升序排列
    sorted_slabs = sorted(slabs, key=lambda x: x[0])
    # 找到满足订单量≥最低值的最大slab（对应最低cps）
    for slab in reversed(sorted_slabs):
        if order_qty >= slab[0]:
            return slab[1]
    # 若订单量小于所有slab最低值，返回最小slab的cps
    return sorted_slabs[0][1]

def calculate_optimal_allocation(total_order, warehouses):
    # 步骤1：初始分配——优先给每个仓库选成本最低的slab（cps最小）
    initial_alloc = {}
    total_initial = 0
    for w, slabs in warehouses.items():
        # 按cps升序排序，取成本最低的slab
        best_slab = min(slabs, key=lambda x: x[1])
        initial_alloc[w] = best_slab[0]
        total_initial += best_slab[0]

    # 处理初始slab总和超过总订单量的情况：逐步降级slab直到总和≤总订单量
    while total_initial > total_order:
        candidates = []
        for w in warehouses:
            current_slab_val = initial_alloc[w]
            sorted_slabs = sorted(warehouses[w], key=lambda x: x[0])
            current_idx = next(i for i, s in enumerate(sorted_slabs) if s[0] == current_slab_val)
            # 只有存在更小的slab时才可以降级
            if current_idx > 0:
                prev_slab = sorted_slabs[current_idx-1]
                # 计算降级后的订单量减少值和成本增加量
                qty_reduce = current_slab_val - prev_slab[0]
                cost_increase = (prev_slab[1] - get_slab_cps(current_slab_val, warehouses[w])) * current_slab_val
                candidates.append((cost_increase, qty_reduce, w))
        # 优先选择成本增加最少、订单量减少最多的仓库降级
        candidates.sort(key=lambda x: (x[0], -x[1]))
        _, qty_reduce, best_w = candidates[0]
        initial_alloc[best_w] = sorted(warehouses[best_w], key=lambda x: x[0])[next(i for i, s in enumerate(sorted(warehouses[best_w], key=lambda x: x[0])) if s[0] == initial_alloc[best_w])-1][0]
        total_initial -= qty_reduce

    allocations = initial_alloc.copy()
    remaining = total_order - total_initial

    # 步骤2：分配剩余订单——全量分配给成本最优的仓库
    if remaining > 0:
        cost_options = []
        for w in warehouses:
            current_qty = allocations[w]
            new_qty = current_qty + remaining
            # 计算分配剩余订单后的成本变化
            cost_change = new_qty * get_slab_cps(new_qty, warehouses[w]) - current_qty * get_slab_cps(current_qty, warehouses[w])
            cost_options.append((cost_change, w))
        # 选择成本增加最少的仓库
        _, best_w = min(cost_options)
        allocations[best_w] += remaining

    # 计算最终总成本
    total_cost = sum(qty * get_slab_cps(qty, slabs) for qty, slabs in zip(allocations.values(), warehouses.values()))
    return allocations, total_cost

# 测试示例输入
warehouses = {
    'A': [(50, 2), (100, 1.8), (200, 1.5)],
    'B': [(30, 2.2), (80, 2), (150, 1.7)],
    'C': [(40, 2.1), (90, 1.9), (180, 1.6)]
}
total_order = 1000
allocations, total_cost = calculate_optimal_allocation(total_order, warehouses)
print(f"最优分配：{allocations}")
print(f"总成本：{total_cost}")

修正说明 #

1. 新增get_slab_cps函数：根据订单量正确匹配对应slab的cps，确保达到门槛后享受最低成本
2. 优化初始分配逻辑：优先选择每个仓库成本最低的slab，同时处理初始总和超量的降级逻辑
3. 剩余订单全量分配：直接计算将剩余订单分配给每个仓库的成本变化，一次性分配给最优仓库，避免增量分配的低效和错误
4. 总成本计算修正：通过get_slab_cps确保每个订单量匹配正确的cps

内容的提问来源于stack exchange，提问作者Abhijit Shinde