API 超时重试熔断计算引擎（专业版）
输入
retry_backoff(attempt,base=1,cap=60)；retry_budget(attempts,timeout_per,base_backoff=1,cap=60)；circuit_state(fail_count,total,fail_threshold=0.5,min_calls=5,opened_at=None,now=None,cooldown=30)
输出
退避秒数；{{total,call_time,backoffs}}；(状态, 说明)
硬性规则
- 退避 min(base·2^(attempt-1),cap)。
- attempts<1 抛错；total<=0 或 total<min_calls 不熔断。
- opened_at 存在且 now-opened_at≥cooldown → HALF_OPEN，否则 OPEN。
示例
retry_budget(3,2,1,60) → {{'total':9,'call_time':6,'backoffs':[1,2]}}
circuit_state(3,5,0.5,5) → ("OPEN", ...)（失败率60%）
circuit_state(0,0) → ("CLOSED", ...)（零样本不除零）
Python 引擎（可独立运行）
# -*- coding: utf-8 -*-
def retry_backoff(attempt, base=1, cap=60):
"""第 attempt 次失败后的退避秒数（attempt 从1起），指数退避封顶 cap。"""
return min(base * (2 ** max(attempt - 1, 0)), cap)
def retry_budget(attempts, timeout_per, base_backoff=1, cap=60):
"""最坏总耗时预算 = 各次超时之和 + 尝试间退避(attempts-1次)。"""
if attempts < 1:
raise ValueError("attempts 必须 >= 1")
if timeout_per < 0 or base_backoff < 0:
raise ValueError("超时与退避基数不能为负")
backoffs = [retry_backoff(i, base_backoff, cap) for i in range(1, attempts)]
call_time = attempts * timeout_per
return {"total": call_time + sum(backoffs), "call_time": call_time,
"backoffs": backoffs}
def circuit_state(fail_count, total, fail_threshold=0.5, min_calls=5,
opened_at=None, now=None, cooldown=30):
"""熔断器状态：CLOSED放行 / OPEN熔断拒绝 / HALF_OPEN冷却后试探。
零样本与样本不足(min_calls)不熔断，避免除零与误判。"""
if total <= 0 or fail_count < 0 or fail_count > total:
return ("CLOSED", "无有效样本，放行")
if opened_at is not None:
if now is None:
now = opened_at + cooldown
elapsed = now - opened_at
if elapsed >= cooldown:
return ("HALF_OPEN", "冷却结束，放行1个试探请求")
return ("OPEN", f"熔断中，冷却剩余{cooldown - elapsed}秒")
if total < min_calls:
return ("CLOSED", f"样本不足({total}<{min_calls})，不熔断")
rate = fail_count / total
if rate >= fail_threshold:
return ("OPEN", f"失败率{rate:.0%}达阈值{fail_threshold:.0%}，熔断")
return ("CLOSED", f"失败率{rate:.0%}低于阈值，放行")
