🧪
结论先拍：1688的基础API（商品搜索/详情/订单/物流）确实免费，但免费≠无限——默认搜索类QPS≈10/s、订单类≈20/s，超了返回 ISP_FLOW_CONTROL_LIMIT 或 HTTP 429，不扣钱但直接失败。破局不是“绕过限制”，而是三层正规手段：客户端令牌桶自律（单Key压8~9 QPS）+ 同企业多AppKey轮询（N×10）+ 买资源包提频到50/100/200并开高级实时库存。做批发防超卖，alibaba.item.get 里的 stock_num 是带缓存的基础库存，不含占用/锁定，真要防超卖必须高级实时库存接口（资源包后开放）。
一、1688“免费”的准确边界
接口类 典型method 是否收调用费 默认QPS 超限表现
商品搜索 alibaba.offer.search / alibaba.product.search 免费 ≈10/s ISP_FLOW_CONTROL_LIMIT
商品详情 alibaba.item.get 免费 ≈10/s 同上（返回stock_num有缓存）
订单/物流 alibaba.trade.buyer.list / alibaba.logistics.trace.get 免费 ≈20/s 同上
高级实时库存 实时可售/占用查询类（非公开统一名，以控制台权限为准） 需资源包 随包提频 无包返回no permission
分销/跨境/数据推送 分销专用族 / 消息订阅高级版 需资源包/认证 随包 同上
官方不承诺固定日总量，但单Key持续打满10 QPS 会被风控审查；实测中小自研（日几千~几万次）零费跑一年没问题。
二、QPS 10 在批发同步里的真实瓶颈
假设代发ERP监控 500个活跃offer，每个offer 5~20个SKU：
• 爆款30s轮询一次 → 需 500/30 ≈ 16.7 次/s，单Key 10 QPS直接限流
• 若每次拉 alibaba.item.get 拿 sku_list.stock_num（基础库存，非强一致，不含占用）→ 免费但占QPS，越拉越堵
- 改高级实时库存接口 → 基础免费额度不覆盖，没买包连调用权限都没有
QPS是并发速率不是日总量：10/s 理论峰值 86万/天，但持续打满必触风控，实际安全日用几万次。
三、破QPS 10 的三层正规解法（不碰灰产）
- 客户端自律：单Key令牌桶压到 8~9 QPS，留20%余量防突发被封。
- 同企业多AppKey轮询：同营业执照下建3~5个“自用型应用”，各10 QPS，网关前做Key轮转（禁止同Key并发伪装、禁止个人号堆砌，否则封应用）。
- 买资源包：基础提频包把单Key升到50/100/200 QPS；实时库存组合包（¥980~2980/年）同时解锁高级接口+提频；大卖接消息订阅Webhook把“拉”变“推”，QPS压力归零。
四、GuardedClient 源码（多Key轮询 + 令牌桶 + 高级接口守卫）
ali1688_guarded_client.py
"""
1688 开放平台 GuardedClient
- 多AppKey轮询（同企业实名）突破单Key QPS10
- 每Key独立令牌桶（保守8 QPS）
- 高级实时库存接口权限守卫（无包直接熔断）
- 捕获 ISP_FLOW_CONTROL_LIMIT / 429 指数退避
"""
import hashlib, time, requests
from typing import Dict, List
from threading import Lock
GW = "https://gw.api.1688.com/openapi/param2/1/"
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 Ali1688GuardedClient:
def init(self, apps: List[Dict], adv_pack: bool = False,
safe_qps: int = 8):
"""
apps: [{"key":..,"secret":..}, ...] 同企业实名多Key
adv_pack: 是否购买了实时库存资源包
"""
self.apps = apps
self.buckets = [TokenBucket(safeqps) for in apps]
self.idx = 0
self.idx_lk = Lock()
self.adv_pack = adv_pack
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=5):
last_err = None
for att in range(max_retry):
i, app, bucket = self._next()
bucket.wait()
# 1688 GET风格：param2/1/{method}/{app_key}
url = f"{GW}{method}/{app['key']}"
params = {
"app_key": app["key"],
"method": method,
"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.get(url, params=params, timeout=15)
d = r.json()
if "error_response" in d:
err = d["error_response"]
sub = str(err.get("sub_code", ""))
msg = str(err.get("msg", ""))
if "FLOW_CONTROL" in sub or "429" in msg or "too many" in msg.lower():
time.sleep(min(2 ** att, 8)) # 指数退避
last_err = f"limit:{sub}"
continue
if "no permission" in msg and "stock" in method.lower():
raise PermissionError("🚨 未购实时库存资源包，高级接口无权限")
raise Exception(f"1688[{sub}]: {msg}")
return d
except requests.RequestException as e:
last_err = str(e)
time.sleep(2 ** att)
continue
raise RuntimeError(f"retry exhausted: {last_err}")
# 基础库存（免费，占QPS10）
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.adv_pack:
raise PermissionError("🚨 未购实时库存资源包，禁止调高级接口；"
"请去控制台服务市场→API资源包购买，或降级用基础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 = Ali1688GuardedClient(apps, adv_pack=False)
# 场景A：基础库存轮询（双Key理论16 QPS，令牌桶各8 → 实际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：没买包硬调高级接口 → 直接PermissionError
try:
cli.get_realtime_stock("612345678901")
except PermissionError as e:
print(e)
五、三种规模的正确架构选型
• <100 SKU监控：单Key + 令牌桶8 QPS + 基础stock_num本地缓存10s → 零费够用。
• 100~1000 SKU代发防超卖：买基础提频+实时库存包（¥980~2980/年），单Key 50 QPS查高级接口，Redis缓存SKU可售5~10s。
- 多店ISV/大卖：3~5个同企业AppKey轮询 + 高级包 + 1688消息订阅（库存变更Webhook） 替代主动轮询，把拉模式切推模式，QPS压力归零。
六、两条必踩坑提醒
• “详情里有stock就不用买包” → 那是基础stock_num，有缓存、不含占用/锁定，大促代发锁库存会超卖，不是省钱是赔钱。
• “多Key堆个人号绕限流” → 必须同企业实名不同应用，个人号堆砌触发风控直接封应用，不是限流是封号。
一句话：1688基础API免费是真的，但QPS 10是硬墙；破墙的正路是“令牌桶自律 + 同企业多Key + 买包提频/开高级库存”，灰产式并发伪装会反手封你应用。做批发同步，防超卖的隐性成本不在调用费，在资源包年费和你敢不敢用基础库存糊弄。
要不要我把上面 Ali1688GuardedClient 改成Redis中心化令牌桶（支持多进程分布式）+ 1688 Webhook库存推送消费端，让“轮询兜底+推送为主”直接可上大促？
