×

《亚马逊SP-API多国站点调度:北美/欧洲/日本/澳洲费率与合规对照》(附Python源码)

万邦科技Lex 万邦科技Lex 发表于2026-08-12 11:34:23 浏览15 评论0

抢沙发发表评论

结论先拍:SP-API按三大区域端点调度(北美NA / 欧洲EU / 远东FE),澳大利亚归在FE端点下(与日本、新加坡同区),API调用费全球统一=0(2026-05-12官方取消原拟0.40/千次超量,第三方ISV当前也0元)。 但“费率一致”不等于“成本一致”——各站点销售佣金/FBA/月租/增值税独立核算,合规层更是三套法域:美国CCPA、欧洲GDPR+EORI+VAT、日本APPI+发票号、澳洲GST+ABN。 做多国调度,重点不是算API钱(为0),是把端点路由+PII留存30天+区域隐私法+站点独立限流编译进Client。


一、三大端点与站点归属(2026现行)

区域端点
GW Base
AWS Region
覆盖站点(含题面四地)
NA 北美
sellingpartnerapi-na.amazon.com
us-east-1
US、CA、MX、BR
EU 欧洲
sellingpartnerapi-eu.amazon.com
eu-west-1
UK、DE、FR、IT、ES、NL、SE、PL、SA、AE、EG、TR、IN、ZA
FE 远东
sellingpartnerapi-fe.amazon.com
us-west-2
JPAU、SG
注意:澳大利亚不在NA也不在EU,落在FE端点(us-west-2),但底层业务数据仍受澳洲GST/ABN监管;日本站同FE端点,受APPI约束。

二、费率对照:API层 vs 业务层

API调用层(SP-API本身)

  • 2026-08今天:卖家自用Private App 0(原拟方案已撤)。

  • 三地端点同价=0,无“北美便宜欧洲贵”之说。

  • 唯一变量是限流配额:Orders API默认40次/秒突发100,按AppKey+区域独立计数,不是按钱。

业务费用层(各站点独立,和API无关但ERP必须算)

站点
月租(专业卖家)
销售佣金典型区间
特殊合规成本
美国US
$39.99
8%~15%
CCPA、W-8BEN-E
加拿大CA
CAD 29.99
8%~15%
GST/HST
日本JP
¥0(按件)或 ¥5,000/月等效
8%~15%
APPI、发票系统登记(適格請求書発行事業者)
澳洲AU
AUD 49.95
8%~15%
GST 10%、ABN、低值进口GST
德国DE
€39.99
8%~15%
VAT+包装法LUCID+EORI
英国UK
£25
8%~15%
VAT+EORI(脱欧后必填)
跨4国跑单店ERP,API费3k~$8k,比原拟SP-API费贵得多——这才是多国架构的真账单。

三、合规对照(PII留存+隐私法+数据驻留)

法域
核心义务
SP-API落地硬约束
美国 CCPA
买家PII可删除、可拒绝出售
订单PII(姓名/地址/电话)订单发货后留存≤30天,之后冷存或删;桌面应用不准落本地明文
欧盟 GDPR
数据最小化、被遗忘权、DPA
EU端点拉到的PII建议存eu-west-1或同区,跨境传美需SCC;留存30天红线同亚马逊AUP
日本 APPI
个人信息定义含买家、需目的明示
日文地址/电话同PII处理;发票号(登録番号)单独落表
澳洲 Privacy Act + GST
ABN关联、GST 10%开票
AU订单PII同30天红线;GST在订单金额字段单独核算
通用 AUP
亚马逊开发者协议
PII不准转售、不准用于营销、STS临时凭证刷新、TLS1.2+
数据驻留建议(非强制但抗审计):
  • NA端点数据 → us-east-1 RDS

  • EU端点数据 → eu-west-1 RDS(GDPR最稳)

  • FE端点(JP/AU/SG)→ us-west-2 RDS,但AU买家PII建议在澳洲本地做一次镜像脱敏备份供ABN审计


四、Python:SpApiRegionRouter(多国调度+端点守卫+PII留存计时+合规标签)

# sp_api_region_router.py
"""
亚马逊SP-API 多国站点调度器
- 站点→端点→AWS Region 路由
- 每区域独立令牌桶(Orders 40/s 突发100)
- PII留存30天自动标记删除
- 合规标签(CCPA/GDPR/APPI/GST)挂站点
"""
import time, hashlib, hmac, requests, json
from typing import Dict, List
from datetime import datetime, timedelta
from threading import Lock

REGION_MAP = {
    "NA": {"gw": "https://sellingpartnerapi-na.amazon.com", "aws": "us-east-1",
           "marketplaces": ["ATVPDKIKX0DER", "A2EUQ1WTGDR2X0", "A1AM78C64UM0Y8"],
           "compliance": ["CCPA"], "pii_retain_days": 30},
    "EU": {"gw": "https://sellingpartnerapi-eu.amazon.com", "aws": "eu-west-1",
           "marketplaces": ["A1F83G8C2ARO7P", "A1PA6795UKMFR9", "A13V1IC380AYAH",
                            "APJ6JRA9NG5V4", "A1RKKUPIHCS9HS", "A21TJRUUN4KGV"],
           "compliance": ["GDPR", "EORI", "VAT"], "pii_retain_days": 30},
    "FE": {"gw": "https://sellingpartnerapi-fe.amazon.com", "aws": "us-west-2",
           "marketplaces": ["A1VC38T7YXB528", "A39IBJ37TRP1C6", "A19VAU5U5O7RUS"],
           "compliance": ["APPI", "GST_ABN"], "pii_retain_days": 30},
}

