Indexer Streams

@dataclass
class IndexerStreams(
  BlockHeight,
  Candles,
  Markets,
  Orders,
  ParentSubaccounts,
  Subaccounts,
  Trades,
):
  """WebSocket indexer stream endpoint group."""

@dataclass
class BlockHeight(StreamsMixin):
  """BlockHeight payload."""
  async def block_height(
    self, *, batched: bool = True, validate: bool | None = None
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to indexer block height updates.

    Args:
      batched: Reduce incoming messages by batching contents.
      validate: Override the client response validation default for this stream.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/websockets#block-height)
    """
    stream = await self.client.subscribe('v4_block_height', {'batched': batched})

    async def parsed_stream() -> AsyncIterable[Notification]:
      """Parsed stream."""
      async for msg in stream:
        data = msg['contents']
        msgs: list[Notification] = data if batched else [data]
        for d in msgs:
          yield notification_adapter.validate_python(d) if self.validate(validate) else d

    c = stream.reply['contents']
    reply: Reply = reply_adapter.validate_python(c) if self.validate(validate) else c
    return Stream(reply, parsed_stream(), stream.unsubscribe)

class Notification(TypedDict):
  """Stream notification payload."""
  blockHeight: str
  time: datetime

class Reply(TypedDict):
  """Reply payload."""
  height: str
  time: datetime

@dataclass
class Markets(StreamsMixin):
  """Markets payload."""
  async def markets(
    self, *, batched: bool = True, validate: bool | None = None,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to market metadata updates.

    Args:
      batched: Reduce incoming messages by batching contents.
      validate: Override the client response validation default for this stream.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/websockets#markets)
    """
    stream = await self.client.subscribe('v4_markets', {'batched': batched})

    async def parsed_stream() -> AsyncIterable[Notification]:
      """Parsed stream."""
      async for msg in stream:
        data = msg['contents']
        msgs: list[Notification] = data if batched else [data]
        for d in msgs:
          yield notification_adapter.validate_python(d) if self.validate(validate) else d

    c = stream.reply['contents']
    reply: Reply = reply_adapter.validate_python(c) if self.validate(validate) else c
    return Stream(reply, parsed_stream(), stream.unsubscribe)

class Notification(TypedDict):
  """Stream notification payload."""
  trading: NotRequired[dict[str, TradingPerpetualMarket]|None]
  oraclePrices: NotRequired[dict[str, OraclePriceMarket]|None]

class OraclePriceMarket(TypedDict):
  """OraclePriceMarket payload."""
  oraclePrice: Decimal
  effectiveAt: datetime
  effectiveAtHeight: str
  marketId: int

class PerpetualMarket(TypedDict):
  """PerpetualMarket payload."""
  clobPairId: str
  ticker: str
  status: Literal['ACTIVE', 'PAUSED', 'CANCEL_ONLY', 'POST_ONLY', 'INITIALIZING', 'FINAL_SETTLEMENT']
  oraclePrice: NotRequired[Decimal | None]
  priceChange24H: Decimal
  volume24H: Decimal
  trades24H: int
  nextFundingRate: Decimal
  initialMarginFraction: Decimal
  maintenanceMarginFraction: Decimal
  openInterest: Decimal
  atomicResolution: int
  quantumConversionExponent: int
  tickSize: Decimal
  stepSize: Decimal
  stepBaseQuantums: int
  subticksPerTick: int
  marketType: Literal['CROSS', 'ISOLATED']
  openInterestLowerCap: NotRequired[Decimal | None]
  openInterestUpperCap: NotRequired[Decimal | None]
  baseOpenInterest: Decimal
  defaultFundingRate1H: NotRequired[Decimal | None]

class Reply(TypedDict):
  """Reply payload."""
  markets: dict[str, PerpetualMarket]

class TradingPerpetualMarket(TypedDict):
  """TradingPerpetualMarket payload."""
  atomicResolution: NotRequired[int|None]
  baseAsset: NotRequired[str|None]
  baseOpenInterest: NotRequired[Decimal|None]
  basePositionSize: NotRequired[Decimal|None]
  clobPairId: NotRequired[str|None]
  id: NotRequired[str|None]
  incrementalPositionSize: NotRequired[Decimal|None]
  initialMarginFraction: NotRequired[Decimal|None]
  maintenanceMarginFraction: NotRequired[Decimal|None]
  marketId: NotRequired[int|None]
  marketType: NotRequired[Literal['PERPETUAL', 'SPOT']|None]
  maxPositionSize: NotRequired[Decimal|None]
  nextFundingRate: NotRequired[Decimal|None]
  openInterest: NotRequired[Decimal|None]
  oraclePrice: NotRequired[Decimal | None]
  priceChange24H: NotRequired[Decimal|None]
  quantumConversionExponent: NotRequired[int|None]
  quoteAsset: NotRequired[str|None]
  status: NotRequired[Literal['ACTIVE', 'PAUSED', 'CANCEL_ONLY', 'POST_ONLY', 'INITIALIZING', 'FINAL_SETTLEMENT']|None]
  stepBaseQuantums: NotRequired[int|None]
  stepSize: NotRequired[Decimal|None]
  subticksPerTick: NotRequired[int|None]
  tickSize: NotRequired[Decimal|None]
  ticker: NotRequired[str|None]
  trades24H: NotRequired[int|None]
  volume24H: NotRequired[Decimal|None]

class Notification(TypedDict):
  """Stream notification payload."""
  trades: list[TradeUpdate]

class Reply(TypedDict):
  """Reply payload."""
  trades: list[Trade]

class Trade(TypedDict):
  """Trade payload."""
  id: str
  side: Literal['BUY', 'SELL']
  size: Decimal
  price: Decimal
  type: Literal['LIMIT', 'LIQUIDATED', 'DELEVERAGED']
  createdAt: datetime
  createdAtHeight: str

class TradeUpdate(TypedDict):
  """TradeUpdate payload."""
  id: str
  createdAt: datetime
  side: Literal['BUY', 'SELL']
  price: Decimal
  size: Decimal
  type: Literal['LIMIT', 'LIQUIDATED', 'DELEVERAGED']

@dataclass
class Trades(StreamsMixin):
  """Trades payload."""
  async def trades(
    self, *, id: str, batched: bool = True, validate: bool | None = None,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to trade updates for a market.

    Args:
      id: Market ticker.
      batched: Reduce incoming messages by batching contents.
      validate: Override the client response validation default for this stream.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/websockets#trades)
    """
    stream = await self.client.subscribe(f'v4_trades:{id}', {'batched': batched})

    async def parsed_stream() -> AsyncIterable[Notification]:
      """Parsed stream."""
      async for msg in stream:
        data = msg['contents']
        msgs: list[Notification] = data if batched else [data]
        for d in msgs:
          yield notification_adapter.validate_python(d) if self.validate(validate) else d

    c = stream.reply['contents']
    reply: Reply = reply_adapter.validate_python(c) if self.validate(validate) else c
    return Stream(reply, parsed_stream(), stream.unsubscribe)

class Candle(TypedDict):
  """Candle payload."""
  startedAt: datetime
  ticker: str
  resolution: Literal['1MIN', '5MINS', '15MINS', '30MINS', '1HOUR', '4HOURS', '1DAY']
  low: Decimal
  high: Decimal
  open: Decimal
  close: Decimal
  baseTokenVolume: Decimal
  usdVolume: Decimal
  trades: int
  startingOpenInterest: Decimal
  orderbookMidPriceOpen: NotRequired[Decimal | None]
  orderbookMidPriceClose: NotRequired[Decimal | None]

@dataclass
class Candles(StreamsMixin):
  """Candles payload."""
  async def candles(
    self,
    market: str,
    *,
    resolution: Resolution,
    validate: bool | None = None,
    batched: bool = True,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to candle updates by market and resolution.

    Args:
      market: Market ticker.
      resolution: Candle resolution.
      validate: Override the client response validation default for this call.
      batched: Request batched WebSocket updates.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.
    """
    return await self.raw_candles(
      id=f'{market}/{resolution}',
      batched=batched,
      validate=validate,
    )

  async def raw_candles(
    self, *, id: str, batched: bool = True, validate: bool | None = None,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to candle updates by raw channel id.

    Args:
      id: Market and candle resolution formatted as {market}/{resolution}.
      batched: Reduce incoming messages by batching contents.
      validate: Override the client response validation default for this stream.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/websockets#candles)
    """
    stream = await self.client.subscribe(f'v4_candles:{id}', {'batched': batched})

    async def parsed_stream() -> AsyncIterable[Notification]:
      """Parsed stream."""
      async for msg in stream:
        data = msg['contents']
        msgs: list[Notification] = data if batched else [data]
        for d in msgs:
          yield notification_adapter.validate_python(d) if self.validate(validate) else d

    c = stream.reply['contents']
    reply: Reply = reply_adapter.validate_python(c) if self.validate(validate) else c
    return Stream(reply, parsed_stream(), stream.unsubscribe)

class Notification(TypedDict):
  """Stream notification payload."""
  startedAt: datetime
  ticker: str
  resolution: Literal['1MIN', '5MINS', '15MINS', '30MINS', '1HOUR', '4HOURS', '1DAY']
  low: Decimal
  high: Decimal
  open: Decimal
  close: Decimal
  baseTokenVolume: Decimal
  usdVolume: Decimal
  trades: int
  startingOpenInterest: Decimal
  orderbookMidPriceOpen: NotRequired[Decimal | None]
  orderbookMidPriceClose: NotRequired[Decimal | None]

class Reply(TypedDict):
  """Reply payload."""
  candles: list[Candle]

class BookEntry(TypedDict):
  """BookEntry payload."""
  price: Decimal
  size: Decimal

class Notification(TypedDict):
  """Stream notification payload."""
  bids: NotRequired[list[NotificationEntry]]
  asks: NotRequired[list[NotificationEntry]]

class NotificationEntry(NamedTuple):
  """NotificationEntry payload."""
  price: Decimal
  size: Decimal

@dataclass
class Orders(StreamsMixin):
  """Orders payload."""
  async def orders(
    self, *, id: str, batched: bool = True, validate: bool | None = None,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to order book updates for a market.

    Args:
      id: Market ticker.
      batched: Reduce incoming messages by batching contents.
      validate: Override the client response validation default for this stream.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/websockets#orders)
    """
    stream = await self.client.subscribe(f'v4_orderbook:{id}', {'batched': batched})

    async def parsed_stream() -> AsyncIterable[Notification]:
      """Parsed stream."""
      async for msg in stream:
        data = msg['contents']
        msgs: list[Notification] = data if batched else [data]
        for d in msgs:
          yield notification_adapter.validate_python(d) if self.validate(validate) else d

    c = stream.reply['contents']
    reply: Reply = reply_adapter.validate_python(c) if self.validate(validate) else c
    return Stream(reply, parsed_stream(), stream.unsubscribe)

class Reply(TypedDict):
  """Reply payload."""
  bids: list[BookEntry]
  asks: list[BookEntry]

class Account(TypedDict):
  """Account payload."""
  address: str
  subaccountNumber: NotRequired[int|None]

class AssetPosition(TypedDict):
  """AssetPosition payload."""
  size: Decimal
  symbol: str
  side: Literal['LONG', 'SHORT']
  assetId: str
  subaccountNumber: int

class AssetPositionSubaccountMessage(TypedDict):
  """AssetPositionSubaccountMessage payload."""
  address: str
  subaccountNumber: int
  positionId: str
  assetId: str
  symbol: str
  side: Literal['LONG', 'SHORT']
  size: Decimal

class FillSubaccountMessage(TypedDict):
  """FillSubaccountMessage payload."""
  id: str
  subaccountId: str
  side: Literal['BUY', 'SELL']
  liquidity: Literal['MAKER', 'TAKER']
  type: Literal['LIMIT', 'LIQUIDATED', 'LIQUIDATION', 'DELEVERAGED', 'OFFSETTING']
  clobPairId: str
  size: Decimal
  price: Decimal
  quoteAmount: Decimal
  eventId: str
  transactionHash: str
  createdAt: datetime
  createdAtHeight: NotRequired[str | None]
  ticker: str
  orderId: NotRequired[str | None]
  clientMetadata: NotRequired[str | None]

class Notification(TypedDict):
  """Stream notification payload."""
  perpetualPositions: NotRequired[list[PerpetualPositionSubaccountMessage]|None]
  assetPositions: NotRequired[list[AssetPositionSubaccountMessage]|None]
  orders: NotRequired[list[OrderSubaccountMessage]|None]
  fills: NotRequired[list[FillSubaccountMessage]|None]
  transfers: NotRequired[list[TransferSubaccountMessage] | TransferSubaccountMessage|None]
  tradingReward: NotRequired[TradingRewardSubaccountMessage|None]
  tradingRewards: NotRequired[list[TradingRewardSubaccountMessage]|None]
  blockHeight: NotRequired[str | None]

class Order(TypedDict):
  """Order payload."""
  id: str
  subaccountId: str
  clientId: str
  clobPairId: str
  side: Literal['BUY', 'SELL']
  size: Decimal
  totalFilled: Decimal
  price: Decimal
  type: Literal['LIMIT', 'MARKET', 'STOP_LIMIT', 'STOP_MARKET', 'TRAILING_STOP', 'TAKE_PROFIT', 'TAKE_PROFIT_MARKET', 'HARD_TRADE', 'FAILED_HARD_TRADE', 'TRANSFER_PLACEHOLDER']
  status: Literal['OPEN', 'FILLED', 'CANCELED', 'BEST_EFFORT_CANCELED', 'UNTRIGGERED', 'BEST_EFFORT_OPENED', 'PENDING']
  timeInForce: Literal['GTT', 'IOC', 'FOK']
  reduceOnly: NotRequired[bool | None]
  orderFlags: str
  goodTilBlock: NotRequired[str | None]
  goodTilBlockTime: NotRequired[datetime | None]
  createdAtHeight: NotRequired[str | None]
  clientMetadata: NotRequired[str | None]
  triggerPrice: NotRequired[Decimal | None]
  postOnly: NotRequired[bool | None]
  ticker: str
  updatedAt: NotRequired[datetime | None]
  updatedAtHeight: NotRequired[str | None]
  subaccountNumber: int
  orderRouterAddress: NotRequired[str|None]
  builderFee: NotRequired[Decimal | None]
  feePpm: NotRequired[Decimal | None]

class OrderSubaccountMessage(TypedDict):
  """OrderSubaccountMessage payload."""
  id: str
  subaccountId: str
  clientId: str
  clobPairId: str
  side: Literal['BUY', 'SELL']
  size: Decimal
  ticker: NotRequired[str | None]
  price: NotRequired[Decimal | None]
  type: Literal['LIMIT', 'MARKET', 'STOP_LIMIT', 'STOP_MARKET', 'TRAILING_STOP', 'TAKE_PROFIT', 'TAKE_PROFIT_MARKET', 'HARD_TRADE', 'FAILED_HARD_TRADE', 'TRANSFER_PLACEHOLDER']
  timeInForce: Literal['GTT', 'IOC', 'FOK']
  postOnly: NotRequired[bool | None]
  reduceOnly: NotRequired[bool | None]
  status: Literal['OPEN', 'FILLED', 'CANCELED', 'BEST_EFFORT_CANCELED', 'UNTRIGGERED', 'BEST_EFFORT_OPENED', 'PENDING']
  orderFlags: NotRequired[str | None]
  totalFilled: NotRequired[Decimal | None]
  totalOptimisticFilled: NotRequired[Decimal | None]
  goodTilBlock: NotRequired[str | None]
  goodTilBlockTime: NotRequired[datetime | None]
  triggerPrice: NotRequired[Decimal | None]
  updatedAt: NotRequired[datetime | None]
  updatedAtHeight: NotRequired[str | None]
  removalReason: NotRequired[str | None]
  createdAtHeight: NotRequired[str | None]
  clientMetadata: NotRequired[str | None]
  subaccountNumber: NotRequired[int | None]
  orderRouterAddress: NotRequired[str | None]

class PerpetualPosition(TypedDict):
  """PerpetualPosition payload."""
  market: str
  status: Literal['OPEN', 'CLOSED', 'LIQUIDATED']
  side: Literal['LONG', 'SHORT']
  size: Decimal
  maxSize: Decimal
  entryPrice: Decimal
  exitPrice: NotRequired[Decimal | None]
  realizedPnl: NotRequired[Decimal | None]
  unrealizedPnl: NotRequired[Decimal | None]
  createdAt: datetime
  createdAtHeight: str
  closedAt: NotRequired[datetime | None]
  sumOpen: Decimal
  sumClose: Decimal
  netFunding: Decimal
  subaccountNumber: int

class PerpetualPositionSubaccountMessage(TypedDict):
  """PerpetualPositionSubaccountMessage payload."""
  market: str
  status: Literal['OPEN', 'CLOSED', 'LIQUIDATED']
  side: Literal['LONG', 'SHORT']
  size: Decimal
  maxSize: Decimal
  entryPrice: Decimal
  exitPrice: NotRequired[Decimal | None]
  realizedPnl: NotRequired[Decimal | None]
  unrealizedPnl: NotRequired[Decimal | None]
  createdAt: NotRequired[datetime | None]
  createdAtHeight: NotRequired[str | None]
  closedAt: NotRequired[datetime | None]
  sumOpen: Decimal
  sumClose: Decimal
  netFunding: Decimal
  subaccountNumber: int

class Reply(TypedDict):
  """Reply payload."""
  subaccount: Subaccount
  orders: list[Order]
  blockHeight: str

class Subaccount(TypedDict):
  """Subaccount payload."""
  address: str
  subaccountNumber: int
  equity: Decimal
  freeCollateral: Decimal
  openPerpetualPositions: PerpetualPositionsMap
  assetPositions: AssetPositionsMap
  marginEnabled: bool
  updatedAtHeight: str
  latestProcessedBlockHeight: str

@dataclass
class Subaccounts(StreamsMixin):
  """Subaccounts payload."""
  async def subaccounts(
    self, address: str, *,
    subaccount: int,
    validate: bool | None = None,
    batched: bool = True,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to subaccount updates.

    Args:
      address: Wallet address.
      subaccount: Subaccount number.
      validate: Override the client response validation default for this call.
      batched: Request batched WebSocket updates.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.
    """
    return await self.raw_subaccounts(
      id=f'{address}/{subaccount}',
      batched=batched,
      validate=validate,
    )

  async def raw_subaccounts(
    self, *, id: str, batched: bool = True, validate: bool | None = None,
  ) -> Stream[Notification, Reply, Unsubscribed]:
    """Subscribe to subaccount updates by raw channel id.

    Args:
      id: Subaccount id formatted as {address}/{subaccount-number}.
      batched: Reduce incoming messages by batching contents.
      validate: Override the client response validation default for this stream.

    Returns:
      A typed stream containing the subscription snapshot, update iterator, and unsubscribe callback.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/websockets#subaccounts)
    """
    stream = await self.client.subscribe(f'v4_subaccounts:{id}', {'batched': batched})

    async def parsed_stream() -> AsyncIterable[Notification]:
      """Parsed stream."""
      async for msg in stream:
        data = msg['contents']
        msgs: list[Notification] = data if batched else [data]
        for d in msgs:
          yield notification_adapter.validate_python(d) if self.validate(validate) else d

    c = stream.reply['contents']
    reply: Reply = reply_adapter.validate_python(c) if self.validate(validate) else c
    return Stream(reply, parsed_stream(), stream.unsubscribe)