🛡️《多平台API配额守卫设计：欠费/限流/超量三重熔断（附Python源码）》

一、三重熔断模型

┌─────────────────────────────────────────────────────────┐
│                  三重熔断守卫链                           │
├─────────────────────────────────────────────────────────┤
│  ① 调用前守卫（Pre-Call Guard）                         │
│     - 日配额水位（免额80%/100%阈值）                    │
│     - 预充值余额（<3天预估→降频，≤0→熔断非核心）      │
│     - 令牌桶（QPS限速，防止突发429）                    │
├─────────────────────────────────────────────────────────┤
│  ② 调用中守卫（In-Call Guard）                          │
│     - 429响应→指数退避+jitter                          │
│     - 5xx响应→快速失败+死信队列                        │
│     - 超时（15s）→重试1次后熔断                        │
├─────────────────────────────────────────────────────────┤
│  ③ 调用后守卫（Post-Call Guard）                        │
│     - 日用量累加→更新Redis计数器                       │
│     - 超量账单估算→若复活原拟SP-API $0.40/千次         │
│     - 告警分级：80% INFO / <3天 WARN / ≤0 CRITICAL     │
└─────────────────────────────────────────────────────────┘

二、Python：TripleGuardClient（生产级守卫骨架）

# triple_guard_client.py
"""
多平台API配额守卫：欠费/限流/超量三重熔断
- PreGuard：日配额+余额+令牌桶
- InGuard：429/5xx/超时退避
- PostGuard：计数+告警
- 支持：淘宝/拼多多/抖店/亚马逊/eBay
"""
import time, hashlib, json, logging
from typing import Dict, Optional, Callable
from datetime import datetime, timedelta
from threading import Lock
from enum import Enum
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("triple_guard")

# ==================== 熔断级别 ====================
class FuseLevel(Enum):
    PASS = "pass"           # 正常通过
    DEGRADE = "degrade"     # 降频（非核心跳过）
    BLOCK = "block"         # 熔断（所有调用拒绝）
    CRITICAL = "critical"   # 严重（电话告警）

# ==================== 平台配置 ====================
PLATFORM_CONFIG = {
    "taobao": {
        "type": "daily_free",       # 日免额模型
        "daily_free": 80_000,
        "unit_price_in": 0.02/100,
        "unit_price_out": 0.20/100,
        "has_prepaid": False,
        "qps": 40,
    },
    "pdd": {
        "type": "prepaid",          # 预充值模型
        "daily_free": 0,
        "unit_price_in": 0.01/100,
        "unit_price_out": 0.10/100,
        "has_prepaid": True,
        "qps": 30,
    },
    "douyin": {
        "type": "prepaid",
        "daily_free": 0,
        "unit_price_in": 0.018/100,
        "unit_price_out": 0.18/100,
        "has_prepaid": True,
        "qps": 50,
    },
    "amazon": {
        "type": "quota",            # 配额模型（Basic 2.5M/月）
        "daily_free": 83_333,       # 2.5M/30 ≈ 83k/天
        "unit_price_in": 0.0,
        "unit_price_out": 0.0,
        "has_prepaid": False,
        "qps": 40,
    },
    "ebay": {
        "type": "daily_quota",      # 日配额5000/API
        "daily_free": 5_000,
        "unit_price_in": 0.0,
        "unit_price_out": 0.0,
        "has_prepaid": False,
        "qps": 20,
    },
}

# ==================== 令牌桶 ====================
class TokenBucket:
    def __init__(self, rate: float, burst: int):
        self.rate = rate
        self.cap = burst
        self.tokens = burst
        self.ts = time.monotonic()
        self.lk = Lock()

    def acquire(self) -> float:
        """返回等待秒数，0表示立即通过"""
        with self.lk:
            now = time.monotonic()
            elapsed = now - self.ts
            self.tokens = min(self.cap, self.tokens + elapsed * self.rate)
            self.ts = now
            if self.tokens < 1:
                wait = (1 - self.tokens) / self.rate + 0.005
                self.tokens = 0
                return wait
            self.tokens -= 1
            return 0.0

# ==================== 预充值守卫 ====================
class PrepaidGuard:
    """拼多多/抖店预充值余额守卫"""
    def __init__(self, initial_balance: float, daily_estimated_cost: float):
        self.balance = initial_balance
        self.daily_cost = daily_estimated_cost
        self.lk = Lock()

    def check(self) -> FuseLevel:
        with self.lk:
            days_left = self.balance / self.daily_cost if self.daily_cost > 0 else float('inf')
            if days_left <= 0:
                logger.critical(f"❌ 余额归零！熔断所有调用")
                return FuseLevel.CRITICAL
            elif days_left < 3:
                logger.warning(f"⚠️ 余额仅够{days_left:.1f}天，降频非核心")
                return FuseLevel.DEGRADE
            elif days_left < 7:
                logger.info(f"ℹ️ 余额{days_left:.1f}天，注意续费")
                return FuseLevel.PASS
            return FuseLevel.PASS

    def deduct(self, amount: float):
        with self.lk:
            self.balance -= amount

    def top_up(self, amount: float):
        with self.lk:
            self.balance += amount

