defunct9

```python
import time
import threading
from typing import Tuple

# 定义三色标记常量
COLOR_GREEN = "GREEN" # C 桶充足，完美放行
COLOR_YELLOW = "YELLOW" # C 桶不足但 E 桶充足，超额放行（突发）
COLOR_RED = "RED" # 双桶皆不足，拒绝请求

class DualBucketThreeColorLimiter:
def __init__(self, bc: float, be: float, cir: float):
"""
:param bc: Committed Burst Size (C 桶容量，承诺突发量)
:param be: Excess Burst Size (E 桶容量，超额突发量)
:param cir: Committed Information Rate (承诺信息速率，每秒恢复的 Token 数)
"""
self.bc = float(bc)
self.be = float(be)
self.cir = float(cir)

# 初始状态：双桶皆满
self.tokens_c = float(bc)
self.tokens_e = float(be)

self.last_tick = time.monotonic()
self._lock = threading.Lock()

def consume(self, weight: float = 1.0) -> Tuple[bool, str]:
"""
尝试消费指定权重的令牌，返回 (是否放行, 流量颜色)
遵循 MEF 10 / RFC 2698 标准的双速率三色标记算法逻辑变体
"""
with self._lock:
now = time.monotonic()
delta = now - self.last_tick
self.last_tick = now

# 1. 延迟计算：向桶内补充令牌
delta_tokens = delta * self.cir

# C 桶溢出的令牌会流入 E 桶
overflow_c = max(0.0, (self.tokens_c + delta_tokens) - self.bc)
self.tokens_c = min(self.bc, self.tokens_c + delta_tokens)
self.tokens_e = min(self.be, self.tokens_e + overflow_c)

# 2. 三色评估与消费逻辑
# 情况 🟢：C 桶令牌足够
if self.tokens_c >= weight:
self.tokens_c -= weight
return True, COLOR_GREEN

# 情况 🟡：C 桶不够，但 E 桶足够（借用突发额度）
elif self.tokens_e >= weight:
self.tokens_e -= weight
return True, COLOR_YELLOW

# 情况 🔴：双桶都不够
else:
return False, COLOR_RED

def sync_from_header(self, server_c_remaining: float, server_e_remaining: float):
"""
量化实战进阶：从交易所 Response Header 权威同步双桶状态
"""
with self._lock:
self.tokens_c = min(self.bc, server_c_remaining)
self.tokens_e = min(self.be, server_e_remaining)
self.last_tick = time.monotonic()
```