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

简介: eBay开发者API免费调用,但严格限流:默认5000次/天/API(PST重置)+ ReviseInventoryStatus 15秒内≤6000次硬限制。多站点共用网关,配额按AppKey+API计,不按国家拆分;OAuth分应用级(7200s)与用户级(refresh最长18个月),需分层缓存。核心成本是配额调度,非账单费用。(239字)

🌐
结论先拍: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)九家调度中台?

相关文章
|
人工智能 Java
零基础五步骤,从零开始天猫精灵
零基础五步骤,从零开始天猫精灵
1780 1
零基础五步骤,从零开始天猫精灵
|
域名解析 网络协议 安全
DNS&BIND9安装配置
DNS概念,BIND9的详细配置安装
2507 2
|
7天前
|
存储 弹性计算 Linux
新手小白如何购买阿里云服务器?2026新版阿里云服务器购买实操教程:从账号准备、参数选型到实例初始化完整指南
很多刚刚接触云服务器的新手,打开ECS购买页面,面对付费类型、地域、实例规格、镜像、存储、安全组等大量配置选项,很容易陷入迷茫。一旦参数选择失误,不仅会造成资金浪费,还会出现网站访问卡顿、外部无法连通服务器、系统存在安全漏洞等一系列问题。本文结合2026年最新的购买页面,完整梳理前期准备工作、三类购买入口的适用人群、每一项核心参数的选择逻辑、下单之后初始化操作以及新手高频踩坑点,零基础用户跟随步骤就可以完成服务器选购与基础配置。本文以ECS云服务器作为讲解对象,同时会区分轻量应用服务器与ECS的适用场景,文中附带可直接复制执行的系统命令,帮助新手完成登录、系统检查、安全防护等实操工作。
140 2
|
1月前
|
人工智能 数据库 Android开发
原生 APP 开发的开发费用
原生APP开发(iOS/Swift+Android/Kotlin)需双端独立编码,成本比跨平台方案高30%~50%。费用取决于功能复杂度(MVP 3万起、企业级100万+)、团队模式及后端支撑,人力占比70%~80%,含双端工程师叠加投入。
|
28天前
|
缓存 NoSQL 关系型数据库
⚡《SP-API收费应对:事件驱动替代轮询,GET调用量砍掉90%的实战》(附Python源码)
SP-API虽取消年费与超量收费,但Basic档250万GET/月仍是硬上限。本文提出“事件驱动+按需拉取+5分钟增量兜底”架构,实测100卖家月调用量从345万锐减至9万,削减97.4%,彻底规避限流、提升大促稳定性,并为未来潜在计费预留最优成本结构。(239字)
|
1月前
|
JavaScript 前端开发 Java
SpringBoot+Vue 医院随访系统完整方案,随访系统源码分享
本系统是基于SpringBoot+Vue的企业级医院随访解决方案,涵盖患者管理、智能随访计划、多渠道执行、数据统计及RBAC权限管理五大模块,集成Redis、RabbitMQ、Quartz等中间件,支持高并发、可落地部署,实现随访数字化、标准化、自动化。
|
1月前
|
JSON 文字识别 API
免费 PDF 文本提取推荐
本文整理2026年实测可用的免费PDF文本提取方案:涵盖万维易源(注册即用)、OCR.space(月2.5万次)、LlamaParse(复杂版面)、Doc2X(公式/多栏)等在线API,以及pdfminer.six、PyMuPDF、pdftotext等本地开源库,兼顾隐私、效率与场景适配。
516 0
|
24天前
|
存储 监控 API
二手 ERP 系统对接闲鱼 API:落地方案、接口边界与业务踩坑
二手ERP对接闲鱼需分清两大接口:闲管家开放平台(仅管理自有店铺,支持商品/订单/物流双向操作)与第三方数据接口(仅读取全网商品,用于选品估价)。二者职责分明,混用将导致业务故障。务必通过正规渠道接入,避免逆向抓包。
167 0
|
28天前
|
弹性计算 安全 Linux
阿里云服务器免费领取保姆级教程,新手零成本上云完整实操
对于刚接触云计算的个人开发者、在校学生以及小型技术爱好者来说,直接付费购买云服务器存在一定试错成本。阿里云面向新用户群体开放免费试用权益,能够零成本体验ECS云服务器与轻量应用服务器,用来做代码调试、个人项目测试、小型网站搭建、学习Linux命令等场景,是入门云计算非常理想的途径。但很多新手在领取过程中,常常会遇到资格不通过、开通之后意外扣费、服务器无法远程连接、试用期到期忘记释放资源等各类问题。本文完整梳理从账号准备、资格校验、实例申领、远程登录、命令行实操、安全配置,再到到期资源释放全流程,同时整理大量高频踩坑点,附带可直接复制执行的命令,帮助大家安全稳妥拿到免费服务器资源,规避不必要的账
474 0