仿真系统平台

Posted Jan 14, 2025

By jelech

25 min read

仿真系统平台

参考文档

开源项目

https://github.com/matsim-org/matsim-libs

https://github.com/eclipse-sumo/sumo

https://simpy.readthedocs.io/en/latest/

https://github.com/NetLogo/NetLogo?tab=readme-ov-file

https://github.com/matsim-org/matsim-libs

https://github.com/OpenModelica/OpenModelica

什么是物流仿真

当我们将物流系统搬进计算机世界，就诞生了“物流仿真”这一概念。它是一种通过数字化模型来模拟真实物流流程的技术手段。具体而言，企业会将仓库布局、货品分布、运输路线及人工/设备配置等细节输入到仿真软件中，再用算法“跑”出系统运作的过程。这样一来，我们不仅可以全局观察物流的流转，还能基于结果数据对方案进行调优。

为什么需要物流仿真？ 首先，通过仿真可以规避高昂的现实试错成本。很多物流决策（如新仓库选址、自动化设备采购、配送网络规划等）牵涉大笔资金与资源，一旦投入就难以轻易回头。仿真技术提供了一个安全的“虚拟实验场”，让人们在付出极低代价的前提下反复演练不同方案，找出更优解。其次，仿真输出的数据往往十分精细——包含车辆利用率、拣货等待时间、库位周转率等关键指标。如此“可视化”与“量化”的结果，有助于管理者精准地定位瓶颈，进而进行针对性改进。

值得一提的是，物流仿真并不只关注“货物搬来搬去”的片段化场景，它更强调对整个供应链体系的综合分析。随着智能设备和大数据技术的崛起，企业能更便利地采集到订单、库存、运力、运费等实时信息，并快速构建高保真的仿真模型。如此一来，从仓储机器人调度，到跨城物流路线设计，再到多仓协同分拣，都能在数字世界中率先“跑通”，并不断在实践中得到验证与迭代。

常用物流场景

仓库设计与布局: 通过仿真工具评估仓库的货架布局、堆垛机数量、拣货路径规划等方案，找出最佳的布局方式与调度策略。
运输与配送网络优化: 评估不同车辆数量、运力分配、线路规划以及订单分配策略，对比哪种方案更经济、更高效。
库存与供应链管理: 模拟原材料、半成品、成品在整个供应链中的流动，评估安全库存策略，平衡成本与服务水平。
人力与自动化设备调度: 研究在不同工作时段、人员排班或自动化设备配置下的产能与效率，以便更好地进行资源配置。

常用物流仿真方法

离散事件仿真（Discrete Event Simulation）: 对仓库进出货、排队、车辆装载、订单处理等离散过程进行模拟，工具包括 SimPy、JaamSim 等。
多智能体仿真（Multi-agent Simulation）: 对自动化设备、不同决策主体之间的交互建模，尤其适用于大规模复杂网络，如配送系统或车队调度。
网络/交通仿真: 对车辆、司机、城市路网进行模拟，代表工具有 MATSim 等，可研究城市级别或区域级别的物流运作。
机器人与3D物理仿真: 用于自动化仓储、机械臂拣选、AGV 调度场景，典型工具包括 Gazebo、PyBullet 等，可与机器人操作系统（ROS）或强化学习结合。

示例

以下为用simpy实现的仓储补货逻辑

简单模型

  
import simpy
import random

# -------------------------------
# 参数设置
# -------------------------------
RANDOM_SEED = 42
SIM_DAYS = 30                # 模拟天数
INITIAL_STOCK = 50           # 仓库初始库存
SAFETY_STOCK = 20            # 安全库存水平
ORDER_COST = 100             # 单位补货成本（示例，若无此需求，可忽略）
MAX_CAPACITY = 200           # 仓库最大容量（若有容量限制）
DAILY_DEMAND_MIN = 5         # 日需求最小值
DAILY_DEMAND_MAX = 15        # 日需求最大值

