Node Orders

@dataclass
class Orders(
  PlaceOrder,
  CancelOrder,
  BatchCancelOrders,
):
  """Composed order helper surface."""

  context: NodeContext
  """Shared node context used for wallet and chain access."""
  tx: Tx
  """Transaction helper used for signing and broadcasting."""

@dataclass
class PlaceOrder:
  """Order placement endpoint."""

  context: NodeContext
  """Shared node context used for wallet and chain access."""
  tx: Tx
  """Transaction helper used for signing and broadcasting."""

  async def default_good_til(self, flags: Flags) -> tuple[int | None, int | None]:
    """Resolve default order expiry for an order flag.

    Args:
      flags: Order flag determining whether expiry is block-based or
        timestamp-based.

    Returns:
      Pair of `good_til_block` and `good_til_block_time` values. One side of
      the pair is populated depending on the order type.

    Raises:
      BadRequest: Raised when latest block metadata needed for expiry is
        unavailable.
    """
    latest_block = await self.context.chain.tendermint.get_latest_block()
    block = latest_block.block
    if block is None or block.header is None:
      raise BadRequest('Latest block response did not include a block header')
    if flags == 'SHORT_TERM':
      return block.header.height + SHORT_BLOCK_WINDOW, None
    if flags in {'LONG_TERM', 'CONDITIONAL'}:
      if block.header.time is None:
        raise BadRequest('Latest block response did not include block time')
      return None, int(block.header.time.timestamp()) + STATEFUL_ORDER_TIME_WINDOW
    return None, None

  async def build_order(
    self,
    *,
    market: Market | PerpetualMarket,
    order: OrderParams,
    subaccount: int = 0,
    good_til_block: int | None = None,
    good_til_block_time: int | None = None,
  ) -> clob.Order:
    """Build a protocol order without broadcasting it.

    Args:
      market: Market metadata used to convert price and size into protocol
        subticks and quantums.
      order: Ergonomic order parameters for the CLOB message.
      subaccount: dYdX subaccount number owned by the wallet.
      good_til_block: Optional explicit short-term expiry block.
      good_til_block_time: Optional explicit stateful expiry timestamp.

    Returns:
      dYdX protocol order message.
    """
    wallet = self.context.require_wallet()
    market_info = Market.from_payload(market)
    flags = order['flags']
    if good_til_block is None and good_til_block_time is None:
      good_til_block = order.get('good_til_block')
      good_til_block_time = order.get('good_til_block_time')
    if good_til_block is None and good_til_block_time is None:
      good_til_block, good_til_block_time = await self.default_good_til(flags)
    client_id = order.get('client_id') or random_client_id()
    order_id = market_info.order_id(
      owner=wallet.address,
      subaccount_number=subaccount,
      client_id=client_id,
      order_flags=parse_flags(flags),
    )
    return clob.Order(
      order_id=order_id,
      side=parse_side(order['side']),
      quantums=market_info.calculate_quantums(order['size']),
      subticks=market_info.calculate_subticks(order['price']),
      good_til_block=good_til_block,
      good_til_block_time=good_til_block_time,
      time_in_force=parse_time_in_force(order.get('time_in_force')),
      reduce_only=order.get('reduce_only', False),
      client_metadata=order.get('client_metadata', 0),
      condition_type=parse_condition_type(order.get('condition_type')),
      conditional_order_trigger_subticks=order.get('conditional_order_trigger_subticks', 0),
    )

  async def place_order(
    self,
    market: Market | PerpetualMarket,
    *,
    order: OrderParams,
    subaccount: int = 0,
    mode: tx_proto.BroadcastMode = tx_proto.BroadcastMode(2),
    simulate: bool = False,
  ) -> PlacedOrder:
    """Build, sign, and broadcast a long-term dYdX order.

    Args:
      market: Market metadata used to convert price and size into protocol
        subticks and quantums. Accepts either a prepared `Market` or the
        indexer `PerpetualMarket` payload.
      order: Order parameters for the CLOB message. dYdX only allows batched
        placement for stateful long-term or conditional orders.
      subaccount: dYdX subaccount number owned by the wallet.
      mode: Cosmos broadcast mode.
      simulate: Whether to simulate the transaction instead of broadcasting it.

    Returns:
      The broadcast response and protocol order that was signed.
    """
    order_message = await self.build_order(
      market=market,
      order=order,
      subaccount=subaccount,
    )
    response = await self.tx.sign_and_broadcast(
      [clob.MsgPlaceOrder(order=order_message)],
      mode=mode,
      simulate=simulate,
    )
    return PlacedOrder(tx=response, order=order_message)

  async def place_orders(
    self,
    orders: Sequence[OrderPlacement],
    *,
    mode: tx_proto.BroadcastMode = tx_proto.BroadcastMode(2),
    simulate: bool = False,
  ) -> PlacedOrders:
    """Build, sign, and broadcast multiple stateful dYdX orders.

    dYdX rejects transactions containing multiple short-term `MsgPlaceOrder`
    messages. This helper only accepts long-term or conditional orders, which
    are stateful and can be submitted together in one transaction.

    Args:
      orders: Order placement items. Each item carries its market metadata,
        order parameters, and optional subaccount number.
      mode: Cosmos broadcast mode.
      simulate: Whether to simulate the transaction before broadcasting.

    Returns:
      The broadcast response and protocol orders that were signed.

    Raises:
      BadRequest: Raised when no orders are supplied or any order is short-term.
    """
    if not orders:
      raise BadRequest('At least one order is required')
    if any(item['order']['flags'] == 'SHORT_TERM' for item in orders):
      raise BadRequest('place_orders only supports long-term or conditional orders')
    order_messages = [
      await self.build_order(
        market=item['market'],
        order=item['order'],
        subaccount=item.get('subaccount', 0),
      )
      for item in orders
    ]
    response = await self.tx.sign_and_broadcast(
      [clob.MsgPlaceOrder(order=order) for order in order_messages],
      mode=mode,
      simulate=simulate,
    )
    return PlacedOrders(tx=response, orders=order_messages)

