×

《eBay API跨境实战:美/英/德/澳多站点令牌管理与限流策略》(附Python源码)

万邦科技Lex 万邦科技Lex 发表于2026-08-13 13:47:33 浏览26 评论0

抢沙发发表评论

结论先拍:eBay开发者计划API调用本身$0,不按量扣费,和淘宝/拼多多/抖店基因完全不同——它的成本是配额管理成本而非账单成本。 默认每AppKey每API 5000次/天(PST午夜重置),短窗口还有 ReviseInventoryStatus 任意15秒窗口≤6000次的硬速率墙,超了不是扣钱是Block 15秒。 多站点(US/GB/DE/AU)共用网关 api.ebay.com,靠 X-EBAY-C-MARKETPLACE-ID 头切换,配额按AppKey+API名计,不按站点拆;OAuth分应用级(client_credentials,7200s)和用户级(authorization_code,refresh最长18个月),必须分层缓存。


一、2026现行配额与限流参数(官方锚点)

维度
规则
来源
开发者订阅费
$0

默认日配额
5000次/天/AppKey/每API(Browse/Sell/Commerce REST同此)

日重置点
太平洋时间午夜(PST)

短时速率墙
ReviseInventoryStatus 任意15s窗口 ≤6000次,超则Block 15s

提额路径
Application Growth Check(原Compatible App Check)→ 10万/天 → 再高交Developer Questionnaire

OAuth应用令牌
client_credentials换,有效期 7200秒,按AppKey缓存

OAuth用户令牌
authorization_code换,access 2h,refresh最长18个月

429处理
Retry-After,指数退避+jitter

站点路由头
X-EBAY-C-MARKETPLANCE-ID: EBAY_US/EBAY_GB/EBAY_DE/EBAY_AU

关键认知:eBay的“限流”是配额墙不是钱墙。你不会因超调被扣$0.01,但会被429/Block拖垮大促同步;多国调度核心是把“5000/天/API”当硬预算切分,而不是担心账单。

二、美/英/德/澳四站路由模型

  • 网关统一:https://api.ebay.com(生产) / https://api.sandbox.ebay.com(沙箱)

  • 站点→Marketplace ID:

    • US → EBAY_US(siteId 0)

    • GB → EBAY_GB(siteId 3)

    • DE → EBAY_DE(siteId 77)

    • AU → EBAY_AU(siteId 15)

  • 配额不按站点独立:同一AppKey调Browse API,US 2000 + GB 2000 + DE 1000 = 5000/天命中上限,第四站再调就429。

  • 旧Trading API用 siteId 字段,REST用 X-EBAY-C-MARKETPLACE-ID 头,两者别混。


三、令牌管理分层(应用级+用户级分离)

AppKey(CID+SEC)
 ├─ 应用令牌 app_token(client_credentials,7200s,按AppKey缓存1份)
 └─ 用户令牌池 {seller1: (access, refresh, exp), seller2: ...}(每商家授权独立)
  • 应用令牌用于Browse/公开读,不占用户授权。

  • 库存/订单/发货(Sell-Inventory/Fulfillment)必须用户级token,refresh前主动换。

  • 多容器部署:应用令牌用Redis集中缓存,用户令牌按seller_id存PG/Redis,别每次call都换。


四、Python:EbayMultiSiteGuardClient(四站路由+双令牌桶+15s/天双维限流+Growth预警)

# ebay_multisite_guard_client.py
"""
eBay 2026 多站点(US/GB/DE/AU)守卫客户端
- 单网关 + MARKETPLACE头路由
- 双令牌桶:日桶(5000/API/AppKey PST重置) + 15s短窗桶(ReviseInventoryStatus≤6000)
- 应用级/用户级 OAuth 分层缓存
- 429/Retry-After/15s-Block 退避
"""
import time, hashlib, requests, json
from typing import Dict, Optional
from datetime import datetime, timezone, timedelta
from threading import Lock

GW_PROD = "https://api.ebay.com"
SITE_HEADER = {
    "US": "EBAY_US", "GB": "EBAY_GB", "DE": "EBAY_DE", "AU": "EBAY_AU",
}
SITE_ID = {"US": 0, "GB": 3, "DE": 77, "AU": 15}

