Indexer Data

@dataclass
class IndexerData(
  GetAssetPositions,
  GetCandlesPaged,
  GetComplianceScreen,
  GetFillsPaged,
  GetFundingPaymentsPaged,
  GetFundingPaymentsForParentSubaccount,
  GetHeight,
  GetHistoricalFundingPaged,
  GetHistoricalPnl,
  GetMarket,
  GetMegavaultHistoricalPnl,
  GetOrder,
  GetOrderBook,
  GetParentAssetPositions,
  GetParentFills,
  GetParentHistoricalPnl,
  GetParentSubaccount,
  GetParentTransfers,
  GetRewards,
  GetRewardsAggregated,
  GetScreen,
  GetSparklines,
  GetSubaccount,
  GetSubaccounts,
  GetTime,
  GetTrades,
  GetTransfersPaged,
  GetTransfersBetween,
  GetVaultsHistoricalPnl,
  ListOrders,
  ListParentOrders,
  ListParentPositions,
  GetOpenPosition,
):
  """HTTP indexer data endpoint group."""

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

@dataclass
class GetAssetPositions(IndexerMixin):
  """Endpoint mixin for asset_positions."""
  async def get_asset_positions(
    self,
    address: str,
    *,
    subaccount: int,
    status: Literal['OPEN', 'CLOSED', 'LIQUIDATED'] | None = None,
    limit: int | None = None,
    created_before_or_at_height: int | None = None,
    created_before_or_at: datetime | None = None,
    validate: bool | None = None
  ) -> list[AssetPosition]:
    """Get asset positions.

    Args:
      address: Wallet address that owns the account.
      subaccount: Subaccount number.
      status: Position status filter.
      limit: Maximum number of results to return.
      created_before_or_at_height: Restrict results to entries created at or before a specific block height.
      created_before_or_at: Restrict results to entries created at or before a specific timestamp.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http/accounts#get-asset-positions)
    """
    params: dict[str, object] = {
      'address': address,
      'subaccountNumber': subaccount,
    }
    if status is not None:
      params['status'] = status
    if limit is not None:
      params['limit'] = limit
    if created_before_or_at_height is not None:
      params['createdBeforeOrAtHeight'] = created_before_or_at_height
    if created_before_or_at is not None:
      params['createdBeforeOrAt'] = ts.dump(created_before_or_at)
    r = await self.request('GET', '/v4/assetPositions', params=params)
    return parse_response(r, validate=self.validate(validate))

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]

class CandlesResponse(TypedDict):
  """Candles response payload."""
  candles: list[Candle]

@dataclass
class GetCandles(IndexerMixin):
  """Endpoint mixin for candles."""
  async def get_candles(
    self,
    market: str,
    *,
    resolution: Literal[
      '1MIN', '5MINS', '15MINS', '30MINS', '1HOUR', '4HOURS', '1DAY'
    ],
    from_iso: datetime | None = None,
    to_iso: datetime | None = None,
    limit: int | None = None,
    validate: bool | None = None
  ) -> CandlesResponse:
    """Fetch candles for a perpetual market.

    Args:
      market: Perpetual market ticker.
      resolution: Candle resolution.
      from_iso: Start timestamp in ISO 8601 format.
      to_iso: End timestamp in ISO 8601 format.
      limit: Maximum number of candles to return.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http#get-candles)
    """
    params: dict[str, object] = {
      'resolution': resolution,
    }
    if from_iso is not None:
      params['fromISO'] = ts.dump(from_iso)
    if to_iso is not None:
      params['toISO'] = ts.dump(to_iso)
    if limit is not None:
      params['limit'] = limit
    r = await self.request('GET', f'/v4/candles/perpetualMarkets/{market}', params=params)
    return parse_response(r, validate=self.validate(validate))

@dataclass
class GetCandlesPaged(GetCandles):
  """Endpoint mixin for candles_paged."""
  async def get_candles_paged(
    self,
    market: str,
    resolution: Resolution,
    *,
    start: datetime | None = None,
    end: datetime | None = None,
    limit: int | None = None,
  ) -> AsyncIterable[Sequence[Candle]]:
    """Page through candles for a perpetual market.

    Args:
      market: The market ticker (e.g. `'BTC-USD'`).
      resolution: The resolution of the candles.
      start: If given, fetches candles starting from the given timestamp.
      end: If given, fetches candles up to and including the given timestamp.
      limit: The max. number of candles to retrieve per request (default: 1000, max: 1000).

    Returns:
      An async iterable of candle pages. Each request advances by the last candle timestamp until an empty page is returned.
    """
    last_time = end
    while True:
      response = await self.get_candles(
        market,
        resolution=resolution,
        from_iso=start,
        to_iso=last_time,
        limit=limit,
      )
      candles = response['candles']
      if not candles:
        break
      yield candles
      last_time = candles[-1]['startedAt']

