Python实现小店进销存：用移动加权平均法自动生成利润表

小店进销存系统做到后面，真正考验设计能力的往往不是库存数量显示得准不准，而是利润表能不能自动算对。很多小店老板到现在还在月底手工算账：这个月卖了多少钱很好统计，难的是每一笔货的成本。上个月进货 3 元一件，这个月进货 3.5 元一件，那已经卖掉的那一批到底按哪个价格算成本？如果你也在做类似的小店管理系统，或者正准备给家里的小生意写一套管理工具，这篇文章会从数据模型、成本算法、Python 实现到利润表生成，完整走一遍。

这篇文章会实现一个最小可用的进销存系统：商品管理、采购入库、销售出库、库存维护、利润表生成。代码不依赖第三方库，只用 Python 标准库自带的 sqlite3 就能跑通。内容适合刚接触业务系统的开发者，也适合想把手动 Excel 记账改成系统的店主参考。核心结论先放在这里：想让利润表不再手算，关键不是写一条复杂的报表 SQL，而是先在销售明细里把“这批货卖出去时的成本”固定下来。

1. 先理解手工利润表为什么难算：核心问题是销售成本

1.1 收入很好统计，成本才是难点

小店每天的流水可以这样描述：卖了多少件、单价多少、收了多少钱。收入是明确的，因为每一笔销售都有成交价和数量。

真正模糊的是成本。同一个商品，不同批次进货价很可能不一样。今天进的苹果 3 元一斤，下个星期变成 3.5 元一斤，那么现在卖掉的一斤苹果，成本到底是 3 元还是 3.5 元？如果店里有三种商品、每个月有几十张采购单和几百条销售记录，手工要按“每一笔卖出时实际对应的进价”去算，基本不现实。

实际经营中，手工记账常见的做法有几种：

所以，进销存系统要解决的第一个问题不是“能不能显示利润”，而是“用什么口径计算销售成本”。口径不定义清楚，报表做得再好看也是错的数字。

1.2 移动加权平均法：利润计算的推荐口径

小店场景下最合适的成本口径通常是“移动加权平均法”。

它的规则很简单：每次采购入库后，重新计算一次该商品的平均成本。

计算公式：

新的平均成本 = (当前库存数量 × 当前平均成本 + 本次采购数量 × 本次采购单价) / (当前库存数量 + 本次采购数量)

销售出库时，按当前的平均成本结转销售成本。平均成本只在采购入库时变化，销售出库只减少数量、不改变平均成本。

举个例子。苹果第一次进货 100 斤，单价 3 元，平均成本就是 3 元。卖出 30 斤，销售成本按 3 元算，库存剩 70 斤，平均成本仍是 3 元。此时再进货 50 斤，单价 3.5 元，新的平均成本是：

(70 × 3 + 50 × 3.5) / (70 + 50) = 385 / 120 ≈ 3.2083 元

之后再卖出，成本就按 3.2083 元结转。

移动加权平均的优点是小店容易理解、计算稳定、不会因为库存批次太少而无法处理；缺点是价格波动大时，成本会比最新进价滞后。它有保质期的商品可以考虑先进先出法（FIFO），但作为一套通用小店系统，移动加权平均足够覆盖大多数场景。

1.3 自动化的目标：从“记流水”升级成“出报表”

手工记账的本质是流水记录：今天进了什么、今天卖了什么。自动化的目标是让系统在流水之上直接产出三个结果：实时库存余额、按商品汇总的毛利、按时间段汇总的利润表。

要做到这一点，系统必须完成三条数据链路：

1. 采购单录入时，更新库存数量，同时重算移动平均成本。
2. 销售单录入时，把“当前平均成本”写入销售明细，作为这一笔的历史成本快照。
3. 报表查询时，用销售明细中的售价和成本快照汇总，算出收入、成本和毛利。

人工只需要录入采购单和销售单，剩下的加权平均计算、库存扣减、利润汇总全部交给程序。这就是“利润表不用自己算了”的技术本质。

2. 数据模型设计：把进销存变成可以计算的表结构

2.1 六张核心表和它们的分工

实现这套逻辑只需要六张表。它们分别承担商品主数据、采购、销售、库存余额四种职责。

