《亚马逊 SP-API：落地方案、接口边界与业务踩坑 —— “年费 $1400+超量$0.40/千次” 的真相与 2026 反转》（附 Python 源码）

0. 先纠正一个时间线（很重要）

• 2026-01：延期

• 2026-03：无限期推迟

• 2026-05-12：Amazon 正式邮件宣布 “will not move forward with the SP-API usage and annual fees at this time”

把 $1400/$0.40 当“成本模型教材”和“未来可能再来一次的防御设计”，但不要把生产架构建立在这个账单一定存在上。

Amazon 原话留了 at this time，以后换种形态收费完全可能。

一、SP-API 和前面几家根本不是一类东西

• Auth：Login with Amazon（LWA）→ access_token（约 1h）+ refresh_token（长期）

• Selling Partner API：orders / catalog / listings / inventory / feeds / reports / notifications / finances / shipments

• EventBridge / SQS 通知：ORDER_CHANGE、REPORT_PROCESSING_FINISHED 等

• Reports API：建报告 → 轮询状态 → 下载文档（大批量正道）

• Dynamic Usage Plan（DUP）：你看到的文档限流是最低档，Amazon 会根据历史自动调

二、订单接口边界（最容易被国内团队写错）

每 5 秒调一次 getOrders 查新单 → 10 秒内把 burst 打光 → 后面全 429

日单 < 2 万：getOrders + getOrderItems 可行

大单量：ORDER_CHANGE 通知 + Orders Report

LastUpdatedAfter 查到的时间是 “当前时间 - 2 分钟”，新单有 2 分钟延迟。

三、生产架构：通知优先，Reports 批量，REST 兜底

EventBridge / SQS
  └─ ORDER_CHANGE (orderId, changeType)
        └─ 入队 (amazonOrderId)

Worker:
  getOrder(orderId)         # 订单头
  getOrderItems(orderId)    # 订单行
  → 写 order 快照

每 30~60 min：
  Orders Report (GET_FLAT_FILE_ORDER_REPORT)
    → 批量对账，补漏

财务/库存：
  GET_FBA_MYI_UNSUPPRESSED_INVENTORY_DATA
  GET_LEDGER_SUMMARY
  → Reports API，不逐 SKU 调 getInventorySummaries

通知做触发器，Report 做批量真相，getOrder 做单条补全，轮询 getOrders 只做兜底。*

四、Python：LWA 令牌管理（refresh 不互踢）

# spapi/auth.py
import time
import requests


class LwaTokenProvider:
    def __init__(self, client_id, client_secret, refresh_token,
                 redis=None, ttl_margin=300):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.redis = redis
        self.margin = ttl_margin
        self._cache = {}

    def _redis_key(self):
        return f"spapi:lwa:access:{self.client_id}"

    def get_access_token(self) -> str:
        if self.redis:
            tok = self.redis.get(self._redis_key())
            if tok:
                return tok

        resp = requests.post("https://api.amazon.com/auth/o2/token", data={
            "grant_type": "refresh_token",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "refresh_token": self.refresh_token,
        }, timeout=5)
        resp.raise_for_status()
        data = resp.json()
        token = data["access_token"]
        expires_in = int(data.get("expires_in", 3600))

        if self.redis:
            self.redis.set(self._redis_key(), token, ex=expires_in - self.margin)
        return token

def refresh_with_lock(provider, r):
    lock = r.lock("spapi:lwa:refresh:lock", timeout=10)
    if lock.acquire(blocking=False):
        try:
            r.delete(provider._redis_key())
            provider.get_access_token()
        finally:
            lock.release()

五、Python：限流感知的 SP-API 客户端

# spapi/client.py
import time
import requests


class SpApiThrottleError(Exception):
    def __init__(self, retry_after):
        super().__init__("throttled")
        self.retry_after = retry_after


class SpApiClient:
    def __init__(self, token_provider, region_endpoint,
                 rate_per_sec=0.0167, burst=20):
        self.tp = token_provider
        self.base = region_endpoint   # NA/EU/FE 不同
        self.rate = rate_per_sec
        self.burst = burst
        self._tokens = burst
        self._last = time.monotonic()

    def _acquire(self):
        while True:
            now = time.monotonic()
            self._tokens = min(
                self.burst,
                self._tokens + (now - self._last) * self.rate
            )
            self._last = now
            if self._tokens >= 1:
                self._tokens -= 1
                return
            time.sleep(0.2)

    def _request(self, method, path, params=None, retries=5):
        for i in range(retries):
            self._acquire()
            token = self.tp.get_access_token()
            headers = {
                "x-amz-access-token": token,
                "Accept": "application/json",
            }
            url = f"{self.base}{path}"
            resp = requests.request(method, url, headers=headers,
                                    params=params, timeout=10)
            if resp.status_code == 429:
                ra = int(resp.headers.get("Retry-After", 60))
                time.sleep(ra + (i * 2))
                continue
            if resp.status_code >= 500:
                time.sleep(2 ** i)
                continue
            resp.raise_for_status()
            return resp.json()
        raise RuntimeError("sp-api retry exhausted")

    def get_orders(self, last_updated_after, next_token=None):
        p = {"MarketplaceIds": "...", "LastUpdatedAfter": last_updated_after}
        if next_token:
            p["NextToken"] = next_token
        return self._request("GET", "/orders/v0/orders", p)

    def get_order_items(self, amazon_order_id):
        return self._request("GET", f"/orders/v0/orders/{amazon_order_id}/orderItems")

