🚦 淘宝TOP API调用限流（QPS/日上限）与429错误规避方案（附Python源码）

• QPS限制（每秒查询率）—— 超了返回 code=7 / ISP_FLOW_CONTROL_LIMIT或 HTTP 429

• 日调用量限制（按自然日）—— 超了返回 accesscontrol.limited-by-app-access-count

一、TOP限流典型数值（企业应用参考）

⚠️ 个人开发者应用QPS更低（通常≤2/s），且部分订单接口无权限。

二、规避429的核心策略

① 客户端令牌桶限速（QPS ≤ 免费上限×0.8）
   ↓
② 捕获限流异常(code=7 / 429)
   ↓
③ 指数退避重试（1s → 2s → 4s，最多3次）
   ↓
④ 日额度耗尽 → 暂停翻页/记录断点次日继续

三、Python：令牌桶 + TOP调用 + 限流重试

# top_rate_limit_client.py
"""
淘宝TOP API Client with:
 - Token Bucket QPS throttle
 - auto-retry on FLOW_CONTROL / 429
 - daily-quota exhausted warning
依赖: top_api_client.TaobaoTopClient (签名+POST封装)
"""
import time
import requests
from typing import Dict, Optional
from top_api_client import TaobaoTopClient
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

class _TokenBucket:
    def __init__(self, rate: float = 4.0, capacity: int = None):
        """
        rate:   允许QPS（建议设低于免费上限，如免费5/s则设4）
        capacity: 桶容量，默认=rate
        """
        self.rate = rate
        self.cap = capacity or int(rate)
        self.tokens = float(self.cap)
        self.ts = time.monotonic()

    def wait(self):
        now = time.monotonic()
        self.tokens = min(self.cap, self.tokens + (now - self.ts) * self.rate)
        self.ts = now
        if self.tokens >= 1:
            self.tokens -= 1
            return
        need = (1 - self.tokens) / self.rate
        time.sleep(need + 0.005)
        self.tokens = 0


class TopThrottledClient(TaobaoTopClient):
    """带QPS限速 + 限流重试的TOP Client"""

    def __init__(self, app_key, app_secret, sandbox=False, qps=4.0,
                 access_token=None):
        super().__init__(app_key, app_secret, sandbox=sandbox)
        self.bucket = _TokenBucket(rate=qps)
        self._access_token = access_token

    def safe_call(self, method: str, biz: Dict,
                   session: str = None, max_retry=3) -> Dict:
        """
        带令牌桶 + 限流自动退避调用
        """
        session = session or self._access_token
        for attempt in range(max_retry):
            self.bucket.wait()   # ← QPS控制点

            try:
                return self.call(method, biz_params=biz, session=session)

            except Exception as e:
                err_str = str(e)
                # ---------- 判断是否为限流 ----------
                is_flow = any(k in err_str for k in [
                    "ISP_FLOW_CONTROL", "FLOW_CONTROL_LIMIT",
                    "code=7", "429"
                ])
                # 日额度耗尽
                if "limited-by-app-access-count" in err_str:
                    raise DailyQuotaExhausted(
                        "TOP日调用额度已耗尽！请明天再跑或申请提额"
                    ) from e

                if is_flow and attempt < max_retry - 1:
                    backoff = 2 ** attempt
                    print(f"⚠️  TOP限流(QPS)，{backoff}s后第{attempt+1}次重试...")
                    time.sleep(backoff)
                    continue

                raise  # 非限流异常直接抛出

        raise RuntimeError("TOP call failed after max retries")


# ── 自定义异常 ──
class DailyQuotaExhausted(Exception):
    pass

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# =========================================================
# 使用示例：同步在售商品（自动避让QPS+限流）
# =========================================================
if __name__ == "__main__":
    client = TopThrottledClient(
        app_key="YOUR_TOP_APP_KEY",
        app_secret="YOUR_TOP_APP_SECRET",
        sandbox=True,
        qps=4,            # ← 免费上限5/s时设4，保守
        access_token="SELLER_ACCESS_TOKEN"
    )

    try:
        # 示例：翻页拉商品（遇429自动重试）
        result = client.safe_call(
            "taobao.items.onsale.get",
            biz={
                "page_no": 1,
                "page_size": 100,
                "fields": "num_iid,title,price,num,outer_id,approve_status"
            }
        )
        items = result.get("items", []) or []
        total = result.get("total_results", 0)
        print(f"✅ 在售商品共 {total} 件，本页 {len(items)} 条")

    except DailyQuotaExhausted as e:
        print(f"🚨 {e}")
        print("→ 建议：记录当前page_no，定时任务明早从此断点继续")
    except Exception as e:
        print("❌", e)

四、翻页同步时的日额度保护

# 在翻页循环中加入日额度保护
try:
    for pg in range(1, max_page+1):
        r = client.safe_call("taobao.items.onsale.get", {...})
        # 处理r ...
        if len(r.get("items",[])) < page_size:
            break
except DailyQuotaExhausted:
    print("日额度耗尽，保存 pg 作为 resume_page 明天继续")
    raise

五、高频避坑清单

六、面试/方案一句话

淘宝TOP API限流分QPS和日调用量两层；客户端用令牌桶限速（QPS≤免费上限×0.8），遇 code=7/ISP_FLOW_CONTROL_LIMIT指数退避重试最多3次，日额度耗尽捕获后断点续跑次日继续，避免触发计费或封禁。