class ComplianceV2Response(TypedDict):
  """Compliance v2 response payload."""
  status: Literal['COMPLIANT', 'FIRST_STRIKE_CLOSE_ONLY', 'FIRST_STRIKE', 'CLOSE_ONLY', 'BLOCKED']
  reason: NotRequired[Literal['MANUAL', 'US_GEO', 'CA_GEO', 'GB_GEO', 'SANCTIONED_GEO', 'COMPLIANCE_PROVIDER'] | None]
  updatedAt: NotRequired[datetime | None]

@dataclass
class GetComplianceScreen(IndexerMixin):
  """Endpoint mixin for compliance_screen."""
  async def get_compliance_screen(
    self,
    address: str,
    validate: bool | None = None
  ) -> ComplianceV2Response:
    """Get compliance screen.

    Args:
      address: EVM or dYdX address.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http/utility#get-compliance-screen)
    """
    r = await self.request('GET', f'/v4/compliance/screen/{address}')
    return parse_response(r, validate=self.validate(validate))

@dataclass
class GetMarkets(IndexerMixin):
  """Endpoint mixin for markets."""
  async def get_markets(
    self,
    *,
    market: str | None = None,
    limit: int | None = None,
    validate: bool | None = None
  ) -> GetMarketsResponse:
    """Fetch perpetual market metadata.

    Args:
      market: Specific market ticker to retrieve, for example `BTC-USD`.
      limit: Maximum number of market entries to return.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http#get-perpetual-markets)
    """
    params: dict[str, object] = {}
    if market is not None:
      params['market'] = market
    if limit is not None:
      params['limit'] = limit
    r = await self.request('GET', '/v4/perpetualMarkets', params=params)
    return parse_response(r, validate=self.validate(validate))

class GetMarketsResponse(TypedDict):
  """Endpoint mixin for markets_response."""
  markets: dict[str, PerpetualMarket]

@dataclass
class GetMarket(GetMarkets):
  """Endpoint mixin for market."""
  async def get_market(
    self,
    market: str,
    *,
    validate: bool | None = None,
  ) -> PerpetualMarket:
    """Fetch one perpetual market by ticker.

    Args:
      market: Market ticker.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.
    """
    response = await self.get_markets(market=market, limit=1, validate=validate)
    return response['markets'][market] if 'markets' in response else response[market]

class Fill(TypedDict):
  """Fill payload."""
  id: str
  side: Literal['BUY', 'SELL']
  liquidity: Literal['MAKER', 'TAKER']
  type: Literal['LIMIT', 'LIQUIDATED', 'LIQUIDATION', 'DELEVERAGED', 'OFFSETTING']
  market: str
  marketType: Literal['PERPETUAL', 'SPOT']
  price: Decimal
  size: Decimal
  fee: Decimal
  affiliateRevShare: Decimal
  createdAt: datetime
  createdAtHeight: str
  orderId: NotRequired[str | None]
  clientMetadata: NotRequired[str | None]
  subaccountNumber: int
  builderFee: NotRequired[Decimal | None]
  builderAddress: NotRequired[str | None]
  positionSizeBefore: NotRequired[Decimal | None]
  entryPriceBefore: NotRequired[Decimal | None]
  positionSideBefore: NotRequired[Literal['LONG', 'SHORT'] | None]

class FillsResponse(TypedDict):
  """Fills response payload."""
  fills: list[Fill]

@dataclass
class GetFills(IndexerMixin):
  """Endpoint mixin for fills."""
  async def get_fills(
    self,
    address: str,
    *,
    subaccount: int,
    market: str | None = None,
    market_type: Literal['PERPETUAL', 'SPOT'] | None = None,
    created_before_or_at_height: int | None = None,
    created_before_or_at: datetime | None = None,
    limit: int | None = None,
    page: int | None = None,
    validate: bool | None = None
  ) -> FillsResponse:
    """Fetch fills for a subaccount.

    Args:
      address: Wallet address that owns the subaccount.
      subaccount: Subaccount number.
      market: Market ticker filter.
      market_type: Market type filter.
      created_before_or_at_height: Latest block height to include.
      created_before_or_at: Latest timestamp to include.
      limit: Maximum number of fills to return.
      page: Page number for paginated results.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http#get-fills)
    """
    params: dict[str, object] = {
      'address': address,
      'subaccountNumber': subaccount,
    }
    if market is not None:
      params['market'] = market
    if market_type is not None:
      params['marketType'] = market_type
    if created_before_or_at_height is not None:
      params['createdBeforeOrAtHeight'] = created_before_or_at_height
    if created_before_or_at is not None:
      params['createdBeforeOrAt'] = ts.dump(created_before_or_at)
    if limit is not None:
      params['limit'] = limit
    if page is not None:
      params['page'] = page
    r = await self.request('GET', '/v4/fills', params=params)
    return parse_response(r, validate=self.validate(validate))

