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

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

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

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

# 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做商品主数据同步 + 订单自动回写，不用花钱。