API 限流策略计算引擎（专业版）
输入
token_bucket(capacity,refill_rate,tokens_needed,current_tokens)；fixed_window(limit,count)；retry_after_header(attempt,base=1,cap=60)
输出
(放行bool,剩余,等待秒数/None)；(放行bool,剩余配额)；退避秒数
硬性规则
- 令牌不超过容量上限；缺口等待=ceil(缺口/速率)；速率0返回None不除零。
- 固定窗口 count≥limit 拒绝，剩余 max(limit-count,0)。
- 退避 min(base·2^(n-1),60)。
示例
token_bucket(10,1,5,2) → (False,2,3)
fixed_window(100,100) → (False,0)
retry_after_header(4) → 8
Python 引擎（可独立运行）
# -*- coding: utf-8 -*-
import math
def token_bucket(capacity, refill_rate, tokens_needed, current_tokens):
"""令牌桶：返回(是否放行, 剩余令牌, 需等待秒数)。refill_rate=令牌/秒。"""
current_tokens = min(current_tokens, capacity)
if tokens_needed <= 0:
return True, current_tokens, 0
if current_tokens >= tokens_needed:
return True, current_tokens - tokens_needed, 0
if refill_rate <= 0:
return False, current_tokens, None
wait = math.ceil((tokens_needed - current_tokens) / refill_rate)
return False, current_tokens, wait
def fixed_window(limit, count_in_window):
"""固定窗口：count 达到 limit 即拒绝，返回(是否放行, 剩余配额)。"""
if limit <= 0:
return False, 0
remaining = limit - count_in_window
return (remaining > 0, max(remaining, 0))
def retry_after_header(attempt, base=1, cap=60):
"""Retry-After 指数退避秒数，封顶 cap。"""
return min(base * (2 ** max(attempt - 1, 0)), cap)
