Handle blocked market data endpoints

app/monitors/market_move.py

@@ -331,25 +331,49 @@ class PublicMarketDataClient:
         self.session = requests.Session()
         self.session.headers.update(legacy.REQ_HEADERS)
         self.last_error_log = {}
+        self.blocked_until = {}
         self._upbit_krw_usdt = None
 
     def update_config(self, config):
         self.config = normalize_config(config)
 
-    def _log_error_once(self, key, message, cooldown=600):
+    def _log_error_once(self, key, message, cooldown=600, level="error"):
         now = time.time()
         if now - self.last_error_log.get(key, 0) >= cooldown:
             self.last_error_log[key] = now
-            self.logger(message, level="error")
+            self.logger(message, level=level)
+
+    def _blocked_key(self, method, base, path):
+        return method, base.rstrip("/"), path
+
+    def _block_for_status(self, method, base, path, status_code):
+        if status_code not in (403, 451, 429):
+            return
+        cooldown = 3600 if status_code in (403, 451) else 120
+        self.blocked_until[self._blocked_key(method, base, path)] = time.time() + cooldown
+        mins = max(1, int(cooldown / 60))
+        self._log_error_once(
+            (base, path, status_code),
+            f"{base}{path} 状态码 {status_code}，临时跳过 {mins} 分钟并使用其他交易所行情",
+            cooldown=cooldown,
+            level="info",
+        )
+
+    def _is_blocked(self, method, base, path):
+        return time.time() < self.blocked_until.get(self._blocked_key(method, base, path), 0)
 
     def _get_json(self, base, path, params=None, timeout=12):
+        if self._is_blocked("GET", base, path):
+            return None
         url = base.rstrip("/") + path
         last_exc = None
         for attempt in range(2):
             try:
                 resp = self.session.get(url, params=params, timeout=timeout)
                 if resp.status_code != 200:
-                    self._log_error_once((base, path, resp.status_code), f"{base}{path} 状态码 {resp.status_code}")
+                    self._block_for_status("GET", base, path, resp.status_code)
+                    if resp.status_code not in (403, 451, 429):
+                        self._log_error_once((base, path, resp.status_code), f"{base}{path} 状态码 {resp.status_code}")
                     return None
                 return resp.json()
             except Exception as exc:
@@ -360,13 +384,17 @@ class PublicMarketDataClient:
         return None
 
     def _post_json(self, base, path, body=None, timeout=12):
+        if self._is_blocked("POST", base, path):
+            return None
         url = base.rstrip("/") + path
         last_exc = None
         for attempt in range(2):
             try:
                 resp = self.session.post(url, json=body or {}, timeout=timeout)
                 if resp.status_code != 200:
-                    self._log_error_once((base, path, resp.status_code), f"{base}{path} 状态码 {resp.status_code}")
+                    self._block_for_status("POST", base, path, resp.status_code)
+                    if resp.status_code not in (403, 451, 429):
+                        self._log_error_once((base, path, resp.status_code), f"{base}{path} 状态码 {resp.status_code}")
                     return None
                 return resp.json()
             except Exception as exc:

coinpush.py

@@ -1373,17 +1373,22 @@ class CoreMarketAggregator:
 
     def _get_json(self, base, path, params=None, method="GET", body=None, timeout=6):
         url = base.rstrip("/") + path
-        try:
-            if method == "POST":
-                resp = self.session.post(url, json=body or {}, timeout=timeout)
-            else:
-                resp = self.session.get(url, params=params, timeout=timeout)
-            if resp.status_code != 200:
-                return None
-            return resp.json()
-        except Exception as exc:
-            self._log_error_once((base, path, "exception"), f"⚠️ 聚合行情 {base}{path} 异常: {exc}")
-            return None
+        last_exc = None
+        for attempt in range(2):
+            try:
+                if method == "POST":
+                    resp = self.session.post(url, json=body or {}, timeout=timeout)
+                else:
+                    resp = self.session.get(url, params=params, timeout=timeout)
+                if resp.status_code != 200:
+                    return None
+                return resp.json()
+            except Exception as exc:
+                last_exc = exc
+                if attempt == 0:
+                    time.sleep(0.2)
+        self._log_error_once((base, path, "exception"), f"⚠️ 聚合行情 {base}{path} 异常: {last_exc}")
+        return None
 
     def aggregate(self, coin_name, coin_cfg, snapshot=None):
         symbol = (coin_cfg or {}).get("symbol")