《亚马逊 SP-API 订单同步：getOrders 实时查询 vs Reports 批量报告 —— 成本/限流/时延三维落地方案》（附 Python 源码）
0. 先更新一个前提（2026-10 今天）
你标题里“年费 0.40/千次”现在不收了：
2025-11-03 公布
2026-01 延期
2026-03 无限期推迟
2026-05-12 Amazon 官方：will not move forward with the SP-API usage and annual fees at this time
所以今天写这篇文章，不是“为了省 $0.40/千次”，而是：
亚马逊订单同步的真正货币是“限流配额 + 时延 + 数据完整性”，不是美元。就算不收钱，你用getOrders硬轮询也会先被 429 打挂。
一、两种取数方式的本质差异
维度 | getOrders + getOrderItems | Orders Report |
|---|---|---|
模式 | 同步 REST 拉取 | 异步：create → poll → download |
时延 | 秒级（但订单数据有 ~2 分钟延迟） | 5–45 分钟 |
限流 | getOrders 0.0167 rps = 1/min，burst 20；getOrderItems 0.5 rps | createReport 也有限流，但一次报告顶几万行 |
适合 | 单订单补全、通知后拉详情、低频对账 | 大批量历史/日级全量/多 marketplace 对账 |
不适合 | 全量轮询“有没有新单” | 实时发货、支付后立刻处理 |
字段 | 全（含 PII，要 RDT） | 大部分有，个别促销/礼品字段可能不全 |
官方自己给的场景边界：
日单 < 2 万、无大促 →
getOrders + getOrderItems可行日单 > 2 万 →
ORDER_CHANGE通知 + Reports 对账
二、架构结论（一句话）
通知做触发器，getOrders 做单条补全，Reports 做批量真相与漏单兜底，本地状态机做仲裁。这和淘宝 DSS / 京东 JMQ / 拼多多增量 / 抖店推送是同构的，只是亚马逊把“限流”做得更狠。
EventBridge / SQS ORDER_CHANGE │ ▼ order_id 入队 │ ▼ getOrder + getOrderItems （受 0.5 rps 限制，令牌桶） │ ▼ order 状态机 upsert │ 每 30~60 min / 每天 02:00 ▼ GET_FLAT_FILE_ALL_ORDERS_DATA_BY_LAST_UPDATE_GENERAL │ ▼ TSV 解析 → 和本地订单 diff → 补丢失/修正状态
三、为什么不能“每 30 秒 getOrders 一次”
getOrders 默认：持续速率：1 次/分钟
突发：20 次
突发用完后，再调就是 429
QuotaExceeded
你写：
while True: getOrders(LastUpdatedAfter=now-30s) sleep(30)
前 20 次很爽，第 21 次开始全 429，新单反而进不来。
亚马逊订单还有一条隐藏规则：
订单创建/更新后，数据大约 2 分钟才可通过 Orders API 查到
所以“30 秒轮询”既浪费配额，又查不到刚发生的单。
四、Python：限流感知的 Orders Client
# spapi/orders_client.py
import time
import requests
class ThrottledOrdersClient:
def __init__(self, token_provider, base,
get_orders_rps=0.0167, get_order_rps=0.5,
get_items_rps=0.5):
self.tp = token_provider
self.base = base
self.buckets = {
"getOrders": [get_orders_rps, get_orders_rps, time.monotonic()],
"getOrder": [get_order_rps, get_order_rps, time.monotonic()],
"getOrderItems":[get_items_rps, get_items_rps, time.monotonic()],
}
def _acquire(self, op):
rate, _, last = self.buckets[op]
cap = rate # burst ≈ rate 的近似；严格实现可单独存 burst
while True:
now = time.monotonic()
tokens = min(rate, self.buckets[op][1] + (now - last) * rate)
self.buckets[op][2] = now
if tokens >= 1:
self.buckets[op][1] = tokens - 1
return
time.sleep(0.05)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def _get(self, op, path, params=None, max_retries=6):
for attempt in range(max_retries):
self._acquire(op)
token = self.tp.get_access_token()
headers = {"x-amz-access-token": token, "Accept": "application/json"}
r = requests.get(self.base + path, headers=headers, params=params, timeout=10)
if r.status_code == 429:
wait = int(r.headers.get("Retry-After", 60))
time.sleep(wait + attempt * 2)
continue
if r.status_code >= 500:
time.sleep(2 ** attempt)
continue
r.raise_for_status()
return r.json()
raise RuntimeError(f"{op} exhausted")
def get_orders(self, marketplace_ids, last_updated_after, next_token=None):
p = {
"MarketplaceIds": ",".join(marketplace_ids),
"LastUpdatedAfter": last_updated_after,
}
if next_token:
p["NextToken"] = next_token
return self._get("getOrders", "/orders/v0/orders", p)
def get_order(self, order_id):
return self._get("getOrder", f"/orders/v0/orders/{order_id}")
def get_order_items(self, order_id):
return self._get("getOrderItems", f"/orders/v0/orders/{order_id}/orderItems")五、通知驱动：ORDER_CHANGE 只取 id
# spapi/order_notification.py
def handle_sqs_message(body: dict):
if body.get("NotificationType") != "ORDER_CHANGE":
return
payload = body["Payload"]
order_id = payload["OrderId"]
# 不信任通知里的状态，只当触发器
queue.publish({"order_id": order_id, "event_time": payload.get("EventTime")})
return "success"规则：
通知 0 次 GET 调用
通知只告诉你要看哪个 order_id
真数据用
getOrder + getOrderItems
六、Reports：批量对账（少调用、多数据）
# spapi/orders_report.py
import time
import requests
class OrdersReportSync:
def __init__(self, token_provider, base):
self.tp = token_provider
self.base = base
def _post(self, path, json_body):
token = self.tp.get_access_token()
r = requests.post(
self.base + path,
headers={"x-amz-access-token": token, "Content-Type": "application/json"},
json=json_body, timeout=10,
)
# 报告状态轮询也别狂打：用指数退避
return r.json()
def create_report(self, marketplace_ids, start, end):
return self._post("/reports/2021-06-30/reports", {
"reportType": "GET_FLAT_FILE_ALL_ORDERS_DATA_BY_LAST_UPDATE_GENERAL",
"marketplaceIds": marketplace_ids,
"dataStartTime": start,
"dataEndTime": end,
})
def wait_until_done(self, report_id):
for i in range(40):
token = self.tp.get_access_token()
status = requests.get(
f"{self.base}/reports/2021-06-30/reports/{report_id}",
headers={"x-amz-access-token": token}, timeout=10,
).json()["payload"]
if status["processingStatus"] == "DONE":
return status["reportDocumentId"]
if status["processingStatus"] in ("FATAL", "CANCELLED"):
raise RuntimeError(status)
time.sleep(min(120, 10 * (2 ** (i // 3))))
raise TimeoutError("report timeout")
def download_tsv(self, doc_id):
token = self.tp.get_access_token()
meta = requests.get(
f"{self.base}/reports/2021-06-30/documents/{doc_id}",
headers={"x-amz-access-token": token}, timeout=10,
).json()["payload"]
return requests.get(meta["url"]).text成本对比（日单 1 万）：
轮询式 getOrders：可能几千~几万次 GET
通知 + 单条补全：几十~几百次 GET
每天 1 次 Orders Report：create 1 次 + 状态轮询 N 次 + 下载 1 次
七、成本/配额权衡矩阵（生产决策用）
场景 | 推荐 |
|---|---|
日单 < 2k，MFN 自发货 | ORDER_CHANGE + getOrder/getOrderItems，每 30–60min 小窗补偿 |
日单 2k–2w | 通知为主，getOrders 只做 30min 时间窗兜底 |
日单 > 2w / Prime Day | 通知 + 每日 Orders Report 对账，禁止全量轮询 |
多 marketplace（NA/EU/FE） | 每个 region 独立端点、独立限流、独立 Report |
要买家姓名/地址/电话 | getOrderBuyerInfo / getOrderAddress + RDT，Report 不含 PII |
财务/促销明细 | Orders Report + Finances API，不要靠 getOrderItems 凑 |
历史回填 | Reports + Data Kiosk，不要拿 getOrders 翻几个月 |
八、补偿轮询的正确写法
def poll_compensation(client, marketplace_ids, now):
end = now - timedelta(minutes=2) # 订单有 2 分钟延迟
start = end - timedelta(minutes=30)
next_token = None
while True:
resp = client.get_orders(marketplace_ids, start.isoformat(), next_token)
for o in resp["payload"]["Orders"]:
enqueue_detail(o["AmazonOrderId"])
next_token = resp["payload"].get("NextToken")
if not next_token:
break
# getOrders 默认 1/min：翻页必须睡，不然 burst 打光
time.sleep(60)规则：
LastUpdatedAfter不是CreatedAfter：订单状态变更也算结束时间 减 2 分钟
翻页别快于 1 次/分钟
补偿不是主通道，是“防丢单”
九、状态机：订单轴 / 发货轴 / 财务轴分开
ORDER_FSM = {
None: ["Pending", "PendingAvailability"],
"Pending": ["Unshipped", "Canceled"],
"Unshipped":["PartiallyShipped", "Shipped", "Canceled"],
"PartiallyShipped": ["Shipped", "Canceled"],
"Shipped": ["InvoiceUnconfirmed", "Closed"],
"Canceled": [],
}
def can_transition(old, new):
return new in ORDER_FSM.get(old, [])订单表只存投影：
order_statusfulfillment_channel(AFN/MFN)payment_methodlatest_ship_by
财务/退款走 Finances / Settlements，不混在订单表里。
十、v2026-01-01 的变量（必须提一句）
亚马逊新 Orders API：
searchOrders：0.0056 rpsgetOrder：0.5 rps，单次返回订单+行项目v0 的
getOrders/getOrderItems计划 2027-03 移除
新版本好处：
原来 2 次调用（order + items）→ 1 次
getOrder支持
includedData按需取字段PII 访问模型也改了
新项目直接按getOrder(searchOrders 定位)写，别再死绑 v0。
十一、落地检查表
[ ] 不把 getOrders 当“实时监听”
[ ] ORDER_CHANGE 通知 0 成本触发
[ ] getOrders 翻页 ≥ 60s/次
[ ] 订单数据 2 分钟延迟写进时间窗
[ ] PII 用 RDT / 新版本 PII 模型，不裸调
[ ] 日级对账用 Orders Report，不堆 GET
[ ] Report 状态轮询指数退避，不每秒查
[ ] 多 region 独立限流桶
[ ] 订单/售后/财务三轴分离
[ ] v0 预留 2027 迁移开关
[ ] 就算不收 $0.40/千次，也按“配额即钱”来设计
十二、一句话收口
国内平台是“花钱买调用”，亚马逊是“限流逼你改架构”。getOrders不是不能用来同步订单，而是不能用来“监听”订单；Reports 不是慢就没用，而是大批量场景下它才是最便宜的那条路。成熟亚马逊订单管道只有九个字：通知叫你，getOrder 补刀，Report 对账。谁用while True: getOrders()写亚马逊 OMS，谁就在给 429 交智商税。