# 设定随机种子，保证复现结果
random.seed(RANDOM_SEED)


class Warehouse:
    """
    一个基础的仓库类，用于存储单一品类的库存量。
    """
    def __init__(self, env, name, initial_stock=INITIAL_STOCK, capacity=MAX_CAPACITY):
        self.env = env
        self.name = name
        self.capacity = capacity
        # 这里可以使用 SimPy 的 Container 资源来管理库存
        self.stock = simpy.Container(env, capacity=capacity, init=initial_stock)
        
    def __str__(self):
        return f"<Warehouse {self.name}, current={self.stock.level}, capacity={self.capacity}>"

    
def daily_demand():
    """
    返回某个品类的日需求量，这里使用均匀分布进行模拟。
    可替换成真实历史数据或其他分布模型（正态分布、泊松分布等）。
    """
    return random.randint(DAILY_DEMAND_MIN, DAILY_DEMAND_MAX)


def replenish_strategy(warehouse, daily_consumption, safety_stock=SAFETY_STOCK):
    """
    简易补货策略示例：
    若(当前库存 - 当日需求) < 安全库存，则补到最大容量的80%（或自定义逻辑）。
    返回需要补货的数量。若无需补货则返回 0。
    """
    current_level = warehouse.stock.level - daily_consumption
    if current_level < safety_stock:
        # 补货到容量的 80%
        target_level = 0.8 * warehouse.capacity
        order_qty = max(0, target_level - current_level)
        return int(order_qty)
    else:
        return 0


def daily_process(env, warehouses):
    """
    每天的主要流程：
      1. 所有仓库处理一天的需求；
      2. 在当日结束时做补货决策并执行补货；
      3. 统计或记录每日数据；
      4. 进入下一天。
    """
    while True:
        # 1. 处理需求
        daily_consumptions = {}
        for w in warehouses:
            demand = daily_demand()
            # 如果库存不够消耗，则只能消耗到 0
            actual_consumption = min(w.stock.level, demand)
            yield w.stock.get(actual_consumption)
            daily_consumptions[w.name] = actual_consumption
        
        # 2. 做补货决策 & 执行
        for w in warehouses:
            # 简单策略：如果低于安全库存，就补一些
            order_qty = replenish_strategy(w, daily_consumptions[w.name], SAFETY_STOCK)
            if order_qty > 0:
                # 此处可以模拟补货延迟(运输时间)，本例中直接即时补货
                yield w.stock.put(order_qty)
        
        # 3. 记录数据（可进一步扩展：写到文件、或将数据存到列表中）
        print(f"Day {env.now+1} Summary:")
        for w in warehouses:
            print(f"  {w.name}: Level after demand & replenish = {w.stock.level}")
        
        # 4. 等待下一天
        yield env.timeout(1)


def run_simulation():
    # 创建仿真环境
    env = simpy.Environment()

    # 创建多个仓库实例（可根据业务需求决定数量）
    warehouse_A = Warehouse(env, name="WH_A", initial_stock=INITIAL_STOCK)
    warehouse_B = Warehouse(env, name="WH_B", initial_stock=INITIAL_STOCK)
    warehouses = [warehouse_A, warehouse_B]
    
    # 将一个“每日流程”进程加入环境
    env.process(daily_process(env, warehouses))
    
    # 运行仿真
    print("Start simulation...")
    env.run(until=SIM_DAYS)
    print("Simulation finished.")


if __name__ == "__main__":
    run_simulation()

多个SKU模型

  
import simpy
import random

# -------------------------------
# 参数设置
# -------------------------------
RANDOM_SEED = 42
SIM_DAYS = 30                 # 模拟天数
SAFETY_STOCK = 20            # 安全库存水平
ORDER_UOM = 10               # 订货最小单位，例如最小起订量为 10 件/箱
WAREHOUSE_CAPACITY = 1000    # 仓库整体最大容量，这里简单处理（亦可细分到 SKU 层面）
LEAD_TIME = 2                # 补货交期，单位：天

