Streams

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

  Args:
    coin: Asset symbol.

  References:
    - [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.

  Args:
    user: Account address.
    coin: Asset symbol.

  References:
    - [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.

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

  References:
    - [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 all_dexs_asset_ctxs(self, *, validate: bool | None = None):
  """Stream asset contexts for all perp dexs.

  Args:
    validate: Override response validation for this call. When omitted,
      the client-level validation setting is used.

  References:
    - [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions#subscriptions)
  """
  should_validate = self.validate if validate is None else validate
  stream = await self.subscribe('allDexsAssetCtxs')
  def mapper(msg) -> AllDexsAssetCtxsData:
    return adapter.validate_python(msg) if should_validate else msg
  return stream.map(mapper)

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

  Args:
    coin: Asset symbol.

  References:
    - [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.

  Args:
    coin: Asset symbol.
    interval: Candle interval.

  References:
    - [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.

  Args:
    user: Account address.
    dex: Perp dex name.

  References:
    - [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:
    data = msg.get('clearinghouseState', msg)
    return adapter.validate_python(data) if self.validate else data
  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.

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

  References:
    - [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions#subscription-messages)
  """
  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)
  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.

  Args:
    user: User address.

  References:
    - [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.

  Args:
    user: Account address.
    dex: Perp dex name.

  References:
    - [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.

  Args:
    user: Account address.

  References:
    - [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.

  Args:
    coin: Asset symbol.

  References:
    - [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.

  Args:
    user: Account address.
    dex: Perp dex name.

  References:
    - [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.

  Args:
    user: Account address.

  References:
    - [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/websocket#subscriptions)
  """
  stream = await self.subscribe('user', {'user': user}, request_channel='userEvents')
  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.

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

  References:
    - [Hyperliquid API docs](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/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.

  Args:
    user: Account address.

  References:
    - [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.

  Args:
    user: Account address.

  References:
    - [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.

  Args:
    user: Account address.

  References:
    - [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.

  Args:
    user: Account address.

  References:
    - [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.

  Args:
    user: Account address.

  References:
    - [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)