# 站点→区域
SITE_TO_REGION = {
    "US": "NA", "CA": "NA", "MX": "NA", "BR": "NA",
    "UK": "EU", "DE": "EU", "FR": "EU", "IT": "EU", "ES": "EU",
    "JP": "FE", "AU": "FE", "SG": "FE",
}

class TokenBucket:
    def __init__(self, rate, burst):
        self.rate = rate; self.cap = burst; self.tokens = burst
        self.ts = time.monotonic(); self.lk = Lock()
    def wait(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:
                time.sleep((1-self.tokens)/self.rate + 0.005)
                self.tokens = 0
            else:
                self.tokens -= 1

class SpApiRegionRouter:
    def __init__(self, lwa_client_id, lwa_secret, refresh_token):
        self.cid = lwa_client_id; self.csec = lwa_secret; self.rt = refresh_token
        self.buckets = {r: TokenBucket(40, 100) for r in REGION_MAP}  # Orders档40/s突发100
        self.tokens = {}  # region -> (access_token, exp)

    # ---- 站点→端点解析 ----
    def resolve(self, site: str) -> Dict:
        region = SITE_TO_REGION.get(site.upper())
        if not region:
            raise KeyError(f"站点{site}未配置区域映射")
        return {"site": site.upper(), "region": region, **REGION_MAP[region]}

    # ---- LWA换票(按region缓存) ----
    def _token(self, region: str) -> str:
        if region in self.tokens:
            tk, exp = self.tokens[region]
            if time.time() < exp - 300:
                return tk
        # 简化:实际应POST https://api.amazon.com/auth/o2/token
        tk = "FAKE_ACCESS_TOKEN_" + region
        self.tokens[region] = (tk, time.time()+3600)
        return tk

    # ---- PII留存标记 ----
    @staticmethod
    def pi_i_retain_until(order_date: datetime) -> datetime:
        return order_date + timedelta(days=30)

    def safe_call(self, site: str, path: str, params: Dict, is_pii: bool = True):
        meta = self.resolve(site)
        self.buckets[meta["region"]].wait()
        token = self._token(meta["region"])
        url = meta["gw"] + path
        headers = {"Authorization": f"Bearer {token}",
                   "x-amz-access-token": token,
                   "Content-Type": "application/json"}
        # 合规头:EU强制标记用途
        if "GDPR" in meta["compliance"]:
            headers["x-data-purpose"] = "order_fulfillment_only"
        r = requests.get(url, params=params, headers=headers, timeout=15)
        d = r.json()
        if "errors" in d:
            code = d["errors"][0].get("code","")
            if code in ("QuotaExceeded", "Throttling"):
                time.sleep(2); return self.safe_call(site, path, params, is_pii)
            raise Exception(f"SP-API {meta['region']}: {d['errors']}")
        # PII留存计时(订单场景)
        if is_pii and "payload" in d and "orderId" in d["payload"]:
            d["_pii_delete_after"] = self.pi_i_retain_until(
                datetime.utcnow()).isoformat()
        d["_region"] = meta["region"]
        d["_compliance"] = meta["compliance"]
        return d

    def list_orders(self, site: str, created_after: str):
        return self.safe_call(site, "/orders/v0/orders",
                              {"MarketplaceIds": self._mp(site),
                               "CreatedAfter": created_after})

    def _mp(self, site: str) -> str:
        region = SITE_TO_REGION[site.upper()]
        return REGION_MAP[region]["marketplaces"][0]  # 演示取首站

if __name__ == "__main__":
    router = SpApiRegionRouter("LWA_ID", "LWA_SEC", "RT")
    for site in ("US", "DE", "JP", "AU"):
        m = router.resolve(site)
        print(f"{site:3} → 区域{m['region']} 端点{m['gw']} AWS{m['aws']} "
              f"合规{m['compliance']} PII留存{m['pii_retain_days']}天")
    # 演示调用(假token)
    try:
        r = router.list_orders("JP", "2026-08-01T00:00:00Z")
        print("JP订单调用区域:", r.get("_region"), "合规:", r.get("_compliance"))
    except Exception as e:
        print("演示调用(无真token):", e)

五、多国调度架构铁律(5条)

  1. 端点按站点分,不按AppKey混:US/CA走NA,DE/UK走EU,JP/AU走FE;同一AppKey可授权多站点,但调用时必须按marketplaceId路由到对应端点,错端点返回InvalidMarketplaceId

  2. 令牌桶按区域独立:NA/EU/FE各40/s突发100,AU虽在FE但不该和JP抢同一桶——可在FE桶内再按站点子队列。

  3. PII留存30天全局红线:订单发货+30天自动删或冷归档,AU/JP/UK一视同仁;桌面端不准存明文。

  4. 合规标签随站点走:EU订单落库加gdpr=true列,AU订单加gst=0.1, abn=xxx,JP加invoice_reg_no,US加ccpa_purge_at

  5. API费0:VAT代理/ABN/GST开票/发票号维护/数据驻留RDS多副本,才是多国ERP的真实OPEX。


六、和国内五家放一起的CTO定性

亚马逊SP-API多国调度的复杂度不在钱(API费0)而在法:三大端点路由是技术问题,CCPA/GDPR/APPI/GST是审计问题。把“端点守卫+30天PII删期+合规标签”编译进Client,跨4国(US/DE/JP/AU)单店月API成本=3k~$8k;国内五家(淘宝/拼多多/抖店/1688/京东)则是反过来的故事——API按量几块到几百块,但无跨法域隐私法负担。跨境ERP的隐式成本永远在合规不在网关
要不要我把 SpApiRegionRouter 扩成 Redis中心化按区域令牌桶(多容器共享)+ PII删期定时任务(30天扫描PG)+ 各站点GST/VAT/发票号落库schema,直接并排进你前面那套九家(国内5+亚马逊+微店+苏宁+快手)中台调度器?


群贤毕至

访客