结论先拍：1688基础API（商品搜索/详情/订单）本身免费，但默认搜索类QPS≈10/s、订单类≈20/s。 做“批发同步”时瓶颈不在钱，而在速率——QPS 10 意味着单Key理论峰值 600次/分钟、3.6万次/小时，但凡你监控几百个SKU、多店轮询、大促冲一下，立刻 ISP_FLOW_CONTROL_LIMIT。而高级实时库存（区别于详情里带出的基础stock_num）和提频能力，必须买资源包（年费几百~几千档），不是“可买可不买”，是“要实时防超卖就必须买”。
一、免费QPS 10 在批发同步里的真实瓶颈
假设你是一件代发/批发采销ERP，同步逻辑是：
- 监控 500 个活跃offer（爆款SKU）
- 每个offer有 5~20 个SKU，需查实时可售库存
- 爆款轮询周期 30s，普通品 5min
算一笔账：
- 爆款500个 × 30s一轮 → 每秒需查 500/30 ≈ 16.7 次/s（已超过单Key 10 QPS）
- 如果为保险每次查
alibaba.item.get拿sku_list.stock基础库存（免费但占QPS），单Key直接限流 - 若改用高级实时库存接口（更准、区分占用/锁定/在途），基础免费额度不覆盖，必须资源包授权才能调
免费QPS 10 的真实含义：适合“每日几千次增量同步”的中小B2B，不适合“几百SKU秒级防超卖”的批发前端。后者要么多Key轮询（同企业实名下，合规但运维复杂），要么买包提频到50/100/200 QPS。
二、为什么要单独买“实时库存资源包”？
1688的库存数据分两层：
- 详情携带基础库存：
alibaba.item.get返回的sku_list.stock_num，免费但非强一致（有缓存、不含锁定占用），适合“大概有货”展示。 - 高级实时库存：高级实时库存接口（如实时可售/占用查询类），返回实时扣减占用后的可售数，做代发锁库存必须用它——这套能力在资源包/增值权限后开放，不在基础免费范围内。
买资源包拿到两件事：
- QPS提频（10 → 50/100/200，按包档位）
- 高级实时库存/数据推送权限（普通Key调高级接口返回无权限）
价格参考（2026）：提频+实时库存组合包 ¥980~2980/年 常见，纯提频低些，跨境寻源增强包同区间。
三、Python：QPS 10 瓶颈模拟 + 多Key轮询 + 高级接口守卫
# ali1688_stock_sync_guard.py
"""
1688批发同步守卫
- 单Key令牌桶限速(保守8/s，留余量)
- 多Key轮询突破QPS 10
- 高级实时库存接口权限守卫（无包直接熔断）
- 捕获 ISP_FLOW_CONTROL_LIMIT / 429 退避
"""
import hashlib, time, requests, json
from typing import Dict, List
from threading import Lock
GW = "https://gw.api.1688.com/openapi/param2/1/"
DEFAULT_QPS = 10
SAFE_QPS = 8 # 单Key保守值
class TokenBucket:
def __init__(self, rate):
self.rate = rate
self.tokens = rate
self.ts = time.monotonic()
self.lk = Lock()
def wait(self):
with self.lk:
now = time.monotonic()
self.tokens = min(self.rate, self.tokens + (now-self.ts)*self.rate)
self.ts = now
if self.tokens < 1:
time.sleep((1-self.tokens)/self.rate + 0.005)
self.tokens = 0
else:
self.tokens -= 1
class Ali1688Guard:
def __init__(self, apps: List[Dict], use_advance_stock=False):
"""
apps: [{"key":..,"secret":..}, ...] 同企业实名多Key
use_advance_stock: 是否调高级实时库存（需资源包）
"""
self.apps = apps
self.buckets = [TokenBucket(SAFE_QPS) for _ in apps]
self.idx = 0
self.idx_lk = Lock()
self.use_adv = use_advance_stock
self.has_adv_pack = False # 默认没买包，调高级接口直接拦
def _next(self):
with self.idx_lk:
i = self.idx
self.idx = (self.idx + 1) % len(self.apps)
return i, self.apps[i], self.buckets[i]
def _sign(self, params: Dict, secret: str) -> str:
f = sorted((k,v) for k,v in params.items()
if v is not None and str(v).strip()!="" and k!="sign")
qs = "".join(f"{k}{v}" for k,v in f)
return hashlib.md5(f"{secret}{qs}{secret}".encode()).hexdigest().upper()
def _call(self, method: str, biz: Dict, token=None, max_retry=4):
for att in range(max_retry):
i, app, bucket = self._next()
bucket.wait()
url = f"{GW}{method}/{app['key']}"
params = {
"method": method,
"app_key": app['key'],
"timestamp": str(int(time.time()*1000)),
"format": "json", "v": "2.0", "sign_method": "md5"
}
if token: params["access_token"] = token
params.update(biz)
params["sign"] = self._sign(params, app['secret'])
try:
化
r = requests.post(url, data=params, timeout=15)
d = r.json()
if "error_response" in d:
err = d["error_response"]
sub = err.get("sub_code","")
msg = err.get("msg","")
if "FLOW_CONTROL" in sub or "ISP_FLOW_CONTROL_LIMIT" in sub or "429" in msg:
time.sleep(min(2**att, 8))
continue
if "no permission" in msg and self.use_adv:
raise PermissionError("❌ 未购实时库存资源包，高级接口无权限")
raise Exception(f"1688[{sub}]: {msg}")
return d
except requests.exceptions.RequestException:
time.sleep(2**att)
continue
raise RuntimeError("多次重试仍限流")
# 基础库存（免费，占QPS）
def get_basic_stock(self, item_id: str) -> Dict:
d = self._call("alibaba.item.get",
{"item_id": item_id, "fields": "sku_list"})
return d.get("alibaba_item_get_response", {}).get("result", {})
# 高级实时库存（需资源包）
def get_realtime_stock(self, item_id: str):
if not self.use_adv:
raise ValueError("业务未开启高级库存模式")
if not self.has_adv_pack:
raise PermissionError("🚨 未购买实时库存资源包，无法调用高级接口；请去控制台购包或降级基础stock_num")
# 示意method（以官方文档为准）
return self._call("alibaba.item.stock.realtime.get",
{"item_id": item_id})
if __name__ == "__main__":
apps = [{"key":"AK1","secret":"AS1"}, {"key":"AK2","secret":"AS2"}]
cli = Ali1688Guard(apps, use_advance_stock=True)
# 场景A：基础库存轮询（免费，受QPS 10限制，双Key理论16/s）
for iid in ["612345678901", "612345678902"]:
try:
res = cli.get_basic_stock(iid)
print("基础库存SKU数:", len(res.get("sku_list",[])))
except Exception as e:
print("ERR", e)
# 场景B：高级实时库存（没买包直接拦）
try:
cli.get_realtime_stock("612345678901")
except PermissionError as e:
print(e)四、批发同步的三种可行架构
- 极小规模（<100 SKU监控）：单Key + 令牌桶8 QPS + 基础
stock_num缓存10秒，零费够用。 - 中等代发（100~1000 SKU防超卖）：买实时库存资源包（提频到50QPS）+ 基础接口拿规格，高级接口查可售，本地Redis缓存SKU库存5~10秒。
- 大卖/多店ISV：多Key轮询（同企业实名3~5个Key）+ 高级包 + 1688消息订阅（Webhook库存变更推送） 替代主动轮询，把“拉”变成“推”，QPS压力直接归零。
五、避坑
- “详情里有stock就不用买包” → 那是基础库存，有缓存、不含占用，做代发锁库存会超卖。
- “QPS 10 乘86400天调用够” → QPS是并发速率不是日总量，10/s 峰值=每天理论86万但持续打满必被风控，实际安全日用几万次。
- “多Key绕开限制” → 可以，但必须同企业实名、不同应用，禁止同Key并发伪装，否则触发风控封应用。
- 资源包过期 → 自动降级回QPS 10，大促当天“突然变慢”的经典事故源。
一句话：1688批发同步的瓶颈不是钱是速率，免费QPS 10只能喂中小B2B；要实时防超卖，高级实时库存资源包 = 提频 + 高级权限二合一，不是智商税。
要不要我帮你把上面
Ali1688Guard 改成Redis中心化令牌桶（支持多进程分布式调度）+ 1688 Webhook库存推送消费端，把轮询模式切到“推送为主、轮询兜底”？