fix: harden market data and fast analysis safeguards

Author: Henry

backend_api_python/app/data_sources/crypto.py

@@ -10,6 +10,7 @@
 import ccxt
 
 from app.data_sources.base import BaseDataSource, TIMEFRAME_SECONDS
+from app.data_sources.errors import MarketDataFailure, classify_market_data_failure
 from app.utils.logger import get_logger
 from app.config import CCXTConfig, APIKeys
 
@@ -154,6 +155,7 @@ def __init__(self):
         self._scoped_market_type = "spot"
         self._preferred_public_exchange_id = ""
         self._markets_load_lock = threading.Lock()
+        self._failure_local = threading.local()
         default_ex = (CCXTConfig.DEFAULT_EXCHANGE or "binance").strip().lower()
         if default_ex == "huobi":
             default_ex = "htx"
@@ -178,6 +180,7 @@ def for_exchange(cls, exchange_id: str, market_type: str = "swap") -> "CryptoDat
         inst._scoped_market_type = mt
         inst._preferred_public_exchange_id = ""
         inst._markets_load_lock = threading.Lock()
+        inst._failure_local = threading.local()
         inst._init_ccxt_exchange(ccxt_id, options)
         _SCOPED_INSTANCES[cache_key] = inst
         logger.info(
@@ -227,10 +230,43 @@ def for_public_market(
         inst._scoped_market_type = mt
         inst._preferred_public_exchange_id = ""
         inst._markets_load_lock = threading.Lock()
+        inst._failure_local = threading.local()
         inst._init_ccxt_exchange(ccxt_id, options)
         _PUBLIC_MARKET_INSTANCES[cache_key] = inst
         return inst
 
+    def _clear_last_failure(self) -> None:
+        local = getattr(self, "_failure_local", None)
+        if local is None:
+            local = threading.local()
+            self._failure_local = local
+        local.value = None
+
+    def _set_last_failure(
+        self,
+        error: Any,
+        *,
+        symbol: str,
+        timeframe: str,
+    ) -> MarketDataFailure:
+        failure = classify_market_data_failure(
+            error,
+            exchange_id=getattr(self, "_scoped_exchange_id", "") or getattr(self.exchange, "id", ""),
+            market_type=getattr(self, "_scoped_market_type", "") or "spot",
+            symbol=symbol,
+            timeframe=timeframe,
+        )
+        local = getattr(self, "_failure_local", None)
+        if local is None:
+            local = threading.local()
+            self._failure_local = local
+        local.value = failure
+        return failure
+
+    def get_last_failure(self) -> Optional[MarketDataFailure]:
+        local = getattr(self, "_failure_local", None)
+        return getattr(local, "value", None) if local is not None else None
+
     def _init_ccxt_exchange(self, ccxt_exchange_id: str, options: Optional[Dict[str, Any]] = None) -> None:
         config: Dict[str, Any] = {
             "timeout": CCXTConfig.TIMEOUT,
@@ -503,6 +539,7 @@ def get_kline(
         after_time: Optional[int] = None,
     ) -> List[Dict[str, Any]]:
         """获取加密货币K线数据"""
+        self._clear_last_failure()
         klines = []
         symbol_pair = ""
 
@@ -548,6 +585,11 @@ def get_kline(
             if exchange_timeframes and ccxt_timeframe not in exchange_timeframes:
                 picked = self._pick_resample_source(ccxt_timeframe, exchange_timeframes)
                 if picked is None:
+                    self._set_last_failure(
+                        f"Unsupported timeframe {ccxt_timeframe} on {self.exchange.id}",
+                        symbol=symbol,
+                        timeframe=timeframe,
+                    )
                     logger.warning(
                         f"Exchange '{self.exchange.id}' cannot serve timeframe '{ccxt_timeframe}' "
                         f"and no finer supported granularity is available for resampling. "
@@ -568,10 +610,18 @@ def get_kline(
             symbol_pair = self._symbol_for_scoped_market(symbol)
 
             if not symbol_pair:
+                self._set_last_failure(
+                    f"Invalid symbol: {symbol}", symbol=symbol, timeframe=timeframe
+                )
                 logger.warning(f"Failed to normalize symbol for K-line: {symbol}")
                 raise _PublicKlineUnavailable
 
             if self._is_invalid_symbol_cached(symbol_pair):
+                self._set_last_failure(
+                    f"Symbol not found (cached): {symbol_pair}",
+                    symbol=symbol_pair,
+                    timeframe=timeframe,
+                )
                 raise _PublicKlineUnavailable
 
             ohlcv = self._fetch_ohlcv(
@@ -580,6 +630,12 @@ def get_kline(
             )
 
             if not ohlcv:
+                if self.get_last_failure() is None:
+                    self._set_last_failure(
+                        "Exchange returned no K-line rows",
+                        symbol=symbol_pair,
+                        timeframe=timeframe,
+                    )
                 logger.warning(f"CCXT returned no K-lines: {symbol_pair}")
                 raise _PublicKlineUnavailable
 
@@ -628,8 +684,14 @@ def get_kline(
                     pass
 
         except _PublicKlineUnavailable:
-            pass
+            if self.get_last_failure() is None:
+                self._set_last_failure(
+                    "No usable market data",
+                    symbol=symbol_pair or symbol,
+                    timeframe=timeframe,
+                )
         except Exception as e:
+            self._set_last_failure(e, symbol=symbol_pair or symbol, timeframe=timeframe)
             logger.error(f"Failed to fetch crypto K-lines {symbol}: {str(e)}")
             import traceback
             logger.error(traceback.format_exc())
@@ -876,6 +938,7 @@ def _fetch_ohlcv(
         except Exception as e:
             if _is_symbol_not_found_error(e):
                 self._mark_invalid_symbol(symbol_pair, e)
+                self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
                 return []
             partial_rows = locals().get("all_ohlcv") or []
             if partial_rows:
@@ -889,6 +952,7 @@ def _fetch_ohlcv(
                 by_ts = {int(row[0]): row for row in partial_rows if row and len(row) >= 6}
                 return sorted(by_ts.values(), key=lambda row: row[0])
             logger.warning(f"CCXT fetch_ohlcv failed: {str(e)}; trying fallback")
+            self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
             return self._fetch_ohlcv_fallback(
                 symbol_pair, ccxt_timeframe, limit, before_time, timeframe, after_time
             )
@@ -935,7 +999,9 @@ def _fetch_ohlcv_fallback(
         except Exception as e:
             if _is_symbol_not_found_error(e):
                 self._mark_invalid_symbol(symbol_pair, e)
+                self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
                 return []
+            self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
             logger.warning("Requested-window fallback failed for %s: %s", symbol_pair, str(e))
 
         try:
@@ -957,4 +1023,5 @@ def _fetch_ohlcv_fallback(
                 self._mark_invalid_symbol(symbol_pair, e)
             else:
                 logger.error("Recent-candle fallback also failed for %s: %s", symbol_pair, str(e))
+            self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
         return []

backend_api_python/app/data_sources/errors.py

@@ -1,14 +1,127 @@
 from __future__ import annotations
 
+import re
+from dataclasses import dataclass
+from typing import Any, Mapping
+
 
 class DataSourceError(Exception):
     """Base class for data source related errors."""
 
 
+@dataclass(frozen=True)
+class MarketDataFailure:
+    """Structured public-market-data failure safe to expose to the UI."""
+
+    code: str
+    message: str
+    technical_detail: str = ""
+    exchange_id: str = ""
+    market_type: str = ""
+    symbol: str = ""
+    timeframe: str = ""
+    retryable: bool = True
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "code": self.code,
+            "message": self.message,
+            "technical_detail": self.technical_detail,
+            "exchange_id": self.exchange_id,
+            "market_type": self.market_type,
+            "symbol": self.symbol,
+            "timeframe": self.timeframe,
+            "retryable": self.retryable,
+        }
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "MarketDataFailure":
+        return cls(
+            code=str(value.get("code") or "no_market_data"),
+            message=str(value.get("message") or "No usable market data is available."),
+            technical_detail=str(value.get("technical_detail") or "")[:500],
+            exchange_id=str(value.get("exchange_id") or ""),
+            market_type=str(value.get("market_type") or ""),
+            symbol=str(value.get("symbol") or ""),
+            timeframe=str(value.get("timeframe") or ""),
+            retryable=bool(value.get("retryable", True)),
+        )
+
+
+class MarketDataUnavailableError(DataSourceError):
+    """Carries a categorized failure through strategy-data loading layers."""
+
+    def __init__(self, failure: MarketDataFailure):
+        self.failure = failure
+        super().__init__(f"marketData.{failure.code}")
+
+
+def classify_market_data_failure(
+    error: Any,
+    *,
+    exchange_id: str = "",
+    market_type: str = "",
+    symbol: str = "",
+    timeframe: str = "",
+) -> MarketDataFailure:
+    """Map provider-specific errors to a small stable frontend contract."""
+    detail = str(error or "").strip()
+    detail = re.sub(
+        r"(?i)(https?://)([^/@\s:]+):([^/@\s]+)@",
+        r"\1***:***@",
+        detail,
+    )
+    text = detail.lower()
+    if any(token in text for token in (
+        "451",
+        "restricted location",
+        "legal reasons",
+        "region restricted",
+        "block access from your country",
+        "blocked access from your country",
+    )):
+        code = "region_restricted"
+        message = "The exchange market-data endpoint is unavailable in this region."
+        retryable = False
+    elif any(token in text for token in ("proxyerror", "proxy error", "proxyconnect", "proxy connection", "tunnel connection", "socks")):
+        code = "proxy_failure"
+        message = "The market-data proxy could not connect to the exchange."
+        retryable = True
+    elif any(token in text for token in ("does not have market symbol", "symbol not found", "invalid symbol", "market does not exist", "trading pair not found")):
+        code = "symbol_not_found"
+        message = "The trading pair does not exist for this exchange and market type."
+        retryable = False
+    elif any(token in text for token in ("429", "too many requests", "rate limit", "ratelimit")):
+        code = "rate_limited"
+        message = "The exchange rate limit was reached. Market data will be retried."
+        retryable = True
+    elif any(token in text for token in ("timeout", "timed out", "network error", "connection reset", "connection refused", "exchange not available", "service unavailable", "502", "503", "504")):
+        code = "exchange_unavailable"
+        message = "The exchange market-data service is temporarily unreachable."
+        retryable = True
+    elif "timeframe" in text and any(token in text for token in ("unsupported", "not support", "cannot serve")):
+        code = "unsupported_timeframe"
+        message = "This exchange does not provide the requested K-line timeframe."
+        retryable = False
+    else:
+        code = "no_market_data"
+        message = "The exchange returned no usable market data."
+        retryable = True
+    return MarketDataFailure(
+        code=code,
+        message=message,
+        technical_detail=detail[:500],
+        exchange_id=str(exchange_id or "").strip().lower(),
+        market_type=str(market_type or "").strip().lower(),
+        symbol=str(symbol or ""),
+        timeframe=str(timeframe or ""),
+        retryable=retryable,
+    )
+
+
 class UnsupportedMarketError(DataSourceError):
     """Raised when a requested market type is not supported by DataSourceFactory."""
 
     def __init__(self, market: str):
         self.market = str(market or "")
         super().__init__(f"Unsupported market type: {self.market}")
-

backend_api_python/app/data_sources/factory.py

@@ -8,7 +8,11 @@
 from typing import Dict, List, Any, Optional
 
 from app.data_sources.base import BaseDataSource
-from app.data_sources.errors import UnsupportedMarketError
+from app.data_sources.errors import (
+    MarketDataFailure,
+    UnsupportedMarketError,
+    classify_market_data_failure,
+)
 from app.utils.logger import get_logger
 from app.utils.resource_guard import (
     ResourceExhaustedError,
@@ -237,15 +241,53 @@ def get_kline(
         Returns:
             K线数据列表
         """
+        rows, _failure = cls.get_kline_with_diagnostics(
+            market=market,
+            symbol=symbol,
+            timeframe=timeframe,
+            limit=limit,
+            before_time=before_time,
+            after_time=after_time,
+            exchange_id=exchange_id,
+            market_type=market_type,
+        )
+        return rows
+
+    @classmethod
+    def get_kline_with_diagnostics(
+        cls,
+        *,
+        market: str,
+        symbol: str,
+        timeframe: str,
+        limit: int,
+        before_time: Optional[int] = None,
+        after_time: Optional[int] = None,
+        exchange_id: Optional[str] = None,
+        market_type: Optional[str] = None,
+    ) -> tuple[List[Dict[str, Any]], Optional[MarketDataFailure]]:
+        """Fetch K-lines and retain a structured provider failure when rows are empty."""
         m = cls.normalize_market(market or "")
         try:
             assert_fd_available(f"market-data kline {m}:{symbol}")
             source = cls._resolve_source(m, exchange_id=exchange_id, market_type=market_type)
             klines = source.get_kline(symbol, timeframe, limit, before_time, after_time)
-            
+
             klines.sort(key=lambda x: x['time'])
-            
-            return klines
+            failure = None
+            if not klines:
+                get_last_failure = getattr(source, "get_last_failure", None)
+                if callable(get_last_failure):
+                    failure = get_last_failure()
+                if failure is None:
+                    failure = classify_market_data_failure(
+                        "Exchange returned no K-line rows",
+                        exchange_id=exchange_id or getattr(getattr(source, "exchange", None), "id", ""),
+                        market_type=market_type or "",
+                        symbol=symbol,
+                        timeframe=timeframe,
+                    )
+            return klines, failure
         except ResourceExhaustedError as e:
             cls._log_limited(
                 "error",
@@ -255,7 +297,13 @@ def get_kline(
                 symbol,
                 str(e),
             )
-            return []
+            return [], classify_market_data_failure(
+                e,
+                exchange_id=exchange_id or "",
+                market_type=market_type or "",
+                symbol=symbol,
+                timeframe=timeframe,
+            )
         except Exception as e:
             if is_fd_exhaustion(e):
                 mark_fd_exhausted(e)
@@ -268,7 +316,13 @@ def get_kline(
                 m,
                 str(e),
             )
-            return []
+            return [], classify_market_data_failure(
+                e,
+                exchange_id=exchange_id or "",
+                market_type=market_type or "",
+                symbol=symbol,
+                timeframe=timeframe,
+            )
 
     @classmethod
     def _resolve_source(