把前几篇拆开的计费口径倒推回来，多平台ERP的最优架构不是“能调通就行”，而是用各家的收费模型反推部署位置、同步方式与熔断策略。五家（淘宝TOP / 京东JOS / 1688 / 拼多多 / 抖店）的收费基因不同，但收敛出来是一套通用骨架。
一、五家收费模型 → 架构约束倒推
平台 收费基因 对架构的硬约束
淘宝TOP 基础超量¥0.02/百次(塔内)·¥0.20(塔外)；增值禁外调；DSS推送¥0.12/百单 必须聚石塔内；订单优先DSS推送+增量API兜底；CRM/罗盘类增值必须签约入塔
京东JOS 联盟免费(抽佣)；商家基础有日免额、超量¥0.02~0.10/百次；电子面单按单；增值包年 联盟与商家Key物理隔离；订单增量pop.order.search；库存用ware.read.get拿真stockNum
1688 基础免费+QPS 10/20；高级实时库存/提频买资源包(几百~几千/年) 批发同步别硬轮询，爆款用高级包或Webhook；多Key同企业轮询仅作兜底
拼多多 预充值；基础云内¥0.01/百次·云外¥0.10；增值禁外调 必须拼多多云内+余额守卫；欠费硬切断非限流，需本地计数器兜底
抖店 基础云内¥0.018/百次·云外¥0.18；增值禁外调；预充值 必须抖店云内；2026.7起商品发布也收费，发布流程合并调用
收敛出的三条铁律：
阿里系+抖系=强制云内（外调贵10倍或禁止），拼多多=强制云内（欠费断气）。
订单同步能推不拉——淘宝DSS、1688 Webhook、抖店消息推送优先，API增量仅作补偿。
配额/余额守卫是ERP的一部分，不是外围脚本。
二、最优整体架构（分层+按平台着色）
┌─────────────────────────────────────────────┐
│ 业务层 ERP/OMS/WMS（统一订单/库存/商品域） │
└───────────────┬─────────────────────────────┘
│ 统一事件（OrderCreated/StockChanged）
┌───────────────┴─────────────────────────────┐
│ 适配层 Adapter（每平台一个Client） │
│ TaobaoAdapter(聚石塔内) JdAdapter │
│ Ali1688Adapter(高级包/Webhook) │
│ PddAdapter(云内+余额守卫) DyAdapter(云内) │
└───────────────┬─────────────────────────────┘
│ MQ(Kafka) 削峰 + 令牌桶限速
┌───────────────┴─────────────────────────────┐
│ 调度层：增量时间窗 + 推送消费 + 失败补偿队列 │
│ Redis：幂等键 / 库存原子预扣 / 日调用计数 │
└───────────────┬─────────────────────────────┘
部署：淘宝→聚石塔ECS │ 抖店→抖店云 │ 拼多多→拼多多云
京东→京东云/聚石塔 │ 1688→阿里云(同主体VPC)
核心模块：
• 统一模型：平台订单→内部OrderDTO，平台SKU→内部SkuInventory（区分可售/锁定/在途）。
• 推送为主：淘宝DSS、抖店/1688消息订阅、拼多多订单推送（若有）走MQ consumer；API增量每5~30min补偿一次。
- 令牌桶按Key：淘宝/京东/1688单Key限速在免费QPS的80%（如1688搜8/s、订单16/s）。
• 余额/配额守卫：拼多多余额<3天预估消耗熔断非核心；淘宝/京东日调用达免额80%切纯增量。
三、Python：多平台统一 Guard 骨架（可直插ERP）
multi_platform_erp_guard.py
"""五家API统一守卫：云内判定 + 日配额 + 余额预警 + 限流退避"""
import time, json, hashlib, requests
from abc import ABC, abstractmethod
from threading import Lock
class PlatformQuota:
"""各平台计费/限额元数据"""
META = {
"taobao": {"free_daily": 80_000, "in_price": 0.02, "out_price": 0.20, "must_tower": True},
"jd": {"free_daily": 50_000, "in_price": 0.05, "out_price": 0.15, "must_tower": False},
"ali1688": {"free_daily": 100_000, "qps": 10, "adv_pack": False, "must_tower": False},
"pdd": {"unit_in": 0.01/100, "unit_out": 0.10/100, "prepaid": True, "must_tower": True},
"douyin": {"unit_in": 0.018/100, "unit_out": 0.18/100, "prepaid": True, "must_tower": True},
}
class BaseGuard(ABC):
def init(self, name, app_key, app_secret, in_cloud: bool):
self.name = name
self.ak, self.as = app_key, app_secret
self.in_cloud = in_cloud
self.today_calls = 0
self._lock = Lock()
meta = PlatformQuota.META[name]
if meta.get("must_tower") and not in_cloud:
raise RuntimeError(f"{name} 必须云内部署，否则高价/禁调/欠费断气")
def _roll_day(self):
pass # 简化：实际用日期判断重置today_calls
def before_call(self, is_value=False):
self._roll_day()
m = PlatformQuota.META[self.name]
# 1. 增值接口云外禁调
if is_value and not self.in_cloud and m.get("must_tower"):
raise PermissionError(f"{self.name} 增值接口禁止云外")
# 2. 日免额预警（淘宝/京东/1688）
if "free_daily" in m:
if self.today_calls >= m["free_daily"]:
print(f"⚠️ {self.name} 超免费日额，后续按量扣费")
elif self.today_calls == int(m["free_daily"] * 0.8):
print(f"⚠️ {self.name} 已达免费额80%，切纯增量")
# 3. 拼多多余额守卫（外部注入）
if self.name == "pdd" and hasattr(self, "balance"):
est_day = (self.today_calls + 1) * m["unit_in" if self.in_cloud else m["unit_out"]
if self.balance <= est_day * 3:
raise RuntimeError("pdd 余额<3天预估，熔断非核心调用")
def after_call(self, cost_unit=False):
with self._lock:
self.today_calls += 1
@abstractmethod
def sign(self, params): ...
def safe_post(self, url, params, is_value=False, max_retry=4):
self.before_call(is_value)
params["sign"] = self.sign(params)
for att in range(max_retry):
try:
r = requests.post(url, data=params, timeout=15)
d = r.json()
if "error_response" in d or "errorResponse" 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.after_call()
return d
except requests.RequestException:
time.sleep(2**att); continue
raise RuntimeError("retry exhausted")
class TaobaoGuard(BaseGuard):
def sign(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.as}{qs}{self.as}".encode()).hexdigest().upper()
class PddGuard(BaseGuard):
def init(self, ak, ask, in_cloud, balance=None):
super().init("pdd", ak, ask, in_cloud)
self.balance = balance
def sign(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.as}{qs}{self.as}".encode()).hexdigest().upper()
工厂
def make_guard(name, ak, ask, in_cloud, **kw):
return {
"taobao": TaobaoGuard(name, ak, ask, in_cloud),
"pdd": PddGuard(ak, ask, in_cloud, balance=kw.get("balance")),
}.get(name, BaseGuard(name, ak, ask, in_cloud))
if name == "main":
# 淘宝必须聚石塔内
tb = make_guard("taobao", "AK", "AS", in_cloud=True)
# 拼多多云内+余额守卫
pdd = make_guard("pdd", "CK", "CS", in_cloud=True, balance=8.5)
print("guards ready, 云内强制校验通过")
这段不是完整SDK，而是把“云内强制、免额预警、增值禁外、拼多多余额熔断”收进同一个before_call，接进你现有Client即可。
四、按规模给选型结论
• 单店/小卖（日单<1000）：各家都在免费额度内，本地服务端+增量modified拉取即可，不用迁云（但拼多多/抖店若用云外会悄悄烧钱，建议至少把拼多多/抖店丢进对应云）。
• 中型（日单1万~5万）：订单走推送（DSS/Webhook）+ 每5min增量补偿；淘宝/抖店/拼多多必须云内；1688买个提频包；京东商家Key单用。月API总账参考前篇：淘宝0/京东0/1688≈0/拼多多≈37/抖店≈12。
- ISV/多店（>10店）：每平台多AppKey轮询+Redis中心化令牌桶；增值数据（CRM/罗盘/竞品）单独签协议入塔；成本大头不是API费，是云主机+服务市场抽成（ISV）。
五、一句话收口
多平台ERP的最优解 = 阿里/抖/拼强制云内 + 京东联盟商家分流 + 1688高级包破QPS + 订单能推不拉 + 配额/余额守卫编进Client；架构不是被API能力推着走，是把五家收费模型当需求文档反推出来的。
要不要我接着把上面 BaseGuard 扩成带 Redis 日计数 + 按平台自动切换“推送消费/增量拉取”的调度骨架，直接可套进 APScheduler / Celery 里跑？
