企业如何免费调用1688基础API？每日限额与QPS限制详解（附Python源码）

1688开放平台的基础API（商品搜索、详情、订单、物流查询）对企业开发者完全免费，只要完成企业实名认证并创建「自用型应用」，即可直接调用。限制主要体现在QPS（每秒查询率）和单日调用总量，理解这两个阈值才能做好系统调度设计。

一、 免费额度的准确数字（2026版参考）

✅结论：中小企业做商品同步（每日几千~几万次）+ 订单回写，免费额度完全够用。

二、 受控调用的Python封装（含QPS限速 + 限流重试）

下面代码在之前API客户端基础上加入令牌桶限速和限流自动退避，确保不触发1688流控：

# ali1688_free_api.py import hashlib import time import requests import urllib.parse from typing import Dict, List, Optional from datetime import datetime # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex # ───────────────────────────────────────────── # 简易令牌桶（控制QPS不超过免费限额） # ───────────────────────────────────────────── class TokenBucket: def __init__(self, rate: float, capacity: int = None): self.rate = rate # 令牌生成速率 = 允许QPS self.capacity = capacity or int(rate) self.tokens = float(self.capacity) self.last_time = time.monotonic() def consume(self, n: int = 1): now = time.monotonic() elapsed = now - self.last_time self.tokens = min(self.capacity, self.tokens + elapsed * self.rate) self.last_time = now if self.tokens >= n: self.tokens -= n return True else: # 计算需等待时间 wait = (n - self.tokens) / self.rate time.sleep(wait + 0.01) self.tokens = 0 return True class Ali1688FreeClient: """ 1688 基础免费API客户端 支持：商品搜索(alibaba.offer.search)、商品详情(alibaba.item.get) 订单列表(alibaba.trade.buyer.list)、订单详情(alibaba.trade.get) """ GATEWAY_SEARCH = "https://gw.open.1688.com/openapi/param2/2/alibaba.offer.search/2.0" GATEWAY_COMMON = "https://gw.open.1688.com/openapi/http/2/1" def __init__(self, app_key: str, app_secret: str, access_token: str = None, qps: int = 8): self.app_key = app_key self.app_secret = app_secret self.access_token = access_token self.bucket = TokenBucket(rate=qps) # 默认按8 QPS限速（留余量） # ───────────── 签名 ───────────── def _sign(self, params: Dict) -> str: filtered = sorted((k, v) for k, v in params.items() if v is not None) qs = ''.join(f"{k}{v}" for k, v in filtered) raw = f"{self.app_secret}{qs}{self.app_secret}" return hashlib.md5(raw.encode('utf-8')).hexdigest().upper() def _call_get(self, url: str, method: str, biz: Dict) -> Dict: self.bucket.consume() # ← QPS控制点 api_params = { "method": method, "app_key": self.app_key, "timestamp": str(int(time.time() * 1000)), "format": "json", "v": "2.0", "sign_method": "md5", } if self.access_token: api_params["session"] = self.access_token api_params["param2" if "param2" in url else "param"] = \ urllib.parse.quote_plus(str(biz).replace("'", '"')) api_params["sign"] = self._sign(api_params) for attempt in range(3): resp = requests.get(url, params=api_params, timeout=15) resp.raise_for_status() data = resp.json() # 限流检测 → 指数退避重试 if "error_response" in data: err_code = str(data["error_response"].get("code", "")) if "FLOW_CONTROL" in err_code or err_code == "429": wait = 2 ** attempt print(f"⚠️ 触发限流，{wait}s后重试(第{attempt+1}次)...") time.sleep(wait) continue else: err = data["error_response"] raise Exception(f"1688 Err[{err.get('code')}]: {err.get('msg')}") # 提取结果 result_key = [k for k in data if k != "error_response"] if result_key: return data[result_key[0]] return data raise Exception("API调用失败：持续被限流，请降低QPS或购买资源包") # ───────────── 商品搜索 ───────────── def search_products(self, keyword: str, page_no: int = 1, page_size: int = 40, price_min=None, price_max=None) -> Dict: biz = { "keywords": keyword, "pageNo": page_no, "pageSize": min(page_size, 50), "sortType": "booked" } if price_min is not None: biz["beginPrice"] = str(int(price_min * 100)) if price_max is not None: biz["endPrice"] = str(int(price_max * 100)) return self._call_get(self.GATEWAY_SEARCH, "alibaba.offer.search", biz) # ───────────── 商品详情 ───────────── def get_product_detail(self, offer_id: str, fields: str = None) -> Dict: biz = {"item_id": offer_id} if fields: biz["fields"] = fields return self._call_get( self.GATEWAY_COMMON, "alibaba.item.get", biz ).get("alibaba_item_get_response", {}).get("item", {}) # ───────────── 订单列表（简易封装）──────────── def list_orders(self, status: str = "waitsellersend", hours_back: int = 48, page_no: int = 1) -> List[Dict]: biz = { "orderStatus": status, "gmtCreateStart": (datetime.now() .replace(microsecond=0).__class__.__sub__(datetime.now(), hours=hours_back) .strftime("%Y-%m-%d %H:%M:%S").replace("__sub__", "-")), "gmtCreateEnd": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "pageNo": page_no, "pageSize": 50 } # 简化：实际建议用之前专用订单client from datetime import timedelta biz["gmtCreateStart"] = (datetime.now() - timedelta(hours=hours_back)).strftime("%Y-%m-%d %H:%M:%S") res = self._call_get(self.GATEWAY_COMMON, "alibaba.trade.buyer.list", biz) return res.get("alibaba_trade_buyer_list_response", {}).get("tradeModelList", []) or [] # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex # ========================================================= # 使用示例 # ========================================================= if __name__ == "__main__": client = Ali1688FreeClient( app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET", # access_token=None # 商品搜索可不传；订单查询需传session key qps=8 # 留余量，低于免费上限10 ) try: # 搜索 result = client.search_products("不锈钢保温杯", page_no=1, price_min=15, price_max=60) offers = result.get("offers", []) total = result.get("totalResult", 0) print(f"✅ 找到 {total} 条，本页 {len(offers)} 条") if offers: offer_id = offers[0].get("offerId") # 详情 detail = client.get_product_detail(str(offer_id)) print(f"商品: {detail.get('title')} | 起批价: {detail.get('price')}") except Exception as e: print(f"❌ {e}")

三、 免费调用四大注意事项（避坑）

1. QPS必须主动限速：免费应用无弹性，超频立刻返回ISP_FLOW_CONTROL_LIMIT。上述TokenBucket按8 QPS限速是安全值。

2. Access Token区分场景：

   • 商品搜索/详情 →可不传（公开数据）

   • 订单/物流/购物车 →必须传（买家登录态 session key）

3. 企业认证是前提：个人开发者应用无法调用订单类接口，且部分商品字段被裁剪。

4. 全量同步错峰：每日全量商品同步放凌晨2~4点，分页步进sleep≥0.2s，避免短时突发超QPS。

四、 什么时候需要买资源包？

绝大多数中小ERP做商品主数据同步 + 订单自动回写，不用花钱。

如需我补充OAuth2获取Access Token完整代码或每日增量同步定时任务(Crontab/APScheduler)，直接说 👍