1. pegasusWMS到底是什么为什么说它适合做Jupyter范例仓库管理系统WMS听上去完全谈不上酷但只要是做电商、制造或者三方仓的项目它都是绕不开的核心。pegasusWMS是我整理的一套教学型WMS项目它用Python把仓库最常见的几条流程——收货、上架、拣货、出库、盘点以及库存分析——全部做成了可调用的模块并且配套一整套Jupyter范例notebook。最有意思的是你打开任何一个范例先在单元格里跑一段初始化再跑一段入库库存表就会真实地变化库位分配结果也会直接打印出来。对于整天面对黑盒系统的仓储运营人员这几乎是用来搞懂WMS底层逻辑的最佳方式对于Python学习者它又是一个有真实业务背景、有完整流程、能直接上手改参数的实战项目。1.1 它覆盖了WMS的哪些核心场景pegasusWMS没有试图去堆砌商业软件里那些复杂到令人头皮发麻的模块而是老老实实把仓库日常运转最核心的几条主线抽了出来入库、上架、出库、拣货、盘点和库存分析。每一个主流程都对应一个独立的notebook比如「入库流程与自动库位分配」「出库波次生成与批次优先」「月度库存周转率分析」「循环盘点与差异处理」。每个notebook里除了代码还会在关键步骤输出业务说明、中间表格和可视化图保证你不仅是“跑了代码”而是真的看懂了数据是怎么流转的。项目的数据模型也不复杂但足够完整核心实体包括商品SKU、库位、库存、入库单、出库单和库存流水。这里我直接列一张表方便你理解字段之间的关联实体关键字段作用SKUsku_id, name, category, weight定义商品主数据Locationlocation_id, zone, row, col, capacity定义库位三维编号和最大容量Inventorysku_id, location_id, qty, batch_date当前库存余额粒度到SKU库位InboundOrderorder_id, sku_id, qty, receive_date入库单据OutboundOrderorder_id, sku_id, qty, priority出库单据StockTransactiontx_id, sku_id, location_id, qty, action_type每次出入库的流水记录库存表并不是一个静态的数字每一次出入库都会先写入流水再更新余额。这样做的目的只有一个所有业务步骤都可以审计、可以回放。如果你在真实WMS里遇到过库存对不上账的情况就知道流水表有多重要。1.2 为什么选择Jupyter来承载范例说实话我最早也想过写一个命令行Demo直接打印结果就行了但做到一半发现这样根本没有学习效果。WMS的业务流程是有严格前后依赖的你入库之后库存才会增加有了库存才能出库出了库才能算周转。这种依赖关系用Jupyter来表达特别自然前面单元格执行完下一个单元格才能看到变化的库存表你改一个参数重新运行后续所有结果全部跟着变。它不像传统脚本那样一次性输出一堆结果而是像在做一个可交互的实验。打个比方商业WMS就像一座装修完好的样板间你只看得到漂亮的界面看不到墙壁里面的管线。pegasusWMS的Jupyter范例则是一套积木搭建的模型房每个承重墙、每条水管都暴露在外面你甚至可以自己把墙推倒重砌一遍看看会不会塌。对于学习者这种“看得见内部结构”的价值比功能全不全重要得多。另外Jupyter对非程序员也比较友好。仓储运营人员大多习惯用Excel看数据而Jupyter输出的DataFrame从视觉上就跟Excel很像还能用几行代码做出图表。范例里刻意在每个关键业务步骤后面附带一个表格或图形就是为了降低认知门槛让完全没写过代码的人也能放心地点“运行”按钮。1.3 它适合什么样的人我的目标读者大致分成三类。第一类是仓储或供应链从业者他们天天用WMS但系统在他们眼里是个黑盒这类人可以通过范例理解批次、库位分配、拣货波次背后的计算逻辑。第二类是数据分析方向的Python学习者他们不应该永远只拿鸢尾花和房价数据练手WMS这种有明确业务含义的数据集反而更能锻炼数据清洗和分析思维。第三类是准备做WMS二次开发的工程师在接真实项目之前先用这套轻量模型验证自己的分配策略或库存计算方案成本很低。我这里也给一个提醒千万别把pegasusWMS当成商业WMS的替代品它在并发、事务、权限控制上都没做工业级设计。它更像一架教学用的模型飞机你用它学会飞行原理后再去碰真正的驾驶舱才不会慌。2. 范例的整体架构与Notebook编排思路一套好的Jupyter范例绝不仅仅是把代码从一个脚本拆进多个单元格。pegasusWMS在架构上做了三个关键决策引擎代码与notebook分离、初始化数据统一管理、每个notebook都有清晰的叙事顺序。这三个决策直接决定了系统是否好维护、范例是否容易复现。2.1 模块与Notebook是如何分工的项目的目录结构大概是这样的pegasusWMS/ ├── wms_core/ │ ├── __init__.py │ ├── models.py # 数据实体定义 │ ├── services.py # 业务服务函数 │ └── analytics.py # 数据分析函数 ├── notebooks/ │ ├── 01_入库流程_自动库位分配.ipynb │ ├── 02_出库波次与库存分配.ipynb │ ├── 03_库存周转率分析.ipynb │ └── 04_盘点差异处理.ipynb ├── data/ │ └── demo_data.csv └── README.mdNotebook只负责做参数设置、调用服务、展示结果真正的业务逻辑全部沉淀在wms_core里。拿入库这个功能来说你在notebook里看到的只是receive_goods(context, ...)一行调用但services.py里包含了库位筛选、容量校验、库存更新、流水生成这四步完整逻辑。这样设计最大的好处是复用同一个allocate_inventory()函数可以同时被入库notebook和出库notebook调用改一处业务逻辑所有示范文档自动同步更新。如果代码全都写在单元格里前三个notebook用的还是旧逻辑第四个已经改成新逻辑这项目很快就没人敢维护了。2.2 统一的数据初始化流程Jupyter项目最典型的一个坑是你在某个notebook里改了全局变量然后切到另一个notebook继续跑发现所有状态全乱了。为了消除这个问题pegasusWMS的所有notebook都会在第一个主干单元格执行统一的init_demo_data()函数。这个函数会生成若干个SKU、几十个库位、一批初始库存以及几张空白订单表最后返回一个叫context的字典后续所有操作都围绕这个字典展开。你在新notebook里导入项目、初始化数据拿到的状态和之前的notebook完全一致不存在“手手相传”的变量依赖。代码大概长这个样子from wms_core.demo_data import init_demo_data ctx init_demo_data( sku_count8, location_count36, initial_quantity120, max_stock_per_location50, seed42 )这里的seed参数很关键它控制了随机种子。Jupyter范例最怕读者每次跑出来的数据都不一样这样你写起来费劲读者看起来也费劲。设置固定种子以后无论谁跑初始数据都一样后续讨论“为什么分配到这个库位”才有共同基础。如果你的业务是探索型分析不希望结果固定那可以把seedNone让每次运行都生成完全不同的一组数据做一些随机场景测试。2.3 Notebook的叙事顺序是如何设计的每个notebook基本遵循一个讲故事的结构先介绍背景数据再执行核心流程接着解释“这个结果为什么合理”最后用一张图收尾。以出库流程为例它的顺序是先展示当前的库存余额表让读者知道“现在仓库里有什么”再创建一批出库订单打印订单明细然后运行allocate_inventory()展示每个订单行被分配到哪个库位、扣了多少数量最后画一张拣货数量的柱状图直观呈现这次波次的工作量。这种顺序不是随意的而是对应了WMS系统里的人在接到一个出库任务时的思考路径先查库存、再做计划、最后执行。如果你自己写Jupyter教学文档也建议按“输入数据 - 处理逻辑 - 输出结果 - 可视化验证”的顺序组织而不是一上来就砸一堆函数读者根本不知道目标是什么。3. 快速跑通第一个范例入库流程与自动库位分配理论说再多都不如亲手跑一遍。这一章带你从零把「入库流程与自动库位分配」这个notebook跑起来我会把每一步为什么这么做也一并讲清楚。3.1 环境准备别跳过虚拟环境我自己见过太多人倒在环境配置这一步直接在系统Python里装了一大堆包结果版本冲突导致Jupyter内核起不来。强烈建议新建虚拟环境不要怕麻烦后面能省事无数倍。# 创建虚拟环境 python -m venv .venv # 激活虚拟环境Windows .venv\Scripts\activate # 激活虚拟环境macOS/Linux source .venv/bin/activate # 安装依赖 pip install pandas numpy matplotlib seaborn jupyter # 启动Jupyter jupyter notebook依赖版本方面pandas建议用1.5以上numpy建议用1.24以上因为太老版本在DataFrame的展示和数据类型推断上有一些坑。启动Jupyter之后在浏览器中进入notebooks/01_入库流程_自动库位分配.ipynb。如果你在Jupyter内核列表里找不到当前虚拟环境的Python需要先执行python -m ipykernel install --user --namewms-env然后在Jupyter右上角切换内核为wms-env否则会出现“明明装了pandas但import失败”这种很诡异的问题。3.2 关键代码与参数计算过程打开notebook后依次执行单元格。前面几个单元格是环境导入和初始化下面的代码是核心入参from wms_core.demo_data import init_demo_data from wms_core.services import receive_goods from wms_core.analytics import stock_value_report ctx init_demo_data( sku_count8, location_count36, initial_quantity120, max_stock_per_location50, seed42 ) print(库存总余额, stock_value_report(ctx[inventory]))运行后会输出类似这样的一行库存总余额 960之所以是960因为8个SKU乘以120件初始库存等于960件。这个数字本身不是重点重点是你把它和后面的入库操作对比时能清楚看到库存增加了多少。接下来执行入库result receive_goods(ctx, sku_idSKU001, qty30) result[inventory_after].head(10)receive_goods函数内部会先找到当前可用的库位规则是“未满载-剩余容量最大-库位编号升序”。然后做两件事更新库存表插入一条流水记录。运行后你会看到SKU001在某个库位的库存从before_qty变成after_qty如果连续执行两次入库第二次入库很可能被分配到另一个库位因为第一次已经占用了部分容量。这就是库位分配策略的直观体现。这里有一个非常重要的参数关系max_stock_per_location50意味着单库位不能超过50件。如果你一次性入库60件receive_goods会自动拆分成两条库存记录分别放到两个库位。如果你还想验证异常场景把max_stock_per_location改成20再跑一次入库系统可能会提示“库位不足”原因是当前所有可用库位剩余空间都不足以放下30件商品。这种模拟就是Jupyter范例的价值——你可以用最小成本压测业务逻辑而不是等上线后才让用户发现问题。3.3 可视化库位状态入库完成后notebook里有一个单元格会生成库位热力图。实现方式很简单就是把所有库位的当前占用率映射成一个二维矩阵再用seaborn.heatmap画出来。如果你在自己的项目里也想复现参考下面这段伪代码思路import numpy as np import seaborn as sns import matplotlib.pyplot as plt loc_matrix np.zeros((6, 6)) for _, row in ctx[inventory].groupby(location_id)[qty].sum().items(): # 假设库位编号形式是 A01-01 zone row[zone] r, c int(row[row]), int(row[col]) loc_matrix[r-1, c-1] row[qty] sns.heatmap(loc_matrix, annotTrue, fmt.0f, cmapYlGnBu) plt.title(库位占用热力图) plt.show()从热力图上你能很直观地看出哪些区域的库存密度高哪些区域还是空的。真实的WMS管理看板里经常有这种热力图用于指导补货和库位规划。每次入库后重新运行这一个单元格热力图颜色都会变化这种即时反馈比文字描述有说服力得多。4. 深入拆解出库波次与库存分配逻辑入库只是开始出库才是WMS里真正的核心难点。一个仓库有几百几千个SKU每个SKU分布在几十个库位当一张包含多种商品的订单进来系统到底从谁那里拿货这一章我用范例里的“批次优先、库位均衡”策略来拆解。4.1 订单分配的完整链路范例中的出库流程分两步先建波次再做分配。所谓波次就是把多个出库订单按某种规则合并成一次拣货任务目的是提高拣货效率。真实WMS里波次可以按承运商、按截单时间、按库区过滤但在pegasusWMS里简化成了“直接把所有订单行聚合成一张拣货需求表”。创建波次之后核心分配函数长这样from wms_core.services import create_wave, allocate_inventory wave create_wave(ctx[open_orders]) allocations allocate_inventory(ctx, wave) allocations[[order_line_id, sku_id, from_location, alloc_qty, batch_date]]allocate_inventory()的执行逻辑是这样的过滤当前可用的库存记录排除已锁定或质量状态不佳的库存。按SKU分组把同一SKU的库存按batch_date升序排列。依次扣减库存优先使用生产日期最早的批次。这就是“先进先出”策略。如果同一个批次分布在多个库位则按库位编号顺序依次扣减尽量避免碎片化拣货。我们用一个表来对比几种常见的分配策略策略名称排序规则优点缺点先进先出FIFO批次日期升序防止物料过期、呆滞可能增加跨库位拣货路径库位优先库位编号升序拣货路径集中可能忽略批次属性指定批次订单指定批次号满足特殊质量要求强制选择容易造成其他批次占用库存充足优先剩余库存多者优先散件减少需要额外库存统计pegasusWMS默认使用第一种也是大多数制造业仓库应该采用的策略。如果你做的是电商仓对效期没那么敏感可以改成第二种但要记得先确认商品没有批次追溯需求。4.2 库存流水如何记录与校验分配完成后库存表会被扣减同时stock_transactions表会插入多条记录每一条都记录着这次分配的来源库位、目标订单、扣减数量和操作前后的库存值。这是一个非常严谨的设计因为有了流水任何一单业务都能被追溯。我在范例里专门写了一个校验单元格把所有流水按SKU分组求和再和当前库存表加上期初库存的差异做比较。这个方法虽然简单但非常实用。你在真实系统里如果遇到库存对不上账第一反应应该是对比流水汇总和余额表而不是直接去改库存表。我见过太多团队因为怕麻烦直接手工改库存结果漏洞越补越多。校验代码的核心是check ( ctx[inventory].groupby(sku_id)[qty].sum() - ctx[stock_transactions].groupby(sku_id)[qty].sum() - ctx[inventory_begin].groupby(sku_id)[qty].sum() ) assert (check.abs() 1e-6).all(), 库存台账异常如果这里报错说明要么有流水缺失要么有库存记录没有被流水驱动更新。这是我个人认为整个Jupyter范例里最有价值的一段代码建议你仔细理解并背下来。4.3 库存周转率一个你一定会用到的分析维度出库流程跑通之后第三个notebook会引导你计算库存周转率。公式非常简单周转次数 期间出库总件数 / 期间平均库存件数假设某个SKU在一个月内累计出库600件期初库存是200件期末库存是100件平均库存按(期初期末)/2计算就是150件周转次数是4次。代码可以这样写out_qty outbound.groupby(sku_id)[qty].sum() avg_inventory (inventory_begin[qty] inventory_end[qty]) / 2 turnover (out_qty / avg_inventory).sort_values(ascendingFalse) turnover.plot.bar(titleSKU库存周转次数)跑完这张图之后你会发现有的SKU周转次数极高比如每天都出库的爆款有的SKU几乎不动周转次数接近0。后者就是典型的“呆滞库存”继续占用库位和资金。这个指标不需要高深的算法但在实际业务中说服力很强无论在周报还是经营分析会上一张周转率直方图比一堆文字描述都有用得多。5. 常见报错、排查经验与避坑清单任何项目在首次跑通的时候都会遇到问题pegasusWMS也不例外。这一章我把自己反复踩过的坑整理成速查表希望能让你少浪费几个晚上。5.1 典型错误与解决方法速查报错现象根本原因解决办法ModuleNotFoundError: No module named pandasJupyter内核使用的是系统Python不是虚拟环境里的Python激活虚拟环境后执行python -m ipykernel install --user --namewms-env然后在Jupyter里切换内核KeyError: from_location前置单元格没执行或初始化数据被覆盖从头重新执行所有单元格先运行init_demo_data()再跑业务函数图表中文乱码Matplotlib默认字体没有中文字体在代码块前面加plt.rcParams[font.sans-serif][SimHei]macOS可用Arial Unicode MSValueError: Buffer dtype mismatchpandas/Numpy版本冲突pip install --upgrade pandas numpy必要时重建虚拟环境库存分配为负数超卖检查被绕过检查allocate_inventory是否在扣减前判断可用库存真实环境需要加锁或乐观锁特别提醒一个很常见的操作误区不要跳过任何单元格直接运行后面的业务代码。Jupyter是一种“易失状态”环境如果初始化单元格没有执行后面的变量都是空的自然报错。这不是项目坏了而是你把它当成普通文档在翻页。5.2 我从这些坑里总结的经验第一一定要用虚拟环境。我最初为了省事直接在全局Python里跑后来装了一堆包系统和项目互相干扰最后不得不花更多时间重装。第二Jupyter里提供了几个很实用的魔法命令极大提升调试效率。%pdb可以在出错时自动进入调试器%debug则可以针对最近一次错误启动事后检查我在调试波次分配函数时就靠它逐行看中间变量。第三当你想对比不同初始化参数的影响时建议把seed固定否则换一组随机数据结果完全不一样你很难判断变化到底是参数引起的还是随机噪声引起的。另外小技巧是每个notebook里的中间结果都可以用.to_csv()保存下来。比如跑完入库流程把ctx[inventory]保存成inventory_before.csv跑完出库流程再保存一份inventory_after.csv然后用Excel对比差异。这样即使你哪一天把内核切坏了之前的快照还在至少不会前功尽弃。5.3 检查库存流水的调试技巧在真实项目中我遇到过很多次库存余额和流水不一致的情况。如果你也想排查类似问题分享一个很直接的操作把stock_transactions表按“SKU库位”分组汇总净变化量再和库存表的当前值做外连接。凡是对不上的SKU和库位就是问题定位点。然后用tx时间序列往前翻看是哪个环节开始对不上。在Jupyter里做这个分析方便到令人发指几行代码就能看全貌。我甚至会把这段调试分析代码保存成一个独立notebook作为项目里的标准排查工具遇到问题直接打开跑一遍。6. 从Jupyter到生产环境工程化改造建议Jupyter范例解决了“理解”的问题但真实业务中还需要把分析能力固化到线上。我在这套项目上摸索出几条比较顺的改造路径这里一并分享。6.1 导出脚本与定时分析Jupyter Notebook本质上是一个包含输出结果的Python脚本所以你可以用命令直接把它转换成.py文件jupyter nbconvert --to script notebooks/03_库存周转率分析.ipynb转换出来的脚本里会保留所有代码单元并自动去掉输出内容。接下来你可以把它放到调度器里比如每隔一天跑一次生成新的周转率报表。这里提醒一句脚本里如果有init_demo_data()记得在工程化时替换成从数据库读取否则你的定时任务永远用的是同一批模拟数据没有任何生产意义。6.2 接入真实数据库或API如果你想把范例变成一套内部运营分析台最简单的方式是新增一个数据加载模块把init_demo_data()替换成从MySQL/PostgreSQL读取库存快照和流水表。核心逻辑不变只需要把ctx里的inventory、stock_transactions等表格换成数据库查询结果即可。我在一次内部改造里就是这么做的把某个大型商超的脱敏库存数据映射到pegasusWMS的字段结构然后不修改任何业务函数只改了前两个单元格的数据来源后面所有分析逻辑照跑。这种“数据替换、逻辑复用”的模式可以极大减少二次开发的工作量。6.3 用轻量应用把结果展示给同事如果你的同事不写Python也不想打开Jupyter看代码那可以考虑做一个30分钟能搞定的内部工具用Streamlit包一层Web页面。比如选择SKU、查看当前库存分布、输入订单号查看分配结果这些都可以在Streamlit里用少量代码复刻。底层还是调用wms_core的同一个业务函数只是把输入框换成页面组件、把DataFrame换成页面表格。这样的工具不需要专门的前端开发也不需要部署复杂服务直接在服务器上一个streamlit run app.py就完事了。我从Jupyter原型走到这个轻量应用前后不到两天效果却比我做过的很多大系统实用得多。7. 关于这套范例我的几句实在话跑了这么多遍这些notebook我最深的体会是WMS并不是什么高深莫测的系统只要把数据模型和流程规则想清楚用Python模拟起来并不难。难的是如何在真实业务中保持对库存数据和流水的耐心。Jupyter范例给我最大的帮助不是“学会了函数”而是让我敢动手去验证各种业务假设。以前同事说“某个商品卡在库位里出不来”我只能去商业系统里层层翻菜单现在我用Jupyter查一下库存流水、批次日期、库位占用情况十秒钟就能得出结论。这种“一切皆可查、一切皆可复现”的能力真的非常解压。最后再分享一个小习惯每当你用这套范例跑完一条新流程顺手把输入和输出的DataFrame存一份快照命名带上日期和参数。比如inventory_20250601_qty120.csv。一个月后当你需要复现当时那个决策场景这些快照能帮你省大量时间。我就靠这个习惯在一次内部盘点对账时快速定位了问题单据不至于从头再演一遍。希望这套pegasusWMS的Jupyter范例也能帮你把业务和代码之间那层窗户纸捅破。