@dataclass
class PlacedOrder:
  """Result of a placed order."""

  tx: tx_proto.BroadcastTxResponse
  """Transaction broadcast response returned by the chain."""
  order: clob.Order
  """Protocol order message that was signed and submitted."""

@dataclass
class PlacedOrders:
  """Result of a multi-order placement transaction."""

  tx: tx_proto.BroadcastTxResponse
  """Transaction broadcast response returned by the chain."""
  orders: list[clob.Order]
  """Protocol order messages that were signed and submitted."""

@dataclass
class CancelOrder:
  """Order cancellation endpoint."""

  context: NodeContext
  """Shared node context used for wallet and chain access."""
  tx: Tx
  """Transaction helper used for signing and broadcasting."""

  async def cancel_order(
    self,
    order_id: clob.OrderId,
    *,
    good_til_block: int | None = None,
    good_til_block_time: int | None = None,
    mode: tx_proto.BroadcastMode = tx_proto.BroadcastMode.SYNC,
    simulate: bool = False,
  ) -> tx_proto.BroadcastTxResponse:
    """Build, sign, and broadcast an order cancellation.

    Args:
      order_id: Protocol order identifier to cancel.
      good_til_block: Optional short-term cancellation expiry block. Loaded
        from the latest chain block when omitted for short-term orders.
      good_til_block_time: Optional stateful cancellation expiry timestamp.
        Loaded from the latest chain block time when omitted for stateful
        orders.
      mode: Cosmos broadcast mode.
      simulate: Whether to simulate the transaction before broadcasting.

    Returns:
      Cosmos broadcast response.

    Raises:
      BadRequest: Raised when latest block metadata needed for default expiry
        is unavailable.
    """
    if good_til_block is None and good_til_block_time is None:
      latest_block = await self.context.chain.tendermint.get_latest_block()
      block = latest_block.block
      if block is None or block.header is None:
        raise BadRequest('Latest block response did not include a block header')
      if order_id.order_flags == 0:
        good_til_block = block.header.height + SHORT_BLOCK_WINDOW
      else:
        if block.header.time is None:
          raise BadRequest('Latest block response did not include block time')
        good_til_block_time = int(block.header.time.timestamp()) + STATEFUL_ORDER_TIME_WINDOW
    message = clob.MsgCancelOrder(
      order_id=order_id,
      good_til_block=good_til_block,
      good_til_block_time=good_til_block_time,
    )
    return await self.tx.sign_and_broadcast([message], mode=mode, simulate=simulate)

