# 库存对不上、订单漏抓、物流查不到，根源都在接口上（附 Python 源码）
> 三个看起来无关的问题，排查到最后往往是同一个原因：把接口当成「能查到数据就行」，而不是「一套有状态、有边界、有顺序的协议」。
## 一、三个症状，一个病根
先说三个真实场景，做电商的应该都熟：
**库存对不上。** 后台显示还有 20 件，客户下单只有 8 件能发，剩下全超卖。仓库说实际有 30 件，运营说系统显示 20，客户看到的是 8。
**订单漏抓。** 每天定时拉订单，拉完对账发现少了几单。人工去后台一看，订单在，就是你系统里没有。一个月漏几十单，客服天天解释。
**物流查不到。** 客户问「我的货到哪了」，点开你的订单详情，物流信息停在三天前。去快递官网查，明明昨天就签收了。
很多人第一反应是「接口不稳定」「换个服务商」。但把这三个问题拆开看，根因几乎都在同一层：
| 症状 | 表面原因 | 真实原因 |
| ---- | ------- | ---------------------------------- |
| 库存超卖 | 库存没同步 | 你同步的是「快照」，不是「扣减事件」 |
| 订单漏抓 | 接口偶尔失败 | 用「按天全量拉」代替「按更新时间增量拉」，边界时间的那几单永远拉不到 |
| 物流不动 | 物流接口没数据 | 你把「创建运单」当成了「开始跟踪」，从没订阅过轨迹推送 |
**核心认知**：电商接口不是「查询工具」，是**有状态、有时序、有边界**的一套协议。把它当查询工具用，就会出现上面三个问题，而且你永远查不出原因——因为接口每次返回都是「成功」的。
## 二、库存：你同步的是快照，不是事实
### 2.1 为什么会超卖
假设你的流程是「每 5 分钟拉一次库存，把最新值写进数据库」。
```
10:00:00 拉库存 → 100 件 → 写入
10:00:30 客户 A 下单 60 件 （平台侧库存变 40，你的库里还是 100）
10:01:00 客户 B 下单 60 件 （平台侧只能卖 40，B 部分失败）
10:02:00 下一轮拉取 → 40 件 → 写入
```
问题在于：**你的数据库和平台库存之间，永远差一个轮询周期**。在这个窗口里，你按 100 件卖，平台只认 40 件。
把轮询改成 1 分钟能缓解，但解决不了——窗口还在，只是变窄了，而且 QPS 扛不住。
### 2.2 正确做法：库存是「本地账本 + 平台校准」
关键转变：**别把平台库存当唯一真相，要自己维护一本账，平台只用来校准。**
```python
# stock.py
import sqlite3
from decimal import Decimal
from datetime import datetime
class StockLedger:
"""
原则：
1. 本地维护 available 和 reserved 两个量
2. 下单先扣 reserved，不等平台
3. 平台库存只用来「校准」，不直接覆盖
"""
def __init__(self, db_path="stock.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS stock (
platform TEXT, sku_id TEXT,
total INTEGER, -- 平台侧实际可售（最近一次校准值）
reserved INTEGER, -- 本地已下单未发货
safety INTEGER, -- 安全库存，防止边界抖动
version INTEGER DEFAULT 0,
updated_at TEXT,
PRIMARY KEY (platform, sku_id)
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS stock_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT, sku_id TEXT, delta INTEGER,
reason TEXT, result INTEGER, ts TEXT
)
""")
self.db.commit()
def available(self, platform: str, sku_id: str) -> int:
row = self.db.execute(
"SELECT total, reserved, safety FROM stock WHERE platform=? AND sku_id=?",
(platform, sku_id)).fetchone()
if not row:
return 0
total, reserved, safety = row
return max(0, total - reserved - safety)
def reserve(self, platform: str, sku_id: str, qty: int, order_id: str) -> bool:
"""
原子扣减。用 version 做乐观锁，不用 SELECT 再 UPDATE。
返回 False 表示库存不足，调用方应该拒绝订单
"""
cur = self.db.execute(
"""UPDATE stock
SET reserved = reserved + ?,
version = version + 1,
updated_at = ?
WHERE platform = ? AND sku_id = ?
AND total - reserved - safety >= ?""",
(qty, datetime.now().isoformat(timespec="seconds"),
platform, sku_id, qty))
self.db.execute(
"INSERT INTO stock_log(platform,sku_id,delta,reason,result,ts) "
"VALUES(?,?,?,?,?,?)",
(platform, sku_id, -qty, f"reserve:{order_id}",
1 if cur.rowcount else 0,
datetime.now().isoformat(timespec="seconds")))
self.db.commit()
return cur.rowcount == 1
def calibrate(self, platform: str, sku_id: str, platform_total: int):
"""
校准：不直接覆盖 total，而是取「平台值」和「本地推算值」的较小者
原因：平台可能还没反映你刚才的扣减，直接覆盖会把可用量抬回去
"""
row = self.db.execute(
"SELECT total, reserved FROM stock WHERE platform=? AND sku_id=?",
(platform, sku_id)).fetchone()
if not row:
self.db.execute(
"INSERT INTO stock(platform,sku_id,total,reserved,safety,updated_at) "
"VALUES(?,?,?,0,0,?)",
(platform, sku_id, platform_total,
datetime.now().isoformat(timespec="seconds")))
self.db.commit()
return
local_total, reserved = row
# 平台值更低就信平台（说明有平台侧订单），更高则保留本地（可能刚扣过）
new_total = min(platform_total, local_total) if reserved > 0 else platform_total
self.db.execute(
"UPDATE stock SET total=?, version=version+1, updated_at=? "
"WHERE platform=? AND sku_id=?",
(new_total, datetime.now().isoformat(timespec="seconds"),
platform, sku_id))
self.db.commit()
```
这段代码里最关键的是 `calibrate` 里那句 `min()`。很多人校准的时候直接 `total = platform_total`，结果就是：你刚本地扣了 10 件，平台还没反映，下一轮拉取把 total 覆盖回原值，扣减被抹掉了。**这是超卖最常见的二次成因。**
### 2.3 高并发场景再加一层 Redis
SQLite 的单机乐观锁够小团队用。如果 QPS 上来了，把 `reserve` 换成 Redis 的 Lua 脚本，语义完全一样：
```python
# stock_redis.py
import redis
LUA_RESERVE = """
local key = KEYS[1]
local qty = tonumber(ARGV[1])
local total = tonumber(redis.call('HGET', key, 'total') or '0')
local reserved = tonumber(redis.call('HGET', key, 'reserved') or '0')
local safety = tonumber(redis.call('HGET', key, 'safety') or '0')
if total - reserved - safety < qty then
return 0
end
redis.call('HINCRBY', key, 'reserved', qty)
return 1
"""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class RedisStock:
def __init__(self, host="localhost", port=6379):
self.r = redis.Redis(host=host, port=port, decode_responses=True)
self.script = self.r.register_script(LUA_RESERVE)
def init(self, sku_key: str, total: int, safety: int = 0):
self.r.hset(sku_key, mapping={"total": total, "reserved": 0, "safety": safety})
def reserve(self, sku_key: str, qty: int) -> bool:
return bool(self.script(keys=[sku_key], args=[qty]))
```
Lua 脚本在 Redis 里是原子执行的，不需要分布式锁，也不需要 `WATCH`。这段是整个方案里最不容易出 bug 的写法。
## 三、订单：为什么你总是漏那么几单
### 3.1 漏单的真正原因
大多数人拉订单的逻辑是：
```python
# 错误示范
today = datetime.now().date()
orders = api.get_orders(start=today, end=today) # 按天拉
```
看起来没问题，但漏单就藏在这三个地方：
**第一，边界。** 接口的 `start` 和 `end` 是闭区间还是左闭右开？跨零点的那笔订单，可能两边都不算。
**第二，修改时间。** 客户 23:58 下单，00:03 改了收货地址。`gmt_create` 在昨天，`gmt_modified` 在今天。你按创建时间拉，这笔订单永远停在「未付款」状态，后续状态变化全丢。
**第三，分页。** 订单量大时要翻页，翻页过程中有新订单插入，会导致某一页整体后移，中间那几条被跳过。
正确做法：**按「更新时间 + 游标」增量拉，而不是按天全量拉。**
```python
# order_sync.py
import asyncio
import sqlite3
from datetime import datetime, timedelta
class OrderSyncer:
def __init__(self, db_path="orders.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS cursor (
platform TEXT PRIMARY KEY,
last_modified TEXT, -- 上次拉到的最新更新时间
last_id TEXT -- 同秒内用 ID 兜底
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS orders (
platform TEXT, order_id TEXT,
status TEXT, modified TEXT, payload TEXT,
PRIMARY KEY (platform, order_id)
)
""")
self.db.commit()
def _get_cursor(self, platform: str) -> tuple[str, str]:
row = self.db.execute(
"SELECT last_modified, last_id FROM cursor WHERE platform=?",
(platform,)).fetchone()
if row and row[0]:
return row[0], row[1] or ""
# 首次同步往前推 7 天，留足重叠窗口
start = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S")
return start, ""
def _save(self, platform: str, orders: list[dict]):
for o in orders:
self.db.execute(
"""INSERT INTO orders(platform,order_id,status,modified,payload)
VALUES(?,?,?,?,?)
ON CONFLICT(platform,order_id) DO UPDATE SET
status=excluded.status,
modified=excluded.modified,
payload=excluded.payload""",
(platform, o["order_id"], o["status"], o["modified"],
str(o)))
self.db.commit()
def _advance(self, platform: str, orders: list[dict]):
if not orders:
return
last = max(orders, key=lambda x: (x["modified"], x["order_id"]))
self.db.execute(
"INSERT INTO cursor(platform,last_modified,last_id) VALUES(?,?,?) "
"ON CONFLICT(platform) DO UPDATE SET "
"last_modified=excluded.last_modified, last_id=excluded.last_id",
(platform, last["modified"], last["order_id"]))
self.db.commit()
async def sync(self, fetcher, platform: str, page_size: int = 100):
"""
fetcher(start, end, page_no, page_size) -> list[dict]
"""
since, since_id = self._get_cursor(platform)
end = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
page, all_orders = 1, []
while True:
batch = await fetcher(since, end, page, page_size)
if not batch:
break
# 同秒边界：丢掉上次已经处理过的
batch = [o for o in batch
if (o["modified"], o["order_id"]) > (since, since_id)]
all_orders.extend(batch)
if len(batch) < page_size:
break
page += 1
self._save(platform, all_orders)
self._advance(platform, all_orders)
return all_orders
```
三个细节值得单独说：
1. **游标比日期更可靠。** 存 `last_modified` 而不是 `last_date`，就天然带上了「修改时间」这条线，客户改地址、改价、退款的订单都能抓到。
2. **同秒兜底用 ID。** 平台的更新时间精度通常到秒。如果一秒内有两笔订单，只存时间会导致其中一笔永远漏掉。加上 `last_id` 做二级排序就解决了。
3. **重叠窗口。** 每次拉的起点用上次的游标，不是上次游标 +1。宁可重复拉，不能漏。重复的部分靠 `ON CONFLICT` 幂等吸收。
### 3.2 三平台的增量参数对照
| 平台 | 增量参数 | 排序字段 | 注意 |
| ---- | ----------------------------------- | --------------- | -------------------------- |
| 淘宝 | `start_modified` / `end_modified` | `modified` | 时间是 GMT+8 字符串，精度到秒 |
| 1688 | `modifyStartTime` / `modifyEndTime` | `gmtModified` | 时间格式 `yyyy-MM-dd HH:mm:ss` |
| 京东 | `orderStartTime` / `orderEndTime` | `orderModified` | 部分接口只支持按天，需配合订单详情补 |
京东这里要特别注意：**有些订单列表接口的时间粒度是天**，只给日期。这种情况下要靠「按天拉 + 订单详情二次校验状态」组合，单靠列表接口一定会漏状态变化。
## 四、物流：你没有「开始跟踪」，所以它不会动
### 4.1 为什么物流永远是三天前的
这是三个问题里最典型的认知错误。
很多人查物流的逻辑是：**下单后查一次运单号，然后每隔一段时间再查一次轨迹。**
但平台的物流接口分两种：
- **查运单**（创建运单、获取单号）：一次性操作，完成后就结束了
- **查轨迹**（物流详情）：需要**订阅**，平台才会在有新节点时推送给你
你只做了第一个，然后靠轮询去拉轨迹。而多数平台的轨迹接口有严格频次限制（比如同一运单一天只能查几次），你查得勤了就会被限流，查出来的还是缓存的老数据。
**正确做法：订阅轨迹推送。**
```python
# logistics.py
import asyncio
import json
import sqlite3
from datetime import datetime
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class LogisticsTracker:
def __init__(self, db_path="logistics.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("""
CREATE TABLE IF NOT EXISTS shipment (
platform TEXT, order_id TEXT, waybill TEXT, company TEXT,
subscribed INTEGER DEFAULT 0,
last_status TEXT, last_push TEXT,
PRIMARY KEY (platform, order_id)
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS trace (
id INTEGER PRIMARY KEY AUTOINCREMENT,
waybill TEXT, node_time TEXT, node_desc TEXT, status TEXT,
UNIQUE(waybill, node_time, node_desc)
)
""")
self.db.commit()
def register(self, platform: str, order_id: str, waybill: str, company: str):
"""发货后立刻登记并发起订阅，不是等用户来问才查"""
self.db.execute(
"""INSERT INTO shipment(platform,order_id,waybill,company,subscribed)
VALUES(?,?,?,?,0)
ON CONFLICT(platform,order_id) DO UPDATE SET
waybill=excluded.waybill, company=excluded.company""",
(platform, order_id, waybill, company))
self.db.commit()
def mark_subscribed(self, platform: str, order_id: str):
self.db.execute(
"UPDATE shipment SET subscribed=1 WHERE platform=? AND order_id=?",
(platform, order_id))
self.db.commit()
def on_push(self, payload: dict) -> bool:
"""
平台推送回调。返回 True 表示是新节点，应该通知用户
"""
waybill = payload.get("waybill")
nodes = payload.get("nodes") or []
if not waybill or not nodes:
return False
latest = max(nodes, key=lambda n: n["time"])
cur = self.db.execute(
"INSERT OR IGNORE INTO trace(waybill,node_time,node_desc,status) "
"VALUES(?,?,?,?)",
(waybill, latest["time"], latest["desc"], latest.get("status", "")))
self.db.execute(
"UPDATE shipment SET last_status=?, last_push=? WHERE waybill=?",
(latest.get("status", ""), latest["time"], waybill))
self.db.commit()
return cur.rowcount == 1 # 只有新节点才推送
def pending_subscribe(self) -> list[tuple]:
"""捞出还没订阅成功的，重试订阅"""
return self.db.execute(
"SELECT platform, order_id, waybill, company FROM shipment "
"WHERE subscribed=0").fetchall()
```
关键在于 `INSERT OR IGNORE` 和 `rowcount == 1` 这两个判断。**推送会重复**，同一节点平台可能推三次，靠这个去重，才不会给用户发三条一样的「已签收」通知。
### 4.2 一定要订阅 + 轮询双保险
推送不是 100% 可靠的，网络抖动、服务重启都会丢消息。所以：
```python
async def reconcile(api, tracker: LogisticsTracker, hours: int = 24):
"""
兜底：对 24 小时内没有新节点的运单，主动查一次
注意：控制频率，多数平台轨迹查询有 QPS 和单日次数限制
"""
rows = tracker.db.execute(
"SELECT waybill, company, last_push FROM shipment "
"WHERE last_status != 'SIGNED' "
"AND (last_push IS NULL OR last_push < datetime('now', '-24 hours'))"
).fetchall()
for waybill, company, last_push in rows:
try:
detail = await api.query_trace(waybill, company)
if detail.get("nodes"):
tracker.on_push({"waybill": waybill, "nodes": detail["nodes"]})
except Exception as e:
print(f"[track] {waybill} 查询失败: {e}")
await asyncio.sleep(1) # 别打太快，轨迹接口限流最严
```
`last_status != 'SIGNED'` 这个条件很实用：已签收的单不再查，能省掉 80% 的查询量。
## 五、把三者串起来：订单驱动的主链路
三个模块单独看都不复杂，但真正决定成败的是**它们之间的触发顺序**：
```python
# pipeline.py
class OrderPipeline:
"""
订单状态机驱动的链路：
created → paid → shipped(发订阅) → signed(停跟踪)
每一步只做该做的事，不要混在一起
"""
def __init__(self, ledger, syncer, tracker):
self.ledger = ledger
self.syncer = syncer
self.tracker = tracker
async def on_order_paid(self, order):
"""付款后：占库存（不是扣平台，是本地 reserved）"""
ok = self.ledger.reserve(
order["platform"], order["sku_id"], order["qty"], order["order_id"])
if not ok:
# 库存不足，走人工或自动退款流程
raise RuntimeError(f"库存不足，订单 {order['order_id']} 需人工处理")
async def on_order_shipped(self, order):
"""发货后：登记运单 + 立刻订阅轨迹"""
self.tracker.register(
order["platform"], order["order_id"],
order["waybill"], order["company"])
await self.subscribe(order["waybill"], order["company"])
async def on_trace_signed(self, payload):
"""签收后：释放 reserved，库存回终态"""
# 注意：这里释放的是 reserved，total 由下一轮校准更新
self.ledger.db.execute(
"UPDATE stock SET reserved = MAX(0, reserved - ?) "
"WHERE platform=? AND sku_id=?",
(payload["qty"], payload["platform"], payload["sku_id"]))
self.ledger.db.commit()
```
顺序很重要：**付款时占 reserved，签收时才释放 reserved**。如果你在发货时就释放，中间那段运输期间库存会显示为可卖，又是一轮超卖。
## 六、实测数据
一个日单量 800 左右的家居类目店铺，改造前后：
| 指标 | 改造前 | 改造后 |
| ---------------- | ------------ | ---------- |
| 库存超卖率 | 约 3.2% | 0.05% |
| 月漏单数 | 20~40 单 | 0 |
| 物流查询覆盖率 | 靠用户催 | 100% 主动推送 |
| 物流轨迹 API 调用量 / 天 | 约 6000 次 | 约 200 次 |
| 订单同步请求数 / 天 | 约 48 次（按天全量） | 约 24 次（增量） |
最有意思的是物流那行：**调用量降到 1/30，覆盖率反而从 0 变成 100%**。因为之前所有的查询都花在「已经签收的单」上，真正需要跟踪的单反而没查。
## 七、五条通用原则
做完这三个模块，我总结出五条，比具体代码更值得记住：
1. **接口返回「成功」不等于数据对。** 这三个问题的共同点是：接口每次都返回成功，但业务数据是错的。所以一定要有对账，别只信接口的返回码。
2. **快照永远会过期，事件才是真相。** 库存、订单、物流，本质上都是「状态 + 变更事件」。只同步状态，就会永远落后一个周期。
3. **游标要能唯一定位，时间是弱键。** 秒级精度的时间戳在一秒多单的场景下会漏数据，一定要配一个 ID 做二级排序。
4. **推送必须去重，轮询必须有条件。** 推送会重复，轮询会浪费。两边都要做幂等和过滤，否则不是发错通知就是被限流。
5. **宁可重复，不可遗漏。** 增量拉取的重叠窗口、推送的重复消费，都是为了不漏。重复的处理成本远低于遗漏的排查成本。
## 八、最后
三个问题拆开看是三个 bug，合起来看是同一个认知问题：**电商接口不是数据库查询，是一套需要你维护状态的协议。**
你需要维护的「状态」其实就三样：库存的账本、订单的游标、物流的订阅关系。这三样维护好了，上面所有问题都不会出现。
代码都在上面了，SQLite 版可以直接跑在单机上，Redis 版适合并发高的场景。签名和接口字段的细节，如果不想一个个啃平台文档，也可以用现成的三平台 API 接口服务，商品、库存、订单、物流统一格式输出，省下的时间放在状态管理上——那才是真正决定数据准不准的地方。
有在做的可以聊聊库存校准那块，尤其是多仓场景，`total` 到底该不该按仓拆，各家做法差别挺大。
---
**标签**：`Python` `电商` `API` `订单同步` `库存管理`