# ==================== 日配额守卫 ====================
class DailyQuotaGuard:
    """日免额/配额水位守卫"""
    def __init__(self, daily_limit: int):
        self.limit = daily_limit
        self.used = 0
        self.reset_ts = self._next_midnight()
        self.lk = Lock()

    def _next_midnight(self) -> float:
        now = datetime.now()
        return (now + timedelta(days=1)).replace(
            hour=0, minute=0, second=0, microsecond=0
        ).timestamp()

    def check(self) -> FuseLevel:
        with self.lk:
            if time.time() >= self.reset_ts:
                self.used = 0
                self.reset_ts = self._next_midnight()
            ratio = self.used / self.limit if self.limit > 0 else 0
            if ratio >= 1.0:
                logger.error(f"❌ 日配额{self.limit}耗尽！熔断")
                return FuseLevel.BLOCK
            elif ratio >= 0.8:
                logger.warning(f"⚠️ 日配额已达{ratio:.0%}，降频非核心")
                return FuseLevel.DEGRADE
            elif ratio >= 0.6:
                logger.info(f"ℹ️ 日配额{ratio:.0%}，注意")
                return FuseLevel.PASS
            return FuseLevel.PASS

    def increment(self):
        with self.lk:
            self.used += 1

# ==================== 三重守卫客户端 ====================
@dataclass
class GuardResult:
    passed: bool
    level: FuseLevel
    wait_seconds: float = 0.0
    message: str = ""

class TripleGuardClient:
    def __init__(self, platform: str, app_key: str,
                 initial_balance: float = 0.0,
                 daily_estimated_cost: float = 0.0):
        self.platform = platform
        self.app_key = app_key
        config = PLATFORM_CONFIG[platform]

        # 令牌桶（QPS限速）
        self.bucket = TokenBucket(config["qps"], int(config["qps"] * 2))

        # 日配额守卫
        self.quota = DailyQuotaGuard(config["daily_free"])

        # 预充值守卫（仅拼多多/抖店）
        self.prepaid = PrepaidGuard(initial_balance, daily_estimated_cost) if config["has_prepaid"] else None

        # 统计
        self.total_calls = 0
        self.blocked_calls = 0
        self.degraded_calls = 0
        self.lk = Lock()

    # ==================== ① 调用前守卫 ====================
    def pre_guard(self, is_core: bool = True) -> GuardResult:
        """调用前三重检查"""
        # 1. 令牌桶（QPS限速）
        wait = self.bucket.acquire()
        if wait > 0:
            time.sleep(wait)

        # 2. 预充值守卫
        if self.prepaid:
            level = self.prepaid.check()
            if level == FuseLevel.CRITICAL:
                return GuardResult(False, FuseLevel.CRITICAL, message="余额归零")
            if level == FuseLevel.DEGRADE and not is_core:
                return GuardResult(False, FuseLevel.DEGRADE, message="余额不足，非核心跳过")

        # 3. 日配额守卫
        level = self.quota.check()
        if level == FuseLevel.BLOCK:
            return GuardResult(False, FuseLevel.BLOCK, message="日配额耗尽")
        if level == FuseLevel.DEGRADE and not is_core:
            return GuardResult(False, FuseLevel.DEGRADE, message="日配额80%，非核心跳过")

        return GuardResult(True, FuseLevel.PASS)

    # ==================== ② 调用中守卫 ====================
    def in_guard(self, func: Callable, max_retries: int = 3) -> Dict:
        """带退避的重试守卫"""
        last_error = None
        for attempt in range(max_retries):
            try:
                result = func()
                return result
            except Exception as e:
                last_error = e
                status = getattr(e, 'status_code', 0) if hasattr(e, 'status_code') else 0
                if status == 429:
                    wait = (2 ** attempt) + (hash(str(time.time())) % 100) / 1000
                    logger.warning(f"429限流，等待{wait:.2f}s重试")
                    time.sleep(wait)
                elif status >= 500:
                    logger.error(f"5xx服务器错误，快速失败")
                    raise
                else:
                    time.sleep(0.5)
        raise last_error

    # ==================== ③ 调用后守卫 ====================
    def post_guard(self, cost: float = 0.0):
        """调用后计数+扣费+告警"""
        with self.lk:
            self.total_calls += 1
            self.quota.increment()
            if self.prepaid and cost > 0:
                self.prepaid.deduct(cost)

        # 超量账单估算（若原拟SP-API复活）
        if self.platform == "amazon":
            monthly = self.quota.used * 30
            if monthly > 2_500_000:
                over = monthly - 2_500_000
                over_fee = over / 1000 * 0.40
                logger.warning(f"亚马逊超量预估: {over:,} GET, 超量费${over_fee:.2f}/月")

    # ==================== 安全调用 ====================
    def safe_call(self, func: Callable, is_core: bool = True,
                  cost: float = 0.0, max_retries: int = 3) -> Optional[Dict]:
        """三重守卫封装的一次安全调用"""
        # 调用前
        guard = self.pre_guard(is_core)
        if not guard.passed:
            with self.lk:
                if guard.level in (FuseLevel.BLOCK, FuseLevel.CRITICAL):
                    self.blocked_calls += 1
                else:
                    self.degraded_calls += 1
            logger.warning(f"⏸ {self.platform}/{self.app_key}: {guard.message}")
            return None

        # 调用中
        try:
            result = self.in_guard(func, max_retries)
        except Exception as e:
            logger.error(f"❌ {self.platform}/{self.app_key}: {e}")
            return None

        # 调用后
        self.post_guard(cost)
        return result

    # ==================== 报告 ====================
    def report(self) -> Dict:
        with self.lk:
            return {
                "platform": self.platform,
                "app_key": self.app_key,
                "total_calls": self.total_calls,
                "blocked_calls": self.blocked_calls,
                "degraded_calls": self.degraded_calls,
                "block_rate": round(self.blocked_calls / max(1, self.total_calls) * 100, 2),
                "quota_used": self.quota.used,
                "quota_limit": self.quota.limit,
                "quota_ratio": round(self.quota.used / max(1, self.quota.limit) * 100, 2),
                "balance": self.prepaid.balance if self.prepaid else None,
                "days_left": round(self.prepaid.balance / max(1, self.prepaid.daily_cost), 1) if self.prepaid else None,
            }