# 模拟多个 SKU 的需求区间（最小值、最大值）
# Key: SKU 名称, Value: (最小需求, 最大需求)
SKU_DEMAND_RANGES = {
    'SKU_A': (5, 10),
    'SKU_B': (2, 8),
    'SKU_C': (0, 5),
}

# 每个 SKU 的初始库存量（可根据实际情况调整）
INITIAL_STOCKS = {
    'SKU_A': 50,
    'SKU_B': 30,
    'SKU_C': 10,
}

# 设定随机种子，保证结果可复现
random.seed(RANDOM_SEED)


class Warehouse:
    """
    仓库类，管理多个商品（SKU）的库存。
    对每个 SKU 使用一个 simpy.Container 或者使用字典管理库存量。
    """
    def __init__(self, env, name, sku_list, capacity=WAREHOUSE_CAPACITY):
        self.env = env
        self.name = name
        self.capacity = capacity  # 简化处理：仓库对所有 SKU 的总容量限制
        self.sku_list = sku_list
        
        # stock_dict 用来记录不同 SKU 的库存容器
        self.stock_dict = {}
        # 这里也可以给每个 SKU 设置各自的最大 capacity；本示例中统一使用 capacity
        for sku in sku_list:
            init_level = INITIAL_STOCKS.get(sku, 0)
            self.stock_dict[sku] = simpy.Container(env, capacity=capacity, init=init_level)
    
    def __repr__(self):
        return f"<Warehouse {self.name}>"

    def total_stock(self):
        """返回仓库内所有 SKU 当前总库存（可用于与capacity比较）。"""
        return sum([c.level for c in self.stock_dict.values()])


def daily_demand(sku):
    """
    返回指定 SKU 的日需求量，这里根据 SKU_DEMAND_RANGES 来随机生成。
    若需要不同分布，可自行替换。
    """
    low, high = SKU_DEMAND_RANGES[sku]
    return random.randint(low, high)


def replenish_strategy(warehouse, sku, daily_consumption, safety_stock=SAFETY_STOCK):
    """
    简易补货策略示例：
    若(当前库存 - 当日需求) < 安全库存，则订货量 = max(0, 补货到安全库存 + safety_stock)。
    亦可加入更多逻辑，比如固定订货批量 ORDER_UOM、或者一次性补满等。
    """
    current_level = warehouse.stock_dict[sku].level - daily_consumption
    if current_level < safety_stock:
        # 这里简单假设：补货目标 = 2 * safety_stock
        # 或者也可以 “补到 capacity 的 80%”、或 “一次性补 100 件”等
        target_level = 2 * safety_stock  
        order_qty = max(0, target_level - current_level)
        
        # 结合最小订货单位 ORDER_UOM 做向上取整
        # 例如最小单位 10，order_qty=12 -> 实际需下单 20
        if order_qty > 0:
            # 向上对齐到订货单位 ORDER_UOM
            remainder = order_qty % ORDER_UOM
            if remainder != 0:
                order_qty = order_qty + (ORDER_UOM - remainder)
                
        return int(order_qty)
    else:
        return 0


def ordering_process(env, warehouse, sku, order_qty):
    """
    模拟下单并等待 LEAD_TIME 天后将库存放入仓库。
    """
    print(f"[t={env.now}] {warehouse.name} 下单: SKU={sku}, 数量={order_qty}")
    yield env.timeout(LEAD_TIME)  # 等待 LEAD_TIME 天
    # 若仓库总容量允许，放入库存；如果有更复杂的容量逻辑，可再增加判断。
    # 这里简单示范一下容量判断：
    can_put = min(order_qty, warehouse.capacity - warehouse.total_stock())
    if can_put < order_qty:
        print(f"[t={env.now}] {warehouse.name} 受总容量限制，实际只能入库 {can_put}/{order_qty}")
    yield warehouse.stock_dict[sku].put(can_put)
    print(f"[t={env.now}] {warehouse.name} 收货完成: SKU={sku}, 入库数量={can_put}")