表之间的关系如下：purchase_orders 与 purchase_items 是一对多，sales_orders 与 sales_items 是一对多，products 与 inventory 是一对一，inventory 只保存每个商品当前的库存数量和当前平均成本。

这里有一个容易忽略的设计点：purchase_items 只保存采购数量和采购单价，不承担实时成本计算；sales_items 保存售价的同时，还要保存 cost_price 字段，记录卖出那一刻的移动平均成本。

2.2 关键设计：在销售明细里快照成本价

这是整套系统最核心的设计决定。

如果只把平均成本存在 inventory 表里，销售时实时读取，那么一旦后续有新的采购入库，平均成本就会变化。这时再去查历史销售，会发现历史利润也跟着变了。进了一批价格不同的货，上个月的利润数字却改变了，这在财务上完全不可接受。

解决办法就是“快照”：销售出库发生的那一刻，把当时的 avg_cost 复制到 sales_items.cost_price 字段。之后无论采购价格怎么变，历史销售的成本都保持不变。利润表只依赖 sales_items 表里的 cost_price，不再回头读 inventory 表的实时平均成本。

这个原则在系统后续扩展时也要守住：历史单据不允许重算成本，新单据只按新的平均成本记账。

2.3 SQLite 建表脚本

下面的脚本可以直接保存为 db.py。SQLite 是 Python 自带的标准库，不需要额外安装数据库服务。

import sqlite3 DB_PATH = "shop.db" def get_conn(db_path=DB_PATH): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row # SQLite 默认不启用外键约束，这里必须手动打开 conn.execute("PRAGMA foreign_keys = ON") return conn def init_db(conn): conn.executescript( """ CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT UNIQUE NOT NULL, name TEXT NOT NULL, category TEXT DEFAULT '', unit TEXT DEFAULT '件', sale_price REAL NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS purchase_orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT UNIQUE NOT NULL, supplier TEXT DEFAULT '', order_date TEXT NOT NULL, remark TEXT DEFAULT '' ); CREATE TABLE IF NOT EXISTS purchase_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, product_id INTEGER NOT NULL, quantity REAL NOT NULL, unit_cost REAL NOT NULL, FOREIGN KEY (order_id) REFERENCES purchase_orders(id), FOREIGN KEY (product_id) REFERENCES products(id) ); CREATE TABLE IF NOT EXISTS sales_orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT UNIQUE NOT NULL, customer TEXT DEFAULT '', order_date TEXT NOT NULL, remark TEXT DEFAULT '' ); CREATE TABLE IF NOT EXISTS sales_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, product_id INTEGER NOT NULL, quantity REAL NOT NULL, price REAL NOT NULL, cost_price REAL NOT NULL, FOREIGN KEY (order_id) REFERENCES sales_orders(id), FOREIGN KEY (product_id) REFERENCES products(id) ); CREATE TABLE IF NOT EXISTS inventory ( product_id INTEGER PRIMARY KEY, quantity REAL NOT NULL DEFAULT 0, avg_cost REAL NOT NULL DEFAULT 0, FOREIGN KEY (product_id) REFERENCES products(id) ); """ ) conn.commit()

建表脚本里有三个细节要注意。

第一，order_no 设置了 UNIQUE，这是防止重复单号的最底层防线。第二，inventory 表用 product_id 做主键，保证一个商品只对应一条库存余额记录。第三，开了 PRAGMA foreign_keys，让 SQLite 真正检查外键约束，避免出现“销售明细指向不存在的商品”这类脏数据。

关于数量字段：称重类商品数量可能是小数，所以这里统一使用 REAL。按件销售的纯整数商品也可以使用 INTEGER，但为了简化类型处理，示例统一用 REAL。

3. Python 实现核心逻辑：入库、出库和库存维护

3.1 环境准备和项目结构

运行环境只需要 Python 3.8 及以上版本，sqlite3 是标准库，不需要 pip 安装任何第三方包。建议按下面的目录组织文件：

shop-psi/ ├── db.py # 连接、建表 ├── core.py # 商品、采购、销售、库存核心逻辑 ├── report.py # 利润表查询 ├── demo.py # 造数并运行验证 └── shop.db # 运行后自动生成

