修复 load_markets 超时后狂打交易所:失败退避 90 秒,避免反复请求导致封 IP。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-13 00:47:47 +08:00
parent 39878de7fc
commit 702d9d3be8
5 changed files with 138 additions and 6 deletions
+40
View File
@@ -0,0 +1,40 @@
"""markets load backoff guard."""
from __future__ import annotations
import unittest
from lib.trade.markets_load_guard_lib import MarketsLoadGuard
class TestMarketsLoadGuard(unittest.TestCase):
def test_success_sets_loaded(self):
g = MarketsLoadGuard(backoff_sec=60.0, label="t")
calls = {"n": 0}
def load():
calls["n"] += 1
g.ensure(load)
self.assertTrue(g.loaded)
g.ensure(load)
self.assertEqual(calls["n"], 1)
def test_failure_backs_off(self):
g = MarketsLoadGuard(backoff_sec=60.0, label="t")
calls = {"n": 0}
def load():
calls["n"] += 1
raise TimeoutError("boom")
with self.assertRaises(TimeoutError):
g.ensure(load)
self.assertFalse(g.loaded)
self.assertEqual(calls["n"], 1)
with self.assertRaises(RuntimeError):
g.ensure(load)
self.assertEqual(calls["n"], 1)
if __name__ == "__main__":
unittest.main()