《从0到1搭多平台电商中台：九家API收费模型倒推出来的最优解》（附python源码）

一、九家收费基因 → 架构硬约束（倒推结果）

收敛三条铁律（来自收费模型反推）：

1. 阿里系+抖系+拼系 = 强制云内，外调×10或禁止，不是“贵”是“不通”。

2. 订单能推不拉：淘宝DSS、拼多多同步服务、抖店Webhook、1688消息订阅、亚马逊Notification、eBay Platform Notifications。

3. 守卫是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):
    CREATED="CREATED"; PAID="PAID"; SHIPPED="SHIPPED"
    SIGNED="SIGNED"; REFUNDING="REFUNDING"; CLOSED="CLOSED"

@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_factory=dict)

    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_cloud=True,
                 prepaid=False, balance=None):
        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_value=False):
        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_calls+1)*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_value=False, max_retry=4):
        self.g.acquire(is_value)
        params["sign"] = self._sign_top(params)
        for att in range(max_retry):
            try:
                r = requests.post(url, data=params, timeout=15)
                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, page=1) -> list[StandardOrder]:
        ...

class TaobaoAdapter(BaseAdapter):
    GW="https://gw.api.taobao.com/router/rest"
    def pull_increment(self, shop_id, token, start, end, page=1):
        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)
        d=self._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"]),
                status=STATUS_MAP["taobao"].get(t["status"],"CREATED"),
                pay_amount=float(t.get("payment",0)),post_fee=float(t.get("post_fee",0)),
                modified_at=t.get("modified"),raw=t))
        return out

class PddAdapter(BaseAdapter):
    GW="https://gw-api.pinduoduo.com/api/router"
    def pull_increment(self, shop_id, token, start, end, page=1):
        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}
        d=self._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"],
                status=STATUS_MAP["pdd"].get(str(o["order_status"]),"CREATED"),
                pay_amount=float(o.get("pay_amount",0)),modified_at=o.get("updated_at"),raw=o))
        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, prepaid=False, bal=None):
        self.channel=channel; self.shop_id=shop_id; self.adapter=adapter
        self.token=token
        self.guard=KeyRateGuard(channel, adapter.ak, qps, free_daily, in_cloud, prepaid, bal)

SHOPS=[
    Shop("taobao","A",TaobaoAdapter(KeyRateGuard("taobao","AK_TB",8,80000,in_cloud=True),"AK_TB","AS_TB"),
         "TB_TOKEN",8,80000,True),
    Shop("pdd","B",PddAdapter(KeyRateGuard("pdd","AK_PD",8,0,in_cloud=True,prepaid=True,balance=8.5),
         "AK_PD","AS_PD"),"PD_TOKEN",8,0,True,prepaid=True,bal=8.5),
]

def sync_loop():
    while True:
        end=datetime.now(); start=end-timedelta(minutes=5)   # 重叠防漏
        for s in SHOPS:
            try:
                orders=s.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()

六、从0到1落地顺序（避坑路径）

1. 资质先行：淘宝/抖店/拼多多订单必须企业自研；1688高级库存买包；京东商家+联盟Key隔离；亚马逊建Private App（自家用豁免年费）；eBay开发者账号免费。

2. 部署着色：淘宝→聚石塔ECS，抖店→抖店云，拼多多→拼多多云，1688/京东→同主体阿里云/京东云VPC，亚马逊/eBay无强制云但出网走固定IP。

3. 先推后拉：每家开推送/订阅，写StandardOrder落库；增量modified每5min兜底。

4. 守卫编进Client：免额80%降频、100%熔断；拼多多余额<3天预估断非核心；亚马逊ISV预算留$1400/年后手。

5. 监控面板：每Key日调用/剩余免额/拼多多余额/令牌桶等待长度 → 企微告警。

七、一句话收口

九家没有一家“API全程免费”，中台最优解不是选最便宜的，而是把90%流量压进免费层（免额内/云内/推送/私有App豁免）+ 把守卫编译进Client让超量/欠费/禁外调在发起前就被拦。架构做对，国内五家单店月API费可压到¥8.4（拼3+抖5.4），跨境四家当前¥0；架构做错，0.01元/百次也能在大促半夜烧穿四位数。