下面开始实现 core.py。所有函数都接收 conn 作为第一个参数，这样方便在调用方统一控制事务，也方便以后替换数据库连接方式。

3.2 采购入库：更新库存并重算移动平均成本

采购入库要完成两件事：写入采购单和采购明细；更新 inventory 表的库存数量和平均成本。

如果商品还没有库存记录，直接以本次采购单价作为平均成本。如果已经有库存记录，按移动加权平均公式重算。

# core.py def add_product(conn, sku, name, category="", unit="件", sale_price=0): conn.execute( """ INSERT INTO products (sku, name, category, unit, sale_price) VALUES (?, ?, ?, ?, ?) """, (sku, name, category, unit, sale_price), ) conn.commit() def purchase_in(conn, order_no, supplier, order_date, items): cur = conn.cursor() cur.execute( """ INSERT INTO purchase_orders (order_no, supplier, order_date) VALUES (?, ?, ?) """, (order_no, supplier, order_date), ) order_id = cur.lastrowid for item in items: product_id = item["product_id"] quantity = item["quantity"] unit_cost = item["unit_cost"] cur.execute( """ INSERT INTO purchase_items (order_id, product_id, quantity, unit_cost) VALUES (?, ?, ?, ?) """, (order_id, product_id, quantity, unit_cost), ) inv = cur.execute( "SELECT quantity, avg_cost FROM inventory WHERE product_id = ?", (product_id,), ).fetchone() if inv is None: # 首次入库，平均成本就是本次采购单价 cur.execute( """ INSERT INTO inventory (product_id, quantity, avg_cost) VALUES (?, ?, ?) """, (product_id, quantity, unit_cost), ) else: old_qty = inv["quantity"] old_avg_cost = inv["avg_cost"] new_qty = old_qty + quantity new_avg_cost = (old_qty * old_avg_cost + quantity * unit_cost) / new_qty # 平均成本保留 4 位小数，减少金额累加误差 cur.execute( """ UPDATE inventory SET quantity = ?, avg_cost = ? WHERE product_id = ? """, (new_qty, round(new_avg_cost, 4), product_id), ) conn.commit()

这里的关键行为是 new_avg_cost 的计算。它不是在原来平均成本上简单取平均值，而是用“库存金额”作为权重：原来的库存金额加上新采购金额，除以总数量。这样先进的价格和新的价格都会按数量占比影响最终成本，符合业务直觉。

3.3 销售出库：扣减库存并写入本次成本

销售出库的流程比采购多一个校验：必须先确认当前库存足够，否则直接回滚，不允许出现负库存。

扣减库存时，要把当前的 avg_cost 写入销售明细的 cost_price。这个动作就是前面说的“成本快照”。

# core.py def sale_out(conn, order_no, customer, order_date, items): cur = conn.cursor() cur.execute( """ INSERT INTO sales_orders (order_no, customer, order_date) VALUES (?, ?, ?) """, (order_no, customer, order_date), ) order_id = cur.lastrowid for item in items: product_id = item["product_id"] quantity = item["quantity"] price = item["price"] inv = cur.execute( "SELECT quantity, avg_cost FROM inventory WHERE product_id = ?", (product_id,), ).fetchone() if inv is None or inv["quantity"] < quantity: conn.rollback() current_qty = 0 if inv is None else inv["quantity"] raise ValueError( f"库存不足：product_id={product_id}，需要={quantity}，现有={current_qty}" ) # 关键：把当前平均成本快照到销售明细 cur.execute( """ INSERT INTO sales_items (order_id, product_id, quantity, price, cost_price) VALUES (?, ?, ?, ?, ?) """, (order_id, product_id, quantity, price, inv["avg_cost"]), ) cur.execute( """ UPDATE inventory SET quantity = quantity - ? WHERE product_id = ? """, (quantity, product_id), ) conn.commit()

注意 sale_out 里调用了 conn.rollback() 再抛出异常。这是因为一张销售单可能包含多个商品，如果后面的商品库存不足，前面的写入也必须全部撤销。不这样做，就会留下只有半张单的脏数据。

