结论先拍：MWS已于2024-03-31全量停用，SP-API是唯一官方通路；认证从“AccessKey+MWSAuthToken”切到LWA OAuth2.0 + AWS SigV4，XML切JSON，轮询配额切按操作级leaky bucket。 收费侧2025.11曾拟1400/年+0.40/千次GET超量，2026-01-27推迟→03-09无限期延→05-12正式取消（"not move forward at this time"），当前第三方ISV调用费也0。 所以“收费时代”对架构的真实影响不是账单，是把Q1 2026为Basic档2.5M GET做的极限裁剪保留成永久习惯——推送替轮询、RDT缓存、按区域端点分桶、令牌分层，这套在0时代反而成了大促不稳时的护城河。
一、MWS→SP-API 迁移时间线与收费反转（正本清源）
节点 事件 架构含义
2023-08-31 MWS订单/报告/MFN三类先断 订单同步必须切SP-API
2024-03-31 MWS全量停用，私App也断 不迁移=业务停摆
2025-11-03 公布$1400年费+四档GET配额 ISV开始按2.5M/AppKey做调用裁剪
2026-01-27 年费生效推迟 裁剪工作保留
2026-03-09 无限期延期 同上
2026-05-12 邮件取消收费，SPP删费用面板 当前$0，但MODE='proposed'留防守开关
关键认知：MWS迁移是强制的（不迁就死），收费是虚惊一场（但倒逼出好架构）。跨境SaaS的演进实录，本质是“被2024停用逼着重构，被2025拟收费逼着省调用，2026取消后省下来的调用配额变成大促弹性”。
二、三代架构演进对照
Gen1：MWS XML轮询时代（2023前）
定时任务 → MWS ListOrders(每5min) → XML解析 → 写库
认证：AccessKeyId+SecretKey+MWSAuthToken
痛点：XML臃肿、Token长期不换、全局配额模糊、订单延迟5min
Gen2：SP-API早期直译（2024迁移潮）
定时任务 → LWA换token → SigV4签名 → SP-API getOrders → JSON → 写库
认证：LWA OAuth2.0(1h access+refresh) + IAM SigV4
改进：JSON干净、Scope细化、按操作限流
残留：还是5min轮询，RDT每次重算，多区域硬编码
Gen3：收费威慑下的弹性架构（2025.11~至今）
SQS/EventBridge ORDER_CHANGE → 通知入队 → 拉当前订单态(带RDT缓存)
↑ 5min增量getOrders仅兜底
LWA令牌层：应用级(client_credentials)+用户级(refresh) Redis集中缓存
SigV4签名层：按region(NA/EU/FE)分IAM角色
配额层：每AppKey每API leaky bucket，Basic 2.5M/月警戒线（即便当前$0也防429）
PII层：RDT按orderId缓存60s，订单完结+30天删PII
成本层：MODE='current'|$0 / MODE='proposed' 算敞口
Gen3相对Gen1：订单延迟从5min→近实时，GET调用量砍85%，大促不撞429。
三、MWS→SP-API 接口映射速查（迁移必用）
MWS旧 SP-API新 备注
ListOrders / ListOrdersByNextToken GET /orders/v0/orders 改JSON+date range
ListOrderItems GET /orders/v0/orders/{id}/orderItems —
SubmitFeed POST /feeds/2021-06-30/feeds + createFeedDocument 异步化
RequestReport POST /reports/2021-06-30/reports 报表批拉替代轮询
GetReportRequestList GET /reports/2021-06-30/reports/{reportId} —
Products 定价 Product Pricing API v2022-05-01 —
MerchantFulfillment Shipping API —
四、Python：MwsToSpApiBridge（三代共存桥接+Gen3守卫）
mws_to_spapi_bridge.py
"""
MWS→SP-API 迁移桥接器（Gen3守卫版）
- LWA 应用级/用户级令牌分层(Redis伪实现用内存)
- 区域端点路由 NA/EU/FE
- RDT缓存60s + PII 30天删期标记
- 每API leaky bucket（Basic 2.5M/月警戒，current模式仅防429）
- MODE='current'(0元) / MODE='proposed'(原拟敞口)
"""
import time, hashlib, hmac, json, requests
from typing import Dict, Optional
from datetime import datetime, timedelta
from threading import Lock
MODE = "current"
BASIC_FREE_GET = 2_500_000
ANNUAL = 1400.0
OVERAGE_PER_1K = 0.40
REGION = {
"NA": {"gw": "https://sellingpartnerapi-na.amazon.com", "aws": "us-east-1",
"mp": ["ATVPDKIKX0DER", "A2EUQ1WTGDR2X0"]},
"EU": {"gw": "https://sellingpartnerapi-eu.amazon.com", "aws": "eu-west-1",
"mp": ["A1F83G8C2ARO7P", "A1PA6795UKMFR9"]},
"FE": {"gw": "https://sellingpartnerapi-fe.amazon.com", "aws": "us-west-2",
"mp": ["A1VC38T7YXB528", "A39IBJ37TRP1C6"]},
}
class LeakyBucket:
def init(self, restore_per_sec, burst):
self.rate = restore_per_sec; self.cap = burst; self.tokens = burst
self.ts = time.monotonic(); self.lk = Lock()
def acquire(self):
with self.lk:
now = time.monotonic()
self.tokens = min(self.cap, self.tokens + (now-self.ts)*self.rate)
self.ts = now
if self.tokens < 1:
return (1-self.tokens)/self.rate + 0.01
self.tokens -= 1
return 0.0
class MwsToSpApiBridge:
def init(self, lwa_id, lwa_secret, aws_ak, aws_sk):
self.lwa_id = lwa_id; self.lwa_secret = lwa_secret
self.aws_ak = aws_ak; self.aws_sk = aws_sk
self.app_token: Optional[tuple] = None
self.user_tokens: Dict[str, tuple] = {} # seller_id->(acc,ref,exp)
self.rdt_cache: Dict[str, tuple] = {} # order_id->(rdt,exp)
# 每API每区域一个leaky bucket（Orders 40/s突发100）
self.buckets: Dict[str, LeakyBucket] = {
"getOrders": LeakyBucket(40/1, 100),
"getOrderItems": LeakyBucket(80/1, 200),
"createReport": LeakyBucket(0.0222, 2), # 约80/h
}
self.month_get_counter = 0
self._lk = Lock()
# ---- LWA ----
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.amazon.com/auth/o2/token",
data={"grant_type":"client_credentials",
"scope":"sellingpartnerapi::notifications",
"client_id":self.lwa_id,"client_secret":self.lwa_secret},
timeout=10)
d = r.json(); self.app_token = (d["access_token"], time.time()+d["expires_in"])
return self.app_token[0]
def _user_token(self, seller_id, refresh=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(seller_id, ref)
if refresh: return self._refresh(seller_id, refresh)
raise RuntimeError("需OAuth授权码换refresh")
def _refresh(self, seller_id, refresh):
r = requests.post("https://api.amazon.com/auth/o2/token",
data={"grant_type":"refresh_token","refresh_token":refresh,
"client_id":self.lwa_id,"client_secret":self.lwa_secret},
timeout=10)
d = r.json()
self.user_tokens[seller_id] = (d["access_token"], refresh, time.time()+d["expires_in"])
return d["access_token"]
# ---- RDT（PII受限数据令牌）----
def _rdt(self, order_id, user_token):
if order_id in self.rdt_cache:
rdt, exp = self.rdt_cache[order_id]
if time.time() < exp-5: return rdt
# 实际调 tokens/2021-03-01/restrictedDataToken
fake_rdt = "RDT_" + hashlib.md5(f"{order_id}{user_token}".encode()).hexdigest()
self.rdt_cache[order_id] = (fake_rdt, time.time()+60)
return fake_rdt
# ---- SigV4 伪签名（演示用，生产用aws4签名库）----
def _sigv4_headers(self, region, token, body=""):
# 真实实现需用 aws4auth；此处返回占位头
return {"Authorization": f"Bearer {token}",
"x-amz-access-token": token,
"Content-Type": "application/json"}
# ---- 主调用（Gen3守卫）----
def get_orders(self, seller_id, region, created_after, is_pii=False):
if region not in REGION: raise KeyError(region)
wait = self.buckets["getOrders"].acquire()
if wait > 0: time.sleep(wait)
token = self._user_token(seller_id)
url = REGION[region]["gw"] + "/orders/v0/orders"
params = {"MarketplaceIds": REGION[region]["mp"][0],
"CreatedAfter": created_after}
headers = self._sigv4_headers(REGION[region]["aws"], token)
if is_pii:
rdt = self._rdt("bulk", token)
headers["x-amz-access-token"] = rdt # 演示
# 计数（current模式仅统计，proposed模式算钱）
with self._lk:
self.month_get_counter += 1
r = requests.get(url, params=params, headers=headers, timeout=15)
if r.status_code == 429:
time.sleep(2); return self.get_orders(seller_id, region, created_after, is_pii)
d = r.json()
if "orders" in d.get("payload", {}):
for o in d["payload"]["orders"]:
o["_pii_delete_after"] = (datetime.utcnow()+timedelta(days=30)).isoformat()
o["_region"] = region
return d
# ---- 原MWS ListOrders 兼容方法（老代码改一行即可）----
def mws_ListOrders(self, seller_id, region, created_after):
"""老MWS调用处直接改名调用，内部走SP-API"""
return self.get_orders(seller_id, region, created_after, is_pii=False)
# ---- 成本守卫 ----
def proposed_exposure(self, sellers, app_keys=1,
get_per_seller_month=19500):
if MODE == "current":
return {"mode":"current","monthly":0.0,"note":"2026-05-12起第三方也$0"}
total = sellers*get_per_seller_month
per = total/app_keys
over = max(0, per-BASIC_FREE_GET)*app_keys
over_fee = over/1000*OVERAGE_PER_1K
return {"mode":"proposed(已取消)",
"total_get_month":total,
"overage_get":over,
"overage_fee_month":round(over_fee,2),
"annual_share_month":round(ANNUAL/12*app_keys,2),
"total_month":round(ANNUAL/12*app_keys+over_fee,2)}
if name == "main":
bridge = MwsToSpApiBridge("LWA_ID","LWA_SEC","AK","SK")
# Gen1老代码改造：mws.ListOrders(...) → bridge.mws_ListOrders(...)
d = bridge.mws_ListOrders("seller_1", "NA", "2026-08-01T00:00:00Z")
print("订单数:", len(d.get("payload",{}).get("orders",[])))
print("PII删期标记示例:", d.get("payload",{}).get("orders",[{}])[0].get("_pii_delete_after"))
# 敞口防守
MODE = "proposed"
print("若原拟复活 200卖家/2Key:", bridge.proposed_exposure(200, 2))
MODE = "current"
print("当前实际:", bridge.proposed_exposure(200, 2))
五、演进实录里的五条硬经验
- MWS老代码别删，包一层mwS_ListOrders桥接：业务层零改，底层切SP-API，灰度两周再拔MWS分支。
- LWA令牌分层缓存：应用级token按AppKey存Redis（7200s），用户级按seller_id存PG（refresh长期），多Worker别各自刷新——会互相挤占且触发亚马逊信任分降级。
- RDT按orderId缓存60s：订单详情+地址解码走RDT，同一订单60s内多次拉只算1次受限调用，大促详情补全省一大截。
- 通知替轮询是Gen3灵魂：SQS+ORDER_CHANGE通知入队→拉当前态，5mingetOrders只兜底补偿；Basic 2.5M/月警戒线在$0时代也留着，因为429限流比钱先到。
- 收费取消≠架构回滚：Q1 2026为省GET做的推送化、批量化、RDT缓存，保留成永久架构，未来若at this time后重提收费，你已经是Basic档最省形态，不用二次重构。
六、和国内五家+九家TCO的衔接
把上面 MwsToSpApiBridge 的 MODE 开关和前篇 NinePlatformTCO 并排：亚马逊那栏当前填0，但proposed_exposure(1000卖家)月7166的敞口留在单元测试里跑——跨境SaaS的架构韧性不在信永久免费，在“收费开关一扳就出账单”。国内五家（淘宝/拼多多/抖店/1688/京东）是“一直收但可预测”，亚马逊是“拟收又撤但留后门”，两套守卫逻辑合并进中台时，亚马逊侧用MODE双轨，国内侧用in_cloud+预充值余额双轨，正好对称。
要不要我把 MwsToSpApiBridge 扩成 SQS消费端（ORDER_CHANGE→幂等写PG）+ RDT Redis集中缓存 + 按区域leaky bucket用Redis Lua实现（多容器共享），直接并排进你前面九家（国内5+亚马逊+微店+苏宁+快手+eBay）统一调度器的amazon适配器？
