Streams

@classmethod
def of(cls, ws: SocketClient, *, validate: bool = True):
  return cls(client=ws, validate=validate)

@classmethod
def new(cls, *, mainnet: bool = True, timeout: timedelta = timedelta(seconds=10), validate: bool = True):
  domain = HYPERLIQUID_MAINNET if mainnet else HYPERLIQUID_TESTNET
  ws = SocketClient(url=f'wss://{domain}/ws', timeout=timeout)
  return cls.of(ws, validate=validate)

async def active_asset_ctx(self, coin: str):
  """Stream active asset context.

  - `coin`: Asset symbol.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('activeAssetCtx', {'coin': coin})
  coin_l = coin.lower()
  stream = stream.filter(lambda msg: msg.get('coin', '').lower() == coin_l)
  def mapper(msg) -> ActiveAssetCtxData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def active_asset_data(self, user: str, coin: str):
  """Stream active asset data for a user.

  - `user`: Account address.
  - `coin`: Asset symbol.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('activeAssetData', {'user': user, 'coin': coin})
  user_l = user.lower()
  coin_l = coin.lower()
  def match(msg):
    return msg.get('user', '').lower() == user_l and msg.get('coin', '').lower() == coin_l
  stream = stream.filter(match)
  def mapper(msg) -> ActiveAssetDataData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def all_mids(self, dex: str | None = None):
  """Stream mids for all coins.

  - `dex`: Perp dex name. Defaults to the empty string which represents the
    first perp dex.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  params: AllMidsParams | None = None
  if dex is not None:
    params = {'dex': dex}
  stream = await self.subscribe('allMids', params)
  def mapper(msg) -> AllMidsData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def bbo(self, coin: str):
  """Stream best-bid-offer updates.

  - `coin`: Asset symbol.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('bbo', {'coin': coin})
  coin_l = coin.lower()
  stream = stream.filter(lambda msg: msg.get('coin', '').lower() == coin_l)
  def mapper(msg) -> BboData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def candle(self, coin: str, interval: str):
  """Stream candle updates for a coin.

  - `coin`: Asset symbol.
  - `interval`: Candle interval.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('candle', {'coin': coin, 'interval': interval})
  coin_l = coin.lower()
  interval_l = interval.lower()
  def match(msg):
    return msg.get('s', '').lower() == coin_l and msg.get('i', '').lower() == interval_l
  stream = stream.filter(match)
  def mapper(msg) -> CandleData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def clearinghouse_state(self, user: str, dex: str):
  """Stream clearinghouse state for a user.

  - `user`: Account address.
  - `dex`: Perp dex name.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('clearinghouseState', {'user': user, 'dex': dex})
  def mapper(msg) -> ClearinghouseStateData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def l2_book(
  self, coin: str, *,
  n_sig_figs: int | None = None,
  mantissa: int | None = None
):
  """Stream L2 book updates.

  - `coin`: Asset symbol.
  - `n_sig_figs`: Aggregate levels to significant figures.
  - `mantissa`: Only allowed when `n_sig_figs` is 5.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  params: L2BookParams = {'coin': coin}
  if n_sig_figs is not None:
    params['nSigFigs'] = n_sig_figs
  if mantissa is not None:
    params['mantissa'] = mantissa
  stream = await self.subscribe('l2Book', params)
  coin_l = coin.lower()
  stream = stream.filter(lambda msg: msg.get('coin', '').lower() == coin_l)
  def mapper(msg) -> L2BookData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def notification(self, user: str):
  """Stream notifications for a user.

  - `user`: User address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('notification', {'user': user})
  def mapper(msg) -> NotificationData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def open_orders(self, user: str, dex: str):
  """Stream open orders for a user.

  - `user`: Account address.
  - `dex`: Perp dex name.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('openOrders', {'user': user, 'dex': dex})
  user_l = user.lower()
  dex_l = dex.lower()
  def match(msg):
    return msg.get('user', '').lower() == user_l and msg.get('dex', '').lower() == dex_l
  stream = stream.filter(match)
  def mapper(msg) -> OpenOrdersData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def order_updates(self, user: str):
  """Stream order updates for a user.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('orderUpdates', {'user': user})
  def mapper(msg) -> OrderUpdatesData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def trades(self, coin: str):
  """Stream trades for a coin.

  - `coin`: Asset symbol.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('trades', {'coin': coin})
  coin_l = coin.lower()
  def match(msg):
    if not msg:
      return False
    first = msg[0] if isinstance(msg, list) else None
    if not isinstance(first, dict):
      return False
    return first.get('coin', '').lower() == coin_l
  stream = stream.filter(match)
  def mapper(msg) -> TradesData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def twap_states(self, user: str, dex: str):
  """Stream TWAP states for a user.

  - `user`: Account address.
  - `dex`: Perp dex name.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('twapStates', {'user': user, 'dex': dex})
  user_l = user.lower()
  dex_l = dex.lower()
  def match(msg):
    return msg.get('user', '').lower() == user_l and msg.get('dex', '').lower() == dex_l
  stream = stream.filter(match)
  def mapper(msg) -> TwapStatesData:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def user_events(self, user: str):
  """Stream user events.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('userEvents', {'user': user})
  def mapper(msg) -> WsUserEvent:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def user_fills(self, user: str, *, aggregate_by_time: bool | None = None):
  """Stream user fills.

  - `user`: Account address.
  - `aggregate_by_time`: Aggregate partial fills in the same block.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  params: UserFillsParams = {'user': user}
  if aggregate_by_time is not None:
    params['aggregateByTime'] = aggregate_by_time
  stream = await self.subscribe('userFills', params)
  user_l = user.lower()
  stream = stream.filter(lambda msg: msg.get('user', '').lower() == user_l)
  def mapper(msg) -> WsUserFills:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def user_fundings(self, user: str):
  """Stream user funding updates.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('userFundings', {'user': user})
  user_l = user.lower()
  def match(msg):
    if not isinstance(msg, dict):
      return True
    user_val = msg.get('user')
    return user_val is None or str(user_val).lower() == user_l
  stream = stream.filter(match)
  def mapper(msg) -> WsUserFundings:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def user_non_funding_ledger_updates(self, user: str):
  """Stream non-funding ledger updates.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('userNonFundingLedgerUpdates', {'user': user})
  user_l = user.lower()
  def match(msg):
    if not isinstance(msg, dict):
      return True
    user_val = msg.get('user')
    return user_val is None or str(user_val).lower() == user_l
  stream = stream.filter(match)
  def mapper(msg) -> WsUserNonFundingLedgerUpdates:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def user_twap_history(self, user: str):
  """Stream TWAP history for a user.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('userTwapHistory', {'user': user})
  user_l = user.lower()
  stream = stream.filter(lambda msg: msg.get('user', '').lower() == user_l)
  def mapper(msg) -> WsUserTwapHistory:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def user_twap_slice_fills(self, user: str):
  """Stream TWAP slice fills for a user.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('userTwapSliceFills', {'user': user})
  user_l = user.lower()
  stream = stream.filter(lambda msg: msg.get('user', '').lower() == user_l)
  def mapper(msg) -> WsUserTwapSliceFills:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)

async def web_data3(self, user: str):
  """Stream WebData3 updates for a user.

  - `user`: Account address.

  > [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('webData3', {'user': user})
  user_l = user.lower()
  def match(msg):
    user_state = msg.get('userState') if isinstance(msg, dict) else None
    return isinstance(user_state, dict) and user_state.get('user', '').lower() == user_l
  stream = stream.filter(match)
  def mapper(msg) -> WebData3Data:
    return adapter.validate_python(msg) if self.validate else msg
  return stream.map(mapper)