3.4 为什么必须用事务保护出入库

采购和销售本质上都是“一张单据 + 多条明细 + 一次库存变动”的组合操作。任何一步失败，单据和库存就可能不一致。

Python 的 sqlite3 默认在一个连接上开启隐式事务，commit 之前的所有写操作都属于同一个事务。上面代码里出现异常时执行 rollback，能把订单头、订单明细和库存变更全部回滚。这个设计在正式数据库里同样成立，只是底层事务管理换成了框架能力。

注意：不要为了省事把一条销售拆分成多次独立的 commit。比如先插入销售单再单独更新库存，一旦更新库存时程序崩溃，就会出现“已收款但没扣库存”的情况。

4. 利润表生成：从销售明细汇总到商品维度和时间维度

4.1 用一条 SQL 按商品汇总利润

有了销售明细里的 price 和 cost_price，利润汇总就变成纯粹的 SQL 聚合。

单笔销售的毛利公式：

毛利 = (售价 - 成本价) × 销售数量

按商品汇总时，对每个商品累加销售收入、销售成本和毛利。

# report.py def profit_statement(conn, start_date, end_date): rows = conn.execute( """ SELECT p.id, p.sku, p.name, SUM(si.quantity) AS sale_quantity, ROUND(SUM(si.price * si.quantity), 2) AS sale_amount, ROUND(SUM(si.cost_price * si.quantity), 2) AS sale_cost, ROUND(SUM((si.price - si.cost_price) * si.quantity), 2) AS gross_profit FROM sales_items si JOIN sales_orders so ON si.order_id = so.id JOIN products p ON si.product_id = p.id WHERE so.order_date BETWEEN ? AND ? GROUP BY p.id, p.sku, p.name ORDER BY gross_profit DESC """, (start_date, end_date), ).fetchall() return rows

这段 SQL 的关键是只从 sales_items 取成本，完全不碰 inventory 表。因为成本已经在销售时被快照，报表查询是确定性的：同一条销售记录任何时候查，收入和成本都一样。

4.2 按月份汇总，满足“月底看利润”场景

如果要做月度利润表，可以改成按 strftime 格式化订单日期后分组。日期范围的起始值建议统一包含边界，例如 2025-01-01 到 2025-01-31。

SELECT strftime('%Y-%m', so.order_date) AS month, COUNT(DISTINCT so.id) AS order_count, ROUND(SUM(si.quantity), 2) AS sale_quantity, ROUND(SUM(si.price * si.quantity), 2) AS sale_amount, ROUND(SUM(si.cost_price * si.quantity), 2) AS sale_cost, ROUND(SUM((si.price - si.cost_price) * si.quantity), 2) AS gross_profit FROM sales_items si JOIN sales_orders so ON si.order_id = so.id GROUP BY month ORDER BY month;

这条 SQL 适合在后台管理页面按月展示，也适合后续导出成 Excel 给店主对账。注意 month 字段的边界：如果存在跨天补录的单据，要统一以 order_date 的业务日期为准，而不是录入时间。

4.3 输出格式与调用方式

profit_statement 返回的是 sqlite3.Row 列表。每个元素可以通过字段名访问，例如 r["sku"]、r["gross_profit"]。这样在 Web 项目里可以直接转成 JSON，在脚本项目里可以直接打印成文本表格。

5. 造一组模拟数据验证利润表

5.1 模拟数据场景

为了让验证有说服力，造一组能覆盖“多次进货、价格变化、跨日期销售”的数据：

• 商品：苹果（A001）、牛奶（A002）
• 苹果分两次进货，第一次 3 元，第二次 3.5 元，中间有销售，结果应该体现移动加权平均
• 牛奶只有一次进货、一次销售，结果应该简单可验算