# ==================== 演示 ====================
def mock_api_call():
    """模拟一次API调用"""
    time.sleep(0.01)
    return {"success": True, "data": "mock_result"}

if __name__ == "__main__":
    # 创建各平台守卫
    guards = {
        "taobao": TripleGuardClient("taobao", "TB_KEY"),
        "pdd": TripleGuardClient("pdd", "PDD_KEY", initial_balance=50.0, daily_estimated_cost=1.0),
        "douyin": TripleGuardClient("douyin", "DY_KEY", initial_balance=100.0, daily_estimated_cost=2.0),
        "amazon": TripleGuardClient("amazon", "AMZ_KEY"),
        "ebay": TripleGuardClient("ebay", "EBAY_KEY"),
    }

    # 模拟100次调用
    for i in range(100):
        for name, guard in guards.items():
            # 核心调用（订单同步）
            result = guard.safe_call(mock_api_call, is_core=True, cost=0.01)
            # 非核心调用（商品浏览）
            if i % 3 == 0:
                guard.safe_call(mock_api_call, is_core=False, cost=0.005)

    # 报告
    print("\n=== 三重守卫运行报告 ===")
    for name, guard in guards.items():
        r = guard.report()
        print(f"\n{name:8}")
        print(f"  总调用: {r['total_calls']}  熔断: {r['blocked_calls']}  降频: {r['degraded_calls']}")
        print(f"  配额: {r['quota_used']}/{r['quota_limit']} ({r['quota_ratio']}%)")
        if r['balance'] is not None:
            print(f"  余额: ¥{r['balance']:.2f}  可用: {r['days_left']}天")

三、三重熔断配置速查（各平台）

四、落地清单

1. 调用前守卫是命脉：预充值余额<3天降频非核心，≤0熔断所有——拼多多/抖店欠费不是限流是停服

2. 429退避必须带jitter：2^attempt + random(0~100ms)，否则多个Worker同时重试再撞429

3. 日配额80%预警+100%熔断：淘宝/京东免额内0元，超了就是钱；亚马逊Basic 2.5M当前$0但防429

4. 非核心降频是省钱杠杆：商品浏览/报表历史/日志查询这些非核心，在配额紧张时直接跳过，保住订单/库存核心链路

5. 告警分级：80%→企微INFO、<3天→WARN、≤0→电话CRITICAL

五、和前几篇的衔接

把本篇 TripleGuardClient 的 safe_call 塞进前篇 four_platform_middleware 的每个Adapter：

• 淘宝Adapter：self.guard.safe_call(lambda: self._do_call(...), is_core=True, cost=0.02/100)

• 拼多多Adapter：self.guard.safe_call(..., is_core=is_order_sync, cost=0.01/100)

• 亚马逊Adapter：self.guard.safe_call(..., is_core=True, cost=0.0)
  一个守卫装饰器，覆盖所有平台的欠费/限流/超量三种死法。