class _Bucket:
    def __init__(self, day_limit, short_window_limit, short_window_sec=15):
        self.day_limit = day_limit
        self.short_limit = short_window_limit
        self.short_sec = short_window_sec
        self.day_used = 0
        self.short_used = 0
        self.short_ts = time.monotonic()
        self.day_reset = self._next_pst_midnight()
        self.lk = Lock()
    def _next_pst_midnight(self):
        # 简化:用UTC-8估算PST午夜,生产应接pytz
        now = datetime.now(timezone.utc)
        pst = now.astimezone(timezone(timedelta(hours=-8)))
        nxt = pst.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
        return nxt.timestamp()
    def _tick(self):
        now = time.monotonic()
        if now - self.short_ts >= self.short_sec:
            self.short_used = 0
            self.short_ts = now
        if time.time() >= self.day_reset:
            self.day_used = 0
            self.day_reset = self._next_pst_midnight()
    def acquire(self, short_window=False):
        with self.lk:
            self._tick()
            if short_window and self.short_used >= self.short_limit:
                sleep_for = self.short_sec - (time.monotonic()-self.short_ts) + 0.05
                return max(0.05, sleep_for)
            if self.day_used >= self.day_limit:
                return max(0.05, self.day_reset - time.time())  # 等PST重置
            self.day_used += 1
            if short_window:
                self.short_used += 1
            return 0.0

class EbayMultiSiteGuardClient:
    def __init__(self, client_id, client_secret,
                 day_limit=5000, short_limit=6000, short_sec=15):
        self.cid = client_id
        self.csec = client_secret
        # 每API名一个日桶;ReviseInventoryStatus额外挂短窗桶
        self.day_buckets: Dict[str, _Bucket] = {}
        self.short_bucket = _Bucket(day_limit, short_limit, short_sec)
        self.app_token: Optional[tuple] = None  # (token, exp)
        self.user_tokens: Dict[str, tuple] = {}  # seller_id -> (access, refresh, exp)
        self._lk = Lock()

    # ---- 令牌 ----
    def _app_token(self):
        if self.app_token:
            tk, exp = self.app_token
            if time.time() < exp - 300:
                return tk
        r = requests.post("https://api.ebay.com/identity/v1/oauth2/token",
                         data={"grant_type": "client_credentials",
                               "scope": "https://api.ebay.com/oauth/api_scope"},
                         auth=(self.cid, self.csec), timeout=10)
        d = r.json()
        tk = d["access_token"]
        self.app_token = (tk, time.time() + d.get("expires_in", 7200))
        return tk

    def _user_token(self, seller_id, refresh_token=None):
        if seller_id in self.user_tokens:
            acc, ref, exp = self.user_tokens[seller_id]
            if time.time() < exp - 300:
                return acc
            if ref:
                return self._refresh_user(seller_id, ref)
        if refresh_token:
            return self._refresh_user(seller_id, refresh_token)
        raise RuntimeError(f"无 {seller_id} 用户令牌,需走OAuth授权码流程")

    def _refresh_user(self, seller_id, refresh_token):
        r = requests.post("https://api.ebay.com/identity/v1/oauth2/token",
                         data={"grant_type": "refresh_token",
                               "refresh_token": refresh_token,
                               "scope": "https://api.ebay.com/oauth/api_scope/sell.fulfillment"},
                         auth=(self.cid, self.csec), timeout=10)
        d = r.json()
        self.user_tokens[seller_id] = (d["access_token"], refresh_token,
                                      time.time() + d.get("expires_in", 7200))
        return d["access_token"]

    # ---- 站点路由 ----
    def _bucket_for(self, api_name):
        if api_name not in self.day_buckets:
            self.day_buckets[api_name] = _Bucket(5000, 6000, 15)
        return self.day_buckets[api_name]

    def safe_call(self, site, api_name, path, params=None, *,
                  seller_id=None, is_revise=False, max_retry=4):
        if site not in SITE_HEADER:
            raise KeyError(f"未支持站点 {site}")
        # 日桶守卫
        b = self._bucket_for(api_name)
        wait = b.acquire(short_window=False)
        if wait > 0:
            raise RuntimeError(f"⏸ {api_name} 日配额5000耗尽,PST重置需等{wait:.0f}s")
        # 15s短窗(仅ReviseInventoryStatus类)
        if is_revise:
            w2 = self.short_bucket.acquire(short_window=True)
            if w2 > 0:
                time.sleep(w2)
        # 选令牌
        token = self._user_token(seller_id) if seller_id else self._app_token()
        url = GW_PROD + path
        headers = {
            "Authorization": f"Bearer {token}",
            "X-EBAY-C-MARKETPLACE-ID": SITE_HEADER[site],
            "Content-Type": "application/json",
        }
        for att in range(max_retry):
            r = requests.get(url, params=params, headers=headers, timeout=15)
            if r.status_code == 429:
                ra = int(r.headers.get("Retry-After", 2 ** att))
                time.sleep(ra + 0.1); continue
            if r.status_code == 200:
                return r.json()
            d = r.json()
            if "errors" in d:
                code = d["errors"][0].get("errorId", "")
                msg = d["errors"][0].get("message", "")
                if "exceeded" in msg.lower() or "call limit" in msg.lower():
                    time.sleep(2 ** att); continue
                raise Exception(f"eBay {site}/{api_name}: {msg}")
            return d
        raise RuntimeError("eBay retry exhausted")

    # 业务方法
    def browse_search(self, site, q, limit=50):
        return self.safe_call(site, "browse",
                              "/buy/browse/v1/item_summary/search",
                              {"q": q, "limit": limit})

    def get_orders(self, seller_id, site, create_after):
        return self.safe_call(site, "fulfillment",
                              "/sell/fulfillment/v1/order",
                              {"create_date_from": create_after, "limit": 50},
                              seller_id=seller_id)

    def revise_inventory(self, seller_id, site, sku, qty):
        # 演示:Trading旧版走XML,这里用Sell-Inventory REST示意
        return self.safe_call(site, "sell-inventory",
                              f"/sell/inventory/v1/inventory_item/{sku}",
                              {"sku": sku, "qty": qty},
                              seller_id=seller_id, is_revise=True)