def daily_process(env, warehouses):
    """
    每天的主要流程：
      1. 所有仓库对每个 SKU 处理一天的需求；
      2. 在当日结束时做补货决策（若需补货则启动一个下单流程）；
      3. 记录/输出仓库的库存信息；
      4. 进入下一天。
    """
    while True:
        for warehouse in warehouses:
            for sku in warehouse.sku_list:
                # 1. 处理日需求
                demand = daily_demand(sku)
                actual_consumption = min(warehouse.stock_dict[sku].level, demand)
                yield warehouse.stock_dict[sku].get(actual_consumption)
        
        # 2. 做补货决策 (在当日结束时统一做)
        for warehouse in warehouses:
            for sku in warehouse.sku_list:
                # 可自行根据实时需求，或记录的当日销量来决定
                # 这里简单将一天前面消耗量视为 daily_consumption
                daily_consumption = None  # 若要精确得到当日消耗量，可使用记录方式
                # 简化处理：就用当日 demand 直接做策略
                # （若需要精确计算，可在上面保存 demand 到字典，然后在这里读取）
                demand_today = daily_demand(sku)  # 这里仅用作估计
                order_qty = replenish_strategy(warehouse, sku, demand_today, SAFETY_STOCK)
                if order_qty > 0:
                    # 发起一个异步的订货进程
                    env.process(ordering_process(env, warehouse, sku, order_qty))
        
        # 3. 输出每日结束时的库存信息
        print(f"=== End of Day {env.now+1} ===")
        for warehouse in warehouses:
            for sku in warehouse.sku_list:
                print(f"  {warehouse.name}, {sku}: {warehouse.stock_dict[sku].level}")
        
        # 4. 进入下一天
        yield env.timeout(1)


def run_simulation():
    # 创建仿真环境
    env = simpy.Environment()

    # 假设我们有两个仓库，每个仓库都管理 [SKU_A, SKU_B, SKU_C]
    warehouse_A = Warehouse(env, name="WH_A", sku_list=['SKU_A', 'SKU_B', 'SKU_C'])
    warehouse_B = Warehouse(env, name="WH_B", sku_list=['SKU_A', 'SKU_B', 'SKU_C'])
    warehouses = [warehouse_A, warehouse_B]
    
    # 将一个“每日流程”进程加入环境
    env.process(daily_process(env, warehouses))
    
    # 运行仿真
    print("Start simulation...")
    env.run(until=SIM_DAYS)
    print("Simulation finished.")


if __name__ == "__main__":
    run_simulation()

仓间调度模型

  
import simpy
import random

# -------------------------------
# 参数设置
# -------------------------------
RANDOM_SEED = 42
SIM_DAYS = 30                 # 模拟天数
SAFETY_STOCK = 20            # 安全库存水平
ORDER_UOM = 10               # 订货最小单位
WAREHOUSE_CAPACITY = 1000    # 仓库整体最大容量（可细分到 SKU 层）
LEAD_TIME = 2                # 外部采购补货交期，单位：天
TRANSFER_LEAD_TIME = 1       # 仓间调拨时间（运输时间），单位：天

# SKU 需求区间（最小值、最大值）
SKU_DEMAND_RANGES = {
    'SKU_A': (5, 10),
    'SKU_B': (2, 8),
    'SKU_C': (0, 5),
}

# 每个 SKU 的初始库存量（可根据实际情况调整）
INITIAL_STOCKS = {
    'SKU_A': 50,
    'SKU_B': 30,
    'SKU_C': 10,
}

# 设定随机种子，保证结果可复现
random.seed(RANDOM_SEED)