@dataclass
class GetFillsPaged(GetFills):
  """Endpoint mixin for fills_paged."""
  def get_fills_paged(
    self, address: str, *,
    subaccount: int,
    market: str | None = None,
    market_type: MarketType | None = None,
    created_before_or_at_height: int | None = None,
    created_before_or_at: datetime | None = None,
    limit: int | None = None,
    validate: bool | None = None,
  ) -> PaginatedResponse[Fill, int]:
    """Page through fills for a subaccount.

    Args:
      address: The wallet address that owns the account.
      subaccount: The identifier for the specific subaccount within the wallet address.
      market: The market name (e.g. `'BTC-USD'`).
      market_type: The market type (`'PERPETUAL'` or `'SPOT'`). Must be provided if `market` is provided.
      created_before_or_at_height: If given, fetches fills up to and including the given block height.
      created_before_or_at: If given, fetches fills up to and including the given timestamp.
      limit: The max. number of fills to retrieve (default: 1000, max: 1000).
      validate: Override the client response validation default for this call.

    Returns:
      A paginated response. Each request advances by page number until the endpoint returns no items.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http#get-fills)
    """
    async def next(page: int) -> tuple[list[Fill], int | None]:
      """Next.

      Args:
        page: Page number to request.
      """
      response = await self.get_fills(
        address,
        subaccount=subaccount,
        market=market,
        market_type=market_type,
        created_before_or_at_height=created_before_or_at_height,
        created_before_or_at=created_before_or_at,
        limit=limit,
        page=page,
        validate=validate,
      )
      fills = response['fills']
      next_page = page + 1 if fills else None
      return fills, next_page

    return PaginatedResponse(1, next)

class FundingPayment(TypedDict):
  """FundingPayment payload."""
  createdAt: datetime
  createdAtHeight: str
  perpetualId: str
  ticker: str
  oraclePrice: Decimal
  size: Decimal
  side: Literal['LONG', 'SHORT']
  rate: Decimal
  payment: Decimal
  subaccountNumber: str
  fundingIndex: Decimal

class FundingPaymentsResponse(TypedDict):
  """Funding payments response payload."""
  fundingPayments: list[FundingPayment]

@dataclass
class GetFundingPayments(IndexerMixin):
  """Endpoint mixin for funding_payments."""
  async def get_funding_payments(
    self,
    address: str,
    *,
    subaccount: int,
    ticker: str | None = None,
    after_or_at: datetime | None = None,
    limit: int | None = None,
    page: int | None = None,
    validate: bool | None = None
  ) -> FundingPaymentsResponse:
    """Fetch funding payments for a subaccount.

    Args:
      address: Wallet address that owns the subaccount.
      subaccount: Subaccount number.
      ticker: Market ticker filter.
      after_or_at: Earliest timestamp to include.
      limit: Maximum number of funding payments to return.
      page: Page number for paginated results.
      validate: Override the client response validation default for this call.

    Returns:
      The validated indexer response payload.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http#get-funding-payments)
    """
    params: dict[str, object] = {
      'address': address,
      'subaccountNumber': subaccount,
    }
    if ticker is not None:
      params['ticker'] = ticker
    if after_or_at is not None:
      params['afterOrAt'] = ts.dump(after_or_at)
    if limit is not None:
      params['limit'] = limit
    if page is not None:
      params['page'] = page
    r = await self.request('GET', '/v4/fundingPayments', params=params)
    return parse_response(r, validate=self.validate(validate))

@dataclass
class GetFundingPaymentsPaged(GetFundingPayments):
  """Endpoint mixin for funding_payments_paged."""
  def get_funding_payments_paged(
    self,
    address: str,
    *,
    subaccount: int,
    ticker: str | None = None,
    after_or_at: datetime | None = None,
    limit: int | None = None,
    validate: bool | None = None,
  ) -> PaginatedResponse[FundingPayment, int]:
    """Page through funding payments for a subaccount.

    Args:
      address: The wallet address that owns the account.
      subaccount: The identifier for the specific subaccount within the wallet address.
      ticker: The market ticker (e.g. `'BTC-USD'`).
      after_or_at: If given, fetches funding payments starting from the given timestamp.
      limit: The max. number of funding payments to retrieve.
      validate: Override the client response validation default for this call.

    Returns:
      A paginated response. Each request advances by page number until the endpoint returns no items.

    References:
      - [dYdX API docs](https://docs.dydx.xyz/indexer-client/http#get-funding-payments)
    """
    async def next(page: int) -> tuple[list[FundingPayment], int | None]:
      """Next.

      Args:
        page: Page number to request.
      """
      response = await self.get_funding_payments(
        address,
        subaccount=subaccount,
        ticker=ticker,
        after_or_at=after_or_at,
        limit=limit,
        page=page,
        validate=validate,
      )
      funding_payments = response['fundingPayments']
      next_page = page + 1 if funding_payments else None
      return funding_payments, next_page

    return PaginatedResponse(1, next)