if __name__ == "__main__":
    cli = EbayMultiSiteGuardClient("CID", "CSEC")
    # 四站Browse各跑(共耗日桶browse 4次)
    for s in ("US", "GB", "DE", "AU"):
        try:
            r = cli.browse_search(s, "drone", 3)
            print(s, "OK total=", r.get("total", "?"),
                  "日桶browse剩余≈", 5000 - cli._bucket_for("browse").day_used)
        except RuntimeError as e:
            print(s, e)
    # 模拟Revise短窗
    try:
        cli.revise_inventory("seller_1", "US", "SKU_A", 10)
        print("Revise OK,短窗已用", cli.short_bucket.short_used)
    except Exception as e:
        print("Revise ERR", e)

五、四站调度的三条铁律

  1. 配额按API名切,不按站点切:Browse 5000/天是US+GB+DE+AU共享,做四站ERP时每站预留1250次/天或按店铺GMV动态分配,别假设“每站5000”。

  2. Revise类走短窗桶ReviseInventoryStatus 任意15s≤6000是硬墙,多SKU批量改库存用“单call多SKU(最多4个)+ 短窗令牌桶”双控,大促改价别循环单SKU调。

  3. OAuth分层缓存:应用令牌按AppKey缓存7200s(边缘/Redis一份),用户令牌按seller_id落库,refresh前300s主动换;多容器别各自换令牌——会互相挤占且浪费调用(换token不占5000但占连接与限流信任分)。


六、和国内五家+亚马逊对照(CTO记账)

平台
调用费
限额模型
多国/多站成本焦点
淘宝TOP
0.02/百次(塔内)
免额+按量
入塔+ECS
拼多多
0.01/百次
预充值硬断
余额守卫
抖店
0.018/百次
预充值
入云
1688
免费
QPS10
令牌桶
亚马逊SP-API
$0
GET超量(原拟已撤)
端点路由+PII30天
eBay
$0
5000/天/API+15s短窗
配额切分+双令牌+OAuth分层
eBay的隐式成本不在网关账单,在“5000次怎么分给US/GB/DE/AU四站”+“Revise短窗别Block”+“用户令牌别过期”。架构做对,四站单AppKey日调2万次内零元跑稳;架构做错,大促改价被Block 15秒×N次=肉眼可见的超卖。
要不要我把上面 EbayMultiSiteGuardClient 改成 Redis中心化日桶/短窗桶(多容器共享)+ 用户令牌PG落库表结构 + 每API配额水位企微告警,直接并排进你前面那套(淘宝/京东/1688/拼多多/抖店/苏宁/微店/亚马逊SP-API)九家调度中台?


群贤毕至

访客