dekun
dbc86a1ce6
OKX/Binance LIVE share half_open and option_closed_perp_pending repair paths; private REST throttles default to 1s and are tunable in settings. Co-authored-by: Cursor <cursoragent@cursor.com>
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""实盘限流 / 退避单测。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from app.live.rate_limit import (
|
|
LiveRetryGate,
|
|
RateLimitError,
|
|
TradeThrottle,
|
|
get_throttle,
|
|
is_rate_limit_error,
|
|
)
|
|
|
|
|
|
def test_is_rate_limit_error() -> None:
|
|
assert is_rate_limit_error("HTTP 429 too many")
|
|
assert is_rate_limit_error("binance eapi cooldown 12s")
|
|
assert is_rate_limit_error(RateLimitError("x", retry_after=5))
|
|
assert not is_rate_limit_error("保证金不足")
|
|
|
|
|
|
def test_trade_throttle_cooldown() -> None:
|
|
t = TradeThrottle("ut_throttle", min_interval_sec=0.01, cooldown_429_sec=0.3)
|
|
t.before_request()
|
|
t.mark_http(429)
|
|
try:
|
|
t.before_request()
|
|
assert False, "expected RateLimitError"
|
|
except RateLimitError as e:
|
|
assert e.retry_after > 0
|
|
time.sleep(0.35)
|
|
t.before_request() # 冷却结束后可继续
|
|
|
|
|
|
def test_get_throttle_singleton() -> None:
|
|
a = get_throttle("ut_shared_x", min_interval_sec=0.01)
|
|
b = get_throttle("ut_shared_x")
|
|
assert a is b
|
|
|
|
|
|
def test_live_retry_gate_backoff() -> None:
|
|
g = LiveRetryGate(base_sec=0.05, max_sec=0.2, rate_limit_min_sec=0.1, trip_after=100)
|
|
assert g.allow("k")[0] is True
|
|
d1 = g.fail("k")
|
|
assert d1 >= 0.05
|
|
ok, left = g.allow("k")
|
|
assert ok is False
|
|
assert left > 0
|
|
time.sleep(d1 + 0.02)
|
|
assert g.allow("k")[0] is True
|
|
g.success("k")
|
|
assert g.fails("k") == 0
|
|
|
|
|
|
def test_live_retry_gate_rate_limited_longer() -> None:
|
|
g = LiveRetryGate(base_sec=0.01, rate_limit_min_sec=0.2)
|
|
d = g.fail("rl", rate_limited=True)
|
|
assert d >= 0.2
|