Support all-market ticker requests
Этот коммит содержится в:
+16
-40
@@ -9,8 +9,10 @@ import websockets
|
|||||||
|
|
||||||
|
|
||||||
class CoinExClient:
|
class CoinExClient:
|
||||||
def __init__(self, api_base: str) -> None:
|
def __init__(self, api_base: str, access_id: str = '', secret_key: str = '') -> None:
|
||||||
self.api_base = api_base.rstrip('/')
|
self.api_base = api_base.rstrip('/')
|
||||||
|
self.access_id = access_id
|
||||||
|
self.secret_key = secret_key
|
||||||
|
|
||||||
async def get_market_list(self) -> dict[str, Any]:
|
async def get_market_list(self) -> dict[str, Any]:
|
||||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||||
@@ -26,18 +28,14 @@ class CoinExClient:
|
|||||||
return response.json()
|
return response.json()
|
||||||
|
|
||||||
async def get_kline(self, market: str, period: str = '1min', limit: int = 100) -> dict[str, Any]:
|
async def get_kline(self, market: str, period: str = '1min', limit: int = 100) -> dict[str, Any]:
|
||||||
params = {
|
params = {'market': market.upper(), 'period': period, 'limit': max(1, min(limit, 1000))}
|
||||||
'market': market.upper(),
|
|
||||||
'period': period,
|
|
||||||
'limit': max(1, min(limit, 1000)),
|
|
||||||
}
|
|
||||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||||
response = await client.get(f'{self.api_base}/spot/kline', params=params)
|
response = await client.get(f'{self.api_base}/spot/kline', params=params)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
return response.json()
|
return response.json()
|
||||||
|
|
||||||
async def get_ticker(self, market: str) -> dict[str, Any]:
|
async def get_ticker(self, market: str | None = None) -> dict[str, Any]:
|
||||||
params = {'market': market.upper()}
|
params = {'market': market.upper()} if market else None
|
||||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||||
response = await client.get(f'{self.api_base}/spot/ticker', params=params)
|
response = await client.get(f'{self.api_base}/spot/ticker', params=params)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
@@ -55,6 +53,12 @@ class CoinExClient:
|
|||||||
except (TypeError, ValueError):
|
except (TypeError, ValueError):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
def extract_tickers(self, ticker: dict[str, Any]) -> list[dict[str, Any]]:
|
||||||
|
data = ticker.get('data') or []
|
||||||
|
if isinstance(data, dict):
|
||||||
|
data = [data]
|
||||||
|
return [item for item in data if isinstance(item, dict) and item.get('market')]
|
||||||
|
|
||||||
|
|
||||||
class CoinExLiveStream:
|
class CoinExLiveStream:
|
||||||
def __init__(self, ws_url: str, rest_client: CoinExClient) -> None:
|
def __init__(self, ws_url: str, rest_client: CoinExClient) -> None:
|
||||||
@@ -67,22 +71,12 @@ class CoinExLiveStream:
|
|||||||
async for event in self._trades_from_ws(market):
|
async for event in self._trades_from_ws(market):
|
||||||
yield event
|
yield event
|
||||||
except Exception as exc: # noqa: BLE001 - dashboard must keep breathing
|
except Exception as exc: # noqa: BLE001 - dashboard must keep breathing
|
||||||
yield {
|
yield {'type': 'stream_warning', 'market': market, 'message': f'CoinEx WebSocket unavailable, using HTTP fallback: {exc}'}
|
||||||
'type': 'stream_warning',
|
|
||||||
'market': market,
|
|
||||||
'message': f'CoinEx WebSocket unavailable, using HTTP fallback: {exc}',
|
|
||||||
}
|
|
||||||
async for event in self._trades_from_http_fallback(market):
|
async for event in self._trades_from_http_fallback(market):
|
||||||
yield event
|
yield event
|
||||||
|
|
||||||
async def _trades_from_ws(self, market: str) -> AsyncIterator[dict[str, Any]]:
|
async def _trades_from_ws(self, market: str) -> AsyncIterator[dict[str, Any]]:
|
||||||
subscribe_message = {
|
subscribe_message = {'id': int(time.time()), 'method': 'deals.subscribe', 'params': {'market_list': [market]}}
|
||||||
'id': int(time.time()),
|
|
||||||
'method': 'deals.subscribe',
|
|
||||||
'params': {
|
|
||||||
'market_list': [market],
|
|
||||||
},
|
|
||||||
}
|
|
||||||
async with websockets.connect(self.ws_url, ping_interval=20, ping_timeout=20) as websocket:
|
async with websockets.connect(self.ws_url, ping_interval=20, ping_timeout=20) as websocket:
|
||||||
await websocket.send(json.dumps(subscribe_message))
|
await websocket.send(json.dumps(subscribe_message))
|
||||||
async for raw_message in websocket:
|
async for raw_message in websocket:
|
||||||
@@ -99,21 +93,13 @@ class CoinExLiveStream:
|
|||||||
ticker = await self.rest_client.get_ticker(market)
|
ticker = await self.rest_client.get_ticker(market)
|
||||||
price = self.rest_client.extract_price_from_ticker(ticker)
|
price = self.rest_client.extract_price_from_ticker(ticker)
|
||||||
if price is not None:
|
if price is not None:
|
||||||
yield {
|
yield {'type': 'live_price', 'source': 'coinex_http_fallback', 'market': market, 'price': price, 'ts': int(time.time() * 1000), 'raw': ticker}
|
||||||
'type': 'live_price',
|
|
||||||
'source': 'coinex_http_fallback',
|
|
||||||
'market': market,
|
|
||||||
'price': price,
|
|
||||||
'ts': int(time.time() * 1000),
|
|
||||||
'raw': ticker,
|
|
||||||
}
|
|
||||||
await asyncio.sleep(2)
|
await asyncio.sleep(2)
|
||||||
|
|
||||||
def _normalize_deals_message(self, market: str, message: dict[str, Any]) -> dict[str, Any] | None:
|
def _normalize_deals_message(self, market: str, message: dict[str, Any]) -> dict[str, Any] | None:
|
||||||
params = message.get('params')
|
params = message.get('params')
|
||||||
if not params:
|
if not params:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
deals: list[Any] = []
|
deals: list[Any] = []
|
||||||
if isinstance(params, dict):
|
if isinstance(params, dict):
|
||||||
if params.get('market') and params.get('market') != market:
|
if params.get('market') and params.get('market') != market:
|
||||||
@@ -123,7 +109,6 @@ class CoinExLiveStream:
|
|||||||
for item in params:
|
for item in params:
|
||||||
if isinstance(item, dict):
|
if isinstance(item, dict):
|
||||||
deals.extend(item.get('deal_list') or item.get('deals') or [item])
|
deals.extend(item.get('deal_list') or item.get('deals') or [item])
|
||||||
|
|
||||||
prices: list[float] = []
|
prices: list[float] = []
|
||||||
for deal in deals:
|
for deal in deals:
|
||||||
if not isinstance(deal, dict):
|
if not isinstance(deal, dict):
|
||||||
@@ -133,15 +118,6 @@ class CoinExLiveStream:
|
|||||||
prices.append(float(price_raw))
|
prices.append(float(price_raw))
|
||||||
except (TypeError, ValueError):
|
except (TypeError, ValueError):
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if not prices:
|
if not prices:
|
||||||
return None
|
return None
|
||||||
|
return {'type': 'live_price', 'source': 'coinex_ws', 'market': market, 'price': prices[-1], 'ts': int(time.time() * 1000), 'raw': message}
|
||||||
return {
|
|
||||||
'type': 'live_price',
|
|
||||||
'source': 'coinex_ws',
|
|
||||||
'market': market,
|
|
||||||
'price': prices[-1],
|
|
||||||
'ts': int(time.time() * 1000),
|
|
||||||
'raw': message,
|
|
||||||
}
|
|
||||||
|
|||||||
Ссылка в новой задаче
Block a user