spot.streams

@classmethod
def new(
  cls, api_key: str | None = None, api_secret: str | None = None, *,
  api_url: str = MEXC_SPOT_API_BASE,
  ws_url: str = MEXC_SPOT_SOCKET_URL,
):
  import os
  if api_key is None:
    api_key = os.environ['MEXC_ACCESS_KEY']
  if api_secret is None:
    api_secret = os.environ['MEXC_SECRET_KEY']
  auth_http = AuthHttpClient(api_key=api_key, api_secret=api_secret)
  return cls(api_url=api_url, ws_url=ws_url, auth_http=auth_http)

async def create(self) -> str:
  """Create a spot user-stream listen key."""
  return await self.auth_ws.listen_key()

async def keepalive(self, key: str):
  """Keep a spot user-stream listen key alive."""
  return await self.auth_ws.refresh_key(key)

async def close(self, key: str):
  """Close a spot user-stream listen key."""
  return await self.auth_ws.delete_key(key)

async def book_ticker(
  self, symbol: str, aggregation: Aggregation = '100ms',
) -> Stream[PublicAggreBookTickerV3Api, Reply, Reply]:
  """
  Subscribe to best bid/ask updates for a spot symbol.

  Args:
    symbol: Spot symbol, for example `BTCUSDT`.
    aggregation: Update aggregation interval.

  References:
    - [MEXC spot WebSocket market streams](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#bookticker-streams)
  """
  stream = await self.subscribe(channel_name(symbol, aggregation))

  async def parsed_stream():
    async for proto in stream:
      if proto.public_aggre_book_ticker is not None:
        yield proto.public_aggre_book_ticker

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def book_ticker_batch(self, symbol: str) -> Stream[PublicBookTickerBatchV3Api, Reply, Reply]:
  """
  Subscribe to batch best bid/ask updates for a spot symbol.

  Args:
    symbol: Spot symbol, for example `BTCUSDT`.

  References:
    - [MEXC spot WebSocket market streams](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#bookticker-streams)
  """
  stream = await self.subscribe(channel_name(symbol))

  async def parsed_stream():
    async for proto in stream:
      if proto.public_book_ticker_batch is not None:
        yield proto.public_book_ticker_batch

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def candles(self, symbol: str, interval: Interval) -> Stream[PublicSpotKlineV3Api, Reply, Reply]:
  """Subscribe to klines (candles) for a given symbol.

  - `symbol`: The symbol being traded, e.g. `BTCUSDT`.
  - `interval`: The interval of the klines (default: 1m).

  > [MEXC API docs](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#k-line-streams)
  """
  stream = await self.subscribe(channel_name(symbol, interval))
  async def parsed_stream():
    async for proto in stream:
      if proto.public_spot_kline is not None:
        yield proto.public_spot_kline
  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def depth(self, symbol: str, level: Literal[5, 10, 20] = 5) -> Stream[PublicLimitDepthsV3Api, Reply, Reply]:
  """Subscribe to the order book for a given symbol.

  - `symbol`: The symbol being traded, e.g. `BTCUSDT`.
  - `level`: The level of the order book (default: 5).

  > [MEXC API docs](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#partial-book-depth-streams)
  """
  stream = await self.subscribe(channel_name(symbol, level))

  async def parsed_stream():
    async for proto in stream:
      if proto.public_limit_depths is not None:
        yield proto.public_limit_depths

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def depth_updates(
  self, symbol: str, aggregation: Aggregation = '10ms',
) -> Stream[PublicAggreDepthsV3Api, Reply, Reply]:
  """
  Subscribe to aggregated spot order-book delta updates.

  Args:
    symbol: Spot symbol, for example `BTCUSDT`.
    aggregation: Update aggregation interval.

  References:
    - [MEXC spot WebSocket market streams](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#diffdepth-stream)
    - [Maintaining a local order book](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#how-to-properly-maintain-a-local-copy-of-the-order-book)
  """
  stream = await self.subscribe(channel_name(symbol, aggregation))

  async def parsed_stream():
    async for proto in stream:
      if proto.public_aggre_depths is not None:
        yield proto.public_aggre_depths

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def trades(
  self, symbol: str, aggregation: Aggregation = '100ms',
) -> Stream[PublicAggreDealsV3Api, Reply, Reply]:
  """
  Subscribe to aggregated spot trade updates.

  Args:
    symbol: Spot symbol, for example `BTCUSDT`.
    aggregation: Update aggregation interval.

  References:
    - [MEXC spot WebSocket market streams](https://www.mexc.com/api-docs/spot-v3/websocket-market-streams#trade-streams)
  """
  stream = await self.subscribe(channel_name(symbol, aggregation))

  async def parsed_stream():
    async for proto in stream:
      if proto.public_aggre_deals is not None:
        yield proto.public_aggre_deals

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def account(self) -> Stream[PrivateAccountV3Api, Reply, Reply]:
  """
  Subscribe to spot account balance updates.

  References:
    - [MEXC spot WebSocket user streams](https://www.mexc.com/api-docs/spot-v3/websocket-user-data-streams#spot-account-update)
  """
  stream = await self.authed_subscribe('spot@private.account.v3.api.pb')

  async def parsed_stream():
    async for proto in stream:
      if proto.private_account is not None:
        yield proto.private_account

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def trades(self) -> Stream[PrivateDealsV3Api, Reply, Reply]:
  """Subscribe to your trades.

  > [MEXC API docs](https://www.mexc.com/api-docs/spot-v3/websocket-user-data-streams#spot-account-deals)
  """
  stream = await self.authed_subscribe('spot@private.deals.v3.api.pb')
  async def parsed_stream():
    async for proto in stream:
      if proto.private_deals is not None:
        yield proto.private_deals
  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)

async def orders(self) -> Stream[PrivateOrdersV3Api, Reply, Reply]:
  """
  Subscribe to spot order updates.

  References:
    - [MEXC spot WebSocket user streams](https://www.mexc.com/api-docs/spot-v3/websocket-user-data-streams#spot-account-orders)
  """
  stream = await self.authed_subscribe('spot@private.orders.v3.api.pb')

  async def parsed_stream():
    async for proto in stream:
      if proto.private_orders is not None:
        yield proto.private_orders

  return Stream(reply=stream.reply, stream=parsed_stream(), unsubscribe=stream.unsubscribe)