把前八篇拆开的九家计费口径反向编译成中台架构——不是“能调通五个SDK就行”而是用每家收费基因倒推部署位置、同步方式、令牌桶粒度、熔断阈值。下面这套是从0到1可直跑的轻量骨架单进程版生产换Redis/Celery/Kafka即可把“云内强制 / 免额熔断 / 预充值守卫 / 推为主拉兜底”全部焊进Client层。一、九家收费基因 → 架构硬约束倒推结果平台收费基因对中台的倒推约束淘宝TOP免额内0超量塔内¥0.02/百次、外¥0.20增值禁外DSS¥0.12/百单必须聚石塔内订单DSS推送为主增量modified兜底CRM/罗盘另起签协议应用京东JOS联盟免费抽佣商家免额内0超量¥0.02~0.10/百次面单按单联盟Key与商家Key物理隔离库存用ware.read.get拿真stockNum1688基础免费但QPS10/20高级实时库存买包几百~几千/年批发防超卖别硬轮询爆款走高级包/Webhook多Key轮询仅兜底拼多多预充值云内¥0.01/百次、外¥0.10增值禁外欠费硬断非限流​必须拼多多云内余额守卫本地计数器兜底断气即熔断非核心抖店预充值云内¥0.018/百次、外¥0.187.1起商品发布收费增值禁外必须抖店云内发布流程合并调用推送为主亚马逊SP-API自家用Private App豁免三方原1400/年超量0.40/千GET2026.4.30起推迟 indefinite但年费1.31已生效自研走Private App0元ISV预算留$1400/年后手Notification替代轮询eBay开发者账号免费默认5000次/天/IPGrowth Check可提额免费但有道德线走Browse APINotifications别暴拉苏宁基础免额超量参考¥0.02~0.05/百次小生态单店0元够用微店轻量自研免费QPS~5/s重同步买包私域单店0元收敛三条铁律来自收费模型反推阿里系抖系拼系 强制云内外调×10或禁止不是“贵”是“不通”。订单能推不拉淘宝DSS、拼多多同步服务、抖店Webhook、1688消息订阅、亚马逊Notification、eBay Platform Notifications。守卫是Client一部分免额80%降频、100%熔断拼多多余额3天预估断非核心亚马逊留“at this time”后手。二、统一DTO先把九家订单归一# dto.py from dataclasses import dataclass, field from enum import Enum from datetime import datetime class StdStatus(str, Enum): CREATEDCREATED; PAIDPAID; SHIPPEDSHIPPED SIGNEDSIGNED; REFUNDINGREFUNDING; CLOSEDCLOSED dataclass class StandardOrder: channel: str shop_id: str order_id: str idempotency_key: str status: StdStatus StdStatus.CREATED pay_amount: float 0.0 post_fee: float 0.0 item_count: int 0 created_at: datetime None modified_at: datetime None raw: dict field(default_factorydict) def __post_init__(self): if not self.idempotency_key: self.idempotency_key f{self.channel}:{self.shop_id}:{self.order_id} STATUS_MAP { taobao: {WAIT_BUYER_PAY:CREATED,TRADE_PAID:PAID,WAIT_SELLER_SEND_GOODS:PAID,TRADE_BUYER_SIGNED:SIGNED,TRADE_CLOSED:CLOSED}, jd: {10:PAID,20:PAID,30:SHIPPED,40:SIGNED,60:CLOSED}, pdd: {0:CREATED,1:PAID,2:SHIPPED,3:SIGNED,5:REFUNDING}, douyin: {1:CREATED,2:PAID,3:SHIPPED,4:SIGNED,5:CLOSED}, ali1688: {waitbuyerpay:CREATED,waitsellersend:PAID,waitbuyerreceive:SHIPPED,confirm_send:SIGNED,cancel:CLOSED}, amazon: {Pending:CREATED,Unshipped:PAID,Shipped:SHIPPED,Delivered:SIGNED,Canceled:CLOSED}, ebay: {ACTIVE:CREATED,PAID:PAID,SHIPPED:SHIPPED,COMPLETED:SIGNED,CANCELLED:CLOSED}, }三、按Key隔离的计费守卫免额/云内/预充值全编进去# guard.py import time, hashlib, json, requests from datetime import datetime from threading import Lock class KeyRateGuard: 每AppKey独立令牌桶 日免额 拼多多余额熔断 云内强制 def __init__(self, platform, app_key, qps, daily_free, in_cloudTrue, prepaidFalse, balanceNone): self.platform platform self.app_key app_key self.qps qps self.tokens qps self.ts time.monotonic() self.lk Lock() self.day datetime.now().date() self.today_calls 0 self.daily_free daily_free self.in_cloud in_cloud self.prepaid prepaid self.balance balance self.unit_in {pdd:0.01/100,douyin:0.018/100}.get(platform, 0) self.unit_out {pdd:0.10/100,douyin:0.18/100}.get(platform, 0) def _roll_day(self): if datetime.now().date() ! self.day: with self.lk: self.day datetime.now().date() self.today_calls 0 def acquire(self, is_valueFalse): self._roll_day() # 1. 增值/敏感必须云内 if is_value and not self.in_cloud and self.platform in (taobao,pdd,douyin): raise PermissionError(f{self.platform} 增值接口必须云内) # 2. 日免额预警/熔断 if self.today_calls self.daily_free: raise RuntimeError(f{self.app_key} 日免额{self.daily_free}耗尽停调防扣费) elif self.today_calls int(self.daily_free*0.8): print(f⚠️ {self.app_key} 达免额80%切纯增量) # 3. 预充值守卫拼多多/抖店 if self.prepaid and self.balance is not None: unit self.unit_in if self.in_cloud else self.unit_out if self.balance (self.today_calls1)*unit*3: raise RuntimeError(f{self.platform} 余额3天预估熔断非核心) # 4. 令牌桶 with self.lk: now time.monotonic() self.tokens min(self.qps, self.tokens (now-self.ts)*self.qps) self.ts now if self.tokens 1: time.sleep((1-self.tokens)/self.qps 0.005) self.tokens 0 else: self.tokens - 1 self.today_calls 1 def after_ok(self): if self.prepaid and self.balance is not None: unit self.unit_in if self.in_cloud else self.unit_out self.balance - unit四、Adapter抽象 淘宝/拼多多/亚马逊示意实现# adapters.py from abc import ABC, abstractmethod import hashlib, time, json, requests from dto import StandardOrder, STATUS_MAP from guard import KeyRateGuard class BaseAdapter(ABC): def __init__(self, guard: KeyRateGuard, ak, ask): self.g guard; self.ak ak; self.ask ask def _sign_top(self, p): f sorted((k,v) for k,v in p.items() if k!sign and v is not None and str(v)!) qs .join(f{k}{v} for k,v in f) return hashlib.md5(f{self.ask}{qs}{self.ask}.encode()).hexdigest().upper() def _safe(self, url, params, is_valueFalse, max_retry4): self.g.acquire(is_value) params[sign] self._sign_top(params) for att in range(max_retry): try: r requests.post(url, dataparams, timeout15) d r.json() if error_response in d: blob json.dumps(d) if any(k in blob for k in (FLOW_CONTROL,limited-by,50001,no permission)): time.sleep(min(2**att,8)); continue raise Exception(blob) self.g.after_ok() return d except requests.RequestException: time.sleep(2**att); continue raise RuntimeError(retry exhausted) abstractmethod def pull_increment(self, shop_id, token, start, end, page1) - list[StandardOrder]: ... class TaobaoAdapter(BaseAdapter): GWhttps://gw.api.taobao.com/router/rest def pull_increment(self, shop_id, token, start, end, page1): biz{start_modified:start,end_modified:end,page_no:page,page_size:50, fields:tid,status,payment,post_fee,modified} p{method:taobao.trades.sold.increment.get,app_key:self.ak, timestamp:str(int(time.time()*1000)),format:json,v:2.0, sign_method:md5,access_token:token} p.update(biz) dself._safe(self.GW,p) out[] for t in d.get(trades_sold_increment_get_response,{}).get(trades,{}).get(trade,[]): out.append(StandardOrder(taobao,shop_id,str(t[tid]), statusSTATUS_MAP[taobao].get(t[status],CREATED), pay_amountfloat(t.get(payment,0)),post_feefloat(t.get(post_fee,0)), modified_att.get(modified),rawt)) return out class PddAdapter(BaseAdapter): GWhttps://gw-api.pinduoduo.com/api/router def pull_increment(self, shop_id, token, start, end, page1): p{client_id:self.ak,method:pdd.order.number.list.increment.get, timestamp:str(int(time.time())),data_type:JSON,v:V1.0, start_updated_at:int(start),end_updated_at:int(end), page:page,page_size:50,access_token:token} dself._safe(self.GW,p) out[] for o in d.get(order_number_list_increment_get_response,{}).get(order_list,[]): out.append(StandardOrder(pdd,shop_id,o[order_sn], statusSTATUS_MAP[pdd].get(str(o[order_status]),CREATED), pay_amountfloat(o.get(pay_amount,0)),modified_ato.get(updated_at),rawo)) return out # JdAdapter / Ali1688Adapter / DyAdapter / AmazonSpApiAdapter / EbayAdapter 同构 # 亚马逊Private App用LWA拿token调orders/v0/orders?LastUpdatedAfter...Notification替代轮询五、统一调度器增量重叠窗 多店轮转 推送可插拔# scheduler.py import time from datetime import datetime, timedelta from adapters import TaobaoAdapter, PddAdapter from guard import KeyRateGuard class Shop: def __init__(self, channel, shop_id, adapter, token, qps, free_daily, in_cloud, prepaidFalse, balNone): self.channelchannel; self.shop_idshop_id; self.adapteradapter self.tokentoken self.guardKeyRateGuard(channel, adapter.ak, qps, free_daily, in_cloud, prepaid, bal) SHOPS[ Shop(taobao,A,TaobaoAdapter(KeyRateGuard(taobao,AK_TB,8,80000,in_cloudTrue),AK_TB,AS_TB), TB_TOKEN,8,80000,True), Shop(pdd,B,PddAdapter(KeyRateGuard(pdd,AK_PD,8,0,in_cloudTrue,prepaidTrue,balance8.5), AK_PD,AS_PD),PD_TOKEN,8,0,True,prepaidTrue,bal8.5), ] def sync_loop(): while True: enddatetime.now(); startend-timedelta(minutes5) # 重叠防漏 for s in SHOPS: try: orderss.adapter.pull_increment(s.shop_id,s.token, start.strftime(%Y-%m-%d %H:%M:%S),end.strftime(%Y-%m-%d %H:%M:%S)) for o in orders: # Redis幂等 SETNX idempotency_key ; PG upsert ; Kafka order.updated print(f✔ {o.channel}/{o.shop_id}/{o.order_id} {o.status}) except (RuntimeError,PermissionError) as e: print(f⚠️ {s.channel}/{s.shop_id} 守卫拦截: {e}) except Exception as e: print(f❌ {s.channel}/{s.shop_id} {e}) time.sleep(60) # 主控1min拉取窗5min if __name____main__: sync_loop()推送扩展点把pull_increment换成 consumer 即可——淘宝DSS Binlog、拼多多同步服务、抖店Webhook、1688消息订阅、亚马逊POST /notifications/v1/subscriptions、eBayPlatform Notifications统一转StandardOrder走同一条幂等写链路轮询仅作每30min全量校对补偿。六、从0到1落地顺序避坑路径资质先行淘宝/抖店/拼多多订单必须企业自研1688高级库存买包京东商家联盟Key隔离亚马逊建Private App自家用豁免年费eBay开发者账号免费。部署着色淘宝→聚石塔ECS抖店→抖店云拼多多→拼多多云1688/京东→同主体阿里云/京东云VPC亚马逊/eBay无强制云但出网走固定IP。先推后拉每家开推送/订阅写StandardOrder落库增量modified每5min兜底。守卫编进Client免额80%降频、100%熔断拼多多余额3天预估断非核心亚马逊ISV预算留$1400/年后手。监控面板每Key日调用/剩余免额/拼多多余额/令牌桶等待长度 → 企微告警。七、一句话收口九家没有一家“API全程免费”中台最优解不是选最便宜的而是把90%流量压进免费层免额内/云内/推送/私有App豁免 把守卫编译进Client让超量/欠费/禁外调在发起前就被拦。架构做对国内五家单店月API费可压到¥8.4拼3抖5.4跨境四家当前¥0架构做错0.01元/百次也能在大促半夜烧穿四位数。要不要我接着把PushConsumer淘宝DSS/拼多多同步服务/抖店Webhook/亚马逊Notification​ 和PostgreSQL upsert Redis幂等键​ 的落地代码补完整让这套骨架从“轮询兜底版”升级成“推拉一体可上大促”的生产版