class Warehouse:
    """
    仓库类，管理多个商品（SKU）的库存。
    对每个 SKU 使用一个 simpy.Container 或者使用字典管理库存量。
    """
    def __init__(self, env, name, sku_list, capacity=WAREHOUSE_CAPACITY):
        self.env = env
        self.name = name
        self.capacity = capacity  # 简化：总容量限制
        self.sku_list = sku_list
        
        # stock_dict 用来记录不同 SKU 的库存容器
        self.stock_dict = {}
        for sku in sku_list:
            init_level = INITIAL_STOCKS.get(sku, 0)
            self.stock_dict[sku] = simpy.Container(env, capacity=capacity, init=init_level)
    
    def __repr__(self):
        return f"<Warehouse {self.name}>"

    def total_stock(self):
        """返回仓库内所有 SKU 当前总库存，用于对比容量等。"""
        return sum([c.level for c in self.stock_dict.values()])


def daily_demand(sku):
    """
    返回指定 SKU 的日需求量，这里根据 SKU_DEMAND_RANGES 来随机生成。
    """
    low, high = SKU_DEMAND_RANGES[sku]
    return random.randint(low, high)


def replenish_strategy(warehouse, sku, estimated_consumption, safety_stock=SAFETY_STOCK):
    """
    简易补货策略：
    若 (当前库存 - 当日需求估算) < 安全库存，则订货量 = 2 * safety_stock - (当前库存 - 当日需求估算)。
    订货量对齐到最小订货单位 ORDER_UOM 的整数倍。
    """
    current_level = warehouse.stock_dict[sku].level
    post_demand_level = current_level - estimated_consumption
    
    if post_demand_level < safety_stock:
        target_level = 2 * safety_stock
        order_qty = max(0, target_level - post_demand_level)
        
        # 向上对齐到最小订货单位 ORDER_UOM
        if order_qty > 0:
            remainder = order_qty % ORDER_UOM
            if remainder != 0:
                order_qty = order_qty + (ORDER_UOM - remainder)
                
        return int(order_qty)
    else:
        return 0


def ordering_process(env, warehouse, sku, order_qty):
    """
    模拟从外部供应商采购流程：等待 LEAD_TIME 后入库。
    """
    print(f"[t={env.now}] {warehouse.name} 下单外采: SKU={sku}, 数量={order_qty}")
    yield env.timeout(LEAD_TIME)
    
    # 检查仓库剩余可用容量（简单处理：统一容量）
    can_put = min(order_qty, warehouse.capacity - warehouse.total_stock())
    if can_put < order_qty:
        print(f"[t={env.now}] {warehouse.name} 受容量限制，只能入库 {can_put}/{order_qty}")
    yield warehouse.stock_dict[sku].put(can_put)
    print(f"[t={env.now}] {warehouse.name} 收货完成: SKU={sku}, 入库={can_put}")


def transfer_process(env, from_wh, to_wh, sku, transfer_qty):
    """
    模拟仓间调拨：从 from_wh 调拨给 to_wh，等待 TRANSFER_LEAD_TIME 后入库。
    """
    print(f"[t={env.now}] 仓间调拨: {from_wh.name} -> {to_wh.name}, SKU={sku}, 数量={transfer_qty}")
    yield env.timeout(TRANSFER_LEAD_TIME)
    
    # 调出仓库已经在调拨瞬间扣减库存，所以只需在目标仓库增加库存
    can_put = min(transfer_qty, to_wh.capacity - to_wh.total_stock())
    if can_put < transfer_qty:
        print(f"[t={env.now}] {to_wh.name} 因容量限制，只能接收 {can_put}/{transfer_qty}")
    yield to_wh.stock_dict[sku].put(can_put)
    print(f"[t={env.now}] {to_wh.name} 收到调拨: SKU={sku}, 入库={can_put}")


