Fix Binance USD-M perpetual kline data and signal/live safeguards

backend_api_python/app/data_sources/crypto.py

@@ -268,6 +268,69 @@
                 return f"{normalized}:{quote}"
         return normalized
 
+    def _is_binance_btc_usdm_request(self, symbol_pair: str) -> bool:
+        return (
+            not bool(getattr(self, "_allow_public_fallback", False))
+            and
+            str(getattr(self.exchange, "id", "") or "").strip().lower() == "binance"
+            and str(getattr(self, "_scoped_market_type", "") or "").strip().lower() == "swap"
+            and str(symbol_pair or "").strip().upper() == "BTC/USDT:USDT"
+        )
+
+    def _validate_binance_btc_usdm_market(self, symbol_pair: str) -> Dict[str, Any]:
+        """Fail closed unless BTC/USDT resolves to the active linear USDⓈ-M perpetual."""
+        if not self._is_binance_btc_usdm_request(symbol_pair):
+            raise ValueError("Request is not scoped to Binance BTC/USDT USDⓈ-M swap")
+        if str(getattr(self.exchange, "market_type", "") or "").strip().lower() != "swap":
+            raise ValueError("Binance BTC/USDT perpetual requires the swap market adapter")
+        if not self._ensure_markets_loaded():
+            raise ValueError("Unable to validate Binance BTC/USDT perpetual market metadata")
+
+        markets = self._markets_cache or {}
+        market = markets.get("BTC/USDT:USDT") or markets.get("BTC/USDT")
+        if not isinstance(market, dict):
+            raise ValueError("Binance BTCUSDT USDⓈ-M perpetual is not present in exchange metadata")
+        info = market.get("info") or {}
+        valid = (
+            str(market.get("id") or "").upper() == "BTCUSDT"
+            and str(market.get("symbol") or "").upper() == "BTC/USDT:USDT"
+            and str(market.get("base") or "").upper() == "BTC"
+            and str(market.get("quote") or "").upper() == "USDT"
+            and str(market.get("settle") or "").upper() == "USDT"
+            and bool(market.get("active"))
+            and market.get("spot") is False
+            and bool(market.get("swap"))
+            and bool(market.get("contract"))
+            and bool(market.get("linear"))
+            and str(info.get("contractType") or "").upper() == "PERPETUAL"
+            and str(info.get("status") or "").upper() == "TRADING"
+            and str(info.get("quoteAsset") or "").upper() == "USDT"
+            and str(info.get("marginAsset") or "").upper() == "USDT"
+        )
+        if not valid:
+            raise ValueError("Binance BTCUSDT market metadata is not an active linear USDⓈ-M perpetual")
+        return market
+
+    @staticmethod
+    def _filter_closed_ohlcv(
+        ohlcv: List,
+        timeframe: str,
+        server_time_ms: int,
+    ) -> List:
+        """Keep unique, ordered candles whose complete interval has elapsed."""
+        duration_seconds = TIMEFRAME_SECONDS.get(timeframe)
+        if not duration_seconds:
+            raise ValueError(f"Unsupported timeframe for closed-candle validation: {timeframe}")
+        interval_ms = int(duration_seconds) * 1000
+        by_timestamp = {}
+        for candle in ohlcv or []:
+            if not candle or len(candle) < 6:
+                continue
+            timestamp_ms = int(candle[0])
+            if timestamp_ms + interval_ms <= int(server_time_ms):
+                by_timestamp[timestamp_ms] = candle
+        return [by_timestamp[timestamp] for timestamp in sorted(by_timestamp)]
+
     def _invalid_symbol_key(self, symbol_pair: str) -> str:
         exchange_id = getattr(self.exchange, 'id', '').lower()
         mt = getattr(self, "_scoped_market_type", "") or "spot"
@@ -581,6 +644,11 @@
                 logger.warning(f"Failed to normalize symbol for K-line: {symbol}")
                 raise _PublicKlineUnavailable
 
