🧮《API成本归因：九家平台按商户/按接口/按场景的计费分摊模型》（附Python源码）

一、九家计费基因（归因底座）

归因时先把平台分成三类：A类按量云内外差价（淘宝/拼/抖/京东）、B类包年摊销（1688/微店重包）、C类当前0元留敞口（亚马逊/eBay）。混用一套公式会错。

二、五维成本立方体与分摊公式

call_tag = {
  tenant_id, shop_id, platform, api_family, scenario, in_cloud
}

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))

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实锤）

四、归因落地的四条铁律

1. 每次调用必须带tenant_id：网关层ApiGateway.call()强制注入，没带的不许发（前篇ObservabilityMiddleware的Span属性里加tenant_id）。

2. 场景打标早于聚合：stock_poll和stock_on_event必须分开——前者是成本毒药，后者接近0元；不分开会以为"库存同步本来就贵"。

3. 1688/微店包年走摊销不走路过：商户C用了1688高级包，月摊1980/12=165元，不分摊进call_cost，否则基础读显示¥0但商户实际在买单。

4. 亚马逊留proposed敞口列：当前$0，但报表里平行出一列"若原拟复活该商户超量费$XX"，老板问起不抓瞎。

五、和前几篇的衔接

把 ApiCostAttributor 的 record(CallTag) 挂到前篇 ObservabilityMiddleware.wrap_call 的 finally 里——每次真实/模拟调用都吐一条CallTag进归因器；CloudResidencyGuard的in_cloud字段直接喂CallTag.in_cloud；Double11CommandCenter降级时把scenario=stock_poll的calls标degraded=True不参与商户计费（大促护航让利）。

一套标签流，串联：限流守卫 → 云内着色 → 链路追踪 → 成本归因 → 商户账单。