# demo.py from db import get_conn, init_db from core import add_product, purchase_in, sale_out from report import profit_statement def main(): conn = get_conn() init_db(conn) # 商品 add_product(conn, "A001", "苹果", "水果", "斤", 5.0) add_product(conn, "A002", "牛奶", "乳品", "盒", 3.5) # 采购 purchase_in( conn, "PO20250105001", "果农合作社", "2025-01-05", [{"product_id": 1, "quantity": 100, "unit_cost": 3.00}], ) purchase_in( conn, "PO20250106001", "乳品批发", "2025-01-06", [{"product_id": 2, "quantity": 60, "unit_cost": 2.00}], ) purchase_in( conn, "PO20250111001", "果农合作社", "2025-01-11", [{"product_id": 1, "quantity": 50, "unit_cost": 3.50}], ) # 销售 sale_out( conn, "SO20250108001", "散客", "2025-01-08", [{"product_id": 1, "quantity": 30, "price": 5.00}], ) sale_out( conn, "SO20250109001", "便利店", "2025-01-09", [{"product_id": 2, "quantity": 20, "price": 3.50}], ) sale_out( conn, "SO20250112001", "散客", "2025-01-12", [{"product_id": 1, "quantity": 40, "price": 5.50}], ) # 利润表 rows = profit_statement(conn, "2025-01-01", "2025-01-31") print("SKU 商品 数量 销售收入 销售成本 毛利") for r in rows: print( f"{r['sku']:<7}{r['name']:<5}" f"{r['sale_quantity']:>5.0f}{r['sale_amount']:>10.2f}" f"{r['sale_cost']:>10.2f}{r['gross_profit']:>10.2f}" ) if __name__ == "__main__": main()

运行命令：

python demo.py

5.2 运行结果

正常情况下，终端输出的利润表应该是：

SKU 商品 数量 销售收入 销售成本 毛利 A001 苹果 70 370.00 218.33 151.67 A002 牛奶 20 70.00 40.00 30.00

合计销售收入 440.00 元，销售成本 258.33 元，毛利 181.67 元。

5.3 手工验算一遍

按移动加权平均逐步推导苹果的数据：

苹果的销售收入：30 × 5.00 + 40 × 5.50 = 370.00 元。
苹果的销售成本：30 × 3.00 + 40 × 3.2083 = 90 + 128.33 = 218.33 元。
苹果的毛利：370.00 - 218.33 = 151.67 元。

牛奶的验算：60 盒进价 2 元，卖出 20 盒，成本 40 元，收入 70 元，毛利 30 元。结果与程序输出一致。

注意：这里 3.2083 是平均成本保留 4 位小数后的值。40 × 3.2083 等于 128.332，展示时四舍五入为 128.33。这个舍入位置要和业务约定一致，避免对账时出现 0.01 元的差异。

5.4 库存余额验证

利润表正确还不够，还要确认库存余额没有丢数据。执行下面的 SQL 查看实时库存：

def inventory_snapshot(conn): rows = conn.execute( """ SELECT p.sku, p.name, i.quantity, i.avg_cost, ROUND(i.quantity * i.avg_cost, 2) AS stock_amount FROM inventory i JOIN products p ON i.product_id = p.id ORDER BY p.sku """ ).fetchall() return rows

预期结果：

如果这一步数字也对，说明采购入库、销售出库、库存扣减三条链路都正确。

6. 利润数字对不上时的排查路径

6.1 从报表倒推：收入、成本、库存三层核对

利润表出错时，不要直接改 SQL 或改数据。按下面的顺序逐层核对：

1. 先核对收入。把报表里的 sale_amount 与微信/支付宝流水、收款记录比对，确认销售明细没有漏单、重单。
2. 再核对成本。把 sale_cost 与销售数量、成本价相乘的结果比对，重点看是不是某个商品的 cost_price 异常。
3. 最后核对库存。把 inventory 的数量与盘点表比对，如果库存多了或少了，利润一定有问题。

这三个数字是联动的：库存出错，成本必出错；成本出错，毛利必出错。定位到某一层后，再用 SQL 缩小范围。

6.2 排查日常用 SQL 清单

把下面这几条 SQL 存成固定的排查脚本，遇到问题先跑一遍。

-- 1. 查找负库存，这是最常见的成本异常源头 SELECT p.sku, p.name, i.quantity FROM inventory i JOIN products p ON i.product_id = p.id WHERE i.quantity < 0;

-- 2. 查找成本价为 0 的销售明细 SELECT so.order_no, so.order_date, p.sku, p.name, si.quantity, si.price, si.cost_price FROM sales_items si JOIN sales_orders so ON si.order_id = so.id JOIN products p ON si.product_id = p.id WHERE si.cost_price = 0;