@dataclass
class BatchCancelOrders:
  """Batch order cancellation endpoint."""

  context: NodeContext
  """Shared node context used for wallet and chain access."""
  tx: Tx
  """Transaction helper used for signing and broadcasting."""

  async def batch_cancel_orders(
    self,
    order_ids: Sequence[clob.OrderId],
    *,
    good_til_block: int | None = None,
    mode: tx_proto.BroadcastMode = tx_proto.BroadcastMode(2),
    simulate: bool = False,
  ) -> tx_proto.BroadcastTxResponse:
    """Build, sign, and broadcast short-term order cancellations.

    Args:
      order_ids: Short-term order identifiers to cancel. All must belong to
        the same subaccount.
      good_til_block: Optional cancellation expiry block. Loaded from the
        latest chain block when omitted.
      mode: Cosmos broadcast mode.
      simulate: Whether to simulate the transaction before broadcasting.

    Returns:
      Cosmos broadcast response.

    Raises:
      BadRequest: Raised when no orders are provided, orders are not short-term,
        orders belong to different subaccounts, or latest block metadata is
        unavailable.
    """
    if not order_ids:
      raise BadRequest('At least one order ID is required')
    subaccount_id = order_ids[0].subaccount_id
    if subaccount_id is None:
      raise BadRequest('Order IDs must include a subaccount ID')
    orders_by_pair: dict[int, list[int]] = defaultdict(list)
    for order_id in order_ids:
      if order_id.order_flags != 0:
        raise BadRequest('Only short-term orders can be canceled in a batch')
      if order_id.subaccount_id != subaccount_id:
        raise BadRequest('All batch-canceled orders must use the same subaccount')
      orders_by_pair[order_id.clob_pair_id].append(order_id.client_id)
    if good_til_block is None:
      latest_block = await self.context.chain.tendermint.get_latest_block()
      block = latest_block.block
      if block is None or block.header is None:
        raise BadRequest('Latest block response did not include a block header')
      good_til_block = block.header.height + SHORT_BLOCK_WINDOW
    message = clob.MsgBatchCancel(
      subaccount_id=subaccount_id,
      short_term_cancels=[
        clob.OrderBatch(clob_pair_id=clob_pair_id, client_ids=client_ids)
        for clob_pair_id, client_ids in orders_by_pair.items()
      ],
      good_til_block=good_til_block,
    )
    return await self.tx.sign_and_broadcast([message], mode=mode, simulate=simulate)

class OrderParams(TypedDict):
  """Ergonomic dYdX order parameters."""

  side: Side
  """Order side."""
  size: DecimalValue
  """Human-readable base asset size."""
  price: DecimalValue
  """Human-readable order price."""
  flags: Flags
  """dYdX order flag category."""
  time_in_force: NotRequired[TimeInForce]
  """Optional dYdX time-in-force behavior."""
  client_metadata: NotRequired[int]
  """Optional client metadata integer included in the protocol order."""
  condition_type: NotRequired[ConditionType]
  """Optional conditional order trigger type."""
  conditional_order_trigger_subticks: NotRequired[int]
  """Optional conditional trigger price expressed as protocol subticks."""
  client_id: NotRequired[int]
  """Optional client-generated order ID."""
  reduce_only: NotRequired[bool]
  """Whether the order may only reduce exposure."""
  good_til_block: NotRequired[int]
  """Optional short-term expiry block."""
  good_til_block_time: NotRequired[int]
  """Optional stateful expiry timestamp."""

class OrderPlacement(TypedDict):
  """One dYdX order placement inside a multi-order transaction."""

  market: Market | PerpetualMarket
  """Market metadata used to convert price and size into protocol values."""
  order: OrderParams
  """Order parameters for the CLOB message."""
  subaccount: NotRequired[int]
  """Optional dYdX subaccount number under the signing wallet."""