from __future__ import annotations import asyncio import json import threading import time from dataclasses import asdict from datetime import datetime from typing import Any import websockets from crypto_spot_bot.bybit import BybitClient, Instrument, websocket_subscribe_message from crypto_spot_bot.config import Settings from crypto_spot_bot.data_quality import analyze_symbol_quality, market_quality_snapshot from crypto_spot_bot.indicators import add_indicators from crypto_spot_bot.models import Candle, Ticker, utc_now from crypto_spot_bot.storage import Storage POPULAR_FALLBACK = [ "BTCUSDT", "ETHUSDT", "HYPEUSDT", "SOLUSDT", "XRPUSDT", "XPLUSDT", "WLDUSDT", "MNTUSDT", "HUSDT", "XAUTUSDT", "IPUSDT", "AAVEUSDT", ] def _float(value: Any, default: float = 0.0) -> float: try: return float(value) except (TypeError, ValueError): return default class MarketData: def __init__(self, settings: Settings, client: BybitClient, storage: Storage): self.settings = settings self.client = client self.storage = storage self.symbols: list[str] = [] self.instruments: dict[str, Instrument] = {} self.tickers: dict[str, Ticker] = {} self.candles: dict[str, list[Candle]] = {} self.trend_candles: dict[str, list[Candle]] = {} self.orderbook_top: dict[str, tuple[float, float]] = {} self.orderbook_metrics: dict[str, dict[str, Any]] = {} self.patterns: dict[str, dict[str, Any]] = {} self.forecasts: dict[str, dict[str, Any]] = {} self.shadow_forecasts: dict[str, dict[str, Any]] = {} self.last_rest_refresh_at: datetime | None = None self.last_ws_message_at: datetime | None = None self.ws_connected = False self._stop_event = asyncio.Event() self._refresh_lock = threading.Lock() self.rest_error_count = 0 self.last_rest_error = "" self.observation_samples = 0 self.last_observation_at: datetime | None = None self.observation_error_count = 0 self.last_observation_error = "" self._last_observation_monotonic: dict[str, float] = {} async def bootstrap(self) -> None: self.instruments = await asyncio.to_thread(self.client.instruments) if self.settings.symbols: self.symbols = [ symbol for symbol in self.settings.symbols if symbol in self.instruments and self.instruments[symbol].quote_coin == "USDT" ] elif self.settings.auto_select_symbols: self.symbols = await asyncio.to_thread( self.client.popular_spot_symbols, self.settings.top_symbols_count ) if not self.symbols: self.symbols = [ symbol for symbol in POPULAR_FALLBACK[: self.settings.top_symbols_count] if symbol in self.instruments ] self.storage.event("Торговые пары: " + ", ".join(self.symbols)) await asyncio.to_thread(self.refresh_rest, True) def refresh_rest(self, force_candles: bool = False) -> None: if not self._refresh_lock.acquire(blocking=False): return try: ticker_map = {ticker.symbol: ticker for ticker in self.client.spot_tickers()} for symbol in self.symbols: ticker = ticker_map.get(symbol) if ticker: self.tickers[symbol] = ticker try: if force_candles or _candles_due(self.candles.get(symbol, []), self.settings.base_interval): candles = self.client.klines( symbol=symbol, interval=self.settings.base_interval, limit=self.settings.kline_limit, ) candles = _closed_candles(candles, self.settings.base_interval) add_indicators(candles) self.candles[symbol] = candles if force_candles or _candles_due( self.trend_candles.get(symbol, []), self.settings.trend_interval ): trend_candles = self.client.klines( symbol=symbol, interval=self.settings.trend_interval, limit=self.settings.trend_kline_limit, ) trend_candles = _closed_candles(trend_candles, self.settings.trend_interval) add_indicators(trend_candles) self.trend_candles[symbol] = trend_candles bid, bid_size, ask, ask_size = self.client.orderbook_level_one(symbol) self._update_orderbook(symbol, bid, bid_size, ask, ask_size) except Exception as exc: self.rest_error_count += 1 self.last_rest_error = str(exc) self.storage.event(f"{symbol}: ошибка обновления REST данных: {exc}", "ERROR") self.last_rest_refresh_at = utc_now() if ticker_map: self.last_rest_error = "" finally: self._refresh_lock.release() async def websocket_loop(self) -> None: if not self.settings.websocket_enabled: return while not self._stop_event.is_set(): try: async with websockets.connect(self.settings.websocket_url, ping_interval=20) as ws: self.ws_connected = True await ws.send(websocket_subscribe_message(self.symbols, self.settings.base_interval)) self.storage.event("Поток данных Bybit подключен") async for raw in ws: self.last_ws_message_at = utc_now() self._handle_ws_message(raw) if self._stop_event.is_set(): break except asyncio.CancelledError: raise except Exception as exc: self.ws_connected = False self.storage.event(f"Поток данных Bybit отключен: {exc}", "WARN") await asyncio.sleep(5) self.ws_connected = False def stop(self) -> None: self._stop_event.set() def reset_stop(self) -> None: if self._stop_event.is_set(): self._stop_event = asyncio.Event() def _handle_ws_message(self, raw: str) -> None: try: message = json.loads(raw) except json.JSONDecodeError: return topic = str(message.get("topic", "")) data = message.get("data") if topic.startswith("tickers.") and isinstance(data, dict): self._handle_ticker(topic.split(".", 1)[1], data) elif topic.startswith("kline.") and isinstance(data, list): parts = topic.split(".") if len(parts) >= 3: self._handle_kline(parts[2], data) elif topic.startswith("orderbook.") and isinstance(data, dict): parts = topic.split(".") if len(parts) >= 3: self._handle_orderbook( parts[2], data, source_timestamp_ms=int(_float(message.get("ts"))), ) def _handle_ticker(self, symbol: str, data: dict[str, Any]) -> None: current = self.tickers.get(symbol) last_price = _float(data.get("lastPrice"), current.last_price if current else 0.0) if last_price <= 0: return self.tickers[symbol] = Ticker( symbol=symbol, last_price=last_price, bid=_float(data.get("bid1Price"), current.bid if current else 0.0), ask=_float(data.get("ask1Price"), current.ask if current else 0.0), turnover_24h=_float(data.get("turnover24h"), current.turnover_24h if current else 0.0), volume_24h=_float(data.get("volume24h"), current.volume_24h if current else 0.0), change_24h=_float(data.get("price24hPcnt")) * 100 if data.get("price24hPcnt") is not None else (current.change_24h if current else 0.0), ) def _handle_kline(self, symbol: str, rows: list[dict[str, Any]]) -> None: existing = self.candles.get(symbol, []) by_timestamp = {candle.timestamp: candle for candle in existing} for row in rows: start = int(row.get("start", 0)) if start <= 0: continue if not _is_closed_kline_row(row, self.settings.base_interval): continue by_timestamp[start] = Candle( timestamp=start, open=_float(row.get("open")), high=_float(row.get("high")), low=_float(row.get("low")), close=_float(row.get("close")), volume=_float(row.get("volume")), turnover=_float(row.get("turnover")), ) candles = sorted(by_timestamp.values(), key=lambda item: item.timestamp) candles = candles[-self.settings.kline_limit :] add_indicators(candles) self.candles[symbol] = candles def _handle_orderbook( self, symbol: str, data: dict[str, Any], source_timestamp_ms: int = 0, ) -> None: bids = data.get("b") or [] asks = data.get("a") or [] bid = _float(bids[0][0]) if bids else 0.0 bid_size = _float(bids[0][1]) if bids and len(bids[0]) > 1 else 0.0 ask = _float(asks[0][0]) if asks else 0.0 ask_size = _float(asks[0][1]) if asks and len(asks[0]) > 1 else 0.0 self._update_orderbook( symbol, bid, bid_size, ask, ask_size, source_timestamp_ms=source_timestamp_ms, ) def _update_orderbook( self, symbol: str, bid: float, bid_size: float, ask: float, ask_size: float, *, source_timestamp_ms: int = 0, ) -> None: if bid > 0 and ask > 0: self.orderbook_top[symbol] = (bid, ask) current = self.tickers.get(symbol) size_total = max(0.0, bid_size) + max(0.0, ask_size) mid_price = (bid + ask) / 2.0 imbalance = ( (max(0.0, bid_size) - max(0.0, ask_size)) / size_total if size_total > 0 else 0.0 ) microprice = ( (ask * max(0.0, bid_size) + bid * max(0.0, ask_size)) / size_total if size_total > 0 else mid_price ) observed_at = utc_now() metrics = { "bid_price": bid, "bid_size": max(0.0, bid_size), "ask_price": ask, "ask_size": max(0.0, ask_size), "mid_price": mid_price, "microprice": microprice, "spread_bps": ((ask - bid) / mid_price) * 10_000 if mid_price > 0 else 0.0, "imbalance": imbalance, "source_timestamp_ms": max(0, source_timestamp_ms), "observed_at": observed_at.isoformat(), } self.orderbook_metrics[symbol] = metrics if current: self.tickers[symbol] = Ticker( symbol=symbol, last_price=current.last_price, bid=bid, ask=ask, turnover_24h=current.turnover_24h, volume_24h=current.volume_24h, change_24h=current.change_24h, ) self._sample_orderbook(symbol, metrics, current.last_price if current else mid_price, observed_at) def _sample_orderbook( self, symbol: str, metrics: dict[str, Any], last_price: float, observed_at: datetime, ) -> None: if not self.settings.market_observation_enabled: return now = time.monotonic() previous = self._last_observation_monotonic.get(symbol) if previous is not None and now - previous < self.settings.market_observation_sample_seconds: return try: self.storage.insert_market_observation( symbol=symbol, bid_price=float(metrics["bid_price"]), bid_size=float(metrics["bid_size"]), ask_price=float(metrics["ask_price"]), ask_size=float(metrics["ask_size"]), mid_price=float(metrics["mid_price"]), microprice=float(metrics["microprice"]), spread_bps=float(metrics["spread_bps"]), imbalance=float(metrics["imbalance"]), last_price=last_price, source_timestamp_ms=int(metrics["source_timestamp_ms"]), created_at=observed_at, ) except Exception as exc: # Storage errors must not disconnect market data. self.observation_error_count += 1 self.last_observation_error = str(exc) return self._last_observation_monotonic[symbol] = now self.observation_samples += 1 self.last_observation_at = observed_at self.last_observation_error = "" def prices(self) -> dict[str, float]: return {symbol: ticker.last_price for symbol, ticker in self.tickers.items()} def symbol_freshness(self, symbol: str) -> dict[str, Any]: ticker = self.tickers.get(symbol) candles = self.candles.get(symbol, []) ticker_age = (utc_now() - ticker.updated_at).total_seconds() if ticker else None interval_ms = _interval_ms(self.settings.base_interval) candle_age = ( max(0.0, (utc_now().timestamp() * 1000 - candles[-1].timestamp) / 1000) if candles else None ) ticker_ok = ticker_age is not None and ticker_age <= self.settings.market_ticker_max_age_seconds candle_ok = bool( candle_age is not None and interval_ms > 0 and candle_age <= (interval_ms / 1000) * 2.5 ) return { "ok": bool(ticker_ok and candle_ok), "ticker_ok": ticker_ok, "candle_ok": candle_ok, "ticker_age_seconds": round(ticker_age, 3) if ticker_age is not None else None, "candle_age_seconds": round(candle_age, 3) if candle_age is not None else None, } def snapshot(self) -> dict[str, Any]: return { "symbols": self.symbols, "ws_connected": self.ws_connected, "rest_error_count": self.rest_error_count, "last_rest_error": self.last_rest_error, "quality": market_quality_snapshot( symbols=self.symbols, candles_by_symbol=self.candles, tickers=self.tickers, interval=self.settings.base_interval, ), "last_rest_refresh_at": self.last_rest_refresh_at.isoformat() if self.last_rest_refresh_at else None, "last_ws_message_at": self.last_ws_message_at.isoformat() if self.last_ws_message_at else None, "observation_collector": { "enabled": self.settings.market_observation_enabled, "sample_seconds": self.settings.market_observation_sample_seconds, "samples_since_start": self.observation_samples, "last_observation_at": self.last_observation_at.isoformat() if self.last_observation_at else None, "error_count": self.observation_error_count, "last_error": self.last_observation_error, }, "markets": [ { "ticker": self.tickers[symbol].as_dict() if symbol in self.tickers else None, "candles": [candle.as_dict() for candle in self.candles.get(symbol, [])[-120:]], "trend_candles": [candle.as_dict() for candle in self.trend_candles.get(symbol, [])[-5:]], "pattern": self.patterns.get(symbol), "forecast": self.forecasts.get(symbol), "shadow_forecast": self.shadow_forecasts.get(symbol), "orderbook": self.orderbook_metrics.get(symbol), "quality": analyze_symbol_quality( symbol=symbol, candles=self.candles.get(symbol, []), ticker=self.tickers.get(symbol), interval=self.settings.base_interval, ), "instrument": asdict(self.instruments[symbol]) if symbol in self.instruments else None, } for symbol in self.symbols ], } def _closed_candles(candles: list[Candle], interval: str, now_ms: int | None = None) -> list[Candle]: interval_ms = _interval_ms(interval) if interval_ms <= 0: return candles now_ms = now_ms if now_ms is not None else int(utc_now().timestamp() * 1000) return [candle for candle in candles if candle.timestamp + interval_ms <= now_ms] def _is_closed_kline_row(row: dict[str, Any], interval: str) -> bool: confirm = row.get("confirm") if isinstance(confirm, bool): return confirm start = int(row.get("start", 0) or 0) interval_ms = _interval_ms(interval) if start <= 0 or interval_ms <= 0: return True return start + interval_ms <= int(utc_now().timestamp() * 1000) def _interval_ms(interval: str) -> int: normalized = str(interval).strip().upper() if normalized == "D": return 24 * 60 * 60 * 1000 if normalized == "W": return 7 * 24 * 60 * 60 * 1000 if normalized.isdigit(): return int(normalized) * 60 * 1000 return 0 def _candles_due(candles: list[Candle], interval: str, now_ms: int | None = None) -> bool: if not candles: return True interval_ms = _interval_ms(interval) if interval_ms <= 0: return True now_ms = now_ms if now_ms is not None else int(utc_now().timestamp() * 1000) expected_latest_start = (now_ms // interval_ms - 1) * interval_ms return candles[-1].timestamp < expected_latest_start