-- 3. 查找重复单号 SELECT order_no, COUNT(*) AS cnt FROM sales_orders GROUP BY order_no HAVING cnt > 1;

-- 4. 对比采购总数量和销售总数量 SELECT p.sku, p.name, (SELECT COALESCE(SUM(quantity), 0) FROM purchase_items WHERE product_id = p.id) AS purchase_qty, (SELECT COALESCE(SUM(quantity), 0) FROM sales_items WHERE product_id = p.id) AS sale_qty FROM products p;

6.3 常见问题与处理方案

第五条“某些商品没有出现在利润表”在实际运作中很常见。报表只聚合 sales_items，有采购但没卖出的商品自然不会出现。如果目的是看整体经营情况，这是对的；如果目的是盘点“哪些商品动销、哪些不动销”，需要另外写基于 products 的左连接查询。

7. 从演示系统到真正可用的进销存系统

7.1 学习环境与生产环境的差异

本文的示例可以快速跑通，但它只适合学习和小规模自用。真实小店系统至少要在下面几方面升级：

如果在生产环境继续使用 REAL 存金额，累加几十万条记录后出现误差几乎是必然的。推荐的稳妥做法是：数据库字段用 DECIMAL(12,2)，应用层使用 Decimal 类型，只在 JSON 输出或表格展示时转成 float 或格式化字符串。

7.2 退货、调价和盘点：三个绕不开的扩展

小店经营中退货是常态，但退货不能直接删原销售记录，否则会破坏收入、成本和库存三条链路的对账基础。建议单独建一张 sales_return_items 表，用负数量正向记录冲销。

调价要区分两种情况：已经在售但未卖完的库存，调价只影响未来售价；采购价变动，走采购入库的移动加权平均自动处理。不要直接去改 inventory.avg_cost 字段，那样会破坏已卖出记录的成本快照。

盘点差异也要单独记录。盘点时发现库存多了或少了，不能直接改 inventory 表，应该生成盘点单，标明盈亏数量和原因，由程序生成库存调整记录，并在利润表里单独展示“盘盈盘亏”行。

7.3 扩展方向：FIFO、批次、多门店

如果商品有保质期，移动加权平均就不够用了。先进先出（FIFO）需要额外引入批次表，每一步销售都要先匹配最早批次的剩余数量，并记录本次销售消耗了哪个批次。数据结构上需要给采购明细增加批次号、入库日期、有效期字段，销售明细增加批次关联字段。

多门店场景则要引入仓库表和调拨表。库存余额表从 product_id 做主键，改成 product_id + warehouse_id 联合主键，采购、销售、调拨都基于仓库维度记账。利润表汇总时可以选择按门店过滤或按门店分组。

7.4 上线前检查清单

不管系统规模多大，上线前建议过一遍下面的清单：

1. 商品主数据完整：每个商品有唯一 SKU、默认售价、单位；停用商品只下架，不删除。
2. 成本口径明确：移动加权平均还是 FIFO，在代码注释和操作手册里都要写明。
3. 历史成本不可变：销售明细的 cost_price 保存快照，任何新采购不得影响历史利润。
4. 库存校验生效：销售出库和采购入库在同一个事务内完成，失败整体回滚。
5. 对账流程落地：每天核对销售收入、销售成本、采购支出、库存盘点四个数字。
6. 备份可恢复：SQLite 每天复制文件，正式数据库做自动备份并演练恢复。
7. 操作有审计：记录谁在什么时间改了哪张单据，避免无法追溯的改数。
8. 金额精度可控：生产环境不使用浮点类型，统一为精确小数类型。

回到最开始的问题：利润表为什么不用自己算了？核心不是写一条 SQL，而是先定义清楚成本口径，再把成本快照保存到每一笔销售明细里，最后用确定性的查询汇总出不会变来变去的利润数字。把这一点想明白，小店进销存系统的主干就稳了。下一步可以继续做退货单、盘点单、批次保质期和多门店，但无论怎么扩展，销售成本快照这个原则都不要丢掉。