结论先拍：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“免费”的准确边界

官方不承诺固定日总量，但单Key持续打满10 QPS 会被风控审查；实测中小自研（日几千~几万次）零费跑一年没问题。

二、QPS 10 在批发同步里的真实瓶颈

• 爆款30s轮询一次 → 需 500/30 ≈ 16.7 次/s，单Key 10 QPS直接限流

• 若每次拉 alibaba.item.get 拿 sku_list.stock_num（基础库存，非强一致，不含占用）→ 免费但占QPS，越拉越堵

• 改高级实时库存接口 → 基础免费额度不覆盖，没买包连调用权限都没有

三、破QPS 10 的三层正规解法（不碰灰产）

1. 客户端自律：单Key令牌桶压到 8~9 QPS，留20%余量防突发被封。

2. 同企业多AppKey轮询：同营业执照下建3~5个“自用型应用”，各10 QPS，网关前做Key轮转（禁止同Key并发伪装、禁止个人号堆砌，否则封应用）。

3. 买资源包：基础提频包把单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(safe_qps) 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 + 买包提频/开高级库存”，灰产式并发伪装会反手封你应用。做批发同步，防超卖的隐性成本不在调用费，在资源包年费和你敢不敢用基础库存糊弄。