def warehouse_transfer_decision(env, warehouses, safety_stock=SAFETY_STOCK):
    """
    每日或固定时间执行的仓间调拨逻辑：
    1. 若某仓库某 SKU 低于安全库存，则尝试从其他仓库调拨；
    2. 目标是将其补到 >= 安全库存；
    3. 若其他仓库也不足或无库存富余，则无法调拨。
    
    此函数可以在 daily_process 中调用，也可独立设置频率（如每周执行一次）。
    """
    for to_wh in warehouses:
        for sku in to_wh.sku_list:
            to_wh_level = to_wh.stock_dict[sku].level
            if to_wh_level < safety_stock:
                # 需要的数量
                shortage_qty = safety_stock - to_wh_level
                # 从有库存富余的仓库调拨
                for from_wh in warehouses:
                    if from_wh is not to_wh:
                        from_wh_level = from_wh.stock_dict[sku].level
                        # 判断富余库存：简单策略，若 from_wh 库存 > 2 * safety_stock，说明可调拨部分
                        # 当然也可以灵活设置调拨阈值
                        surplus = from_wh_level - 2 * safety_stock
                        if surplus > 0:
                            # 实际可调拨量
                            transfer_qty = min(shortage_qty, surplus)
                            
                            # 在调拨瞬间先从 from_wh 扣减库存
                            yield from_wh.stock_dict[sku].get(transfer_qty)
                            
                            # 启动异步调拨流程
                            env.process(transfer_process(env, from_wh, to_wh, sku, transfer_qty))
                            
                            shortage_qty -= transfer_qty
                            if shortage_qty <= 0:
                                # 已满足调拨需求，不再从其他仓库继续调拨
                                break


def daily_process(env, warehouses):
    """
    每天的主要流程：
      1. 各仓库对每个 SKU 处理一天需求；
      2. 当日结束时做补货决策（若需外部采购则启动 ordering_process）；
      3. 进行仓间调拨决策（示例里放到每日流程，亦可改为其他频率）；
      4. 输出仓库库存信息；
      5. 进入下一天。
    """
    while True:
        # 1. 处理需求
        for warehouse in warehouses:
            for sku in warehouse.sku_list:
                demand = daily_demand(sku)
                actual_consumption = min(warehouse.stock_dict[sku].level, demand)
                yield warehouse.stock_dict[sku].get(actual_consumption)
        
        # 2. 做补货决策
        for warehouse in warehouses:
            for sku in warehouse.sku_list:
                # 估算今日需求(也可以记录第 1 步的实际消耗量，或做更复杂的预测)
                estimated_consumption = daily_demand(sku)  
                order_qty = replenish_strategy(warehouse, sku, estimated_consumption, SAFETY_STOCK)
                if order_qty > 0:
                    env.process(ordering_process(env, warehouse, sku, order_qty))
        
        # 3. 进行仓间调拨决策
        yield env.process(warehouse_transfer_decision(env, warehouses, SAFETY_STOCK))
        
        # 4. 每日结束时输出当前库存信息
        print(f"=== End of Day {env.now+1} ===")
        for warehouse in warehouses:
            for sku in warehouse.sku_list:
                print(f"  {warehouse.name}, {sku}: {warehouse.stock_dict[sku].level}")
        
        # 5. 进入下一天
        yield env.timeout(1)


def run_simulation():
    # 创建仿真环境
    env = simpy.Environment()

    # 假设我们有两个仓库，每个仓库都管理 [SKU_A, SKU_B, SKU_C]
    warehouse_A = Warehouse(env, name="WH_A", sku_list=['SKU_A', 'SKU_B', 'SKU_C'])
    warehouse_B = Warehouse(env, name="WH_B", sku_list=['SKU_A', 'SKU_B', 'SKU_C'])
    warehouses = [warehouse_A, warehouse_B]
    
    # 将一个“每日流程”进程加入环境
    env.process(daily_process(env, warehouses))
    
    # 运行仿真
    print("Start simulation...")
    env.run(until=SIM_DAYS)
    print("Simulation finished.")


if __name__ == "__main__":
    run_simulation()

开源项目学习

开源库学习笔记

This post is licensed under CC BY 4.0 by the author.

仿真系统平台

参考文档

相关博客

开源项目

什么是物流仿真

常用物流场景

常用物流仿真方法

示例

简单模型

多个SKU模型

仓间调度模型

Trending Tags