六、订单同步：通知 + Report + getOrders 兜底

# spapi/order_sync.py
from datetime import datetime, timedelta


class SpOrderSync:
    def __init__(self, client, repo, dedup):
        self.client = client
        self.repo = repo
        self.dedup = dedup

    def on_order_change(self, amazon_order_id: str):
        # 通知只给 id，不信任通知体里的状态
        self.dedup.mark(f"order_change:{amazon_order_id}")
        self.fill_order(amazon_order_id)

    def fill_order(self, amazon_order_id):
        head = self.client._request("GET", f"/orders/v0/orders/{amazon_order_id}")
        items = self.client.get_order_items(amazon_order_id)
        snap = self.normalize(head, items)
        self.apply(snap)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def poll_compensation(self, marketplace_ids, now: datetime):
        # 结束时间留 2 分钟缓冲（Amazon 订单有 2 分钟延迟）
        end = now - timedelta(minutes=2)
        start = end - timedelta(minutes=30)
        next_token = None
        while True:
            resp = self.client.get_orders(
                last_updated_after=start.isoformat(),
                next_token=next_token,
            )
            for o in resp["payload"]["Orders"]:
                self.on_order_change(o["AmazonOrderId"])
            next_token = resp["payload"].get("NextToken")
            if not next_token:
                break
            # getOrders 默认 1/min，别在循环里硬跑：
            time.sleep(60)

七、Reports 模式：大单量别用 getOrders

# spapi/reports.py
class SpReportClient:
    def __init__(self, client):
        self.client = client

    def pull_orders_report(self, marketplace_ids, start, end):
        create = self.client._request("POST", "/reports/2021-06-30/reports", {
            "reportType": "GET_FLAT_FILE_ORDER_REPORT",
            "marketplaceIds": marketplace_ids,
            "dataStartTime": start.isoformat(),
            "dataEndTime": end.isoformat(),
        })
        report_id = create["reportId"]

        # 轮询状态：别狂轮，指数退避
        for _ in range(30):
            status = self.client._request(
                "GET", f"/reports/2021-06-30/reports/{report_id}"
            )["payload"]
            if status["processingStatus"] == "DONE":
                return self.download(status["reportDocumentId"])
            time.sleep(min(60, 5 * (2 ** _)))
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def download(self, doc_id):
        doc = self.client._request(
            "GET", f"/reports/2021-06-30/documents/{doc_id}"
        )["payload"]
        # doc["url"] 是带签名的临时地址
        r = requests.get(doc["url"])
        return r.text  # TSV/CSV

• 1 万订单：getOrderItems 可能 1 万次 GET

• Orders Report：建 1 次 + 状态轮询 ~5 次 + 下载 1 次 = 7 次调用

八、限流 / 错误分类（Amazon 特有）

九、“如果真收 $0.40/千次” 的防御代码

# spapi/cost_meter.py
class SpApiGetMeter:
    def __init__(self, redis, app_id, budget_usd_per_month=50.0):
        self.r = redis
        self.key = f"spapi:get:cost:{app_id}:{month_now()}"
        self.price_per_call = 0.40 / 1000

    def count(self, n=1):
        total = self.r.incrby(self.key, n)
        cost = total * self.price_per_call
        if cost > 50.0:
            raise BudgetExceeded(f"SP-API GET 预算将超: ${cost:.2f}")

    def estimated_cost(self):
        return int(self.r.get(self.key) or 0) * self.price_per_call

meter.count(1)
resp = client.get_orders(...)

• GET 才计费，POST/PUT 不按量

• Reports 把 N 次 GET 压成几次 → 天然省钱

• 通知替代轮询 → 把“心跳轮询”成本降到 0

• 多卖家共享 App → 调用量聚合，更容易撞档 → 私域 App / 多 App 分租户要算清

十、和前几篇收口

• 1688：secret 头尾 MD5，QPS 卡脖

• 淘宝：聚石塔内外 10 倍，DSS 推送免费

• 京东：云鼎内稳，POP 时间窗

• 拼多多/抖店：预充值，云外 ×10，增值禁外

• 亚马逊 SP-API：不限国内云内云外概念，限流极狠 + 通知/报告分离 + 原拟 $1400+$0.40 但 2026-05 取消

国内平台是“钱换调用”，亚马逊是“限流换架构”。

你用国内 ERP 的轮询本能写 SP-API，会先被 429 打服，再被（可能的）账单打醒。

SP-API 的正解永远是：EventBridge 叫你、Reports 扛批量、getOrder 只补刀、NextToken 别硬翻、2 分钟延迟写进时间窗。*