+            server_time_ms = None
+            if self._is_binance_btc_usdm_request(symbol_pair):
+                self._validate_binance_btc_usdm_market(symbol_pair)
+                server_time_ms = self.exchange.fetch_server_time()
+
             if self._is_invalid_symbol_cached(symbol_pair):
                 self._set_last_failure(
                     f"Symbol not found (cached): {symbol_pair}",
@@ -594,6 +662,13 @@
                 before_time, fetch_qd_timeframe, after_time,
             )
 
+            if ohlcv and server_time_ms is not None:
+                ohlcv = self._filter_closed_ohlcv(
+                    ohlcv,
+                    fetch_qd_timeframe,
+                    server_time_ms,
+                )
+
             if not ohlcv:
                 if self.get_last_failure() is None:
                     self._set_last_failure(
@@ -675,8 +750,8 @@
             if klines:
                 try:
                     from datetime import datetime as _dt
                     first_ts = _dt.utcfromtimestamp(klines[0]['time']).isoformat()
                     last_ts = _dt.utcfromtimestamp(klines[-1]['time']).isoformat()
                     logger.info(
                         f"[CryptoKline] {symbol} {timeframe} returned {len(klines)} candles, "
                         f"utc_range={first_ts}~{last_ts}, limit={limit}, before_time={before_time}"

backend_api_python/app/data_sources/native_crypto.py

@@ -454,7 +454,7 @@
                 open_price=row.get("open"),
                 high=row.get("high24h"),
                 low=row.get("low24h"),
                 percentage=percentage,
                 quote_volume=row.get("quoteVolume"),
                 timestamp=row.get("ts") or payload.get("requestTime"),
             )
@@ -607,6 +607,20 @@
         unique = {int(row[0]): row for row in parsed if row and int(row[0]) >= lower_bound}
         return sorted(unique.values(), key=lambda row: row[0])[:limit]
 
+    def fetch_server_time(self) -> int:
+        """Return exchange UTC time from a public market-data endpoint."""
+        if self.id != "binance":
+            raise NotImplementedError("Public server time is only implemented for Binance")
+        if self.market_type == "swap":
+            url = "https://fapi.binance.com/fapi/v1/time"
+        else:
+            url = "https://api.binance.com/api/v3/time"
+        payload = self._get(url)
+        server_time = int(payload.get("serverTime") or 0)
+        if server_time <= 0:
+            raise NativeCryptoAPIError("Binance public server time response was invalid")
+        return server_time
+
 
 def create_native_crypto_client(
     exchange_id: str,

backend_api_python/app/routes/kline.py

@@ -9,7 +9,11 @@
 from app.services.kline import KlineService
 from app.utils.logger import get_logger
 from app.services.market.watchlist import validate_watchlist_pair
-from app.services.market_context import MarketContext, SUPPORTED_CRYPTO_EXCHANGE_IDS
+from app.services.market_context import (
+    MarketContext,
+    SUPPORTED_CRYPTO_EXCHANGE_IDS,
+    parse_crypto_identifier,
+)
 from app.utils.request_guard import RequestGuardError, cache_key, guarded_cached
 
 logger = get_logger(__name__)
@@ -83,7 +87,14 @@ def get_kline():
         exchange_id = (request.args.get('exchange_id') or request.args.get('exchangeId') or '').strip() or None
         market_type = (request.args.get('market_type') or request.args.get('marketType') or '').strip() or None
         instrument_id = (request.args.get('instrument_id') or request.args.get('instrumentId') or '').strip()
-        if market == 'Crypto':
+        if not symbol:
+            return jsonify({
+                'code': 0,
+                'msg': 'Missing symbol parameter',
+                'data': None
+            }), 400
+
+        try:
             context = MarketContext.from_mapping({
                 'market': market,
                 'symbol': symbol,
@@ -92,39 +103,55 @@ def get_kline():
                 'instrument_id': instrument_id,
                 'timeframe': timeframe,
             })
+        except ValueError as e:
+            return jsonify({'code': 0, 'msg': str(e), 'data': None}), 400
+
+        service_symbol = context.symbol if market == 'Crypto' else symbol
+        service_exchange_id = context.exchange_id if market == 'Crypto' else exchange_id
+        service_market_type = context.market_type if market == 'Crypto' else market_type
+        if market == 'Crypto':
             if context.exchange_id not in SUPPORTED_CRYPTO_EXCHANGE_IDS:
                 return jsonify({'code': 0, 'msg': 'Unsupported crypto exchange', 'data': None}), 400
+
+            try:
+                parsed_symbol, suffix_market_type, suffix_exchange_id = parse_crypto_identifier(symbol)
+            except ValueError as e:
+                return jsonify({'code': 0, 'msg': str(e), 'data': None}), 400
+            if (
+                parsed_symbol == 'BTC/USDT'
+                and suffix_market_type == 'swap'
+                and not suffix_exchange_id
+                and context.exchange_id != 'binance'
+            ):
+                return jsonify({
+                    'code': 0,
+                    'msg': 'Crypto:BTC/USDT@swap is mapped to Binance USDⓈ-M; use an explicit venue-qualified symbol for another exchange',
+                    'data': None,
+                }), 400
 
         if before_time:
             before_time = int(before_time)
 
-        if not symbol:
-            return jsonify({
-                'code': 0,
-                'msg': 'Missing symbol parameter',
-                'data': None
-            }), 400
-
-        validation_err = validate_watchlist_pair(market, symbol)
+        validation_err = validate_watchlist_pair(market, service_symbol)
         if validation_err:
             return jsonify({'code': 0, 'msg': validation_err, 'data': None}), 400
 
-        logger.info(f"Requesting K-lines: {market}:{symbol}, timeframe={timeframe}, limit={limit}")
+        logger.info(f"Requesting K-lines: {market}:{service_symbol}, timeframe={timeframe}, limit={limit}")
 
         policy = _guard_policy(timeframe, limit, before_time)
         klines = guarded_cached(
             cache_key(
-                "indicator_kline", market, symbol, timeframe, limit, before_time or "",
-                exchange_id or "", market_type or "", instrument_id,
+                "indicator_kline", market, service_symbol, timeframe, limit, before_time or "",
+                service_exchange_id or "", service_market_type or "", instrument_id,
             ),
             lambda: kline_service.get_kline(
                 market=market,
-                symbol=symbol,
+                symbol=service_symbol,
                 timeframe=timeframe,
                 limit=limit,
                 before_time=before_time,
-                exchange_id=exchange_id,
-                market_type=market_type,
+                exchange_id=service_exchange_id,
+                market_type=service_market_type,
                 instrument_id=instrument_id,
             ),
             ttl_sec=policy['ttl_sec'],
@@ -147,14 +174,6 @@ def get_kline():
                 'hint': 'tiingo_subscription' if (market == 'Forex' and timeframe == '1m') else None
             })
 
-        context = MarketContext.from_mapping({
-            'market': market,
-            'symbol': symbol,
-            'exchange_id': exchange_id,
-            'market_type': market_type,
-            'instrument_id': instrument_id,
-            'timeframe': timeframe,
-        })
         return jsonify({
             'code': 1,
             'msg': 'success',

backend_api_python/app/services/market_context.py

@@ -35,11 +35,58 @@ def normalize_market_type(value: Any, *, market: str = "") -> str:
     return market_type if market_type in CRYPTO_MARKET_TYPES else "spot"
 
 
+def parse_crypto_identifier(value: Any) -> tuple[str, str | None, str | None]:
+    """Parse a canonical crypto identifier into symbol, product type, and venue.
+
+    QuantDinger identifiers use ``@swap`` for a perpetual and may qualify the
+    venue as ``@binance:swap``. A CCXT-style settlement suffix such as
+    ``:USDT`` remains part of the pair until it is validated and removed.
+    """
+    identifier = str(value or "").strip().upper()
+    if identifier.startswith("CRYPTO:"):
+        identifier = identifier[len("CRYPTO:"):]
+
+    suffix_market_type = None
+    suffix_exchange_id = None
+    if "@" in identifier:
+        identifier, suffix = identifier.rsplit("@", 1)
+        if "@" in identifier or not identifier or not suffix:
+            raise ValueError(f"Invalid crypto market identifier: {value}")
+        if ":" in suffix:
+            suffix_exchange_id, suffix = suffix.rsplit(":", 1)
+            if not suffix_exchange_id:
+                raise ValueError(f"Invalid crypto exchange in market identifier: {value}")
+            suffix_exchange_id = normalize_exchange_id(suffix_exchange_id)
+        suffix_aliases = {
+            "SPOT": "spot",
+            "SWAP": "swap",
+            "PERP": "swap",
+            "PERPETUAL": "swap",
+            "FUTURE": "swap",
+            "FUTURES": "swap",
+        }
+        suffix_market_type = suffix_aliases.get(suffix)
+        if suffix_market_type is None:
+            raise ValueError(f"Unsupported crypto product suffix: @{suffix}")
+
+    identifier = identifier.strip()
+    if ":" in identifier:
+        pair, settle = identifier.rsplit(":", 1)
+        if "/" in pair:
+            quote = pair.rsplit("/", 1)[1]
+            if settle != quote:
+                raise ValueError(
+                    f"Crypto settlement suffix :{settle} does not match quote {quote}"
+                )
+            identifier = pair
+        else:
+            identifier = pair
+
+    return identifier.strip().upper(), suffix_market_type, suffix_exchange_id
+
+
 def canonical_crypto_symbol(value: Any) -> str:
-    symbol = str(value or "").strip().upper()
-    if ":" in symbol:
-        symbol = symbol.split(":", 1)[0]
-    return symbol
+    return parse_crypto_identifier(value)[0]
 
 
 @dataclass(frozen=True)
@@ -66,18 +113,45 @@ def from_mapping(
             or data.get("marketCategory")
             or ""
         ).strip()
-        symbol = canonical_crypto_symbol(data.get("symbol")) if market == "Crypto" else str(data.get("symbol") or "").strip().upper()
-        exchange_id = normalize_exchange_id(
+        raw_symbol = data.get("symbol")
+        if market == "Crypto":
+            symbol, suffix_market_type, suffix_exchange_id = parse_crypto_identifier(raw_symbol)
+        else:
+            symbol = str(raw_symbol or "").strip().upper()
+            suffix_market_type = None
+            suffix_exchange_id = None
+
+        explicit_exchange_id = normalize_exchange_id(
             data.get("exchange_id")
             or data.get("exchangeId")
             or data.get("exchange")
         )
-        market_type = normalize_market_type(
-            data.get("market_type") or data.get("marketType"),
-            market=market,
-        )
+        if suffix_exchange_id and explicit_exchange_id and suffix_exchange_id != explicit_exchange_id:
+            raise ValueError(
+                f"Market identifier exchange {suffix_exchange_id} conflicts with "
+                f"requested exchange {explicit_exchange_id}"
+            )
+        exchange_id = explicit_exchange_id or suffix_exchange_id or ""
+
+        explicit_market_type = data.get("market_type") or data.get("marketType")
+        normalized_market_type = normalize_market_type(explicit_market_type, market=market)
+        if (
+            suffix_market_type
+            and explicit_market_type
+            and normalized_market_type != suffix_market_type
+        ):
+            raise ValueError(
+                f"Market identifier product {suffix_market_type} conflicts with "
+                f"requested market_type {normalized_market_type}"
+            )
+        market_type = suffix_market_type or normalized_market_type
         if market == "Crypto" and apply_crypto_defaults and not exchange_id:
-            exchange_id = default_crypto_exchange_id()
+            # The unqualified BTC/USDT perpetual identifier is the public
+            # Binance USDⓈ-M instrument used by research and chart requests.
+            if suffix_market_type == "swap" and symbol == "BTC/USDT":
+                exchange_id = "binance"
+            else:
+                exchange_id = default_crypto_exchange_id()
         if market != "Crypto":
             exchange_id = ""
         return cls(