商品详情、SKU、价格、库存，四个高频接口的正确调用姿势（附 Python 源码）
先给一个数字：这四个接口，占我们平台总调用量的 85%。
这个占比意味着两件事。第一，任何调用姿势上的坏习惯，都会被放大 85%——配额浪费、数据错位、深夜告警，基本都出在这四个接口上。第二，优化它们的收益也是 85% 的，比优化任何冷门接口都值。
这篇是手册型文章，和第 11 篇的对照表配套使用：那篇讲三个平台"有什么不同"，这篇讲"每个接口该怎么调"。四个接口各一份错误姿势清单，最后一个总纲。
〇、总纲：四个接口不是四件事，是一棵树
最常见的错误，是把它们当成四个平级接口、用同一个轮询周期去调。
但它们有严格的依赖关系：
商品详情（根）
└── SKU 集合（挂在商品上）
├── 价格（挂在 SKU 上）
└── 库存（挂在 SKU 上）
这个依赖关系推出两条铁律：
铁律一：按变更频率分级调度。 详情一天变不了几次，库存每秒都在变。捆在一个轮询周期里，要么详情白调（浪费配额），要么库存延迟（业务事故）。
铁律二：能用一次详情带出的，不要拆四次请求。 三个平台的详情接口几乎都会返回 SKU 列表和标价——一次请求拿三样东西，这是最大的配额节省。真正需要单独调的只有库存（它要实时）。
| 数据 | 变更频率 | 来源 | 同步方式 |
|---|---|---|---|
| 标题、主图、类目 | 天级 | 详情接口 | 每日增量 |
| SKU 集合 | 小时级 | 详情接口内 | 集合 diff |
| 价格 | 分钟级 | 详情 / 价格接口 | 定时 + 改价回执 |
| 库存 | 秒级 | 库存接口 / 消息推送 | 事件优先，轮询兜底 |
下面四个接口的"正确姿势"，全部从这两条铁律推出来。
一、商品详情：它是根，不是流水账
详情接口的错误姿势，集中在"把它当唯一的真相来源"。
| 错误姿势 | 后果 | 正确姿势 |
|---|---|---|
| 全量轮询所有商品详情 | 配额爆炸，且 90% 请求拿回的是没变的数据 | 列表接口增量筛出变更 ID，再按需调详情 |
不传 fields，全字段拉 |
响应体积差 10 倍，解析和带宽都翻倍 | 按需声明字段，用不到的不拉 |
| 把详情里的价格/库存当业务数据源 | 数据滞后几分钟到几小时 | 详情只管"慢字段"，快字段走分级调度 |
| 下架商品按正常结构解析 | 静默拿到空数据或 KeyError | 先判状态位，软删除单独处理 |
四条里最贵的是第二条。很多人不知道详情接口支持字段裁剪，把整个响应体解析一遍，结果：每次多传 80% 用不到的字节，配额按调用次数算倒是不亏，但带宽、解析时间、落库量全都翻倍。字段白名单就一行的事：
DETAIL_FIELDS = [
"num_iid", "title", "pic_url", "cid",
"skus", "price", # SKU 集合和标价跟着详情一起拿（铁律二）
"list_time", "delist_time",
]
params = {
"num_iid": item_id, "fields": ",".join(DETAIL_FIELDS)}
第四条单独说一句：下架商品在不同平台的表现不同——有的返回错误码，有的返回成功但核心字段为空，有的直接不返回。用第 12 篇的 integrity.validate 兜底，加上一个状态位判断，别让下架商品在你的库里"看起来还活着"。
二、SKU：它是会消失的
SKU 是四个接口里坑最深的，因为它的坑不在"怎么调"，在"调回来之后怎么管"。
| 错误姿势 | 后果 | 正确姿势 |
|---|---|---|
只拿 sku_id 当主键 |
多店铺/多平台必然撞键 | 主键 = (platform, item_id, sku_id) |
| 覆盖式更新 SKU 表 | 商家删掉的 SKU 永远留在你库里 | 集合 diff，消失的标记 inactive |
props.split(":") 硬拆销售属性 |
属性名里含冒号就炸 | 结构化解析（第 2 篇的 _parse_props） |
| 假设每个商品都有 SKU | 单规格商品返回空 | 先判商品规格类型 |
第二条是重灾区。商家改规格是日常操作：把"红色 L"删了换成"红色 XL"。如果你每次拉到 SKU 列表就整表覆盖，或者只往上插入新 SKU——被删掉的那个 SKU 会永远留在你库里，价格库存还挂着旧数据。客户拿它下单，就是一次客诉。
正确做法是集合 diff：
# sku_diff.py
def sku_diff(old_ids: set[str], new_ids: set[str]) -> tuple[set[str], set[str]]:
"""
返回 (新增, 消失)。
消失的 SKU 千万别物理删除——历史订单还引用它。
标记 inactive，保留可查。
"""
return new_ids - old_ids, old_ids - new_ids
async def sync_skus(store, item_id: str, new_skus: list[dict]):
old_ids = set(store.sku_ids(item_id))
new_ids = {
s["sku_id"] for s in new_skus}
added, removed = sku_diff(old_ids, new_ids)
for s in new_skus:
if s["sku_id"] in added:
store.insert_sku(item_id, s)
else:
store.update_sku(item_id, s) # 属性或价格变了
for sku_id in removed:
store.mark_inactive(item_id, sku_id) # 软删，订单引用不断
另外主键那条：sku_id 只在单个商品内唯一。做三平台、多店铺的库，主键从第一天就是三元组，后期改主键的痛苦是接入期的十倍（第 11 篇说 1688 连 SKU 都没有，那它的"规格"要单独建表，别硬凑）。
三、价格：它不是一个字段，是一组字段
价格的问题出在"你以为价格是一个数"。
| 错误姿势 | 后果 | 正确姿势 |
|---|---|---|
| 只取一个价格字段 | 拿到标价，不是成交价 | 价格组归一，算"生效价" |
| float 存价格 | 账对不上，差几分钱 | Decimal，最后一步才转 float |
| 改价后立刻读回来验证 | 读到旧缓存，以为没改成 | 接受几分钟延迟，或以改价接口回执为准 |
| 1688 只取一个价 | 阶梯价拿错档位 | 按起订量取档 |
| 促销价无限期生效 | 过期促销价当现价 | 促销价必须带时效字段 |
第一和第五条组合起来，是最经典的线上事故：拿到促销价 → 没记过期时间 → 促销结束后继续按促销价计算。利润表和实际对不上，查一周。
正确建模是把价格当成一个带时效的小对象：
# price_model.py
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo
TZ = ZoneInfo("Asia/Shanghai")
@dataclass
class Price:
sku_id: str
list_price: Decimal # 标价 / 一口价
promo_price: Decimal | None = None # 促销价
promo_end: datetime | None = None # 促销截止时间
def effective(self, now: datetime | None = None) -> Decimal:
"""生效价：促销在有效期内才生效，过期自动回落标价"""
now = now or datetime.now(TZ)
if self.promo_price is not None:
if self.promo_end is None or self.promo_end > now:
return self.promo_price
return self.list_price
1688 的价格是另一个物种——阶梯价，按起订量分档：
def alibaba_price_tiers(raw: list[dict]) -> list[tuple[int, Decimal]]:
"""
1688 priceRange：[{beginAmount: 2, price: "8.50"}, {beginAmount: 100, price: "7.20"}]
按起订量升序排列，采购 N 件时取满足 N 的最后一档。
"""
tiers = sorted(
((int(x["beginAmount"]), Decimal(x["price"])) for x in raw),
key=lambda t: t[0],
)
return tiers
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def tier_for(tiers: list[tuple[int, Decimal]], qty: int) -> Decimal:
price = tiers[0][1]
for begin, p in tiers:
if qty >= begin:
price = p
return price
还有改价那条：改价接口返回成功 ≠ 价格已生效。平台侧有缓存延迟，短则几秒长则几分钟。以改价接口的响应做业务判断，别改完立刻读回来验证——你会拿到旧价格，然后误判"改价失败"再改一次。
四、库存：只有"可售库存"有意义
库存的问题在于：平台给你好几个数，你用了错的那个。
| 错误姿势 | 后果 | 正确姿势 |
|---|---|---|
| 用"总库存"当可卖数量 | 超卖（锁定的库存又卖了） | 用可售库存字段 |
| 平台库存直接覆盖本地库存 | 抹掉本地扣减，二次超卖 | 本地账本 + 校准（第 3 篇） |
| 和详情同频率轮询 | 延迟大还烧配额 | 事件推送优先，轮询兜底 |
| 库存为 0 就当售罄 | 可能是下架或区域性无货 | 状态位 + 库存一起判断 |
| 硬凑 1688 的库存 | 它没有 SKU 库存概念 | 可售状态 + 阶梯价建模 |
第三条是这篇的落地重点。库存是唯一值得上事件推送的字段——淘宝的 TMC 有库存变更消息，订阅它，库存事件毫秒级到达，配额零消耗。轮询只做兜底（比如每 5 分钟对账一次，防丢消息）：
# stock_sync.py —— 事件优先，轮询兜底
async def on_stock_event(msg: dict):
"""TMC 库存变更消息：毫秒级，零配额"""
await apply_stock(msg["item_id"], msg["sku_id"], msg["quantity"])
async def reconcile_loop(gw, store, item_ids: list[str], interval: float = 300):
"""兜底对账：防消息丢失。只查本地 5 分钟内有订单变动的商品，别全量"""
while True:
hot = store.items_with_recent_orders(minutes=5)
for iid in hot:
raw = await gw.call("stock.get", item_id=iid)
local = store.local_stock(iid)
# 校准取 min：平台还没反映本地刚扣的量（第 3 篇的规则）
safe = min(int(raw["saleable"]), local)
if store.remote_stock(iid) != safe:
store.update_remote_stock(iid, safe)
await asyncio.sleep(interval)
注意兜底循环里那句 hot = items_with_recent_orders(minutes=5)：只对账"最近有单"的商品。全量对账是最容易写出来的版本，也是把库存轮询成本放大一百倍的版本——而真正需要对账的（刚发生扣减、可能不一致的）永远只占极小一部分。
五、把四个接口串起来：分级调度器
最后给一个可跑的调度骨架。它的职责只有一件事：保证每个字段按自己的频率被同步，不多不少。
# scheduler.py
import asyncio
import time
LEVELS = {
"detail": 3600, "sku": 3600, "price": 300, "stock": 30}
class DueScheduler:
def __init__(self, r):
self.r = r
async def due(self, item_id: str, field: str) -> bool:
last = await self.r.get(f"due:{field}:{item_id}")
if last is None:
return True # 从没同步过，先来一次
return time.time() - float(last) >= LEVELS[field]
async def mark(self, item_id: str, field: str):
await self.r.set(f"due:{field}:{item_id}", time.time())
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def sync_item(gw, sched: DueScheduler, store, item_id: str):
# ---- 1) 详情（小时级）：一次拉全 慢字段 + SKU 集合 + 标价 ----
if await sched.due(item_id, "detail"):
d = await gw.call("item.get", item_id=item_id,
fields=",".join(DETAIL_FIELDS))
store.upsert_item(item_id, d)
new_skus = d.get("skus", [])
await sync_skus(store, item_id, new_skus) # 内部做集合 diff
await sched.mark(item_id, "sku")
await sched.mark(item_id, "detail")
# ---- 2) 价格（分钟级）：从详情缓存或价格接口取，归一成 Price 对象 ----
if await sched.due(item_id, "price"):
raw = await gw.call("price.get", item_id=item_id)
for sku_price in normalize_prices(item_id, raw):
store.save_price(sku_price) # effective() 按时效算生效价
await sched.mark(item_id, "price")
# ---- 3) 库存（秒级）：事件推送是主通道，这里只是兜底标记 ----
# 真正的兜底对账在 reconcile_loop，只扫最近有单的商品
async def main(gw, sched, store, item_ids):
# 详情和价格按各自的频率到期才调；库存靠事件
while True:
for iid in item_ids:
await sync_item(gw, sched, store, iid)
await asyncio.sleep(30) # 最细粒度的检查周期，但不是最细粒度的调用频率
跑起来的效果：同一个商品，详情接口一天调 24 次、价格 288 次、库存 0 次（走事件）——对比"四个接口都 30 秒一轮"的写法，配额消耗约降到 1/300，而库存的实时性反而更好（事件是毫秒级的）。
六、四个反直觉的结论
1. 详情接口调得越勤，数据越不准。
因为详情是慢字段的来源，高频调它不会让标题更新更快（它一天就变一次），只会让你的价格库存看起来"应该也是新的"——然后掩盖它们其实滞后的事实。分开调度，延迟才藏不住。
2. 消失的 SKU 比新增的危险。
新增最多是数据多一条；消失的 SKU 还挂着旧价格旧库存，客户能下单。覆盖式更新是这里的万恶之源。
3. 促销价必须带时效，这是建模问题不是字段问题。
没有 promo_end 字段的促销价，就是一颗定时炸弹——促销结束那天准时爆。
4. 库存的正确姿势是"少调"，不是"勤调"。
事件推送零配额、毫秒级；轮询全量、高延迟。轮询不是兜不了底，是兜底时只扫"最近有单"的商品——全量对账是库存成本失控的第一原因。
七、最后
这篇没有新架构，讲的全是已有架构里每个接口的使用纪律。
回看整个系列：第 8 篇给了骨架（统一接口层），第 9、10 篇给了运行时（稳定性、限流），第 12 篇给了选型（为什么用 API），第 11 篇给了差异（平台之间差在哪）。这篇补的是最后一层——每个高频接口的调用纪律。骨架、运行时、选型、差异、纪律，五层齐了。
下一篇写「多店铺授权管理」：token 的存储与加密、并发刷新撞车、单店授权过期的隔离。那是接入体系的最后一个坑区，写完这个系列就真正完整了。
LEVELS 里的数字是起点不是标准：详情 1 小时、价格 5 分钟、库存 30 秒，都是按"多数电商场景"取的。你们的业务如果价格敏感度高（比如比价类），价格调到 60 秒；如果纯铺货场景，详情调到 6 小时都够。关键是分级这件事本身，不是每级的具体数字。
