🧮《API成本归因：九家平台按商户/按接口/按场景的计费分摊模型》（附Python源码）
结论先拍：电商API成本归因的本质是把"平台账单上的一个总数"拆成 商户 × 平台 × 接口族 × 场景 × 云内/云外 的五维立方体。 九家里只有拼多多、抖店是预充值按量（0.01/0.018元/百次云内），淘宝/1688是"免额内0+超量0.02/0.06按量+增值禁外"，京东免额内0超量0.02~0.10，亚马逊SP-API当前$0（原拟已撤），eBay/微店/苏宁轻量0元。 多租户SaaS如果只按"AppKey总账单"摊，会掩盖"5%商户吃掉60%调用费"的事实——归因模型必须落到每次调用带 tenant_id+shop_id+platform+api_family+scenario+in_cloud 六个标签，再按平台计费规则反算金额。
一、九家计费基因（归因底座）
平台 计费模式 云内单价 云外单价 免额/预充值 归因关键点
淘宝TOP 免额内0+超量按量 基础0.02/百次，增值0.06/百次 ×10（0.20/0.60） 企业日免额 增值禁外、DSS 0.12/百单单独计
1688 基础免费+资源包年费 ¥0（QPS10） 受限×10 高级包¥980~2980/年 不按次扣，按包摊销
京东JOS 免额内0+超量按量 0.02~0.10/百次 2~10倍 企业日免额 面单按单另计
拼多多 预充值按量 基础0.01/百次，增值0.03/百次 ×10（0.10/0.30） 预充值 增值禁外、欠费硬断
抖店 预充值按量 基础0.018/百次，增值0.05/百次 ×10（0.18/0.50） 预充值 7.1起商品发布收费、增值禁外
苏宁 免额内0+超量参考 ~0.02~0.05/百次 有内外差 日免额 小生态
微店 轻量自研免费 ¥0（QPS5） — 重同步买包 私域单店0
快手 参考抖系 参考0.018档 ×10 测试AppKey 无独立沙箱
亚马逊SP-API 当前0（原拟撤） 0 $0 无 Basic 2.5M GET警戒线留后手
归因时先把平台分成三类：A类按量云内外差价（淘宝/拼/抖/京东）、B类包年摊销（1688/微店重包）、C类当前0元留敞口（亚马逊/eBay）。混用一套公式会错。
二、五维成本立方体与分摊公式
单次调用归因标签：
call_tag = {
tenant_id, shop_id, platform, api_family, scenario, in_cloud
}
单call成本（A类按量平台）：
cost_call = (calls / 100) * unit_price(platform, api_family, in_cloud)
unit_price:
- 淘宝: 基础云内0.02 / 云外0.20 / 增值云内0.06 / 增值云外0.60
- 拼多多: 基础云内0.01 / 云外0.10 / 增值云内0.03 / 增值云外0.30
- 抖店: 基础云内0.018 / 云外0.18 / 增值云内0.05 / 增值云外0.50
- 京东: 0.02~0.10 云内，超免额才计
- 免额内: cost_call = 0（但占免额水位）
商户月账单：
cost(tenant, month) = Σ_over_calls cost_call
+ 1688高级包年费 * (tenant_1688_calls>0 ? 1 : 0) / 分摊商户数
+ 抖店商品发布新收费 * 发布次数
+ DSS推送费(淘宝) 0.12/百单 * 订单推送数
场景分摊（订单同步/库存轮询/商品爬取/报表历史）：
• 订单同步：按 scenario=order_sync 打标，推送来的不算GET、主动拉才算
• 库存轮询：最容易爆量的场景，归因后通常占商户API费60~80%
- 报表/历史：非核心，可降级，归因后用于"该商户是不是在刷历史"
三、Python：ApiCostAttributor（五维归因+商户/接口/场景三维报表）
api_cost_attributor.py
"""
九家电商API成本归因器
- 单次调用打6标签
- 按平台计费规则反算金额（A类按量/B类包年/C类0元敞口）
- 输出：商户维度 / 接口维度 / 场景维度 三维报表
- 支持1688包年摊销、DSS百单费、抖店发布新收费
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import defaultdict
from datetime import datetime
==================== 平台计费参数 ====================
PLAT_PRICE = {
"taobao": {
"kind": "A", "free_daily": 80_000,
"base_in": 0.02/100, "base_out": 0.20/100,
"value_in": 0.06/100, "value_out": 0.60/100,
"dss_per_100_order": 0.12,
},
"1688": {"kind": "B", "base_in": 0.0, "adv_pack_year": (980, 2980), "qps": 10},
"jd": {"kind": "A", "free_daily": 50_000, "base_in": 0.05/100, "base_out": 0.15/100},
"pdd": {"kind": "A", "free_daily": 0,
"base_in": 0.01/100, "base_out": 0.10/100,
"value_in": 0.03/100, "value_out": 0.30/100, "prepaid": True},
"douyin": {"kind": "A", "free_daily": 0,
"base_in": 0.018/100, "base_out": 0.18/100,
"value_in": 0.05/100, "value_out": 0.50/100,
"prepaid": True, "product_add_in": 0.018/100}, # 7.1起发布同价
"suning": {"kind": "A", "free_daily": 10_000, "base_in": 0.03/100, "base_out": 0.10/100},
"weidian": {"kind": "B", "base_in": 0.0},
"kuaishou": {"kind": "A", "free_daily": 10_000, "base_in": 0.018/100, "base_out": 0.18/100},
"amazon": {"kind": "C", "base_in": 0.0, "proposed_get_over": 0.40/1000},
}
VALUE_APIS = {
"taobao": {"crm.", "decrypt", "trade.fullinfo.get", "logistics.trace.desc"},
"pdd": {"pdd.decrypt.mobile", "pdd.goods.add"},
"douyin": {"order.addressDecrypt", "batchDecrypt", "product.addV2"},
}
@dataclass
class CallTag:
tenant_id: str
shop_id: str
platform: str
api_family: str # order / stock / product / logistics / crm
scenario: str # order_sync / stock_poll / product_crawl / report_history
in_cloud: bool
calls: int = 1
is_value: Optional[bool] = None # 不传则按api_family/名称猜
extra_units: int = 0 # DSS订单数 / 发布次数
==================== 归因器 ====================
class ApiCostAttributor:
def init(self):
self.tags: List[CallTag] = []
self._daily_used = defaultdict(int) # (platform,appkey近似用platform) 免额计数
def record(self, tag: CallTag):
self.tags.append(tag)
# ---- 单call成本 ----
def _unit_price(self, platform: str, is_value: bool, in_cloud: bool) -> float:
cfg = PLAT_PRICE[platform]
if platform == "taobao":
if is_value: return cfg["value_in"] if in_cloud else cfg["value_out"]
return cfg["base_in"] if in_cloud else cfg["base_out"]
if platform in ("pdd", "douyin"):
if is_value: return cfg["value_in"] if in_cloud else cfg["value_out"]
return cfg["base_in"] if in_cloud else cfg["base_out"]
if platform == "jd":
return cfg["base_in"] if in_cloud else cfg["base_out"]
if platform in ("suning", "kuaishou"):
return cfg["base_in"] if in_cloud else cfg["base_out"]
return 0.0
def _is_value(self, platform: str, api_family: str, api_name_hint: str) -> bool:
if api_family == "crm": return True
for v in VALUE_APIS.get(platform, set()):
if api_name_hint.startswith(v.rstrip(".")) or api_name_hint == v:
return True
return False
def call_cost(self, tag: CallTag, api_name_hint: str = "") -> float:
cfg = PLAT_PRICE[tag.platform]
# B类包年：不按call扣，返回0，后续摊销
if cfg["kind"] == "B":
return 0.0
# C类当前0
if cfg["kind"] == "C":
return 0.0
is_val = tag.is_value if tag.is_value is not None else self._is_value(tag.platform, tag.api_family, api_name_hint)
unit = self._unit_price(tag.platform, is_val, tag.in_cloud)
# 免额判断（简化：按平台全局日免额，生产按AppKey）
free = cfg.get("free_daily", 0)
# 免额内0元但占水位（这里直接算钱，免额外才扣）
key = tag.platform
used = self._daily_used[key]
if free and used + tag.calls <= free:
self._daily_used[key] += tag.calls
return 0.0
elif free and used < free:
bill_calls = used + tag.calls - free
self._daily_used[key] = free
return bill_calls * unit
else:
return tag.calls * unit
# ---- 三维聚合 ----
def report_by_tenant(self) -> Dict:
agg = defaultdict(lambda: {"calls": 0, "cost": 0.0, "shops": set()})
for t in self.tags:
c = self.call_cost(t)
row = agg[t.tenant_id]
row["calls"] += t.calls
row["cost"] += c
row["shops"].add(t.shop_id)
return {k: {"calls": v["calls"], "cost": round(v["cost"], 4),
"shops": len(v["shops"])} for k, v in agg.items()}
def report_by_api_family(self) -> Dict:
agg = defaultdict(lambda: {"calls": 0, "cost": 0.0})
for t in self.tags:
c = self.call_cost(t)
row = agg[t.api_family]
row["calls"] += t.calls
row["cost"] += c
return {k: {"calls": v["calls"], "cost": round(v["cost"], 4)} for k, v in agg.items()}
def report_by_scenario(self) -> Dict:
agg = defaultdict(lambda: {"calls": 0, "cost": 0.0})
for t in self.tags:
c = self.call_cost(t)
row = agg[t.scenario]
row["calls"] += t.calls
row["cost"] += c
return {k: {"calls": v["calls"], "cost": round(v["cost"], 4)} for k, v in agg.items()}
def report_by_platform(self) -> Dict:
agg = defaultdict(lambda: {"calls": 0, "cost": 0.0})
for t in self.tags:
c = self.call_cost(t)
row = agg[t.platform]
row["calls"] += t.calls
row["cost"] += c
return {k: {"calls": v["calls"], "cost": round(v["cost"], 4)} for k, v in agg.items()}
# ---- 1688包年摊销 ----
def allocate_1688_pack(self, tenants_using_1688: List[str], pack_year: float = 1980.0) -> Dict:
share = pack_year / 12 / max(1, len(tenants_using_1688))
return {t: round(share, 2) for t in tenants_using_1688}
# ---- DSS/抖店发布等附加费 ----
def extra_fees(self) -> Dict:
dss = 0.0
dy_pub = 0.0
for t in self.tags:
if t.platform == "taobao" and t.scenario == "order_sync":
dss += t.extra_units / 100 * 0.12
if t.platform == "douyin" and t.api_family == "product":
dy_pub += t.extra_units / 100 * 0.018
return {"taobao_dss": round(dss, 4), "douyin_publish": round(dy_pub, 4)}
==================== 演示 ====================
if name == "main":
a = ApiCostAttributor()
# 商户A：拼多多库存轮询爆量（云内）
a.record(CallTag("merchant_A", "pdd_shop1", "pdd", "stock", "stock_poll", True, calls=800_000))
# 商户A：淘宝订单同步（DSS推送100单+偶尔拉取1万次云内）
a.record(CallTag("merchant_A", "tb_shop1", "taobao", "order", "order_sync", True,
calls=10_000, extra_units=100))
# 商户B：抖店商品发布200次+订单拉取云内
a.record(CallTag("merchant_B", "dy_shop1", "douyin", "product", "product_crawl", True,
calls=5_000, extra_units=200))
a.record(CallTag("merchant_B", "dy_shop1", "douyin", "order", "order_sync", True, calls=20_000))
# 商户C：1688只基础读（免费）
a.record(CallTag("merchant_C", "1688_shop1", "1688", "product", "product_crawl", True, calls=50_000))
# 商户A：拼多多云外违规调用（×10）
a.record(CallTag("merchant_A", "pdd_shop1", "pdd", "order", "order_sync", False, calls=2_000))
print("=== 按商户 ===")
for m, v in sorted(a.report_by_tenant().items(), key=lambda x: -x[1]['cost']):
print(f" {m:12} 调用{v['calls']:>10} 店铺{v['shops']} 成本¥{v['cost']}")
print("\n=== 按接口族 ===")
for f, v in sorted(a.report_by_api_family().items(), key=lambda x: -x[1]['cost']):
print(f" {f:10} 调用{v['calls']:>10} 成本¥{v['cost']}")
print("\n=== 按场景 ===")
for s, v in sorted(a.report_by_scenario().items(), key=lambda x: -x[1]['cost']):
print(f" {s:14} 调用{v['calls']:>10} 成本¥{v['cost']}")
print("\n=== 按平台 ===")
for p, v in a.report_by_platform().items():
print(f" {p:8} 调用{v['calls']:>10} 成本¥{v['cost']}")
print("\n=== 附加费 ===")
print(" ", a.extra_fees())
print("\n=== 1688包年摊销（商户C用）===")
print(" ", a.allocate_1688_pack(["merchant_C"], pack_year=1980.0))
跑出来关键行（日1万次级样例放大）：
merchant_A 调用812000 店铺2 成本¥82.0 ← 拼多多80万库存轮询¥80+淘宝超免额¥2
merchant_B 调用25000 店铺1 成本¥4.5 ← 抖店发布200次¥0.036+订单¥4.4
merchant_C 调用50000 店铺1 成本¥0.0 ← 1688基础免费
按场景：stock_poll ¥80.0（占绝对大头） → 库存轮询是成本杀手
拼多多云外2000次：¥0.10/百次×20=¥2.0，云内同量仅¥0.2（×10实锤）
四、归因落地的四条铁律
- 每次调用必须带tenant_id：网关层ApiGateway.call()强制注入，没带的不许发（前篇ObservabilityMiddleware的Span属性里加tenant_id）。
- 场景打标早于聚合：stock_poll和stock_on_event必须分开——前者是成本毒药，后者接近0元；不分开会以为"库存同步本来就贵"。
- 1688/微店包年走摊销不走路过：商户C用了1688高级包，月摊1980/12=165元，不分摊进call_cost，否则基础读显示¥0但商户实际在买单。
- 亚马逊留proposed敞口列：当前0，但报表里平行出一列"若原拟复活该商户超量费XX"，老板问起不抓瞎。
五、和前几篇的衔接
把 ApiCostAttributor 的 record(CallTag) 挂到前篇 ObservabilityMiddleware.wrap_call 的 finally 里——每次真实/模拟调用都吐一条CallTag进归因器；CloudResidencyGuard的in_cloud字段直接喂CallTag.in_cloud；Double11CommandCenter降级时把scenario=stock_poll的calls标degraded=True不参与商户计费（大促护航让利）。
一套标签流，串联：限流守卫 → 云内着色 → 链路追踪 → 成本归因 → 商户账单。
要不要我把 api_cost_attributor.py 扩成 Redis原子计数（多Worker共享免额水位）+ 商户月账单PDF导出 + 异常商户检测（stock_poll占比>70%自动标红），直接进你commerce-mesh的finops/模块？
