API Reference

This reference documents the main classes and methods in the Meridian Python SDK.

Client Classes

AsyncRESTClient

The primary asynchronous client for interacting with the Meridian API via REST endpoints. Recommended for all new applications.

from meridian import AsyncRESTClient

# Create async client (use within async function)
client = await AsyncRESTClient.create({
    "base_url": "https://api.meridian.xyz",
    "chain_config": {
        "rpc_url": "https://rpc.mainnet.chain.robinhood.com",
        "private_key": "your_private_key",  # optional, required for signing
    }
})

# Use the client
products = await client.products()
subaccounts = await client.subaccounts()

# Remember to close when done
await client.close()

meridian.async_rest_client.AsyncRESTClient

Bases: AsyncHTTPClient

Asynchronous REST client for the Meridian API.

Notes for maintainers: - This client composes endpoint functions from the meridian.rest.* modules by assigning them as attributes on the class (see below). Each function expects self to provide get, post, and get_validated from AsyncHTTPClient and to expose _models for the active network. - Network-specific models are accessed via self._models which is set based on the configured network via meridian.models.load_models. - Use AsyncRESTClient.create(...) to ensure async initialization (RPC config, optional chain client) happens before use. Remember to await client.close() to release the underlying httpx.AsyncClient. - We intentionally avoid “async properties”. For convenience methods that derive indices (e.g., products by ticker/id), use explicit async methods like products_by_ticker() or products_by_id() that fetch fresh data each call to keep behavior lightweight and predictable.

Source code in meridian/async_rest_client.py

class AsyncRESTClient(AsyncHTTPClient):
    """Asynchronous REST client for the Meridian API.

    Notes for maintainers:
    - This client composes endpoint functions from the `meridian.rest.*` modules
      by assigning them as attributes on the class (see below). Each function
      expects `self` to provide `get`, `post`, and `get_validated` from
      AsyncHTTPClient and to expose `_models` for the active network.
    - Network-specific models are accessed via `self._models` which is set based
      on the configured network via `meridian.models.load_models`.
    - Use `AsyncRESTClient.create(...)` to ensure async initialization (RPC
      config, optional chain client) happens before use. Remember to `await
      client.close()` to release the underlying `httpx.AsyncClient`.
    - We intentionally avoid “async properties”. For convenience methods that
      derive indices (e.g., products by ticker/id), use explicit async methods
      like `products_by_ticker()` or `products_by_id()` that fetch fresh
      data each call to keep behavior lightweight and predictable.
    """

    list_funding = list_funding
    list_projected_funding = list_projected_funding
    get_order = get_order
    list_fills = list_fills
    list_order_group = list_order_group
    list_orders = list_orders
    list_trades = list_trades
    prepare_order = prepare_order
    sign_order = sign_order
    submit_order = submit_order
    submit_batch_order = submit_batch_order
    prepare_batch_order = prepare_batch_order
    dry_run_order = dry_run_order
    prepare_cancel_order = prepare_cancel_order
    sign_cancel_order = sign_cancel_order
    submit_cancel_order = submit_cancel_order
    prepare_cancel_all_order = prepare_cancel_all_order
    sign_cancel_all_order = sign_cancel_all_order
    submit_cancel_all_order = submit_cancel_all_order
    cancel_all_orders = cancel_all_orders
    get_signer = get_signer
    get_signer_by_address = get_signer_by_address
    get_signer_quota = get_signer_quota
    list_signers = list_signers
    prepare_linked_signer = prepare_linked_signer
    sign_linked_signer = sign_linked_signer
    link_linked_signer = link_linked_signer
    prepare_revoke_linked_signer = prepare_revoke_linked_signer
    sign_revoke_linked_signer = sign_revoke_linked_signer
    revoke_linked_signer = revoke_linked_signer
    prepare_refresh_linked_signer = prepare_refresh_linked_signer
    sign_refresh_linked_signer = sign_refresh_linked_signer
    refresh_linked_signer = refresh_linked_signer
    prepare_extend_linked_signer = prepare_extend_linked_signer
    sign_extend_linked_signer = sign_extend_linked_signer
    extend_linked_signer = extend_linked_signer
    list_positions = list_positions
    get_position = get_position
    get_active_position = get_active_position
    list_position_fills = list_position_fills
    list_position_liquidations = list_position_liquidations
    list_position_fees = list_position_fees
    list_projected_position_fees = list_projected_position_fees
    get_market_liquidity = get_market_liquidity
    get_product = get_product
    list_mark_price_gaps = list_mark_price_gaps
    list_market_prices = list_market_prices
    list_products = list_products
    get_rpc_config = get_rpc_config
    list_subaccounts = list_subaccounts
    list_all_subaccounts = list_all_subaccounts
    get_subaccount = get_subaccount
    get_subaccount_balances = get_subaccount_balances
    get_subaccount_balance_history = get_subaccount_balance_history
    get_subaccount_unrealized_pnl_history = get_subaccount_unrealized_pnl_history
    get_subaccount_volume_history = get_subaccount_volume_history
    get_subaccount_funding_history = get_subaccount_funding_history
    get_subaccount_position_fee_history = get_subaccount_position_fee_history
    get_subaccount_total_volume = get_subaccount_total_volume
    get_referral_summary = get_referral_summary
    list_referrals = list_referrals
    get_maintenance_status = get_maintenance_status
    get_rate_limit_config = get_rate_limit_config
    get_system_time = get_system_time
    get_token = get_token
    list_token_withdraws = list_token_withdraws
    list_tokens = list_tokens
    list_token_transfers = list_token_transfers
    prepare_withdraw_token = prepare_withdraw_token
    sign_withdraw_token = sign_withdraw_token
    get_withdraw_digest = get_withdraw_digest
    submit_withdraw_token = submit_withdraw_token
    prepare_convert_token = prepare_convert_token
    sign_convert_token = sign_convert_token
    submit_convert_token = submit_convert_token
    get_last_price_config = get_last_price_config
    get_last_price_symbol = get_last_price_symbol
    search_last_price_symbols = search_last_price_symbols
    get_last_price_history = get_last_price_history
    get_last_price_time = get_last_price_time
    get_oracle_price_config = get_oracle_price_config
    get_oracle_price_symbol = get_oracle_price_symbol
    search_oracle_price_symbols = search_oracle_price_symbols
    get_oracle_price_history = get_oracle_price_history
    get_oracle_price_time = get_oracle_price_time

    def __init__(self, config: Union[Dict[str, Any], RESTConfig] = {}):
        self.config = RESTConfig.model_validate(config)
        network = self.config.network or "testnet"

        if not self.config.base_url:
            self.config.base_url = AnyHttpUrl(
                NETWORK_URLS.get(network, NETWORK_URLS["testnet"])
            )

        if not self.config.archive_base_url:
            self.config.archive_base_url = AnyHttpUrl(
                ARCHIVE_NETWORK_URLS.get(network, ARCHIVE_NETWORK_URLS["testnet"])
            )

        if not self.config.tv_base_url:
            self.config.tv_base_url = AnyHttpUrl(
                TV_NETWORK_URLS.get(network, TV_NETWORK_URLS["testnet"])
            )

        self._models = load_models("rest", network)
        self._archive_models = load_models("archive", network)
        self._tv_models = load_models("tv", network)
        super().__init__(self.config)

        self._archive_base_url = self.config.archive_base_url
        self._tv_base_url = self.config.tv_base_url
        self.chain: Optional[ChainClient] = None
        self.rpc_config: Optional[RpcConfigDto] = None
        self.private_key: Optional[str] = None
        self.web3: Optional[Any] = None
        self.default_time_in_force = self.config.default_time_in_force
        self.default_post_only = self.config.default_post_only
        self._subaccounts: Optional[List[SubaccountDto]] = None
        self._products: Optional[List[ProductDto]] = None
        self._tokens: Optional[List[TokenDto]] = None
        self._products_by_ticker: Optional[Dict[str, ProductDto]] = None
        self._products_by_id: Optional[Dict[UUID, ProductDto]] = None

    @classmethod
    async def create(
        cls, config: Union[Dict[str, Any], RESTConfig] = {}
    ) -> AsyncRESTClient:
        """Factory method to create and asynchronously initialize the client.

        Args:
            config (Union[Dict[str, Any], RESTConfig], optional): Configuration dictionary or RESTConfig object. Optional fields include:
                private_key (str, optional): The private key.
                base_url (str, optional): Base URL for REST requests. Defaults to testnet.
                timeout (int, optional): Timeout in seconds for REST requests.
                verbose (bool, optional): Enables debug logging. Defaults to False.
                rate_limit_headers (bool, optional): Enables rate limit headers. Defaults to False.
                chain_config (ChainConfig, optional): Chain configuration for signing transactions.

        Returns:
            AsyncRESTClient: Fully initialized async client instance.
        """
        client = cls(config)
        await client._async_init()
        return client

    async def _async_init(self):
        """Asynchronous initialization.

        Loads RPC configuration and (optionally) initializes the chain client if
        `chain_config` is provided. This split constructor pattern keeps
        `__init__` sync, while `create()` ensures the instance is fully ready.
        """
        self.rpc_config = await self.get_rpc_config()
        tokens = await self.list_tokens()
        if self.config.chain_config:
            self._init_chain_client(self.config.chain_config, self.rpc_config, tokens)
        self.private_key = self.chain.private_key if self.chain else None
        self.web3 = self.chain.web3 if self.chain else None
        # No eager cache priming; caches are populated on first access

    def _init_chain_client(
        self,
        config: Union[Dict[str, Any], ChainConfig],
        rpc_config: Optional[RpcConfigDto] = None,
        tokens: Optional[List[TokenDto]] = None,
    ):
        """Initialize the ChainClient for transaction signing.

        Args:
            config (Union[Dict[str, Any], ChainConfig]): The chain configuration.
            rpc_config (RpcConfigDto, optional): RPC configuration. Defaults to None.
            tokens (List[TokenDto], optional): List of token configurations. Defaults to None.
        """
        config = ChainConfig.model_validate(config)
        try:
            self.chain = ChainClient(config, rpc_config, tokens)
            self.logger.debug("Chain client initialized successfully")
        except Exception as e:
            self.logger.debug(f"Failed to initialize chain client: {e}")

    async def _get_pages(
        self,
        endpoint: str,
        request_model: type[BaseModel],
        response_model: type[BaseModel],
        paginate: bool = False,
        max_pages: int = 10,
        **kwargs,
    ) -> Any:
        """Make a GET request with validated parameters and response, handling pagination.

        Args:
            endpoint (str): API endpoint path (e.g. "order" will be appended to the base URL and prefix to form "/v1/order")
            request_model (type[BaseModel]): Pydantic model for request validation
            response_model (type[BaseModel]): Pydantic model for response validation
            paginate (bool, optional): If True, fetch all pages and return aggregated list. Defaults to False.
            max_pages (int, optional): Maximum number of pages to fetch when paginating. Defaults to 10.

        Other Parameters:
            **kwargs: Parameters to validate and include in the request

        Returns:
            Any: List of validated response objects. Single page if paginate=False, all pages if paginate=True.
        """
        first_page = await self.get_validated(
            url_path=f"{API_PREFIX}/{endpoint}",
            request_model=request_model,
            response_model=response_model,
            **kwargs,
        )
        # `get_validated` already returns an instance of `response_model` with
        # `.data`, `.has_next`, and `.next_cursor`.
        if not paginate:
            return list(getattr(first_page, "data"))

        all_items: List[Any] = list(getattr(first_page, "data"))
        current_cursor = getattr(first_page, "next_cursor", None)
        has_next = getattr(first_page, "has_next", False)

        page_count = 1
        while has_next and current_cursor and page_count <= max_pages:
            page = await self.get_validated(
                url_path=f"{API_PREFIX}/{endpoint}",
                request_model=request_model,
                response_model=response_model,
                cursor=current_cursor,
                **kwargs,
            )
            all_items.extend(getattr(page, "data"))
            has_next = getattr(page, "has_next", False)
            current_cursor = getattr(page, "next_cursor", None)
            page_count += 1

        return all_items

    async def subaccounts(self, refresh: bool = False) -> List[SubaccountDto]:
        """Get the list of subaccounts.

        Args:
            refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

        Returns:
            List[SubaccountDto]: List of subaccount objects for the connected wallet address.

        Raises:
            ValueError: If no chain client is configured or address is unavailable.
        """
        if not self.chain or not getattr(self.chain, "address", None):
            raise ValueError("Chain address is required to list subaccounts")
        if refresh or self._subaccounts is None:
            self._subaccounts = await self.list_subaccounts(
                sender=self.chain.address, order_by="createdAt", order="asc"
            )
        return self._subaccounts

    async def products(self, refresh: bool = False) -> List[ProductDto]:
        """Get the list of products.

        Args:
            refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

        Returns:
            List[ProductDto]: List of product objects.
        """
        if refresh or self._products is None:
            self._products = await self.list_products()
            self._products_by_id = None
            self._products_by_ticker = None
        return self._products

    async def tokens(self, refresh: bool = False) -> List[TokenDto]:
        """Get the list of tokens.

        Args:
            refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

        Returns:
            List[TokenDto]: List of token objects.
        """
        if refresh or self._tokens is None:
            self._tokens = await self.list_tokens()
        return self._tokens

    async def products_by_ticker(self, refresh: bool = False) -> Dict[str, ProductDto]:
        """Get the products indexed by ticker.

        Args:
            refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

        Returns:
            Dict[str, ProductDto]: Dictionary of products keyed by ticker.
        """
        if refresh or self._products_by_ticker is None:
            products = await self.products(refresh=refresh)
            self._products_by_ticker = {
                p.ticker: p for p in products if getattr(p, "ticker", None)
            }
        return self._products_by_ticker

    async def products_by_id(self, refresh: bool = False) -> Dict[UUID, ProductDto]:
        """Get the products indexed by ID.

        Args:
            refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

        Returns:
            Dict[UUID, ProductDto]: Dictionary of products keyed by ID.
        """
        if refresh or self._products_by_id is None:
            products = await self.products(refresh=refresh)
            self._products_by_id = {p.id: p for p in products}
        return self._products_by_id

    async def get_maintenance_margin(
        self,
        subaccount_id: UUID,
        positions: Optional[List[PositionDto] | List[Dict[str, Any]]] = None,
        products: Optional[List[ProductDto] | List[Dict[str, Any]]] = None,
        product_ids: Optional[List[UUID]] = None,
    ) -> Decimal:
        """Calculate the an account's maintenance margin for specified positions or products.

        Args:
            subaccount_id (str): Fetch positions for this subaccount when ``positions`` is not supplied.
            positions (List[PositionDto] | List[Dict[str, Any]], optional): Pre-fetched positions to use directly.
            products (List[ProductDto] | List[Dict[str, Any]], optional): Pre-fetched products used to filter
                the calculation.
            product_ids (List[str] | List[UUID], optional): Filters the calculation to these product IDs.

        Returns:
            Decimal: Total maintenance margin for the filtered positions.

        Raises:
            ValueError: If neither positions nor subaccount context is provided, or if any
                referenced product cannot be resolved.
        """

        class PartialPosition(BaseModel):
            product_id: UUID
            cost: str

        class PartialProduct(BaseModel):
            id: UUID
            max_leverage: float
            taker_fee: str

        if positions is None:
            positions = await self.list_positions(subaccount_id=subaccount_id)

        if products and product_ids:
            raise ValueError("Can only specify one of products and product_ids")

        if products is not None:
            valid_products = [
                PartialProduct(
                    **p if isinstance(p, dict) else p.model_dump(mode="json")
                )
                for p in products
            ]
            products_by_id: Dict[UUID, PartialProduct] = {
                product.id: product for product in valid_products
            }
        else:
            raw_products_by_id = await self.products_by_id()
            products_by_id = {
                UUID(str(k)): PartialProduct(
                    **v if isinstance(v, dict) else v.model_dump(mode="json")
                )
                for k, v in raw_products_by_id.items()
            }

        product_ids_set: Optional[Set[UUID]] = None
        if products is not None:
            product_ids_set = set(products_by_id.keys())
        if product_ids is not None:
            product_ids_set = set([UUID(str(pid)) for pid in product_ids])
            products_by_id_keys = set(products_by_id.keys())
            missing_products = product_ids_set - products_by_id_keys
            if missing_products:
                missing = ", ".join(sorted([str(p) for p in missing_products]))
                raise ValueError(f"Products not found for calculation: {missing}")

        valid_positions = [
            PartialPosition(**p if isinstance(p, dict) else p.model_dump(mode="json"))
            for p in positions
        ]
        positions_filtered = [
            position
            for position in valid_positions
            if product_ids_set is None or position.product_id in product_ids_set
        ]

        if not positions_filtered:
            return Decimal("0")

        total_mm = Decimal("0")
        for position in positions_filtered:
            product = products_by_id.get(position.product_id)
            if product is None:
                raise ValueError(
                    f"Product '{position.product_id}' not found for position '{position.product_id}'"
                )

            notional = abs(Decimal(position.cost))
            max_leverage = Decimal(str(product.max_leverage))
            taker_fee_rate = Decimal(product.taker_fee)

            mmr = Decimal("1") / (max_leverage * Decimal("2"))
            total_mm += notional * mmr
            total_mm += notional * taker_fee_rate

        return total_mm

    async def get_tokens(self) -> List[TokenDto]:
        """Return the latest list of tokens (no caching)."""
        return await self.list_tokens()

    async def _resolve_subaccount_name(self, subaccount_id: UUID) -> str:
        """Resolve a subaccount name from its ID, using the cached list first."""
        for sa in await self.subaccounts():
            if sa.id == subaccount_id:
                return sa.name
        return (await self.get_subaccount(id=subaccount_id)).name

    async def _resolve_sender_and_subaccount(
        self, sender: Optional[str] = None, subaccount: Optional[str] = None
    ) -> Tuple[str, str]:
        if sender is None and self.chain:
            sender = self.chain.address
            warnings.warn(
                "Automatically resolving 'sender' parameter is deprecated and will be removed in a future version. "
                "Always explicitly pass 'sender' and 'subaccount' to avoid issues while using multiple account. "
                f"Using sender='{sender}'",
                DeprecationWarning,
                stacklevel=3,
            )
        if sender is None:
            raise ValueError(
                "Could not resolve 'sender'. Either pass 'sender' explicitly or initialize the client with a chain configuration."
            )
        if subaccount is None:
            subaccounts = await self.subaccounts()
            if not subaccounts:
                raise ValueError(
                    "No subaccounts found for this account. Please create a subaccount first."
                )
            self.logger.debug(
                f"First subaccount name: '{subaccounts[0].name}', id: '{subaccounts[0].id}'"
            )
            self.logger.debug(f"All subaccount names: {[s.name for s in subaccounts]}")
            subaccount = subaccounts[0].name
            warnings.warn(
                "Automatically resolving 'subaccount' parameter is deprecated and will be removed in a future version. "
                "Always explicitly pass 'subaccount' and 'subaccount' to avoid issues while using multiple account. "
                f"Using subaccount='{subaccount}'",
                DeprecationWarning,
                stacklevel=3,
            )
        return sender, subaccount

    async def link_signer(
        self,
        signer: str,
        subaccount_id: UUID,
        subaccount: Optional[str] = None,
        sender: Optional[str] = None,
        signer_private_key: Optional[str] = None,
        nonce: Optional[str] = None,
        signed_at: Optional[int] = None,
        sign_sender: bool = True,
        sign_signer: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[SignerDto, LinkSignerDto]:
        """Prepare, sign, and optionally submit a linked signer payload.

        Args:
            signer (str): Address being linked as a delegate signer. Required.
            subaccount_id (UUID): Target subaccount id. Required.
            subaccount (str, optional): Bytes32 subaccount name. If omitted, fetched by id.
            sender (str, optional): Owner address. Defaults to the client's chain address.
            signer_private_key (str, optional): Private key for the signer address.
            nonce (str, optional): Custom nonce for the signature.
            signed_at (int, optional): Seconds since epoch for the signature timestamp.
            sign_sender (bool, optional): Sign with the owner key. Defaults to True.
            sign_signer (bool, optional): Sign with the signer key. Defaults to True.
            submit (bool, optional): Submit to the API. If False, return the prepared DTO.

        Returns:
            Union[SignerDto, LinkSignerDto]: Submitted record or the prepared/signable payload.

        Raises:
            ValueError: If the subaccount name cannot be resolved or signer signature is missing.
        """
        if subaccount is None:
            subaccount = await self._resolve_subaccount_name(subaccount_id)
        sender, subaccount = await self._resolve_sender_and_subaccount(
            sender, subaccount
        )

        link = await self.prepare_linked_signer(
            sender=sender,
            signer=signer,
            subaccount=subaccount,
            subaccount_id=subaccount_id,
            include_signature=False,
            nonce=nonce,
            signed_at=signed_at,
        )

        if sign_sender:
            link = await self.sign_linked_signer(link)

        if sign_signer:
            if signer_private_key:
                link = await self.sign_linked_signer(
                    link, signer_private_key=signer_private_key
                )
            elif not link.signer_signature:
                raise ValueError(
                    "signer_private_key or signer_signature is required when sign_signer=True"
                )

        return link if not submit else await self.link_linked_signer(link, **kwargs)

    async def revoke_signer(
        self,
        signer: str,
        subaccount_id: UUID,
        subaccount: Optional[str] = None,
        sender: Optional[str] = None,
        sign: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[SignerDto, RevokeLinkedSignerDto]:
        """Prepare, sign, and optionally submit a revoke-linked-signer payload.

        Args:
            signer (str): Address of the delegate to revoke. Required.
            subaccount_id (UUID): Target subaccount id. Required.
            subaccount (str, optional): Bytes32 subaccount name. If omitted, fetched by id.
            sender (str, optional): Owner address. Defaults to the client's chain address.
            sign (bool, optional): Sign with the owner key. Defaults to True.
            submit (bool, optional): Submit to the API. If False, return the prepared DTO.

        Returns:
            Union[SignerDto, RevokeLinkedSignerDto]: Submitted record or the prepared/signable payload.

        Raises:
            ValueError: If the subaccount name cannot be resolved.
        """
        if subaccount is None:
            subaccount = await self._resolve_subaccount_name(subaccount_id)
        sender, subaccount = await self._resolve_sender_and_subaccount(
            sender, subaccount
        )

        revoke = await self.prepare_revoke_linked_signer(
            sender=sender,
            signer=signer,
            subaccount=subaccount,
            subaccount_id=subaccount_id,
            include_signature=False,
            **kwargs,
        )

        if sign:
            revoke = await self.sign_revoke_linked_signer(revoke)

        return (
            revoke if not submit else await self.revoke_linked_signer(revoke, **kwargs)
        )

    async def withdraw_token(
        self,
        token_id: UUID,
        amount: int,
        destination_address: str,
        destination_endpoint: int,
        subaccount: Optional[str] = None,
        account: Optional[str] = None,
        nonce: Optional[str] = None,
        signed_at: Optional[int] = None,
        sign: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[WithdrawDto, InitiateWithdrawDto]:
        """Prepares, signs, and optionally submits a token withdrawal.

        This is the primary method for initiating token withdrawals. It handles
        the complete withdrawal flow: preparing the withdrawal payload, signing
        it with EIP-712, and submitting to the API. To submit a payload you
        already prepared and signed, use ``submit_withdraw_token`` instead.

        Example:
            Withdraw 100 USD to the connected wallet::

                result = await client.withdraw_token(
                    token_id=usd_token.id,
                    amount=100,
                    destination_address=client.chain.address,
                    destination_endpoint=0,
                )

            Prepare without submitting (for inspection or manual submission)::

                payload = await client.withdraw_token(
                    token_id=usd_token.id,
                    amount=100,
                    destination_address="0x...",
                    destination_endpoint=0,
                    submit=False,
                )

        Args:
            token_id: UUID of the token to withdraw.
            amount: Amount to withdraw in the token's base units.
            destination_address: The LayerZero destination address where funds
                will be sent. Automatically padded to bytes32 format.
            destination_endpoint: LayerZero endpoint ID for the destination chain.
                Use 0 for same-chain withdrawals.
            subaccount: Hex-encoded subaccount name. If not provided, defaults to
                the first subaccount associated with the connected wallet.
            account: Recipient wallet address. If not provided, defaults to the
                connected wallet's address.
            nonce: Custom nonce for the EIP-712 signature. If not provided, a
                random nonce is generated.
            signed_at: Unix timestamp (seconds) for the signature. If not provided,
                defaults to the current time.
            sign: Whether to sign the withdrawal payload. Defaults to True.
            submit: Whether to submit the withdrawal to the API. Set to False to
                receive the prepared payload without submitting. Defaults to True.

        Returns:
            If ``submit=True``, returns a WithdrawDto containing the withdrawal
            record created by the API. If ``submit=False``, returns the prepared
            InitiateWithdrawDto payload.
        """
        # Resolve token address from token_id
        token = await self.get_token(id=token_id)
        token_addr = token.address

        # Resolve account/subaccount
        account_addr, subaccount_name = await self._resolve_sender_and_subaccount(
            account, subaccount
        )

        withdraw_dto = await self.prepare_withdraw_token(
            subaccount=subaccount_name,
            token=token_addr,
            amount=amount,
            account=account_addr,
            destination_address=destination_address,
            destination_endpoint=destination_endpoint,
            include_signature=False,
            nonce=nonce,
            signed_at=signed_at,
        )

        if sign:
            withdraw_dto = await self.sign_withdraw_token(withdraw_dto)

        if not submit:
            return withdraw_dto
        return await self.submit_withdraw_token(
            dto=withdraw_dto, token_id=token_id, **kwargs
        )

    async def convert_token(
        self,
        from_token_id: UUID,
        to_token_id: UUID,
        amount: int,
        subaccount: Optional[str] = None,
        sender: Optional[str] = None,
        nonce: Optional[str] = None,
        signed_at: Optional[int] = None,
        sign: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[ConversionDto, InitiateConversionDto]:
        """Prepares, signs, and optionally submits a token conversion.

        This is the primary method for converting between tokens. It handles
        the complete conversion flow: preparing the conversion payload, signing
        it with EIP-712, and submitting to the API. To submit a payload you
        already prepared and signed, use ``submit_convert_token`` instead.

        Example:
            Convert 100 of one token into another::

                result = await client.convert_token(
                    from_token_id=from_token.id,
                    to_token_id=to_token.id,
                    amount=100,
                )

            Prepare without submitting (for inspection or manual submission)::

                payload = await client.convert_token(
                    from_token_id=from_token.id,
                    to_token_id=to_token.id,
                    amount=100,
                    submit=False,
                )

        Args:
            from_token_id: UUID of the token to convert from.
            to_token_id: UUID of the token to convert to.
            amount: Amount to convert in the token's base units.
            subaccount: Hex-encoded subaccount name. If not provided, defaults to
                the first subaccount associated with the connected wallet.
            sender: Account or linked signer address placing the request. If not
                provided, defaults to the connected wallet's address.
            nonce: Custom nonce for the EIP-712 signature. If not provided, a
                random nonce is generated.
            signed_at: Unix timestamp (seconds) for the signature. If not provided,
                defaults to the current time.
            sign: Whether to sign the conversion payload. Defaults to True.
            submit: Whether to submit the conversion to the API. Set to False to
                receive the prepared payload without submitting. Defaults to True.

        Returns:
            If ``submit=True``, returns a ConversionDto containing the conversion
            record created by the API. If ``submit=False``, returns the prepared
            InitiateConversionDto payload.
        """
        # Resolve token addresses from token ids
        from_token = await self.get_token(id=from_token_id)
        to_token = await self.get_token(id=to_token_id)

        # Resolve sender/subaccount
        sender_addr, subaccount_name = await self._resolve_sender_and_subaccount(
            sender, subaccount
        )

        conversion_dto = await self.prepare_convert_token(
            subaccount=subaccount_name,
            from_token=from_token.address,
            to_token=to_token.address,
            amount=amount,
            sender=sender_addr,
            include_signature=False,
            nonce=nonce,
            signed_at=signed_at,
        )

        if sign:
            conversion_dto = await self.sign_convert_token(conversion_dto)

        if not submit:
            return conversion_dto
        return await self.submit_convert_token(dto=conversion_dto, **kwargs)

    async def refresh_signer(
        self,
        signer: str,
        subaccount_id: UUID,
        subaccount: Optional[str] = None,
        sender: Optional[str] = None,
        sign: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[SignerDto, RefreshLinkedSignerDto]:
        """Prepare, sign, and optionally submit a refresh-linked-signer payload.

        Args:
            signer (str): Address of the delegate to refresh. Required.
            subaccount_id (UUID): Target subaccount id. Required.
            subaccount (str, optional): Bytes32 subaccount name. If omitted, fetched by id.
            sender (str, optional): Owner address. Defaults to the client's chain address.
            sign (bool, optional): Sign with the owner key. Defaults to True.
            submit (bool, optional): Submit to the API. If False, return the prepared DTO.

        Returns:
            Union[SignerDto, RefreshLinkedSignerDto]: Submitted record or the prepared/signable payload.

        Raises:
            ValueError: If the subaccount name cannot be resolved.
        """
        if subaccount is None:
            subaccount = await self._resolve_subaccount_name(subaccount_id)
        sender, subaccount = await self._resolve_sender_and_subaccount(
            sender, subaccount
        )

        refresh = await self.prepare_refresh_linked_signer(
            sender=sender,
            signer=signer,
            subaccount=subaccount,
            subaccount_id=subaccount_id,
            include_signature=False,
            **kwargs,
        )

        if sign:
            refresh = await self.sign_refresh_linked_signer(refresh)

        return (
            refresh
            if not submit
            else await self.refresh_linked_signer(refresh, **kwargs)
        )

    async def extend_signer(
        self,
        sender: Optional[str] = None,
        sign: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[SignerDto, ExtendLinkedSignerDto]:
        """Prepare, sign, and optionally submit an extend-linked-signer payload.

        Args:
            sender (str, optional): Address of the linked signer extending itself.
                Defaults to the client's chain address.
            sign (bool, optional): Sign with the client's key. Defaults to True.
            submit (bool, optional): Submit to the API. If False, return the prepared DTO.

        Returns:
            Union[SignerDto, ExtendLinkedSignerDto]: Submitted record or the prepared/signable payload.
        """
        if sender is None:
            if not self.chain:
                raise ValueError(
                    "Could not resolve 'sender'. Either pass 'sender' explicitly or initialize the client with a chain configuration."
                )
            sender = self.chain.address

        extend = await self.prepare_extend_linked_signer(
            sender=sender,
            include_signature=False,
            **kwargs,
        )

        if sign:
            extend = await self.sign_extend_linked_signer(extend)

        return (
            extend if not submit else await self.extend_linked_signer(extend, **kwargs)
        )

    async def create_order(
        self,
        order_type: str,
        quantity: float,
        side: int,
        price: Optional[float] = None,
        ticker: Optional[str] = None,
        product_id: Optional[UUID] = None,
        client_order_id: Optional[str] = None,
        sender: Optional[str] = None,
        subaccount: Optional[str] = None,
        time_in_force: Optional[str] = None,
        post_only: Optional[bool] = None,
        reduce_only: Optional[bool] = False,
        close: Optional[bool] = None,
        stop_price: Optional[float] = None,
        stop_type: Optional[int] = None,
        expires_at: Optional[int] = None,
        group_id: Optional[str] = None,
        group_contingency_type: Optional[int] = None,
        sign: bool = True,
        dry_run: bool = False,
        submit: bool = True,
    ) -> Union[SubmitOrderCreatedDto, DryRunOrderCreatedDto, SubmitOrderDto]:
        """Create and submit an order.

        Args:
            order_type (str): 'LIMIT' or 'MARKET'. Required.
            quantity (float): Order size. Required.
            side (int): 0 for buy, 1 for sell. Required.
            price (float, optional): Limit price for LIMIT orders.
            ticker (str, optional): Ticker of the product.
            product_id (str, optional): UUID of the product.
            client_order_id (str, optional): Subaccount-scoped client-generated id (UUID or <=32 alphanumeric).
            sender (str, optional): Address placing the order. Defaults to chain address.
            subaccount (str, optional): Hex-encoded subaccount name. Defaults to first subaccount.
            time_in_force (str, optional): For LIMIT orders (e.g., 'GTC', 'GTD'). Defaults to 'GTC'.
            post_only (bool, optional): For LIMIT orders; rejects if crossing. Defaults to False.
            reduce_only (bool, optional): If True, order only reduces position. Defaults to False.
            close (bool, optional): For MARKET orders; If True, closes the position.
            stop_price (float, optional): Stop trigger price.
            stop_type (int, optional): Stop type, either 0 (take-profit) or 1 (stop-loss), requires non-zero stopPrice.
            expires_at (int, optional): Expiry timestamp for GTD.
            group_id (str, optional): Group Id (UUID) for linking orders together in OCO/OTO relationships.
            group_contingency_type (int, optional): Contingency type for order groups: 0=OTO (Order-Triggers-Order), 1=OCO (One-Cancels-Other).
            sign (bool, optional): If True, sign the payload immediately. Defaults to True.
            dry_run (bool, optional): If True, validate without execution. Defaults to False.
            submit (bool, optional): If True, submit the order. Defaults to True.

        Returns:
            Union[SubmitOrderCreatedDto, DryRunOrderCreatedDto, SubmitOrderDto]: Created order response, dry-run validation result, or prepared order payload.

        Raises:
            ValueError: If neither product_id nor ticker is provided or if order type is invalid.
        """
        sender, subaccount = await self._resolve_sender_and_subaccount(
            sender, subaccount
        )

        if product_id is not None:
            products_by_id = await self.products_by_id()
            onchain_id = products_by_id[product_id].onchain_id
        elif ticker is not None:
            products_by_ticker = await self.products_by_ticker()
            onchain_id = products_by_ticker[ticker].onchain_id
        else:
            raise ValueError("Either product_id or ticker must be provided")

        order_params: Dict[str, Any] = {
            "subaccount": subaccount,
            "side": side,
            "quantity": quantity,
            "onchain_id": onchain_id,
            "order_type": order_type,
            "client_order_id": client_order_id,
            "reduce_only": reduce_only,
            "close": close,
            "stop_price": stop_price,
            "stop_type": stop_type,
            "group_id": group_id,
            "group_contingency_type": group_contingency_type,
        }

        if order_type == "LIMIT":
            order_params.update(
                {
                    "price": price,
                    "time_in_force": time_in_force or self.default_time_in_force,
                    "post_only": post_only or self.default_post_only,
                    "expires_at": expires_at,
                }
            )
        elif order_type != "MARKET":
            raise ValueError("Invalid order type")

        order = await self.prepare_order(sender, **order_params, include_signature=sign)
        if dry_run:
            return await self.dry_run_order(order)
        elif submit:
            return await self.submit_order(order)
        else:
            return order

    async def create_batch_order(
        self,
        instructions: List[Union[SubmitOrderDto, CancelOrderDto]],
        submit: bool = True,
        **kwargs,
    ) -> Union[BatchOrderDto, BatchOrderResponseDto]:
        """Prepare and optionally submit a batch of signed order instructions.

        Args:
            instructions (List[Union[SubmitOrderDto, CancelOrderDto]]): Prepared
                SubmitOrderDto or prepared CancelOrderDto payloads.
            submit (bool, optional): If True, submit the batch order. Defaults to True.

        Other Parameters:
            **kwargs: Additional request parameters accepted by the API.

        Returns:
            BatchOrderResponseDto or BatchOrderDto: Batch response or prepared batch payload.
        """
        batch_order = await self.prepare_batch_order(instructions)
        if not submit:
            return batch_order
        return await self.submit_batch_order(batch_order, **kwargs)

    async def cancel_orders(
        self,
        order_ids: List[UUID],
        sender: Optional[str] = None,
        subaccount: Optional[str] = None,
        client_order_ids: List[str] = [],
        sign: bool = True,
        submit: bool = True,
        **kwargs,
    ) -> Union[List[CancelOrderResultDto], CancelOrderDto]:
        """Prepares and optionally submits a request to cancel multiple orders.

        Args:
            order_ids (List[str]): Order UUIDs to cancel. Required.
            sender (str, optional): Address initiating the cancellation. Defaults to chain address.
            subaccount (str, optional): Hex-encoded subaccount name. Defaults to first subaccount.
            client_order_ids (List[str], optional): Client-generated IDs to cancel. Defaults to empty list.
            sign (bool, optional): If True, sign the payload immediately. Defaults to True.
            submit (bool, optional): If True, submit the request to the API. Defaults to True.

        Other Parameters:
            **kwargs: Additional request parameters accepted by the API.

        Returns:
            Union[List[CancelOrderResultDto], CancelOrderDto]: Cancellation results per order id or prepared cancel payload.

        Raises:
            ValueError: If no order IDs or client order IDs provided for cancellation.
        """
        if len(order_ids) == 0 and len(client_order_ids) == 0:
            raise ValueError(
                "No order IDs or client order IDs provided for cancellation"
            )
        sender, subaccount = await self._resolve_sender_and_subaccount(
            sender, subaccount
        )
        try:
            prepared_cancel = await self.prepare_cancel_order(
                order_ids=order_ids,
                client_order_ids=client_order_ids,
                sender=sender,
                subaccount=subaccount,
                include_signature=sign,
                **kwargs,
            )
        except ValueError as e:
            self.logger.warning(f"Could not prepare/sign order cancellation: {e}")
            raise

        if not submit:
            return prepared_cancel

        result = await self.submit_cancel_order(
            prepared_cancel,
            **kwargs,
        )
        return result

    async def replace_order(
        self,
        order: Optional[OrderDto] = None,
        order_id: Optional[UUID] = None,
        quantity: Optional[float] = None,
        price: Optional[float] = None,
        time_in_force: Optional[str] = None,
        post_only: Optional[bool] = None,
        reduce_only: Optional[bool] = False,
    ) -> Tuple[SubmitOrderCreatedDto, bool]:
        """Replace an existing order with new parameters.

        Args:
            order (OrderDto, optional): Existing order object to replace.
            order_id (str, optional): UUID of the order to replace.
            quantity (float, optional): New order size.
            price (float, optional): New limit price.
            time_in_force (str, optional): New time in force.
            post_only (bool, optional): New post-only flag.
            reduce_only (bool, optional): New reduce-only flag. Defaults to False.

        Returns:
            Tuple[SubmitOrderCreatedDto, bool]: Created order response and success flag.

        Raises:
            ValueError: If neither order nor order_id is provided, or both are provided.
        """
        if order is None and order_id is None:
            raise ValueError("Either order or order_id must be provided")
        elif order is not None and order_id is not None:
            raise ValueError("Only one of order or order_id must be provided")
        elif order is not None:
            old_order = order
        elif order_id is not None:
            old_order = await self.get_order(id=order_id)
        subaccount = await self.get_subaccount(id=old_order.subaccount_id)

        quantity = float(old_order.quantity) if quantity is None else quantity
        price = float(old_order.price) if price is None else price
        time_in_force = (
            old_order.time_in_force.value
            if time_in_force is None and old_order.time_in_force
            else time_in_force
        )
        post_only = old_order.post_only if post_only is None else post_only
        reduce_only = old_order.reduce_only if reduce_only is None else reduce_only

        cancel_result = await self.cancel_orders(
            order_ids=[old_order.id],
            sender=old_order.sender,
            subaccount=subaccount.name,
            sign=True,
            submit=True,
        )
        if not isinstance(cancel_result, list) or len(cancel_result) != 1:
            raise ValueError("Failed to cancel order")
        canceled_order = cancel_result[0]

        if not canceled_order.result.value == "Ok":
            raise ValueError(
                f"Failed to cancel order {order_id}: {canceled_order.result.value}"
            )

        new_order = await self.create_order(
            order_type=old_order.type.value,
            quantity=quantity,
            side=old_order.side.value,
            price=price,
            product_id=old_order.product_id,
            sender=old_order.sender,
            subaccount=subaccount.name,
            time_in_force=time_in_force or self.default_time_in_force,
            post_only=post_only or self.default_post_only,
            reduce_only=reduce_only,
            dry_run=False,
        )
        return self._models.SubmitOrderCreatedDto.model_validate(
            new_order
        ), canceled_order.result.value == "Ok"

cancel_orders(order_ids, sender=None, subaccount=None, client_order_ids=[], sign=True, submit=True, **kwargs) async

Prepares and optionally submits a request to cancel multiple orders.

Parameters:

Name	Type	Description	Default
order_ids	List[str]	Order UUIDs to cancel. Required.	required
sender	str	Address initiating the cancellation. Defaults to chain address.	None
subaccount	str	Hex-encoded subaccount name. Defaults to first subaccount.	None
client_order_ids	List[str]	Client-generated IDs to cancel. Defaults to empty list.	[]
sign	bool	If True, sign the payload immediately. Defaults to True.	True
submit	bool	If True, submit the request to the API. Defaults to True.	True

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API.

Returns:

Type	Description
Union[List[CancelOrderResultDto], CancelOrderDto]	Union[List[CancelOrderResultDto], CancelOrderDto]: Cancellation results per order id or prepared cancel payload.

Raises:

Type	Description
ValueError	If no order IDs or client order IDs provided for cancellation.

Source code in meridian/async_rest_client.py

async def cancel_orders(
    self,
    order_ids: List[UUID],
    sender: Optional[str] = None,
    subaccount: Optional[str] = None,
    client_order_ids: List[str] = [],
    sign: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[List[CancelOrderResultDto], CancelOrderDto]:
    """Prepares and optionally submits a request to cancel multiple orders.

    Args:
        order_ids (List[str]): Order UUIDs to cancel. Required.
        sender (str, optional): Address initiating the cancellation. Defaults to chain address.
        subaccount (str, optional): Hex-encoded subaccount name. Defaults to first subaccount.
        client_order_ids (List[str], optional): Client-generated IDs to cancel. Defaults to empty list.
        sign (bool, optional): If True, sign the payload immediately. Defaults to True.
        submit (bool, optional): If True, submit the request to the API. Defaults to True.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API.

    Returns:
        Union[List[CancelOrderResultDto], CancelOrderDto]: Cancellation results per order id or prepared cancel payload.

    Raises:
        ValueError: If no order IDs or client order IDs provided for cancellation.
    """
    if len(order_ids) == 0 and len(client_order_ids) == 0:
        raise ValueError(
            "No order IDs or client order IDs provided for cancellation"
        )
    sender, subaccount = await self._resolve_sender_and_subaccount(
        sender, subaccount
    )
    try:
        prepared_cancel = await self.prepare_cancel_order(
            order_ids=order_ids,
            client_order_ids=client_order_ids,
            sender=sender,
            subaccount=subaccount,
            include_signature=sign,
            **kwargs,
        )
    except ValueError as e:
        self.logger.warning(f"Could not prepare/sign order cancellation: {e}")
        raise

    if not submit:
        return prepared_cancel

    result = await self.submit_cancel_order(
        prepared_cancel,
        **kwargs,
    )
    return result

convert_token(from_token_id, to_token_id, amount, subaccount=None, sender=None, nonce=None, signed_at=None, sign=True, submit=True, **kwargs) async

Prepares, signs, and optionally submits a token conversion.

This is the primary method for converting between tokens. It handles the complete conversion flow: preparing the conversion payload, signing it with EIP-712, and submitting to the API. To submit a payload you already prepared and signed, use submit_convert_token instead.

Example

Convert 100 of one token into another::

result = await client.convert_token(
    from_token_id=from_token.id,
    to_token_id=to_token.id,
    amount=100,
)

Prepare without submitting (for inspection or manual submission)::

payload = await client.convert_token(
    from_token_id=from_token.id,
    to_token_id=to_token.id,
    amount=100,
    submit=False,
)

Parameters:

Name	Type	Description	Default
from_token_id	UUID	UUID of the token to convert from.	required
to_token_id	UUID	UUID of the token to convert to.	required
amount	int	Amount to convert in the token's base units.	required
subaccount	Optional[str]	Hex-encoded subaccount name. If not provided, defaults to the first subaccount associated with the connected wallet.	None
sender	Optional[str]	Account or linked signer address placing the request. If not provided, defaults to the connected wallet's address.	None
nonce	Optional[str]	Custom nonce for the EIP-712 signature. If not provided, a random nonce is generated.	None
signed_at	Optional[int]	Unix timestamp (seconds) for the signature. If not provided, defaults to the current time.	None
sign	bool	Whether to sign the conversion payload. Defaults to True.	True
submit	bool	Whether to submit the conversion to the API. Set to False to receive the prepared payload without submitting. Defaults to True.	True

Returns:

Type	Description
Union[ConversionDto, InitiateConversionDto]	If submit=True, returns a ConversionDto containing the conversion
Union[ConversionDto, InitiateConversionDto]	record created by the API. If submit=False, returns the prepared
Union[ConversionDto, InitiateConversionDto]	InitiateConversionDto payload.

Source code in meridian/async_rest_client.py

async def convert_token(
    self,
    from_token_id: UUID,
    to_token_id: UUID,
    amount: int,
    subaccount: Optional[str] = None,
    sender: Optional[str] = None,
    nonce: Optional[str] = None,
    signed_at: Optional[int] = None,
    sign: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[ConversionDto, InitiateConversionDto]:
    """Prepares, signs, and optionally submits a token conversion.

    This is the primary method for converting between tokens. It handles
    the complete conversion flow: preparing the conversion payload, signing
    it with EIP-712, and submitting to the API. To submit a payload you
    already prepared and signed, use ``submit_convert_token`` instead.

    Example:
        Convert 100 of one token into another::

            result = await client.convert_token(
                from_token_id=from_token.id,
                to_token_id=to_token.id,
                amount=100,
            )

        Prepare without submitting (for inspection or manual submission)::

            payload = await client.convert_token(
                from_token_id=from_token.id,
                to_token_id=to_token.id,
                amount=100,
                submit=False,
            )

    Args:
        from_token_id: UUID of the token to convert from.
        to_token_id: UUID of the token to convert to.
        amount: Amount to convert in the token's base units.
        subaccount: Hex-encoded subaccount name. If not provided, defaults to
            the first subaccount associated with the connected wallet.
        sender: Account or linked signer address placing the request. If not
            provided, defaults to the connected wallet's address.
        nonce: Custom nonce for the EIP-712 signature. If not provided, a
            random nonce is generated.
        signed_at: Unix timestamp (seconds) for the signature. If not provided,
            defaults to the current time.
        sign: Whether to sign the conversion payload. Defaults to True.
        submit: Whether to submit the conversion to the API. Set to False to
            receive the prepared payload without submitting. Defaults to True.

    Returns:
        If ``submit=True``, returns a ConversionDto containing the conversion
        record created by the API. If ``submit=False``, returns the prepared
        InitiateConversionDto payload.
    """
    # Resolve token addresses from token ids
    from_token = await self.get_token(id=from_token_id)
    to_token = await self.get_token(id=to_token_id)

    # Resolve sender/subaccount
    sender_addr, subaccount_name = await self._resolve_sender_and_subaccount(
        sender, subaccount
    )

    conversion_dto = await self.prepare_convert_token(
        subaccount=subaccount_name,
        from_token=from_token.address,
        to_token=to_token.address,
        amount=amount,
        sender=sender_addr,
        include_signature=False,
        nonce=nonce,
        signed_at=signed_at,
    )

    if sign:
        conversion_dto = await self.sign_convert_token(conversion_dto)

    if not submit:
        return conversion_dto
    return await self.submit_convert_token(dto=conversion_dto, **kwargs)

create(config={}) async classmethod

Factory method to create and asynchronously initialize the client.

Parameters:

Name	Type	Description	Default
config	Union[Dict[str, Any], RESTConfig]	Configuration dictionary or RESTConfig object. Optional fields include: private_key (str, optional): The private key. base_url (str, optional): Base URL for REST requests. Defaults to testnet. timeout (int, optional): Timeout in seconds for REST requests. verbose (bool, optional): Enables debug logging. Defaults to False. rate_limit_headers (bool, optional): Enables rate limit headers. Defaults to False. chain_config (ChainConfig, optional): Chain configuration for signing transactions.	{}

Returns:

Name	Type	Description
AsyncRESTClient	AsyncRESTClient	Fully initialized async client instance.

Source code in meridian/async_rest_client.py

@classmethod
async def create(
    cls, config: Union[Dict[str, Any], RESTConfig] = {}
) -> AsyncRESTClient:
    """Factory method to create and asynchronously initialize the client.

    Args:
        config (Union[Dict[str, Any], RESTConfig], optional): Configuration dictionary or RESTConfig object. Optional fields include:
            private_key (str, optional): The private key.
            base_url (str, optional): Base URL for REST requests. Defaults to testnet.
            timeout (int, optional): Timeout in seconds for REST requests.
            verbose (bool, optional): Enables debug logging. Defaults to False.
            rate_limit_headers (bool, optional): Enables rate limit headers. Defaults to False.
            chain_config (ChainConfig, optional): Chain configuration for signing transactions.

    Returns:
        AsyncRESTClient: Fully initialized async client instance.
    """
    client = cls(config)
    await client._async_init()
    return client

create_batch_order(instructions, submit=True, **kwargs) async

Prepare and optionally submit a batch of signed order instructions.

Parameters:

Name	Type	Description	Default
instructions	List[Union[SubmitOrderDto, CancelOrderDto]]	Prepared SubmitOrderDto or prepared CancelOrderDto payloads.	required
submit	bool	If True, submit the batch order. Defaults to True.	True

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API.

Returns:

Type	Description
Union[BatchOrderDto, BatchOrderResponseDto]	BatchOrderResponseDto or BatchOrderDto: Batch response or prepared batch payload.

Source code in meridian/async_rest_client.py

async def create_batch_order(
    self,
    instructions: List[Union[SubmitOrderDto, CancelOrderDto]],
    submit: bool = True,
    **kwargs,
) -> Union[BatchOrderDto, BatchOrderResponseDto]:
    """Prepare and optionally submit a batch of signed order instructions.

    Args:
        instructions (List[Union[SubmitOrderDto, CancelOrderDto]]): Prepared
            SubmitOrderDto or prepared CancelOrderDto payloads.
        submit (bool, optional): If True, submit the batch order. Defaults to True.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API.

    Returns:
        BatchOrderResponseDto or BatchOrderDto: Batch response or prepared batch payload.
    """
    batch_order = await self.prepare_batch_order(instructions)
    if not submit:
        return batch_order
    return await self.submit_batch_order(batch_order, **kwargs)

create_order(order_type, quantity, side, price=None, ticker=None, product_id=None, client_order_id=None, sender=None, subaccount=None, time_in_force=None, post_only=None, reduce_only=False, close=None, stop_price=None, stop_type=None, expires_at=None, group_id=None, group_contingency_type=None, sign=True, dry_run=False, submit=True) async

Create and submit an order.

Parameters:

Name	Type	Description	Default
order_type	str	'LIMIT' or 'MARKET'. Required.	required
quantity	float	Order size. Required.	required
side	int	0 for buy, 1 for sell. Required.	required
price	float	Limit price for LIMIT orders.	None
ticker	str	Ticker of the product.	None
product_id	str	UUID of the product.	None
client_order_id	str	Subaccount-scoped client-generated id (UUID or <=32 alphanumeric).	None
sender	str	Address placing the order. Defaults to chain address.	None
subaccount	str	Hex-encoded subaccount name. Defaults to first subaccount.	None
time_in_force	str	For LIMIT orders (e.g., 'GTC', 'GTD'). Defaults to 'GTC'.	None
post_only	bool	For LIMIT orders; rejects if crossing. Defaults to False.	None
reduce_only	bool	If True, order only reduces position. Defaults to False.	False
close	bool	For MARKET orders; If True, closes the position.	None
stop_price	float	Stop trigger price.	None
stop_type	int	Stop type, either 0 (take-profit) or 1 (stop-loss), requires non-zero stopPrice.	None
expires_at	int	Expiry timestamp for GTD.	None
group_id	str	Group Id (UUID) for linking orders together in OCO/OTO relationships.	None
group_contingency_type	int	Contingency type for order groups: 0=OTO (Order-Triggers-Order), 1=OCO (One-Cancels-Other).	None
sign	bool	If True, sign the payload immediately. Defaults to True.	True
dry_run	bool	If True, validate without execution. Defaults to False.	False
submit	bool	If True, submit the order. Defaults to True.	True

Returns:

Type	Description
Union[SubmitOrderCreatedDto, DryRunOrderCreatedDto, SubmitOrderDto]	Union[SubmitOrderCreatedDto, DryRunOrderCreatedDto, SubmitOrderDto]: Created order response, dry-run validation result, or prepared order payload.

Raises:

Type	Description
ValueError	If neither product_id nor ticker is provided or if order type is invalid.

Source code in meridian/async_rest_client.py

async def create_order(
    self,
    order_type: str,
    quantity: float,
    side: int,
    price: Optional[float] = None,
    ticker: Optional[str] = None,
    product_id: Optional[UUID] = None,
    client_order_id: Optional[str] = None,
    sender: Optional[str] = None,
    subaccount: Optional[str] = None,
    time_in_force: Optional[str] = None,
    post_only: Optional[bool] = None,
    reduce_only: Optional[bool] = False,
    close: Optional[bool] = None,
    stop_price: Optional[float] = None,
    stop_type: Optional[int] = None,
    expires_at: Optional[int] = None,
    group_id: Optional[str] = None,
    group_contingency_type: Optional[int] = None,
    sign: bool = True,
    dry_run: bool = False,
    submit: bool = True,
) -> Union[SubmitOrderCreatedDto, DryRunOrderCreatedDto, SubmitOrderDto]:
    """Create and submit an order.

    Args:
        order_type (str): 'LIMIT' or 'MARKET'. Required.
        quantity (float): Order size. Required.
        side (int): 0 for buy, 1 for sell. Required.
        price (float, optional): Limit price for LIMIT orders.
        ticker (str, optional): Ticker of the product.
        product_id (str, optional): UUID of the product.
        client_order_id (str, optional): Subaccount-scoped client-generated id (UUID or <=32 alphanumeric).
        sender (str, optional): Address placing the order. Defaults to chain address.
        subaccount (str, optional): Hex-encoded subaccount name. Defaults to first subaccount.
        time_in_force (str, optional): For LIMIT orders (e.g., 'GTC', 'GTD'). Defaults to 'GTC'.
        post_only (bool, optional): For LIMIT orders; rejects if crossing. Defaults to False.
        reduce_only (bool, optional): If True, order only reduces position. Defaults to False.
        close (bool, optional): For MARKET orders; If True, closes the position.
        stop_price (float, optional): Stop trigger price.
        stop_type (int, optional): Stop type, either 0 (take-profit) or 1 (stop-loss), requires non-zero stopPrice.
        expires_at (int, optional): Expiry timestamp for GTD.
        group_id (str, optional): Group Id (UUID) for linking orders together in OCO/OTO relationships.
        group_contingency_type (int, optional): Contingency type for order groups: 0=OTO (Order-Triggers-Order), 1=OCO (One-Cancels-Other).
        sign (bool, optional): If True, sign the payload immediately. Defaults to True.
        dry_run (bool, optional): If True, validate without execution. Defaults to False.
        submit (bool, optional): If True, submit the order. Defaults to True.

    Returns:
        Union[SubmitOrderCreatedDto, DryRunOrderCreatedDto, SubmitOrderDto]: Created order response, dry-run validation result, or prepared order payload.

    Raises:
        ValueError: If neither product_id nor ticker is provided or if order type is invalid.
    """
    sender, subaccount = await self._resolve_sender_and_subaccount(
        sender, subaccount
    )

    if product_id is not None:
        products_by_id = await self.products_by_id()
        onchain_id = products_by_id[product_id].onchain_id
    elif ticker is not None:
        products_by_ticker = await self.products_by_ticker()
        onchain_id = products_by_ticker[ticker].onchain_id
    else:
        raise ValueError("Either product_id or ticker must be provided")

    order_params: Dict[str, Any] = {
        "subaccount": subaccount,
        "side": side,
        "quantity": quantity,
        "onchain_id": onchain_id,
        "order_type": order_type,
        "client_order_id": client_order_id,
        "reduce_only": reduce_only,
        "close": close,
        "stop_price": stop_price,
        "stop_type": stop_type,
        "group_id": group_id,
        "group_contingency_type": group_contingency_type,
    }

    if order_type == "LIMIT":
        order_params.update(
            {
                "price": price,
                "time_in_force": time_in_force or self.default_time_in_force,
                "post_only": post_only or self.default_post_only,
                "expires_at": expires_at,
            }
        )
    elif order_type != "MARKET":
        raise ValueError("Invalid order type")

    order = await self.prepare_order(sender, **order_params, include_signature=sign)
    if dry_run:
        return await self.dry_run_order(order)
    elif submit:
        return await self.submit_order(order)
    else:
        return order

extend_signer(sender=None, sign=True, submit=True, **kwargs) async

Prepare, sign, and optionally submit an extend-linked-signer payload.

Parameters:

Name	Type	Description	Default
sender	str	Address of the linked signer extending itself. Defaults to the client's chain address.	None
sign	bool	Sign with the client's key. Defaults to True.	True
submit	bool	Submit to the API. If False, return the prepared DTO.	True

Returns:

Type	Description
Union[SignerDto, ExtendLinkedSignerDto]	Union[SignerDto, ExtendLinkedSignerDto]: Submitted record or the prepared/signable payload.

Source code in meridian/async_rest_client.py

async def extend_signer(
    self,
    sender: Optional[str] = None,
    sign: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[SignerDto, ExtendLinkedSignerDto]:
    """Prepare, sign, and optionally submit an extend-linked-signer payload.

    Args:
        sender (str, optional): Address of the linked signer extending itself.
            Defaults to the client's chain address.
        sign (bool, optional): Sign with the client's key. Defaults to True.
        submit (bool, optional): Submit to the API. If False, return the prepared DTO.

    Returns:
        Union[SignerDto, ExtendLinkedSignerDto]: Submitted record or the prepared/signable payload.
    """
    if sender is None:
        if not self.chain:
            raise ValueError(
                "Could not resolve 'sender'. Either pass 'sender' explicitly or initialize the client with a chain configuration."
            )
        sender = self.chain.address

    extend = await self.prepare_extend_linked_signer(
        sender=sender,
        include_signature=False,
        **kwargs,
    )

    if sign:
        extend = await self.sign_extend_linked_signer(extend)

    return (
        extend if not submit else await self.extend_linked_signer(extend, **kwargs)
    )

get_maintenance_margin(subaccount_id, positions=None, products=None, product_ids=None) async

Calculate the an account's maintenance margin for specified positions or products.

Parameters:

Name	Type	Description	Default
subaccount_id	str	Fetch positions for this subaccount when positions is not supplied.	required
positions	List[PositionDto] | List[Dict[str, Any]]	Pre-fetched positions to use directly.	None
products	List[ProductDto] | List[Dict[str, Any]]	Pre-fetched products used to filter the calculation.	None
product_ids	List[str] | List[UUID]	Filters the calculation to these product IDs.	None

Returns:

Name	Type	Description
Decimal	Decimal	Total maintenance margin for the filtered positions.

Raises:

Type	Description
ValueError	If neither positions nor subaccount context is provided, or if any referenced product cannot be resolved.

Source code in meridian/async_rest_client.py

async def get_maintenance_margin(
    self,
    subaccount_id: UUID,
    positions: Optional[List[PositionDto] | List[Dict[str, Any]]] = None,
    products: Optional[List[ProductDto] | List[Dict[str, Any]]] = None,
    product_ids: Optional[List[UUID]] = None,
) -> Decimal:
    """Calculate the an account's maintenance margin for specified positions or products.

    Args:
        subaccount_id (str): Fetch positions for this subaccount when ``positions`` is not supplied.
        positions (List[PositionDto] | List[Dict[str, Any]], optional): Pre-fetched positions to use directly.
        products (List[ProductDto] | List[Dict[str, Any]], optional): Pre-fetched products used to filter
            the calculation.
        product_ids (List[str] | List[UUID], optional): Filters the calculation to these product IDs.

    Returns:
        Decimal: Total maintenance margin for the filtered positions.

    Raises:
        ValueError: If neither positions nor subaccount context is provided, or if any
            referenced product cannot be resolved.
    """

    class PartialPosition(BaseModel):
        product_id: UUID
        cost: str

    class PartialProduct(BaseModel):
        id: UUID
        max_leverage: float
        taker_fee: str

    if positions is None:
        positions = await self.list_positions(subaccount_id=subaccount_id)

    if products and product_ids:
        raise ValueError("Can only specify one of products and product_ids")

    if products is not None:
        valid_products = [
            PartialProduct(
                **p if isinstance(p, dict) else p.model_dump(mode="json")
            )
            for p in products
        ]
        products_by_id: Dict[UUID, PartialProduct] = {
            product.id: product for product in valid_products
        }
    else:
        raw_products_by_id = await self.products_by_id()
        products_by_id = {
            UUID(str(k)): PartialProduct(
                **v if isinstance(v, dict) else v.model_dump(mode="json")
            )
            for k, v in raw_products_by_id.items()
        }

    product_ids_set: Optional[Set[UUID]] = None
    if products is not None:
        product_ids_set = set(products_by_id.keys())
    if product_ids is not None:
        product_ids_set = set([UUID(str(pid)) for pid in product_ids])
        products_by_id_keys = set(products_by_id.keys())
        missing_products = product_ids_set - products_by_id_keys
        if missing_products:
            missing = ", ".join(sorted([str(p) for p in missing_products]))
            raise ValueError(f"Products not found for calculation: {missing}")

    valid_positions = [
        PartialPosition(**p if isinstance(p, dict) else p.model_dump(mode="json"))
        for p in positions
    ]
    positions_filtered = [
        position
        for position in valid_positions
        if product_ids_set is None or position.product_id in product_ids_set
    ]

    if not positions_filtered:
        return Decimal("0")

    total_mm = Decimal("0")
    for position in positions_filtered:
        product = products_by_id.get(position.product_id)
        if product is None:
            raise ValueError(
                f"Product '{position.product_id}' not found for position '{position.product_id}'"
            )

        notional = abs(Decimal(position.cost))
        max_leverage = Decimal(str(product.max_leverage))
        taker_fee_rate = Decimal(product.taker_fee)

        mmr = Decimal("1") / (max_leverage * Decimal("2"))
        total_mm += notional * mmr
        total_mm += notional * taker_fee_rate

    return total_mm

get_tokens() async

Return the latest list of tokens (no caching).

Source code in meridian/async_rest_client.py

async def get_tokens(self) -> List[TokenDto]:
    """Return the latest list of tokens (no caching)."""
    return await self.list_tokens()

link_signer(signer, subaccount_id, subaccount=None, sender=None, signer_private_key=None, nonce=None, signed_at=None, sign_sender=True, sign_signer=True, submit=True, **kwargs) async

Prepare, sign, and optionally submit a linked signer payload.

Parameters:

Name	Type	Description	Default
signer	str	Address being linked as a delegate signer. Required.	required
subaccount_id	UUID	Target subaccount id. Required.	required
subaccount	str	Bytes32 subaccount name. If omitted, fetched by id.	None
sender	str	Owner address. Defaults to the client's chain address.	None
signer_private_key	str	Private key for the signer address.	None
nonce	str	Custom nonce for the signature.	None
signed_at	int	Seconds since epoch for the signature timestamp.	None
sign_sender	bool	Sign with the owner key. Defaults to True.	True
sign_signer	bool	Sign with the signer key. Defaults to True.	True
submit	bool	Submit to the API. If False, return the prepared DTO.	True

Returns:

Type	Description
Union[SignerDto, LinkSignerDto]	Union[SignerDto, LinkSignerDto]: Submitted record or the prepared/signable payload.

Raises:

Type	Description
ValueError	If the subaccount name cannot be resolved or signer signature is missing.

Source code in meridian/async_rest_client.py

async def link_signer(
    self,
    signer: str,
    subaccount_id: UUID,
    subaccount: Optional[str] = None,
    sender: Optional[str] = None,
    signer_private_key: Optional[str] = None,
    nonce: Optional[str] = None,
    signed_at: Optional[int] = None,
    sign_sender: bool = True,
    sign_signer: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[SignerDto, LinkSignerDto]:
    """Prepare, sign, and optionally submit a linked signer payload.

    Args:
        signer (str): Address being linked as a delegate signer. Required.
        subaccount_id (UUID): Target subaccount id. Required.
        subaccount (str, optional): Bytes32 subaccount name. If omitted, fetched by id.
        sender (str, optional): Owner address. Defaults to the client's chain address.
        signer_private_key (str, optional): Private key for the signer address.
        nonce (str, optional): Custom nonce for the signature.
        signed_at (int, optional): Seconds since epoch for the signature timestamp.
        sign_sender (bool, optional): Sign with the owner key. Defaults to True.
        sign_signer (bool, optional): Sign with the signer key. Defaults to True.
        submit (bool, optional): Submit to the API. If False, return the prepared DTO.

    Returns:
        Union[SignerDto, LinkSignerDto]: Submitted record or the prepared/signable payload.

    Raises:
        ValueError: If the subaccount name cannot be resolved or signer signature is missing.
    """
    if subaccount is None:
        subaccount = await self._resolve_subaccount_name(subaccount_id)
    sender, subaccount = await self._resolve_sender_and_subaccount(
        sender, subaccount
    )

    link = await self.prepare_linked_signer(
        sender=sender,
        signer=signer,
        subaccount=subaccount,
        subaccount_id=subaccount_id,
        include_signature=False,
        nonce=nonce,
        signed_at=signed_at,
    )

    if sign_sender:
        link = await self.sign_linked_signer(link)

    if sign_signer:
        if signer_private_key:
            link = await self.sign_linked_signer(
                link, signer_private_key=signer_private_key
            )
        elif not link.signer_signature:
            raise ValueError(
                "signer_private_key or signer_signature is required when sign_signer=True"
            )

    return link if not submit else await self.link_linked_signer(link, **kwargs)

products(refresh=False) async

Get the list of products.

Parameters:

Name	Type	Description	Default
refresh	bool	If True, bypass cache and fetch fresh data. Defaults to False.	False

Returns:

Type	Description
List[ProductDto]	List[ProductDto]: List of product objects.

Source code in meridian/async_rest_client.py

async def products(self, refresh: bool = False) -> List[ProductDto]:
    """Get the list of products.

    Args:
        refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

    Returns:
        List[ProductDto]: List of product objects.
    """
    if refresh or self._products is None:
        self._products = await self.list_products()
        self._products_by_id = None
        self._products_by_ticker = None
    return self._products

products_by_id(refresh=False) async

Get the products indexed by ID.

Parameters:

Name	Type	Description	Default
refresh	bool	If True, bypass cache and fetch fresh data. Defaults to False.	False

Returns:

Type	Description
Dict[UUID, ProductDto]	Dict[UUID, ProductDto]: Dictionary of products keyed by ID.

Source code in meridian/async_rest_client.py

async def products_by_id(self, refresh: bool = False) -> Dict[UUID, ProductDto]:
    """Get the products indexed by ID.

    Args:
        refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

    Returns:
        Dict[UUID, ProductDto]: Dictionary of products keyed by ID.
    """
    if refresh or self._products_by_id is None:
        products = await self.products(refresh=refresh)
        self._products_by_id = {p.id: p for p in products}
    return self._products_by_id

products_by_ticker(refresh=False) async

Get the products indexed by ticker.

Parameters:

Name	Type	Description	Default
refresh	bool	If True, bypass cache and fetch fresh data. Defaults to False.	False

Returns:

Type	Description
Dict[str, ProductDto]	Dict[str, ProductDto]: Dictionary of products keyed by ticker.

Source code in meridian/async_rest_client.py

async def products_by_ticker(self, refresh: bool = False) -> Dict[str, ProductDto]:
    """Get the products indexed by ticker.

    Args:
        refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

    Returns:
        Dict[str, ProductDto]: Dictionary of products keyed by ticker.
    """
    if refresh or self._products_by_ticker is None:
        products = await self.products(refresh=refresh)
        self._products_by_ticker = {
            p.ticker: p for p in products if getattr(p, "ticker", None)
        }
    return self._products_by_ticker

refresh_signer(signer, subaccount_id, subaccount=None, sender=None, sign=True, submit=True, **kwargs) async

Prepare, sign, and optionally submit a refresh-linked-signer payload.

Parameters:

Name	Type	Description	Default
signer	str	Address of the delegate to refresh. Required.	required
subaccount_id	UUID	Target subaccount id. Required.	required
subaccount	str	Bytes32 subaccount name. If omitted, fetched by id.	None
sender	str	Owner address. Defaults to the client's chain address.	None
sign	bool	Sign with the owner key. Defaults to True.	True
submit	bool	Submit to the API. If False, return the prepared DTO.	True

Returns:

Type	Description
Union[SignerDto, RefreshLinkedSignerDto]	Union[SignerDto, RefreshLinkedSignerDto]: Submitted record or the prepared/signable payload.

Raises:

Type	Description
ValueError	If the subaccount name cannot be resolved.

Source code in meridian/async_rest_client.py

async def refresh_signer(
    self,
    signer: str,
    subaccount_id: UUID,
    subaccount: Optional[str] = None,
    sender: Optional[str] = None,
    sign: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[SignerDto, RefreshLinkedSignerDto]:
    """Prepare, sign, and optionally submit a refresh-linked-signer payload.

    Args:
        signer (str): Address of the delegate to refresh. Required.
        subaccount_id (UUID): Target subaccount id. Required.
        subaccount (str, optional): Bytes32 subaccount name. If omitted, fetched by id.
        sender (str, optional): Owner address. Defaults to the client's chain address.
        sign (bool, optional): Sign with the owner key. Defaults to True.
        submit (bool, optional): Submit to the API. If False, return the prepared DTO.

    Returns:
        Union[SignerDto, RefreshLinkedSignerDto]: Submitted record or the prepared/signable payload.

    Raises:
        ValueError: If the subaccount name cannot be resolved.
    """
    if subaccount is None:
        subaccount = await self._resolve_subaccount_name(subaccount_id)
    sender, subaccount = await self._resolve_sender_and_subaccount(
        sender, subaccount
    )

    refresh = await self.prepare_refresh_linked_signer(
        sender=sender,
        signer=signer,
        subaccount=subaccount,
        subaccount_id=subaccount_id,
        include_signature=False,
        **kwargs,
    )

    if sign:
        refresh = await self.sign_refresh_linked_signer(refresh)

    return (
        refresh
        if not submit
        else await self.refresh_linked_signer(refresh, **kwargs)
    )

replace_order(order=None, order_id=None, quantity=None, price=None, time_in_force=None, post_only=None, reduce_only=False) async

Replace an existing order with new parameters.

Parameters:

Name	Type	Description	Default
order	OrderDto	Existing order object to replace.	None
order_id	str	UUID of the order to replace.	None
quantity	float	New order size.	None
price	float	New limit price.	None
time_in_force	str	New time in force.	None
post_only	bool	New post-only flag.	None
reduce_only	bool	New reduce-only flag. Defaults to False.	False

Returns:

Type	Description
Tuple[SubmitOrderCreatedDto, bool]	Tuple[SubmitOrderCreatedDto, bool]: Created order response and success flag.

Raises:

Type	Description
ValueError	If neither order nor order_id is provided, or both are provided.

Source code in meridian/async_rest_client.py

async def replace_order(
    self,
    order: Optional[OrderDto] = None,
    order_id: Optional[UUID] = None,
    quantity: Optional[float] = None,
    price: Optional[float] = None,
    time_in_force: Optional[str] = None,
    post_only: Optional[bool] = None,
    reduce_only: Optional[bool] = False,
) -> Tuple[SubmitOrderCreatedDto, bool]:
    """Replace an existing order with new parameters.

    Args:
        order (OrderDto, optional): Existing order object to replace.
        order_id (str, optional): UUID of the order to replace.
        quantity (float, optional): New order size.
        price (float, optional): New limit price.
        time_in_force (str, optional): New time in force.
        post_only (bool, optional): New post-only flag.
        reduce_only (bool, optional): New reduce-only flag. Defaults to False.

    Returns:
        Tuple[SubmitOrderCreatedDto, bool]: Created order response and success flag.

    Raises:
        ValueError: If neither order nor order_id is provided, or both are provided.
    """
    if order is None and order_id is None:
        raise ValueError("Either order or order_id must be provided")
    elif order is not None and order_id is not None:
        raise ValueError("Only one of order or order_id must be provided")
    elif order is not None:
        old_order = order
    elif order_id is not None:
        old_order = await self.get_order(id=order_id)
    subaccount = await self.get_subaccount(id=old_order.subaccount_id)

    quantity = float(old_order.quantity) if quantity is None else quantity
    price = float(old_order.price) if price is None else price
    time_in_force = (
        old_order.time_in_force.value
        if time_in_force is None and old_order.time_in_force
        else time_in_force
    )
    post_only = old_order.post_only if post_only is None else post_only
    reduce_only = old_order.reduce_only if reduce_only is None else reduce_only

    cancel_result = await self.cancel_orders(
        order_ids=[old_order.id],
        sender=old_order.sender,
        subaccount=subaccount.name,
        sign=True,
        submit=True,
    )
    if not isinstance(cancel_result, list) or len(cancel_result) != 1:
        raise ValueError("Failed to cancel order")
    canceled_order = cancel_result[0]

    if not canceled_order.result.value == "Ok":
        raise ValueError(
            f"Failed to cancel order {order_id}: {canceled_order.result.value}"
        )

    new_order = await self.create_order(
        order_type=old_order.type.value,
        quantity=quantity,
        side=old_order.side.value,
        price=price,
        product_id=old_order.product_id,
        sender=old_order.sender,
        subaccount=subaccount.name,
        time_in_force=time_in_force or self.default_time_in_force,
        post_only=post_only or self.default_post_only,
        reduce_only=reduce_only,
        dry_run=False,
    )
    return self._models.SubmitOrderCreatedDto.model_validate(
        new_order
    ), canceled_order.result.value == "Ok"

revoke_signer(signer, subaccount_id, subaccount=None, sender=None, sign=True, submit=True, **kwargs) async

Prepare, sign, and optionally submit a revoke-linked-signer payload.

Parameters:

Name	Type	Description	Default
signer	str	Address of the delegate to revoke. Required.	required
subaccount_id	UUID	Target subaccount id. Required.	required
subaccount	str	Bytes32 subaccount name. If omitted, fetched by id.	None
sender	str	Owner address. Defaults to the client's chain address.	None
sign	bool	Sign with the owner key. Defaults to True.	True
submit	bool	Submit to the API. If False, return the prepared DTO.	True

Returns:

Type	Description
Union[SignerDto, RevokeLinkedSignerDto]	Union[SignerDto, RevokeLinkedSignerDto]: Submitted record or the prepared/signable payload.

Raises:

Type	Description
ValueError	If the subaccount name cannot be resolved.

Source code in meridian/async_rest_client.py

async def revoke_signer(
    self,
    signer: str,
    subaccount_id: UUID,
    subaccount: Optional[str] = None,
    sender: Optional[str] = None,
    sign: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[SignerDto, RevokeLinkedSignerDto]:
    """Prepare, sign, and optionally submit a revoke-linked-signer payload.

    Args:
        signer (str): Address of the delegate to revoke. Required.
        subaccount_id (UUID): Target subaccount id. Required.
        subaccount (str, optional): Bytes32 subaccount name. If omitted, fetched by id.
        sender (str, optional): Owner address. Defaults to the client's chain address.
        sign (bool, optional): Sign with the owner key. Defaults to True.
        submit (bool, optional): Submit to the API. If False, return the prepared DTO.

    Returns:
        Union[SignerDto, RevokeLinkedSignerDto]: Submitted record or the prepared/signable payload.

    Raises:
        ValueError: If the subaccount name cannot be resolved.
    """
    if subaccount is None:
        subaccount = await self._resolve_subaccount_name(subaccount_id)
    sender, subaccount = await self._resolve_sender_and_subaccount(
        sender, subaccount
    )

    revoke = await self.prepare_revoke_linked_signer(
        sender=sender,
        signer=signer,
        subaccount=subaccount,
        subaccount_id=subaccount_id,
        include_signature=False,
        **kwargs,
    )

    if sign:
        revoke = await self.sign_revoke_linked_signer(revoke)

    return (
        revoke if not submit else await self.revoke_linked_signer(revoke, **kwargs)
    )

subaccounts(refresh=False) async

Get the list of subaccounts.

Parameters:

Name	Type	Description	Default
refresh	bool	If True, bypass cache and fetch fresh data. Defaults to False.	False

Returns:

Type	Description
List[SubaccountDto]	List[SubaccountDto]: List of subaccount objects for the connected wallet address.

Raises:

Type	Description
ValueError	If no chain client is configured or address is unavailable.

Source code in meridian/async_rest_client.py

async def subaccounts(self, refresh: bool = False) -> List[SubaccountDto]:
    """Get the list of subaccounts.

    Args:
        refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

    Returns:
        List[SubaccountDto]: List of subaccount objects for the connected wallet address.

    Raises:
        ValueError: If no chain client is configured or address is unavailable.
    """
    if not self.chain or not getattr(self.chain, "address", None):
        raise ValueError("Chain address is required to list subaccounts")
    if refresh or self._subaccounts is None:
        self._subaccounts = await self.list_subaccounts(
            sender=self.chain.address, order_by="createdAt", order="asc"
        )
    return self._subaccounts

tokens(refresh=False) async

Get the list of tokens.

Parameters:

Name	Type	Description	Default
refresh	bool	If True, bypass cache and fetch fresh data. Defaults to False.	False

Returns:

Type	Description
List[TokenDto]	List[TokenDto]: List of token objects.

Source code in meridian/async_rest_client.py

async def tokens(self, refresh: bool = False) -> List[TokenDto]:
    """Get the list of tokens.

    Args:
        refresh (bool, optional): If True, bypass cache and fetch fresh data. Defaults to False.

    Returns:
        List[TokenDto]: List of token objects.
    """
    if refresh or self._tokens is None:
        self._tokens = await self.list_tokens()
    return self._tokens

withdraw_token(token_id, amount, destination_address, destination_endpoint, subaccount=None, account=None, nonce=None, signed_at=None, sign=True, submit=True, **kwargs) async

Prepares, signs, and optionally submits a token withdrawal.

This is the primary method for initiating token withdrawals. It handles the complete withdrawal flow: preparing the withdrawal payload, signing it with EIP-712, and submitting to the API. To submit a payload you already prepared and signed, use submit_withdraw_token instead.

Example

Withdraw 100 USD to the connected wallet::

result = await client.withdraw_token(
    token_id=usd_token.id,
    amount=100,
    destination_address=client.chain.address,
    destination_endpoint=0,
)

Prepare without submitting (for inspection or manual submission)::

payload = await client.withdraw_token(
    token_id=usd_token.id,
    amount=100,
    destination_address="0x...",
    destination_endpoint=0,
    submit=False,
)

Parameters:

Name	Type	Description	Default
token_id	UUID	UUID of the token to withdraw.	required
amount	int	Amount to withdraw in the token's base units.	required
destination_address	str	The LayerZero destination address where funds will be sent. Automatically padded to bytes32 format.	required
destination_endpoint	int	LayerZero endpoint ID for the destination chain. Use 0 for same-chain withdrawals.	required
subaccount	Optional[str]	Hex-encoded subaccount name. If not provided, defaults to the first subaccount associated with the connected wallet.	None
account	Optional[str]	Recipient wallet address. If not provided, defaults to the connected wallet's address.	None
nonce	Optional[str]	Custom nonce for the EIP-712 signature. If not provided, a random nonce is generated.	None
signed_at	Optional[int]	Unix timestamp (seconds) for the signature. If not provided, defaults to the current time.	None
sign	bool	Whether to sign the withdrawal payload. Defaults to True.	True
submit	bool	Whether to submit the withdrawal to the API. Set to False to receive the prepared payload without submitting. Defaults to True.	True

Returns:

Type	Description
Union[WithdrawDto, InitiateWithdrawDto]	If submit=True, returns a WithdrawDto containing the withdrawal
Union[WithdrawDto, InitiateWithdrawDto]	record created by the API. If submit=False, returns the prepared
Union[WithdrawDto, InitiateWithdrawDto]	InitiateWithdrawDto payload.

Source code in meridian/async_rest_client.py

async def withdraw_token(
    self,
    token_id: UUID,
    amount: int,
    destination_address: str,
    destination_endpoint: int,
    subaccount: Optional[str] = None,
    account: Optional[str] = None,
    nonce: Optional[str] = None,
    signed_at: Optional[int] = None,
    sign: bool = True,
    submit: bool = True,
    **kwargs,
) -> Union[WithdrawDto, InitiateWithdrawDto]:
    """Prepares, signs, and optionally submits a token withdrawal.

    This is the primary method for initiating token withdrawals. It handles
    the complete withdrawal flow: preparing the withdrawal payload, signing
    it with EIP-712, and submitting to the API. To submit a payload you
    already prepared and signed, use ``submit_withdraw_token`` instead.

    Example:
        Withdraw 100 USD to the connected wallet::

            result = await client.withdraw_token(
                token_id=usd_token.id,
                amount=100,
                destination_address=client.chain.address,
                destination_endpoint=0,
            )

        Prepare without submitting (for inspection or manual submission)::

            payload = await client.withdraw_token(
                token_id=usd_token.id,
                amount=100,
                destination_address="0x...",
                destination_endpoint=0,
                submit=False,
            )

    Args:
        token_id: UUID of the token to withdraw.
        amount: Amount to withdraw in the token's base units.
        destination_address: The LayerZero destination address where funds
            will be sent. Automatically padded to bytes32 format.
        destination_endpoint: LayerZero endpoint ID for the destination chain.
            Use 0 for same-chain withdrawals.
        subaccount: Hex-encoded subaccount name. If not provided, defaults to
            the first subaccount associated with the connected wallet.
        account: Recipient wallet address. If not provided, defaults to the
            connected wallet's address.
        nonce: Custom nonce for the EIP-712 signature. If not provided, a
            random nonce is generated.
        signed_at: Unix timestamp (seconds) for the signature. If not provided,
            defaults to the current time.
        sign: Whether to sign the withdrawal payload. Defaults to True.
        submit: Whether to submit the withdrawal to the API. Set to False to
            receive the prepared payload without submitting. Defaults to True.

    Returns:
        If ``submit=True``, returns a WithdrawDto containing the withdrawal
        record created by the API. If ``submit=False``, returns the prepared
        InitiateWithdrawDto payload.
    """
    # Resolve token address from token_id
    token = await self.get_token(id=token_id)
    token_addr = token.address

    # Resolve account/subaccount
    account_addr, subaccount_name = await self._resolve_sender_and_subaccount(
        account, subaccount
    )

    withdraw_dto = await self.prepare_withdraw_token(
        subaccount=subaccount_name,
        token=token_addr,
        amount=amount,
        account=account_addr,
        destination_address=destination_address,
        destination_endpoint=destination_endpoint,
        include_signature=False,
        nonce=nonce,
        signed_at=signed_at,
    )

    if sign:
        withdraw_dto = await self.sign_withdraw_token(withdraw_dto)

    if not submit:
        return withdraw_dto
    return await self.submit_withdraw_token(
        dto=withdraw_dto, token_id=token_id, **kwargs
    )

RESTClient

Synchronous wrapper around AsyncRESTClient for backward compatibility. Use AsyncRESTClient for new applications.

from meridian import RESTClient

client = RESTClient({
    "base_url": "https://api.meridian.xyz",
    "chain_config": {
        "rpc_url": "https://rpc.mainnet.chain.robinhood.com",
        "private_key": "your_private_key",  # optional
    }
})

# Property access works for convenience
products = client.products
subaccounts = client.subaccounts

# Remember to close when done
client.close()

meridian.rest.rpc

get_rpc_config(self, **kwargs) async

Gets RPC configuration for EIP-712 signing and contract info.

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
RpcConfigDto	RpcConfigDto	Domain and signature type definitions for signing.

Source code in meridian/rest/rpc.py

async def get_rpc_config(self, **kwargs) -> RpcConfigDto:
    """Gets RPC configuration for EIP-712 signing and contract info.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        RpcConfigDto: Domain and signature type definitions for signing.
    """
    endpoint = f"{API_PREFIX}/rpc/config"

    res = await self.get(endpoint, **kwargs)
    domain = self._models.DomainTypeDto(**res["domain"])
    signature_types = self._models.SignatureTypesDto(**res["signatureTypes"])
    return self._models.RpcConfigDto(domain=domain, signatureTypes=signature_types)

meridian.rest.subaccount

get_subaccount(self, id, **kwargs) async

Gets a specific subaccount by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SubaccountDto	SubaccountDto	Subaccount details.

Source code in meridian/rest/subaccount.py

async def get_subaccount(self, id: UUID, **kwargs) -> SubaccountDto:
    """Gets a specific subaccount by ID.

    Args:
        id (str): UUID of the subaccount. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SubaccountDto: Subaccount details.
    """
    endpoint = f"{API_PREFIX}/subaccount/{id}"
    res = await self.get(endpoint, **kwargs)
    return self._models.SubaccountDto(**res)

get_subaccount_balance_history(self, **kwargs) async

Gets historical subaccount balances from the archive API.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
start_time	float	Range start time in milliseconds since Unix epoch. Required.
end_time	float	Range end time in milliseconds since Unix epoch. Optional.
resolution	str	Data resolution (e.g., 'hour1', 'day1'). Required.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (defaults to 'time'). Optional.
**kwargs		Additional query parameters supported by the API.

Returns:

Type	Description
List[BalanceHistoryDto]	List[BalanceHistoryDto]: Historical balance records ordered per request parameters.

Source code in meridian/rest/subaccount.py

async def get_subaccount_balance_history(self, **kwargs) -> List["BalanceHistoryDto"]:
    """Gets historical subaccount balances from the archive API.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        start_time (float): Range start time in milliseconds since Unix epoch. Required.
        end_time (float, optional): Range end time in milliseconds since Unix epoch. Optional.
        resolution (str): Data resolution (e.g., 'hour1', 'day1'). Required.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (defaults to 'time'). Optional.
        **kwargs: Additional query parameters supported by the API.

    Returns:
        List[BalanceHistoryDto]: Historical balance records ordered per request parameters.
    """

    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/balance",
        request_model=self._archive_models.V1SubaccountBalanceGetParametersQuery,
        response_model=self._archive_models.PageOfBalanceHistoryDtos,
        base_url_override=self._archive_base_url,
        **kwargs,
    )

    data = [
        self._archive_models.BalanceHistoryDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

get_subaccount_balances(self, **kwargs) async

Gets token balances for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'tokenName', 'updatedAt', 'amount'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[SubaccountBalanceDto]	List[SubaccountBalanceDto]: Balances for the subaccount.

Source code in meridian/rest/subaccount.py

async def get_subaccount_balances(self, **kwargs) -> List[SubaccountBalanceDto]:
    """Gets token balances for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'tokenName', 'updatedAt', 'amount'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[SubaccountBalanceDto]: Balances for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/balance",
        request_model=self._models.V1SubaccountBalanceGetParametersQuery,
        response_model=self._models.PageOfSubaccountBalanceDtos,
        **kwargs,
    )
    data = [
        self._models.SubaccountBalanceDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

get_subaccount_funding_history(self, **kwargs) async

Gets historical funding charges for positions in a subaccount from the archive API.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required
start_time	float	Range start time in milliseconds since Unix epoch. Required.	required

Other Parameters:

Name	Type	Description
end_time	float	Range end time in milliseconds since Unix epoch. Optional.
position_ids	List[str]	Filter by specific position IDs (max 128). Optional.
product_ids	List[str]	Filter by specific product IDs. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (defaults to 'time'). Optional.
**kwargs		Additional query parameters supported by the API.

Returns:

Type	Description
List[PositionFundingHistoryDto]	List[PositionFundingHistoryDto]: Historical funding charge records for positions in the subaccount.

Source code in meridian/rest/subaccount.py

async def get_subaccount_funding_history(
    self, **kwargs
) -> List["PositionFundingHistoryDto"]:
    """Gets historical funding charges for positions in a subaccount from the archive API.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.
        start_time (float): Range start time in milliseconds since Unix epoch. Required.

    Other Parameters:
        end_time (float, optional): Range end time in milliseconds since Unix epoch. Optional.
        position_ids (List[str], optional): Filter by specific position IDs (max 128). Optional.
        product_ids (List[str], optional): Filter by specific product IDs. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (defaults to 'time'). Optional.
        **kwargs: Additional query parameters supported by the API.

    Returns:
        List[PositionFundingHistoryDto]: Historical funding charge records for positions in the subaccount.
    """

    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/funding",
        request_model=self._archive_models.V1SubaccountFundingGetParametersQuery,
        response_model=self._archive_models.PageOfPositionFundingHistoryDtos,
        base_url_override=self._archive_base_url,
        **kwargs,
    )

    data = [
        self._archive_models.PositionFundingHistoryDto(
            **model.model_dump(by_alias=True)
        )
        for model in res.data
    ]
    return data

get_subaccount_position_fee_history(self, **kwargs) async

Gets historical position-fee charges for positions in a subaccount from the archive API.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required
start_time	float	Range start time in milliseconds since Unix epoch. Required.	required

Other Parameters:

Name	Type	Description
end_time	float	Range end time in milliseconds since Unix epoch. Optional.
position_ids	List[str]	Filter by specific position IDs (max 128). Optional.
product_ids	List[str]	Filter by specific product IDs. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (defaults to 'time'). Optional.
**kwargs		Additional query parameters supported by the API.

Returns:

Type	Description
List[PositionFeeHistoryDto]	List[PositionFeeHistoryDto]: Historical position-fee charge records for positions in the subaccount.

Source code in meridian/rest/subaccount.py

async def get_subaccount_position_fee_history(
    self, **kwargs
) -> List["PositionFeeHistoryDto"]:
    """Gets historical position-fee charges for positions in a subaccount from the archive API.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.
        start_time (float): Range start time in milliseconds since Unix epoch. Required.

    Other Parameters:
        end_time (float, optional): Range end time in milliseconds since Unix epoch. Optional.
        position_ids (List[str], optional): Filter by specific position IDs (max 128). Optional.
        product_ids (List[str], optional): Filter by specific product IDs. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (defaults to 'time'). Optional.
        **kwargs: Additional query parameters supported by the API.

    Returns:
        List[PositionFeeHistoryDto]: Historical position-fee charge records for positions in the subaccount.
    """

    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/position-fee",
        request_model=self._archive_models.V1SubaccountPositionFeeGetParametersQuery,
        response_model=self._archive_models.PageOfPositionFeeHistoryDtos,
        base_url_override=self._archive_base_url,
        **kwargs,
    )

    data = [
        self._archive_models.PositionFeeHistoryDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

get_subaccount_total_volume(self, **kwargs) async

Gets the total trading volume for a subaccount from the archive API.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional query parameters supported by the API.

Returns:

Name	Type	Description
TotalSubaccountVolumeDto	TotalSubaccountVolumeDto	Total volume in USD for the subaccount.

Source code in meridian/rest/subaccount.py

async def get_subaccount_total_volume(self, **kwargs) -> "TotalSubaccountVolumeDto":
    """Gets the total trading volume for a subaccount from the archive API.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        **kwargs: Additional query parameters supported by the API.

    Returns:
        TotalSubaccountVolumeDto: Total volume in USD for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/total-volume",
        request_model=self._archive_models.V1SubaccountTotalVolumeGetParametersQuery,
        response_model=self._archive_models.TotalSubaccountVolumeDto,
        base_url_override=self._archive_base_url,
        **kwargs,
    )
    return res

get_subaccount_unrealized_pnl_history(self, **kwargs) async

Gets historical unrealized PnL for a subaccount from the archive API.

Source code in meridian/rest/subaccount.py

async def get_subaccount_unrealized_pnl_history(
    self, **kwargs
) -> List["UnrealizedPnlHistoryDto"]:
    """Gets historical unrealized PnL for a subaccount from the archive API."""

    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/unrealized-pnl",
        request_model=self._archive_models.V1SubaccountUnrealizedPnlGetParametersQuery,
        response_model=self._archive_models.PageOfUnrealizedPnlHistoryDtos,
        base_url_override=self._archive_base_url,
        **kwargs,
    )

    data = [
        self._archive_models.UnrealizedPnlHistoryDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

get_subaccount_volume_history(self, **kwargs) async

Gets historical trading volume for a subaccount from the archive API.

Source code in meridian/rest/subaccount.py

async def get_subaccount_volume_history(
    self, **kwargs
) -> List["SubaccountVolumeHistoryDto"]:
    """Gets historical trading volume for a subaccount from the archive API."""

    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/volume",
        request_model=self._archive_models.V1SubaccountVolumeGetParametersQuery,
        response_model=self._archive_models.PageOfSubaccountVolumeHistoryDtos,
        base_url_override=self._archive_base_url,
        **kwargs,
    )

    data = [
        self._archive_models.SubaccountVolumeHistoryDto(
            **model.model_dump(by_alias=True)
        )
        for model in res.data
    ]
    return data

list_all_subaccounts(self, **kwargs) async

Lists all subaccounts.

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[SubaccountDto]	List[SubaccountDto]: Subaccount records.

Source code in meridian/rest/subaccount.py

async def list_all_subaccounts(self, **kwargs) -> List[SubaccountDto]:
    """Lists all subaccounts.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[SubaccountDto]: Subaccount records.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/all",
        request_model=self._models.V1SubaccountAllGetParametersQuery,
        response_model=self._models.PageOfSubaccountDtos,
        **kwargs,
    )
    data = [
        self._models.SubaccountDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_subaccounts(self, **kwargs) async

Lists subaccounts for a given sender (address).

Parameters:

Name	Type	Description	Default
sender	str	Wallet address to query subaccounts for. Required.	required

Other Parameters:

Name	Type	Description
name	str	Filter by subaccount name. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[SubaccountDto]	List[SubaccountDto]: Subaccount records for the sender.

Source code in meridian/rest/subaccount.py

async def list_subaccounts(self, **kwargs) -> List[SubaccountDto]:
    """Lists subaccounts for a given sender (address).

    Args:
        sender (str): Wallet address to query subaccounts for. Required.

    Other Parameters:
        name (str, optional): Filter by subaccount name. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[SubaccountDto]: Subaccount records for the sender.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/",
        request_model=self._models.V1SubaccountGetParametersQuery,
        response_model=self._models.PageOfSubaccountDtos,
        **kwargs,
    )
    data = [
        self._models.SubaccountDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

meridian.rest.product

get_market_liquidity(self, **kwargs) async

Gets market liquidity for a product.

Parameters:

Name	Type	Description	Default
product_id	str	Id representing the registered product. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
MarketLiquidityDto	MarketLiquidityDto	Top-of-book and depth liquidity metrics.

Source code in meridian/rest/product.py

async def get_market_liquidity(self, **kwargs) -> MarketLiquidityDto:
    """Gets market liquidity for a product.

    Args:
        product_id (str): Id representing the registered product. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        MarketLiquidityDto: Top-of-book and depth liquidity metrics.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/product/market-liquidity",
        request_model=self._models.V1ProductMarketLiquidityGetParametersQuery,
        response_model=self._models.MarketLiquidityDto,
        **kwargs,
    )
    return res

get_product(self, id, **kwargs) async

Gets a specific product by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID of the product. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
ProductDto	ProductDto	Product configuration for the specified ID.

Source code in meridian/rest/product.py

async def get_product(self, id: UUID, **kwargs) -> ProductDto:
    """Gets a specific product by ID.

    Args:
        id (str): UUID of the product. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        ProductDto: Product configuration for the specified ID.
    """
    endpoint = f"{API_PREFIX}/product/{id}"
    res = await self.get(endpoint, **kwargs)
    return self._models.ProductDto(**res)

list_mark_price_gaps(self, **kwargs) async

Lists mark price gap windows for one or more products.

Parameters:

Name	Type	Description	Default
product_ids	List[UUID]	List of product IDs (1-50). Required.	required
start_time	float	Range start time in milliseconds since Unix epoch. Required.	required
end_time	float	Range end time in milliseconds since Unix epoch. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by, e.g., 'startTime'. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[MarkPriceGapDto]	List[MarkPriceGapDto]: Mark price gap windows for the requested products.

Source code in meridian/rest/product.py

async def list_mark_price_gaps(self, **kwargs) -> List[MarkPriceGapDto]:
    """Lists mark price gap windows for one or more products.

    Args:
        product_ids (List[UUID]): List of product IDs (1-50). Required.
        start_time (float): Range start time in milliseconds since Unix epoch. Required.
        end_time (float): Range end time in milliseconds since Unix epoch. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by, e.g., 'startTime'. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[MarkPriceGapDto]: Mark price gap windows for the requested products.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/product/mark-price-gap",
        request_model=self._models.V1ProductMarkPriceGapGetParametersQuery,
        response_model=self._models.PageOfMarkPriceGapDtos,
        **kwargs,
    )
    data = [
        self._models.MarkPriceGapDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_market_prices(self, **kwargs) async

Gets market prices for one or more products.

Parameters:

Name	Type	Description	Default
product_ids	List[str]	List of product IDs to query. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[MarketPriceDto]	List[MarketPriceDto]: Best bid/ask prices for the requested products.

Source code in meridian/rest/product.py

async def list_market_prices(self, **kwargs) -> List[MarketPriceDto]:
    """Gets market prices for one or more products.

    Args:
        product_ids (List[str]): List of product IDs to query. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[MarketPriceDto]: Best bid/ask prices for the requested products.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/product/market-price",
        request_model=self._models.V1ProductMarketPriceGetParametersQuery,
        response_model=self._models.ListOfMarketPriceDtos,
        **kwargs,
    )
    data = [
        self._models.MarketPriceDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_products(self, **kwargs) async

Lists all products and their configurations.

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
ticker	str	Filter by product ticker (e.g., 'ETHUSD'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[ProductDto]	List[ProductDto]: Product configuration records.

Source code in meridian/rest/product.py

async def list_products(self, **kwargs) -> List[ProductDto]:
    """Lists all products and their configurations.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        ticker (str, optional): Filter by product ticker (e.g., 'ETHUSD'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[ProductDto]: Product configuration records.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/product",
        request_model=self._models.V1ProductGetParametersQuery,
        response_model=self._models.PageOfProductDtos,
        **kwargs,
    )
    data = [
        self._models.ProductDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

meridian.rest.position

get_active_position(self, **kwargs) async

Gets the active position for a subaccount and product.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required
product_id	str	UUID of the product. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
PositionDto	PositionDto	Active position for the subaccount and product.

Source code in meridian/rest/position.py

async def get_active_position(
    self,
    **kwargs,
) -> PositionDto:
    """Gets the active position for a subaccount and product.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.
        product_id (str): UUID of the product. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        PositionDto: Active position for the subaccount and product.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/position/active",
        request_model=self._models.V1PositionActiveGetParametersQuery,
        response_model=self._models.PositionDto,
        **kwargs,
    )
    return res

get_position(self, id, **kwargs) async

Gets a specific position by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID of the position. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
PositionDto	PositionDto	Position information for the specified ID.

Source code in meridian/rest/position.py

async def get_position(
    self,
    id: UUID,
    **kwargs,
) -> PositionDto:
    """Gets a specific position by ID.

    Args:
        id (str): UUID of the position. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        PositionDto: Position information for the specified ID.
    """
    endpoint = f"{API_PREFIX}/position/{id}"
    res = await self.get(endpoint, **kwargs)
    return self._models.PositionDto(**res)

list_position_fees(self, **kwargs) async

Lists historical position-fee charges for a product over a time range.

Parameters:

Name	Type	Description	Default
product_id	str	Id representing the registered product. Required.	required
start_time	float	Range start time in milliseconds since Unix epoch. Required.	required
end_time	float	Range end time in milliseconds since Unix epoch. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by, e.g., 'chargedAt'. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[PositionFeeDto]	List[PositionFeeDto]: Position-fee charge records for the product.

Source code in meridian/rest/position.py

async def list_position_fees(
    self,
    **kwargs,
) -> List[PositionFeeDto]:
    """Lists historical position-fee charges for a product over a time range.

    Args:
        product_id (str): Id representing the registered product. Required.
        start_time (float): Range start time in milliseconds since Unix epoch. Required.
        end_time (float): Range end time in milliseconds since Unix epoch. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by, e.g., 'chargedAt'. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[PositionFeeDto]: Position-fee charge records for the product.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/position-fee",
        request_model=self._models.V1PositionFeeGetParametersQuery,
        response_model=self._models.PageOfPositionFeeDtos,
        **kwargs,
    )
    data = [
        self._models.PositionFeeDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_position_fills(self, **kwargs) async

Lists fills for a position.

Parameters:

Name	Type	Description	Default
position_id	str	UUID of the position. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[PositionFillDto]	List[PositionFillDto]: Fill records for the position.

Source code in meridian/rest/position.py

async def list_position_fills(
    self,
    **kwargs,
) -> List[PositionFillDto]:
    """Lists fills for a position.

    Args:
        position_id (str): UUID of the position. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[PositionFillDto]: Fill records for the position.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/position/fill",
        request_model=self._models.V1PositionFillGetParametersQuery,
        response_model=self._models.PageOfPositionFillDtos,
        **kwargs,
    )
    data = [
        self._models.PositionFillDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_position_liquidations(self, **kwargs) async

Lists position liquidations.

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[PositionLiquidationDto]	List[PositionLiquidationDto]: List of position liquidation records.

Source code in meridian/rest/position.py

async def list_position_liquidations(
    self,
    **kwargs,
) -> List[PositionLiquidationDto]:
    """Lists position liquidations.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[PositionLiquidationDto]: List of position liquidation records.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/position/liquidation",
        request_model=self._models.V1PositionLiquidationGetParametersQuery,
        response_model=self._models.PageOfPositionLiquidationsDto,
        **kwargs,
    )
    data = [
        self._models.PositionLiquidationDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_positions(self, **kwargs) async

Lists positions for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
product_ids	List[str]	UUIDs of products to filter by. Optional.
open	bool	Filter for open positions. Optional.
created_after	float	Filter positions created after this timestamp. Optional.
created_before	float	Filter positions created before this timestamp. Optional.
side	int	Filter by position side (0 for buy, 1 for sell). Optional.
is_liquidated	bool	Filter liquidated positions. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by, e.g., 'size', 'createdAt', 'updatedAt', or 'realizedPnl'. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[PositionDto]	List[PositionDto]: List of position information for the subaccount.

Source code in meridian/rest/position.py

async def list_positions(
    self,
    **kwargs,
) -> List[PositionDto]:
    """Lists positions for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        product_ids (List[str], optional): UUIDs of products to filter by. Optional.
        open (bool, optional): Filter for open positions. Optional.
        created_after (float, optional): Filter positions created after this timestamp. Optional.
        created_before (float, optional): Filter positions created before this timestamp. Optional.
        side (int, optional): Filter by position side (0 for buy, 1 for sell). Optional.
        is_liquidated (bool, optional): Filter liquidated positions. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by, e.g., 'size', 'createdAt', 'updatedAt', or 'realizedPnl'. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[PositionDto]: List of position information for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/position",
        request_model=self._models.V1PositionGetParametersQuery,
        response_model=self._models.PageOfPositionDtos,
        **kwargs,
    )
    data = [
        self._models.PositionDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_projected_position_fees(self, **kwargs) async

Lists projected position-fee rates for multiple products.

Parameters:

Name	Type	Description	Default
product_ids	List[UUID]	List of product IDs (1-50). Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[ProjectedPositionFeeDto]	List[ProjectedPositionFeeDto]: Projected position-fee rate objects for the products.

Source code in meridian/rest/position.py

async def list_projected_position_fees(
    self,
    **kwargs,
) -> List[ProjectedPositionFeeDto]:
    """Lists projected position-fee rates for multiple products.

    Args:
        product_ids (List[UUID]): List of product IDs (1-50). Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[ProjectedPositionFeeDto]: Projected position-fee rate objects for the products.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/position-fee/projected-rate",
        request_model=self._models.V1PositionFeeProjectedRateGetParametersQuery,
        response_model=self._models.PageOfProjectedPositionFeeDtos,
        **kwargs,
    )
    data = [
        self._models.ProjectedPositionFeeDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

meridian.rest.order

cancel_all_orders(self, subaccount_id, **kwargs) async

Cancel all open orders for a given subaccount via the cancel-all endpoint.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API.

Returns:

Type	Description
List[CancelOrderResultDto]	List[CancelOrderResultDto]: Cancellation results per order id.

Source code in meridian/rest/order.py

async def cancel_all_orders(
    self,
    subaccount_id: UUID,
    **kwargs,
) -> List[CancelOrderResultDto]:
    """Cancel all open orders for a given subaccount via the cancel-all endpoint.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API.

    Returns:
        List[CancelOrderResultDto]: Cancellation results per order id.
    """
    subaccount = await self.get_subaccount(id=subaccount_id)
    prepared = await self.prepare_cancel_all_order(
        sender=subaccount.account,
        subaccount=subaccount.name,
        include_signature=True,
    )
    return await self.submit_cancel_all_order(prepared, **kwargs)

dry_run_order(self, order, **kwargs) async

Submits a prepared order for validation without execution.

Parameters:

Name	Type	Description	Default
order	SubmitOrderDto	Prepared order payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
DryRunOrderCreatedDto	DryRunOrderCreatedDto	Dry-run validation result.

Source code in meridian/rest/order.py

async def dry_run_order(
    self,
    order: SubmitOrderDto,
    **kwargs,
) -> DryRunOrderCreatedDto:
    """Submits a prepared order for validation without execution.

    Args:
        order (SubmitOrderDto): Prepared order payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        DryRunOrderCreatedDto: Dry-run validation result.
    """
    submit_payload = self._models.SubmitDryOrderDto.model_validate(
        {"data": order.data.model_dump(mode="json", by_alias=True, exclude_unset=True)}
    )
    endpoint = f"{API_PREFIX}/order/dry-run"
    res = await self.post(
        endpoint,
        data=submit_payload.model_dump(
            mode="json", by_alias=True, exclude_unset=True, exclude_none=True
        ),
        **kwargs,
    )
    return self._models.DryRunOrderCreatedDto.model_validate(res)

get_order(self, id, **kwargs) async

Gets a specific order by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID of the order. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
OrderDto	OrderDto	Order details.

Source code in meridian/rest/order.py

async def get_order(self, id: UUID, **kwargs) -> OrderDto:
    """Gets a specific order by ID.

    Args:
        id (str): UUID of the order. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        OrderDto: Order details.
    """
    endpoint = f"{API_PREFIX}/order/{id}"
    response = await self.get(endpoint, **kwargs)
    return self._models.OrderDto(**response)

list_fills(self, **kwargs) async

Lists order fills for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
product_ids	List[str]	Filter by one or more product IDs. Optional.
created_after	float	Filter fills created after this timestamp. Optional.
created_before	float	Filter fills created before this timestamp. Optional.
side	int	Filter by order side (0 for buy, 1 for sell). Optional.
include_self_trades	bool	Include self trades in results. Defaults to False. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt', 'productId'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[OrderFillDto]	List[OrderFillDto]: Fill records for the subaccount.

Source code in meridian/rest/order.py

async def list_fills(self, **kwargs) -> List[OrderFillDto]:
    """Lists order fills for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        product_ids (List[str], optional): Filter by one or more product IDs. Optional.
        created_after (float, optional): Filter fills created after this timestamp. Optional.
        created_before (float, optional): Filter fills created before this timestamp. Optional.
        side (int, optional): Filter by order side (0 for buy, 1 for sell). Optional.
        include_self_trades (bool, optional): Include self trades in results. Defaults to False. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt', 'productId'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[OrderFillDto]: Fill records for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/order/fill",
        request_model=self._models.V1OrderFillGetParametersQuery,
        response_model=self._models.PageOfOrderFillDtos,
        **kwargs,
    )
    data = [
        self._models.OrderFillDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_order_group(self, id, **kwargs) async

Lists orders in the same contingency group as an order.

Parameters:

Name	Type	Description	Default
id	str	UUID of the order. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[OrderDto]	List[OrderDto]: Orders in the contingency group.

Source code in meridian/rest/order.py

async def list_order_group(self, id: UUID, **kwargs) -> List[OrderDto]:
    """Lists orders in the same contingency group as an order.

    Args:
        id (str): UUID of the order. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[OrderDto]: Orders in the contingency group.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/order/{id}/group",
        request_model=self._models.V1OrderIdGroupGetParametersQuery,
        response_model=self._models.PageOfOrderDtos,
        **kwargs,
    )
    data = [
        self._models.OrderDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

list_orders(self, **kwargs) async

Lists orders for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
product_ids	List[str]	Filter by one or more product IDs. Optional.
client_order_id	str	Filter by a client-generated order id. Optional.
statuses	List[str]	Filter by status values. Optional.
created_after	float	Filter orders created after this timestamp. Optional.
created_before	float	Filter orders created before this timestamp. Optional.
side	int	Filter by order side (0 for buy, 1 for sell). Optional.
close	bool	Filter by close flag. Optional.
stop_types	List[int]	Filter by stop types (0 for GAIN, 1 for LOSS). Optional.
is_working	bool	Filter working orders. Optional.
is_pending	bool	Filter pending orders. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[OrderDto]	List[OrderDto]: Order records for the subaccount.

Source code in meridian/rest/order.py

async def list_orders(self, **kwargs) -> List[OrderDto]:
    """Lists orders for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        product_ids (List[str], optional): Filter by one or more product IDs. Optional.
        client_order_id (str, optional): Filter by a client-generated order id. Optional.
        statuses (List[str], optional): Filter by status values. Optional.
        created_after (float, optional): Filter orders created after this timestamp. Optional.
        created_before (float, optional): Filter orders created before this timestamp. Optional.
        side (int, optional): Filter by order side (0 for buy, 1 for sell). Optional.
        close (bool, optional): Filter by close flag. Optional.
        stop_types (List[int], optional): Filter by stop types (0 for GAIN, 1 for LOSS). Optional.
        is_working (bool, optional): Filter working orders. Optional.
        is_pending (bool, optional): Filter pending orders. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[OrderDto]: Order records for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/order",
        request_model=self._models.V1OrderGetParametersQuery,
        response_model=self._models.PageOfOrderDtos,
        **kwargs,
    )
    data = [
        self._models.OrderDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

list_trades(self, **kwargs) async

Lists trades for a specific product.

Parameters:

Name	Type	Description	Default
product_id	str	Product ID to query trades for. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[TradeDto]	List[TradeDto]: Trade records.

Source code in meridian/rest/order.py

async def list_trades(self, **kwargs) -> List[TradeDto]:
    """Lists trades for a specific product.

    Args:
        product_id (str): Product ID to query trades for. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[TradeDto]: Trade records.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/order/trade",
        request_model=self._models.V1OrderTradeGetParametersQuery,
        response_model=self._models.PageOfTradeDtos,
        **kwargs,
    )
    data = [
        self._models.TradeDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

prepare_batch_order(self, instructions) async

Prepares a batch order payload from signed order and cancel payloads.

Parameters:

Name	Type	Description	Default
instructions	List[Union[SubmitOrderDto, CancelOrderDto]]	Prepared SubmitOrderDto or prepared CancelOrderDto payloads.	required

Returns:

Name	Type	Description
BatchOrderDto	BatchOrderDto	Prepared batch order payload.

Source code in meridian/rest/order.py

async def prepare_batch_order(
    self,
    instructions: List[Union[SubmitOrderDto, CancelOrderDto]],
) -> "BatchOrderDto":
    """Prepares a batch order payload from signed order and cancel payloads.

    Args:
        instructions: Prepared SubmitOrderDto or prepared CancelOrderDto payloads.

    Returns:
        BatchOrderDto: Prepared batch order payload.
    """
    batch_instructions = []
    for instruction in instructions:
        if isinstance(instruction, self._models.SubmitOrderDto):
            batch_instructions.append(
                self._models.BatchInstructionDto1(
                    data=instruction.data,
                    signature=instruction.signature,
                    type="NEW",
                )
            )
        elif isinstance(instruction, self._models.CancelOrderDto):
            data = instruction.data
            order_ids = data.order_ids or []
            client_order_ids = data.client_order_ids or []

            if len(order_ids) + len(client_order_ids) != 1:
                raise ValueError(
                    "Batch cancel instructions must target exactly one order id "
                    "or client order id"
                )

            batch_instructions.append(
                self._models.BatchInstructionDto2(
                    data=self._models.BatchCancelOrderDtoData.model_validate(
                        {
                            "sender": data.sender,
                            "subaccount": data.subaccount,
                            "nonce": data.nonce,
                            "orderId": order_ids[0] if order_ids else None,
                            "clientOrderId": client_order_ids[0]
                            if client_order_ids
                            else None,
                        }
                    ),
                    signature=instruction.signature,
                    type="CANCEL",
                )
            )
        else:
            raise TypeError(
                "Batch instructions must be prepared SubmitOrderDto or "
                "CancelOrderDto payloads"
            )

    return self._models.BatchOrderDto(instructions=batch_instructions)

prepare_cancel_all_order(self, sender, subaccount, include_signature=False, **kwargs) async

Prepares the payload for canceling all open orders of a subaccount.

Parameters:

Name	Type	Description	Default
sender	str	Address initiating the cancellation. Required.	required
subaccount	str	Hex-encoded subaccount name. Required.	required
include_signature	bool	If True, sign the payload immediately. Optional.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing.
**kwargs		Additional request parameters accepted by the API.

Returns:

Name	Type	Description
CancelAllOrderDto	CancelAllOrderDto	Prepared (and optionally signed) cancel-all payload.

Source code in meridian/rest/order.py

async def prepare_cancel_all_order(
    self,
    sender: str,
    subaccount: str,
    include_signature: bool = False,
    **kwargs,
) -> CancelAllOrderDto:
    """Prepares the payload for canceling all open orders of a subaccount.

    Args:
        sender (str): Address initiating the cancellation. Required.
        subaccount (str): Hex-encoded subaccount name. Required.
        include_signature (bool): If True, sign the payload immediately. Optional.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing.
        **kwargs: Additional request parameters accepted by the API.

    Returns:
        CancelAllOrderDto: Prepared (and optionally signed) cancel-all payload.
    """
    nonce = kwargs.get("nonce", None) or generate_nonce()
    data_model = self._models.CancelAllOrderDtoData(
        sender=sender,
        subaccount=subaccount,
        nonce=nonce,
    )
    result = self._models.CancelAllOrderDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        return await self.sign_cancel_all_order(result)
    return result

prepare_cancel_order(self, sender, subaccount, order_ids=[], client_order_ids=[], include_signature=False, **kwargs) async

Prepares the payload for canceling one or more orders.

Parameters:

Name	Type	Description	Default
sender	str	Address initiating the cancellation. Required.	required
subaccount	str	Hex-encoded subaccount name. Required.	required
order_ids	List[str]	Order UUIDs to cancel. Optional.	[]
client_order_ids	List[str]	Client-generated IDs to cancel. Optional.	[]
include_signature	bool	If True, sign the payload immediately. Optional.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing.
**kwargs		Additional request parameters accepted by the API.

Returns:

Name	Type	Description
CancelOrderDto	CancelOrderDto	Prepared (and optionally signed) cancel payload.

Source code in meridian/rest/order.py

async def prepare_cancel_order(
    self,
    sender: str,
    subaccount: str,
    order_ids: List[UUID] = [],
    client_order_ids: List[str] = [],
    include_signature: bool = False,
    **kwargs,
) -> CancelOrderDto:
    """Prepares the payload for canceling one or more orders.

    Args:
        sender (str): Address initiating the cancellation. Required.
        subaccount (str): Hex-encoded subaccount name. Required.
        order_ids (List[str]): Order UUIDs to cancel. Optional.
        client_order_ids (List[str]): Client-generated IDs to cancel. Optional.
        include_signature (bool): If True, sign the payload immediately. Optional.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing.
        **kwargs: Additional request parameters accepted by the API.

    Returns:
        CancelOrderDto: Prepared (and optionally signed) cancel payload.
    """
    nonce = kwargs.get("nonce", None) or generate_nonce()
    uuid_order_ids = [
        UUID(order_id) if isinstance(order_id, str) else order_id
        for order_id in order_ids
    ]
    data_model = self._models.CancelOrderDtoData(
        sender=sender,
        subaccount=subaccount,
        nonce=nonce,
        orderIds=uuid_order_ids,
        clientOrderIds=client_order_ids,
    )
    result = self._models.CancelOrderDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        return await self.sign_cancel_order(result)
    return result

prepare_order(self, sender, price=None, quantity=None, side=None, subaccount=None, onchain_id=None, order_type=None, client_order_id=None, time_in_force=None, post_only=False, reduce_only=False, close=None, stop_price=None, stop_type=None, group_id=None, group_contingency_type=None, expires_at=None, include_signature=False, **kwargs) async

Prepares the payload for an order, optionally including a signature.

Parameters:

Name	Type	Description	Default
sender	str	Address placing the order. Required.	required
price	Union[str, float, Decimal]	Limit price for LIMIT orders.	None
quantity	Union[str, float, Decimal]	Order size.	None
side	int	0 for buy, 1 for sell.	None
subaccount	str	Hex-encoded subaccount name.	None
onchain_id	float	Product onchain ID.	None
order_type	str	'LIMIT' or 'MARKET'.	None
client_order_id	str	Subaccount-scoped client-generated id (UUID or <=32 alphanumeric).	None
time_in_force	str	For LIMIT orders (e.g., 'GTC', 'GTD').	None
post_only	bool	For LIMIT orders; rejects if crossing.	False
reduce_only	bool	If True, order only reduces position.	False
close	bool	If True, closes the position.	None
stop_price	Union[str, float, Decimal]	Stop trigger price.	None
stop_type	int	Stop type enum value.	None
group_id	str	Contingency group id.	None
group_contingency_type	int	Group contingency type.	None
expires_at	int	Expiry timestamp for GTD.	None
include_signature	bool	If True, sign the payload immediately.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing.
signed_at	int	Seconds since epoch for signature timestamp.
**kwargs		Additional request parameters accepted by the API.

Returns:

Name	Type	Description
SubmitOrderDto	SubmitOrderDto	Prepared (and optionally signed) order payload.

Source code in meridian/rest/order.py

async def prepare_order(
    self,
    sender: str,
    price: Optional[Union[str, float, Decimal]] = None,
    quantity: Optional[Union[str, float, Decimal]] = None,
    side: Optional[int] = None,
    subaccount: Optional[str] = None,
    onchain_id: Optional[int] = None,
    order_type: Optional[str] = None,
    client_order_id: Optional[str] = None,
    time_in_force: Optional[str] = None,
    post_only: Optional[bool] = False,
    reduce_only: Optional[bool] = False,
    close: Optional[bool] = None,
    stop_price: Optional[Union[str, float, Decimal]] = None,
    stop_type: Optional[int] = None,
    group_id: Optional[str] = None,
    group_contingency_type: Optional[int] = None,
    expires_at: Optional[int] = None,
    include_signature: bool = False,
    **kwargs,
) -> SubmitOrderDto:
    """Prepares the payload for an order, optionally including a signature.

    Args:
        sender (str): Address placing the order. Required.
        price (Union[str, float, Decimal], optional): Limit price for LIMIT orders.
        quantity (Union[str, float, Decimal], optional): Order size.
        side (int, optional): 0 for buy, 1 for sell.
        subaccount (str, optional): Hex-encoded subaccount name.
        onchain_id (float, optional): Product onchain ID.
        order_type (str, optional): 'LIMIT' or 'MARKET'.
        client_order_id (str, optional): Subaccount-scoped client-generated id (UUID or <=32 alphanumeric).
        time_in_force (str, optional): For LIMIT orders (e.g., 'GTC', 'GTD').
        post_only (bool, optional): For LIMIT orders; rejects if crossing.
        reduce_only (bool, optional): If True, order only reduces position.
        close (bool, optional): If True, closes the position.
        stop_price (Union[str, float, Decimal], optional): Stop trigger price.
        stop_type (int, optional): Stop type enum value.
        group_id (str, optional): Contingency group id.
        group_contingency_type (int, optional): Group contingency type.
        expires_at (int, optional): Expiry timestamp for GTD.
        include_signature (bool): If True, sign the payload immediately.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing.
        signed_at (int, optional): Seconds since epoch for signature timestamp.
        **kwargs: Additional request parameters accepted by the API.

    Returns:
        SubmitOrderDto: Prepared (and optionally signed) order payload.
    """
    # Generate nonce and signed_at timestamp
    nonce = kwargs.get("nonce", None) or generate_nonce()
    signed_at = kwargs.get("signed_at", None) or int(time.time())

    # Prepare order data with common fields
    order_data = {
        "sender": sender,
        "subaccount": subaccount,
        "quantity": quantity,
        "side": side,
        "engineType": 0,
        "onchainId": onchain_id,
        "nonce": nonce,
        "type": order_type,
        "clientOrderId": client_order_id,
        "reduceOnly": reduce_only,
        "signedAt": signed_at,
        "close": close,
        "stopPrice": stop_price,
        "stopType": stop_type,
        "groupId": group_id,
        "groupContingencyType": group_contingency_type,
        "expiresAt": expires_at,
    }

    # Declare data_model type
    data_model: Union[SubmitOrderLimitDtoData, SubmitOrderMarketDtoData]

    # Create specific order data based on type
    if order_type == "LIMIT":
        order_data.update(
            {
                "price": price,
                "timeInForce": time_in_force,
                "postOnly": post_only,
            }
        )
        data_model = self._models.SubmitOrderLimitDtoData.model_validate(order_data)
    elif order_type == "MARKET":
        data_model = self._models.SubmitOrderMarketDtoData.model_validate(order_data)
    else:
        raise ValueError(f"Invalid order type: {order_type}")

    result = self._models.SubmitOrderDto.model_validate(
        {
            "data": data_model.model_dump(
                mode="json", by_alias=True, exclude_unset=True, exclude_none=True
            ),
            "signature": "",
        }
    )

    if include_signature:
        result = await self.sign_order(result)

    return result

sign_cancel_all_order(self, cancel_all_order, private_key=None) async

Signs a cancel-all order payload using EIP-712.

The payload is signed with the same CancelOrder primary type used for canceling orders by id; the signed message covers sender, subaccount and nonce only.

Parameters:

Name	Type	Description	Default
cancel_all_order	CancelAllOrderDto	Prepared cancel-all payload. Required.	required
private_key	str	Private key override. Defaults to client's key.	None

Returns:

Name	Type	Description
CancelAllOrderDto	CancelAllOrderDto	The same DTO with signature populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/order.py

async def sign_cancel_all_order(
    self,
    cancel_all_order: CancelAllOrderDto,
    private_key: Optional[str] = None,
) -> CancelAllOrderDto:
    """Signs a cancel-all order payload using EIP-712.

    The payload is signed with the same CancelOrder primary type used for
    canceling orders by id; the signed message covers sender, subaccount
    and nonce only.

    Args:
        cancel_all_order (CancelAllOrderDto): Prepared cancel-all payload. Required.
        private_key (str, optional): Private key override. Defaults to client's key.

    Returns:
        CancelAllOrderDto: The same DTO with signature populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key and not self.chain.private_key:
        raise ValueError("No private key available for signing")
    elif not private_key:
        private_key = self.chain.private_key

    message = cancel_all_order.data.model_dump(mode="json", by_alias=True)
    primary_type = "CancelOrder"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)

    cancel_all_order.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return cancel_all_order

sign_cancel_order(self, order_to_cancel, private_key=None) async

Signs a cancel order payload using EIP-712.

Parameters:

Name	Type	Description	Default
order_to_cancel	CancelOrderDto	Prepared cancel payload. Required.	required
private_key	str	Private key override. Defaults to client's key.	None

Returns:

Name	Type	Description
CancelOrderDto	CancelOrderDto	The same DTO with signature populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/order.py

async def sign_cancel_order(
    self,
    order_to_cancel: CancelOrderDto,
    private_key: Optional[str] = None,
) -> CancelOrderDto:
    """Signs a cancel order payload using EIP-712.

    Args:
        order_to_cancel (CancelOrderDto): Prepared cancel payload. Required.
        private_key (str, optional): Private key override. Defaults to client's key.

    Returns:
        CancelOrderDto: The same DTO with signature populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key and not self.chain.private_key:
        raise ValueError("No private key available for signing")
    elif not private_key:
        private_key = self.chain.private_key

    # Prepare message for signing
    message = order_to_cancel.data.model_dump(mode="json", by_alias=True)

    # For cancel orders, orderIds need to be converted to bytes32 format
    order_ids = (
        [uuid_to_bytes32(str(order_id)) for order_id in order_to_cancel.data.order_ids]
        if order_to_cancel.data.order_ids
        else []
    )
    message["orderIds"] = order_ids

    client_order_ids = (
        [client_order_id_to_bytes32(id) for id in order_to_cancel.data.client_order_ids]
        if order_to_cancel.data.client_order_ids
        else []
    )
    message["clientOrderIds"] = client_order_ids

    # Get domain and types for signing
    primary_type = "CancelOrder"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)

    # Sign the message
    order_to_cancel.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return order_to_cancel

sign_order(self, order, private_key=None) async

Signs an order payload using EIP-712.

Parameters:

Name	Type	Description	Default
order	SubmitOrderDto	Prepared order to sign. Required.	required
private_key	str	Private key override. Defaults to client's key.	None

Returns:

Name	Type	Description
SubmitOrderDto	SubmitOrderDto	The same DTO with signature populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/order.py

async def sign_order(
    self, order: SubmitOrderDto, private_key: Optional[str] = None
) -> SubmitOrderDto:
    """Signs an order payload using EIP-712.

    Args:
        order (SubmitOrderDto): Prepared order to sign. Required.
        private_key (str, optional): Private key override. Defaults to client's key.

    Returns:
        SubmitOrderDto: The same DTO with signature populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key and not self.chain.private_key:
        raise ValueError("No private key available for signing")
    elif not private_key:
        private_key = self.chain.private_key

    # The order data fields are wrapped in a root model
    order.data.root.signed_at = int(time.time())

    # Prepare message for signing
    message = order.data.model_dump(mode="json", by_alias=True)

    # Make some adjustments to the message
    message["quantity"] = int(Decimal(message["quantity"]) * Decimal("1e9"))
    message["price"] = int(Decimal(message.get("price", 0)) * Decimal("1e9"))
    message["productId"] = int(message["onchainId"])
    message["signedAt"] = int(message["signedAt"])
    if message.get("clientOrderId"):
        message["clientOrderId"] = client_order_id_to_bytes32(message["clientOrderId"])

    # Get domain and types for signing
    primary_type = "TradeOrder"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)

    # Sign the message
    order.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return order

submit_batch_order(self, batch_order, **kwargs) async

Submits a batch of prepared and signed order instructions.

Parameters:

Name	Type	Description	Default
batch_order	BatchOrderDto	BatchOrderDto or a dict with an instructions list.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
BatchOrderResponseDto	BatchOrderResponseDto	Result for each batch instruction.

Source code in meridian/rest/order.py

async def submit_batch_order(
    self,
    batch_order: "BatchOrderDto",
    **kwargs,
) -> "BatchOrderResponseDto":
    """Submits a batch of prepared and signed order instructions.

    Args:
        batch_order: BatchOrderDto or a dict with an ``instructions`` list.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        BatchOrderResponseDto: Result for each batch instruction.
    """
    batch_order = self._models.BatchOrderDto.model_validate(batch_order)
    res = await self.post(
        f"{API_PREFIX}/order/batch",
        data=batch_order.model_dump(
            mode="json", by_alias=True, exclude_unset=True, exclude_none=True
        ),
        **kwargs,
    )
    return self._models.BatchOrderResponseDto.model_validate(res)

submit_cancel_all_order(self, cancel_all_order, **kwargs) async

Submits a prepared and signed cancel-all order request.

Parameters:

Name	Type	Description	Default
cancel_all_order	CancelAllOrderDto	Prepared and signed cancel-all payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[CancelOrderResultDto]	List[CancelOrderResultDto]: Cancellation results per order id.

Source code in meridian/rest/order.py

async def submit_cancel_all_order(
    self,
    cancel_all_order: CancelAllOrderDto,
    **kwargs,
) -> List[CancelOrderResultDto]:
    """Submits a prepared and signed cancel-all order request.

    Args:
        cancel_all_order (CancelAllOrderDto): Prepared and signed cancel-all payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[CancelOrderResultDto]: Cancellation results per order id.
    """
    endpoint = f"{API_PREFIX}/order/cancel-all"
    res = await self.post(
        endpoint,
        data=cancel_all_order.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return [
        self._models.CancelOrderResultDto.model_validate(item)
        for item in res.get("data", [])
    ]

submit_cancel_order(self, order_to_cancel, **kwargs) async

Submits a prepared and signed cancel order request.

Parameters:

Name	Type	Description	Default
order_to_cancel	CancelOrderDto	Prepared and signed cancel payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[CancelOrderResultDto]	List[CancelOrderResultDto]: Cancellation results per order id.

Source code in meridian/rest/order.py

async def submit_cancel_order(
    self,
    order_to_cancel: CancelOrderDto,
    **kwargs,
) -> List[CancelOrderResultDto]:
    """Submits a prepared and signed cancel order request.

    Args:
        order_to_cancel (CancelOrderDto): Prepared and signed cancel payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[CancelOrderResultDto]: Cancellation results per order id.
    """
    endpoint = f"{API_PREFIX}/order/cancel"
    res = await self.post(
        endpoint,
        data=order_to_cancel.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return [
        self._models.CancelOrderResultDto.model_validate(item)
        for item in res.get("data", [])
    ]

submit_order(self, order, **kwargs) async

Submits a prepared and signed order.

Parameters:

Name	Type	Description	Default
order	SubmitOrderDto	Prepared and signed order payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SubmitOrderCreatedDto	SubmitOrderCreatedDto	Created order response.

Source code in meridian/rest/order.py

async def submit_order(
    self,
    order: SubmitOrderDto,
    **kwargs,
) -> SubmitOrderCreatedDto:
    """Submits a prepared and signed order.

    Args:
        order (SubmitOrderDto): Prepared and signed order payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SubmitOrderCreatedDto: Created order response.
    """
    endpoint = f"{API_PREFIX}/order"
    res = await self.post(
        endpoint,
        data=order.model_dump(
            mode="json", by_alias=True, exclude_unset=True, exclude_none=True
        ),
        **kwargs,
    )
    return self._models.SubmitOrderCreatedDto.model_validate(res)

meridian.rest.funding

list_funding(self, **kwargs) async

Lists historical funding rates for a product over a specified time range.

Parameters:

Name	Type	Description	Default
product_id	str	Id representing the registered product. Required.	required
range	str	Time window to query. One of 'DAY', 'WEEK', or 'MONTH'. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by, e.g., 'createdAt'. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[FundingDto]	List[FundingDto]: Funding rate history objects for the product.

Source code in meridian/rest/funding.py

async def list_funding(self, **kwargs) -> List["FundingDto"]:
    """Lists historical funding rates for a product over a specified time range.

    Args:
        product_id (str): Id representing the registered product. Required.
        range (str): Time window to query. One of 'DAY', 'WEEK', or 'MONTH'. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by, e.g., 'createdAt'. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[FundingDto]: Funding rate history objects for the product.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/funding",
        request_model=self._models.V1FundingGetParametersQuery,
        response_model=self._models.PageOfFundingDtos,
        **kwargs,
    )
    data = [
        self._models.FundingDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

list_projected_funding(self, **kwargs) async

Lists projected funding rates for multiple products.

Parameters:

Name	Type	Description	Default
product_ids	List[UUID]	List of product IDs (1-10). Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[ProjectedFundingDto]	List[ProjectedFundingDto]: Projected funding rate objects for the products.

Source code in meridian/rest/funding.py

async def list_projected_funding(self, **kwargs) -> List["ProjectedFundingDto"]:
    """Lists projected funding rates for multiple products.

    Args:
        product_ids (List[UUID]): List of product IDs (1-10). Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[ProjectedFundingDto]: Projected funding rate objects for the products.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/funding/projected-rate",
        request_model=self._models.V1FundingProjectedRateGetParametersQuery,
        response_model=self._models.PageOfProjectedFundingDtos,
        **kwargs,
    )
    data = [
        self._models.ProjectedFundingDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

meridian.rest.linked_signer

extend_linked_signer(self, dto, **kwargs) async

Submits a prepared and signed extend-linked-signer payload.

Parameters:

Name	Type	Description	Default
dto	ExtendLinkedSignerDto	Prepared and signed extend payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SignerDto	SignerDto	Signer record reflecting the extended expiry.

Source code in meridian/rest/linked_signer.py

async def extend_linked_signer(
    self,
    dto: ExtendLinkedSignerDto,
    **kwargs,
) -> SignerDto:
    """Submits a prepared and signed extend-linked-signer payload.

    Args:
        dto (ExtendLinkedSignerDto): Prepared and signed extend payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SignerDto: Signer record reflecting the extended expiry.
    """
    endpoint = f"{API_PREFIX}/linked-signer/extend"
    res = await self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return self._models.SignerDto.model_validate(res)

get_signer(self, id, **kwargs) async

Gets a specific linked signer by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID of the linked signer. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SignerDto	SignerDto	Linked signer details.

Source code in meridian/rest/linked_signer.py

async def get_signer(
    self,
    id: UUID,
    **kwargs,
) -> SignerDto:
    """Gets a specific linked signer by ID.

    Args:
        id (str): UUID of the linked signer. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SignerDto: Linked signer details.
    """
    endpoint = f"{API_PREFIX}/linked-signer/{id}"
    res = await self.get(endpoint, **kwargs)
    return self._models.SignerDto.model_validate(res)

get_signer_by_address(self, address, **kwargs) async

Gets a specific linked signer by address.

Parameters:

Name	Type	Description	Default
address	str	Address of the linked signer. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SignerDto	SignerDto	Linked signer details.

Source code in meridian/rest/linked_signer.py

async def get_signer_by_address(
    self,
    address: str,
    **kwargs,
) -> SignerDto:
    """Gets a specific linked signer by address.

    Args:
        address (str): Address of the linked signer. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SignerDto: Linked signer details.
    """
    endpoint = f"{API_PREFIX}/linked-signer/address/{address}"
    res = await self.get(endpoint, **kwargs)
    return self._models.SignerDto.model_validate(res)

get_signer_quota(self, **kwargs) async

Gets the signer quota for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
AccountSignerQuotaDto	AccountSignerQuotaDto	Remaining quota information for the subaccount.

Source code in meridian/rest/linked_signer.py

async def get_signer_quota(
    self,
    **kwargs,
) -> AccountSignerQuotaDto:
    """Gets the signer quota for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        AccountSignerQuotaDto: Remaining quota information for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/linked-signer/quota",
        request_model=self._models.V1LinkedSignerQuotaGetParametersQuery,
        response_model=self._models.AccountSignerQuotaDto,
        **kwargs,
    )
    return res

link_linked_signer(self, dto, **kwargs) async

Submits a prepared and signed link-signer payload.

Parameters:

Name	Type	Description	Default
dto	LinkSignerDto	Prepared and signed link payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SignerDto	SignerDto	Linked signer record after submission.

Source code in meridian/rest/linked_signer.py

async def link_linked_signer(
    self,
    dto: LinkSignerDto,
    **kwargs,
) -> SignerDto:
    """Submits a prepared and signed link-signer payload.

    Args:
        dto (LinkSignerDto): Prepared and signed link payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SignerDto: Linked signer record after submission.
    """
    endpoint = f"{API_PREFIX}/linked-signer/link"
    res = await self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return self._models.SignerDto.model_validate(res)

list_signers(self, **kwargs) async

Lists linked signers for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[SignerDto]	List[SignerDto]: Linked signer records.

Source code in meridian/rest/linked_signer.py

async def list_signers(
    self,
    **kwargs,
) -> List[SignerDto]:
    """Lists linked signers for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[SignerDto]: Linked signer records.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/linked-signer",
        request_model=self._models.V1LinkedSignerGetParametersQuery,
        response_model=self._models.PageOfSignersDto,
        **kwargs,
    )
    data = [
        self._models.SignerDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

prepare_extend_linked_signer(self, sender, include_signature=False, **kwargs) async

Prepares the payload for extending a linked signer, optionally signing it.

Parameters:

Name	Type	Description	Default
sender	str	Address of the linked signer extending itself. Required.	required
include_signature	bool	If True, sign with the client's key. Optional.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing. Optional.
signed_at	int	Seconds since epoch for the signature timestamp. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
ExtendLinkedSignerDto	ExtendLinkedSignerDto	Prepared (and optionally signed) extend payload.

Source code in meridian/rest/linked_signer.py

async def prepare_extend_linked_signer(
    self,
    sender: str,
    include_signature: bool = False,
    **kwargs,
) -> ExtendLinkedSignerDto:
    """Prepares the payload for extending a linked signer, optionally signing it.

    Args:
        sender (str): Address of the linked signer extending itself. Required.
        include_signature (bool): If True, sign with the client's key. Optional.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing. Optional.
        signed_at (int, optional): Seconds since epoch for the signature timestamp. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        ExtendLinkedSignerDto: Prepared (and optionally signed) extend payload.
    """
    nonce = kwargs.get("nonce") or generate_nonce()
    signed_at = kwargs.get("signed_at") or int(time.time())
    data = {
        "sender": sender,
        "nonce": nonce,
        "signedAt": signed_at,
    }
    data_model = self._models.ExtendLinkedSignerDtoData.model_validate(data)
    dto = self._models.ExtendLinkedSignerDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = await self.sign_extend_linked_signer(dto)
    return dto

prepare_linked_signer(self, sender, signer, subaccount, subaccount_id, signer_signature='', include_signature=False, **kwargs) async

Prepares the payload for linking a signer, optionally including a signature.

Parameters:

Name	Type	Description	Default
sender	str	Owner address initiating the link. Required.	required
signer	str	Address of the signer being linked. Required.	required
subaccount	str	Hex-encoded subaccount name. Required.	required
subaccount_id	str	UUID of the subaccount. Required.	required
signer_signature	str	Signature from the signer address. Optional.	''
include_signature	bool	If True, sign with the owner's key as well. Optional.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing. Optional.
signed_at	int	Seconds since epoch for the signature timestamp. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
LinkSignerDto	LinkSignerDto	Prepared (and optionally signed) link payload.

Source code in meridian/rest/linked_signer.py

async def prepare_linked_signer(
    self,
    sender: str,
    signer: str,
    subaccount: str,
    subaccount_id: UUID,
    signer_signature: str = "",
    include_signature: bool = False,
    **kwargs,
) -> LinkSignerDto:
    """Prepares the payload for linking a signer, optionally including a signature.

    Args:
        sender (str): Owner address initiating the link. Required.
        signer (str): Address of the signer being linked. Required.
        subaccount (str): Hex-encoded subaccount name. Required.
        subaccount_id (str): UUID of the subaccount. Required.
        signer_signature (str): Signature from the signer address. Optional.
        include_signature (bool): If True, sign with the owner's key as well. Optional.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing. Optional.
        signed_at (int, optional): Seconds since epoch for the signature timestamp. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        LinkSignerDto: Prepared (and optionally signed) link payload.
    """
    nonce = kwargs.get("nonce") or generate_nonce()
    signed_at = kwargs.get("signed_at") or int(time.time())
    data = {
        "sender": sender,
        "signer": signer,
        "subaccount": subaccount,
        "subaccountId": subaccount_id,
        "nonce": nonce,
        "signedAt": signed_at,
    }
    data_model = self._models.LinkSignerDtoData.model_validate(data)

    # Prepare dto
    dto_data = {
        "data": data_model.model_dump(mode="json", by_alias=True),
        "signature": "",
        "signerSignature": signer_signature,
    }
    dto = self._models.LinkSignerDto.model_validate(dto_data, by_alias=True)
    if include_signature:
        dto = await self.sign_linked_signer(dto, private_key=self.chain.private_key)
    return dto

prepare_refresh_linked_signer(self, sender, signer, subaccount, subaccount_id, include_signature=False, **kwargs) async

Prepares the payload for refreshing a linked signer, optionally signing it.

Parameters:

Name	Type	Description	Default
sender	str	Owner address initiating the refresh. Required.	required
signer	str	Signer address being refreshed. Required.	required
subaccount	str	Hex-encoded subaccount name. Required.	required
subaccount_id	str	UUID of the subaccount. Required.	required
include_signature	bool	If True, sign with the owner's key. Optional.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing. Optional.
signed_at	int	Seconds since epoch for the signature timestamp. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
RefreshLinkedSignerDto	RefreshLinkedSignerDto	Prepared (and optionally signed) refresh payload.

Source code in meridian/rest/linked_signer.py

async def prepare_refresh_linked_signer(
    self,
    sender: str,
    signer: str,
    subaccount: str,
    subaccount_id: UUID,
    include_signature: bool = False,
    **kwargs,
) -> RefreshLinkedSignerDto:
    """Prepares the payload for refreshing a linked signer, optionally signing it.

    Args:
        sender (str): Owner address initiating the refresh. Required.
        signer (str): Signer address being refreshed. Required.
        subaccount (str): Hex-encoded subaccount name. Required.
        subaccount_id (str): UUID of the subaccount. Required.
        include_signature (bool): If True, sign with the owner's key. Optional.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing. Optional.
        signed_at (int, optional): Seconds since epoch for the signature timestamp. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        RefreshLinkedSignerDto: Prepared (and optionally signed) refresh payload.
    """
    nonce = kwargs.get("nonce") or generate_nonce()
    signed_at = kwargs.get("signed_at") or int(time.time())
    data = {
        "sender": sender,
        "signer": signer,
        "subaccount": subaccount,
        "subaccountId": subaccount_id,
        "nonce": nonce,
        "signedAt": signed_at,
    }
    data_model = self._models.RefreshLinkedSignerDtoData.model_validate(data)
    dto = self._models.RefreshLinkedSignerDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = await self.sign_refresh_linked_signer(dto)
    return dto

prepare_revoke_linked_signer(self, sender, signer, subaccount, subaccount_id, include_signature=False, **kwargs) async

Prepares the payload for revoking a linked signer, optionally signing it.

Parameters:

Name	Type	Description	Default
sender	str	Owner address initiating the revoke. Required.	required
signer	str	Signer address being revoked. Required.	required
subaccount	str	Hex-encoded subaccount name. Required.	required
subaccount_id	str	UUID of the subaccount. Required.	required
include_signature	bool	If True, sign with the owner's key. Optional.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for signing. Optional.
signed_at	int	Seconds since epoch for the signature timestamp. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
RevokeLinkedSignerDto	RevokeLinkedSignerDto	Prepared (and optionally signed) revoke payload.

Source code in meridian/rest/linked_signer.py

async def prepare_revoke_linked_signer(
    self,
    sender: str,
    signer: str,
    subaccount: str,
    subaccount_id: UUID,
    include_signature: bool = False,
    **kwargs,
) -> RevokeLinkedSignerDto:
    """Prepares the payload for revoking a linked signer, optionally signing it.

    Args:
        sender (str): Owner address initiating the revoke. Required.
        signer (str): Signer address being revoked. Required.
        subaccount (str): Hex-encoded subaccount name. Required.
        subaccount_id (str): UUID of the subaccount. Required.
        include_signature (bool): If True, sign with the owner's key. Optional.

    Other Parameters:
        nonce (str, optional): Custom nonce for signing. Optional.
        signed_at (int, optional): Seconds since epoch for the signature timestamp. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        RevokeLinkedSignerDto: Prepared (and optionally signed) revoke payload.
    """
    nonce = kwargs.get("nonce") or generate_nonce()
    signed_at = kwargs.get("signed_at") or int(time.time())
    data = {
        "sender": sender,
        "signer": signer,
        "subaccount": subaccount,
        "subaccountId": subaccount_id,
        "nonce": nonce,
        "signedAt": signed_at,
    }
    data_model = self._models.RevokeLinkedSignerDtoData.model_validate(data)
    dto = self._models.RevokeLinkedSignerDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = await self.sign_revoke_linked_signer(dto)
    return dto

refresh_linked_signer(self, dto, **kwargs) async

Submits a prepared and signed refresh-linked-signer payload.

Parameters:

Name	Type	Description	Default
dto	RefreshLinkedSignerDto	Prepared and signed refresh payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SignerDto	SignerDto	Signer record reflecting the refreshed expiry.

Source code in meridian/rest/linked_signer.py

async def refresh_linked_signer(
    self,
    dto: RefreshLinkedSignerDto,
    **kwargs,
) -> SignerDto:
    """Submits a prepared and signed refresh-linked-signer payload.

    Args:
        dto (RefreshLinkedSignerDto): Prepared and signed refresh payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SignerDto: Signer record reflecting the refreshed expiry.
    """
    endpoint = f"{API_PREFIX}/linked-signer/refresh"
    res = await self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return self._models.SignerDto.model_validate(res)

revoke_linked_signer(self, dto, **kwargs) async

Submits a prepared and signed revoke-linked-signer payload.

Parameters:

Name	Type	Description	Default
dto	RevokeLinkedSignerDto	Prepared and signed revoke payload. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SignerDto	SignerDto	Signer record reflecting revocation.

Source code in meridian/rest/linked_signer.py

async def revoke_linked_signer(
    self,
    dto: RevokeLinkedSignerDto,
    **kwargs,
) -> SignerDto:
    """Submits a prepared and signed revoke-linked-signer payload.

    Args:
        dto (RevokeLinkedSignerDto): Prepared and signed revoke payload. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SignerDto: Signer record reflecting revocation.
    """
    endpoint = f"{API_PREFIX}/linked-signer/revoke"
    res = await self.delete(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return self._models.SignerDto.model_validate(res)

sign_extend_linked_signer(self, extend_to_sign, private_key=None) async

Signs the extend-linked-signer payload with the linked signer's key.

Parameters:

Name	Type	Description	Default
extend_to_sign	ExtendLinkedSignerDto	Prepared extend payload. Required.	required
private_key	str	Private key override. Defaults to client's key.	None

Returns:

Name	Type	Description
ExtendLinkedSignerDto	ExtendLinkedSignerDto	DTO with signature populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/linked_signer.py

async def sign_extend_linked_signer(
    self,
    extend_to_sign: ExtendLinkedSignerDto,
    private_key: Optional[str] = None,
) -> ExtendLinkedSignerDto:
    """Signs the extend-linked-signer payload with the linked signer's key.

    Args:
        extend_to_sign (ExtendLinkedSignerDto): Prepared extend payload. Required.
        private_key (str, optional): Private key override. Defaults to client's key.

    Returns:
        ExtendLinkedSignerDto: DTO with signature populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key:
        if not self.chain.private_key:
            raise ValueError("No private key available for signing")
        private_key = self.chain.private_key

    # Prepare message for signing
    message = extend_to_sign.data.model_dump(mode="json", by_alias=True)
    message["signedAt"] = int(message["signedAt"])

    primary_type = "ExtendLinkedSigner"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)
    extend_to_sign.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return extend_to_sign

sign_linked_signer(self, link_to_sign, signer_private_key=None, private_key=None) async

Signs the link-signer payload with the signer and/or owner key.

Parameters:

Name	Type	Description	Default
link_to_sign	LinkSignerDto	Prepared link payload. Required.	required
signer_private_key	str	Signer's private key for cosigning. Optional.	None
private_key	str	Owner's private key override. Optional.	None

Returns:

Name	Type	Description
LinkSignerDto	LinkSignerDto	DTO with signature fields populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/linked_signer.py

async def sign_linked_signer(
    self,
    link_to_sign: LinkSignerDto,
    signer_private_key: Optional[str] = None,
    private_key: Optional[str] = None,
) -> LinkSignerDto:
    """Signs the link-signer payload with the signer and/or owner key.

    Args:
        link_to_sign (LinkSignerDto): Prepared link payload. Required.
        signer_private_key (str, optional): Signer's private key for cosigning. Optional.
        private_key (str, optional): Owner's private key override. Optional.

    Returns:
        LinkSignerDto: DTO with signature fields populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key and not self.chain.private_key and not signer_private_key:
        raise ValueError("No private key available for signing")
    elif not private_key:
        private_key = self.chain.private_key

    # Prepare message for signing
    message = link_to_sign.data.model_dump(mode="json", by_alias=True)
    message["signedAt"] = int(message["signedAt"])

    primary_type = "LinkSigner"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)

    if signer_private_key:
        signer_signature = self.chain.sign_message(
            signer_private_key, domain, types, primary_type, message
        )
        link_to_sign.signer_signature = signer_signature
    if private_key:
        signature = self.chain.sign_message(
            private_key, domain, types, primary_type, message
        )
        link_to_sign.signature = signature
    return link_to_sign

sign_refresh_linked_signer(self, refresh_to_sign, private_key=None) async

Signs the refresh-linked-signer payload with the owner's key.

Parameters:

Name	Type	Description	Default
refresh_to_sign	RefreshLinkedSignerDto	Prepared refresh payload. Required.	required
private_key	str	Private key override. Defaults to client's key.	None

Returns:

Name	Type	Description
RefreshLinkedSignerDto	RefreshLinkedSignerDto	DTO with signature populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/linked_signer.py

async def sign_refresh_linked_signer(
    self,
    refresh_to_sign: RefreshLinkedSignerDto,
    private_key: Optional[str] = None,
) -> RefreshLinkedSignerDto:
    """Signs the refresh-linked-signer payload with the owner's key.

    Args:
        refresh_to_sign (RefreshLinkedSignerDto): Prepared refresh payload. Required.
        private_key (str, optional): Private key override. Defaults to client's key.

    Returns:
        RefreshLinkedSignerDto: DTO with signature populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key:
        if not self.chain.private_key:
            raise ValueError("No private key available for signing")
        private_key = self.chain.private_key

    # Prepare message for signing
    message = refresh_to_sign.data.model_dump(mode="json", by_alias=True)
    message["signedAt"] = int(message["signedAt"])

    primary_type = "RefreshLinkedSigner"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)
    refresh_to_sign.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return refresh_to_sign

sign_revoke_linked_signer(self, revoke_to_sign, private_key=None) async

Signs the revoke-linked-signer payload with the owner's key.

Parameters:

Name	Type	Description	Default
revoke_to_sign	RevokeLinkedSignerDto	Prepared revoke payload. Required.	required
private_key	str	Private key override. Defaults to client's key.	None

Returns:

Name	Type	Description
RevokeLinkedSignerDto	RevokeLinkedSignerDto	DTO with signature populated.

Raises:

Type	Description
ValueError	If no chain client or private key is available.

Source code in meridian/rest/linked_signer.py

async def sign_revoke_linked_signer(
    self,
    revoke_to_sign: RevokeLinkedSignerDto,
    private_key: Optional[str] = None,
) -> RevokeLinkedSignerDto:
    """Signs the revoke-linked-signer payload with the owner's key.

    Args:
        revoke_to_sign (RevokeLinkedSignerDto): Prepared revoke payload. Required.
        private_key (str, optional): Private key override. Defaults to client's key.

    Returns:
        RevokeLinkedSignerDto: DTO with signature populated.

    Raises:
        ValueError: If no chain client or private key is available.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key:
        if not self.chain.private_key:
            raise ValueError("No private key available for signing")
        private_key = self.chain.private_key

    # Prepare message for signing
    message = revoke_to_sign.data.model_dump(mode="json", by_alias=True)
    message["signedAt"] = int(message["signedAt"])

    primary_type = "RevokeLinkedSigner"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)
    revoke_to_sign.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return revoke_to_sign

meridian.rest.token

get_token(self, id, **kwargs) async

Gets a specific token by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID for the token. Required.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
TokenDto	TokenDto	Token metadata.

Source code in meridian/rest/token.py

async def get_token(
    self,
    id: UUID,
    **kwargs,
) -> TokenDto:
    """Gets a specific token by ID.

    Args:
        id (str): UUID for the token. Required.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        TokenDto: Token metadata.
    """
    endpoint = f"{API_PREFIX}/token/{id}"
    res = await self.get(endpoint, **kwargs)
    return self._models.TokenDto(**res)

get_withdraw_digest(self, withdraw_dto) async

Computes the EIP-712 digest of a withdrawal payload.

This digest identifies the pending withdrawal on-chain and is the value expected by ChainClient.finalize_withdraw once the withdrawal is ready to be claimed.

Parameters:

Name	Type	Description	Default
withdraw_dto	InitiateWithdrawDto	A prepared withdrawal payload (from prepare_withdraw_token).	required

Returns:

Type	Description
str	The hexadecimal digest string prefixed with '0x'.

Raises:

Type	Description
ValueError	If no chain client is configured on the client.

Source code in meridian/rest/token.py

async def get_withdraw_digest(
    self,
    withdraw_dto: InitiateWithdrawDto,
) -> str:
    """Computes the EIP-712 digest of a withdrawal payload.

    This digest identifies the pending withdrawal on-chain and is the value
    expected by ChainClient.finalize_withdraw once the withdrawal is ready
    to be claimed.

    Args:
        withdraw_dto: A prepared withdrawal payload (from ``prepare_withdraw_token``).

    Returns:
        The hexadecimal digest string prefixed with '0x'.

    Raises:
        ValueError: If no chain client is configured on the client.
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for digest computation")

    message = _withdraw_signing_message(withdraw_dto)
    primary_type = "InitiateWithdraw"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)
    return self.chain.get_typed_data_digest(domain, types, primary_type, message)

list_token_transfers(self, **kwargs) async

Lists token transfers for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
statuses	List[str]	Filter by transfer status (e.g., 'COMPLETED'). Optional.
types	List[str]	Filter by transfer type (e.g., 'WITHDRAW'). Optional.
created_after	float	Filter transfers created after this timestamp. Optional.
created_before	float	Filter transfers created before this timestamp. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[TransferDto]	List[TransferDto]: Transfer records for the subaccount.

Source code in meridian/rest/token.py

async def list_token_transfers(
    self,
    **kwargs,
) -> List[TransferDto]:
    """Lists token transfers for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        statuses (List[str], optional): Filter by transfer status (e.g., 'COMPLETED'). Optional.
        types (List[str], optional): Filter by transfer type (e.g., 'WITHDRAW'). Optional.
        created_after (float, optional): Filter transfers created after this timestamp. Optional.
        created_before (float, optional): Filter transfers created before this timestamp. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[TransferDto]: Transfer records for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/token/transfer",
        request_model=self._models.V1TokenTransferGetParametersQuery,
        response_model=self._models.PageOfTransfersDtos,
        **kwargs,
    )
    data = [
        self._models.TransferDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_token_withdraws(self, **kwargs) async

Lists token withdrawals for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	UUID of the subaccount. Required.	required

Other Parameters:

Name	Type	Description
is_active	bool	Filter active withdrawals. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[WithdrawDto]	List[WithdrawDto]: Withdrawal records for the subaccount.

Source code in meridian/rest/token.py

async def list_token_withdraws(
    self,
    **kwargs,
) -> List[WithdrawDto]:
    """Lists token withdrawals for a subaccount.

    Args:
        subaccount_id (str): UUID of the subaccount. Required.

    Other Parameters:
        is_active (bool, optional): Filter active withdrawals. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[WithdrawDto]: Withdrawal records for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/token/withdraw",
        request_model=self._models.V1TokenWithdrawGetParametersQuery,
        response_model=self._models.PageOfWithdrawDtos,
        **kwargs,
    )
    data = [
        self._models.WithdrawDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

list_tokens(self, **kwargs) async

Lists all supported tokens.

Other Parameters:

Name	Type	Description
deposit_enabled	bool	Filter tokens with deposits enabled. Optional.
withdraw_enabled	bool	Filter tokens with withdrawals enabled. Optional.
order	str	Sort order, 'asc' or 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by (e.g., 'createdAt'). Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[TokenDto]	List[TokenDto]: Token metadata.

Source code in meridian/rest/token.py

async def list_tokens(
    self,
    **kwargs,
) -> List[TokenDto]:
    """Lists all supported tokens.

    Other Parameters:
        deposit_enabled (bool, optional): Filter tokens with deposits enabled. Optional.
        withdraw_enabled (bool, optional): Filter tokens with withdrawals enabled. Optional.
        order (str, optional): Sort order, 'asc' or 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by (e.g., 'createdAt'). Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[TokenDto]: Token metadata.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/token",
        request_model=self._models.V1TokenGetParametersQuery,
        response_model=self._models.PageOfTokensDtos,
        **kwargs,
    )
    data = [
        self._models.TokenDto(**model.model_dump(by_alias=True)) for model in res.data
    ]
    return data

prepare_convert_token(self, subaccount, from_token, to_token, amount, sender, include_signature=False, nonce=None, signed_at=None) async

Prepares a token conversion payload for signing and submission.

Constructs an InitiateConversionDto containing all the data required for a conversion request. The payload can optionally be signed immediately if include_signature=True.

For most use cases, prefer the higher-level convert_token() method on AsyncRESTClient which handles preparation, signing, and submission in a single call.

Parameters:

Name	Type	Description	Default
subaccount	str	Hex-encoded (bytes32) subaccount name to convert within.	required
from_token	str	Address of the token to convert from (zero address for virtual tokens).	required
to_token	str	Address of the token to convert to (zero address for virtual tokens).	required
amount	int	Amount to convert in the token's base units.	required
sender	str	Account or linked signer address placing the conversion request.	required
include_signature	bool	If True, signs the payload immediately using the client's configured private key. Defaults to False.	False
nonce	Optional[str]	Custom nonce for the EIP-712 signature. If not provided, a random nonce is generated.	None
signed_at	Optional[int]	Unix timestamp (seconds) for the signature. If not provided, defaults to the current time.	None

Returns:

Type	Description
InitiateConversionDto	A prepared conversion payload. If include_signature=True, the
InitiateConversionDto	signature field will be populated; otherwise it will be empty.

Source code in meridian/rest/token.py

async def prepare_convert_token(
    self,
    subaccount: str,
    from_token: str,
    to_token: str,
    amount: int,
    sender: str,
    include_signature: bool = False,
    nonce: Optional[str] = None,
    signed_at: Optional[int] = None,
) -> InitiateConversionDto:
    """Prepares a token conversion payload for signing and submission.

    Constructs an InitiateConversionDto containing all the data required for a
    conversion request. The payload can optionally be signed immediately if
    ``include_signature=True``.

    For most use cases, prefer the higher-level ``convert_token()`` method
    on AsyncRESTClient which handles preparation, signing, and submission
    in a single call.

    Args:
        subaccount: Hex-encoded (bytes32) subaccount name to convert within.
        from_token: Address of the token to convert from (zero address for
            virtual tokens).
        to_token: Address of the token to convert to (zero address for virtual
            tokens).
        amount: Amount to convert in the token's base units.
        sender: Account or linked signer address placing the conversion request.
        include_signature: If True, signs the payload immediately using the
            client's configured private key. Defaults to False.
        nonce: Custom nonce for the EIP-712 signature. If not provided, a
            random nonce is generated.
        signed_at: Unix timestamp (seconds) for the signature. If not provided,
            defaults to the current time.

    Returns:
        A prepared conversion payload. If ``include_signature=True``, the
        signature field will be populated; otherwise it will be empty.
    """
    nonce = nonce or generate_nonce()
    signed_at = signed_at or int(time.time())

    data = {
        "sender": sender,
        "subaccount": subaccount,
        "fromToken": from_token,
        "toToken": to_token,
        "amount": amount,
        "nonce": nonce,
        "signedAt": signed_at,
    }
    data_model = self._models.InitiateConversionDtoData.model_validate(data)
    dto = self._models.InitiateConversionDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = await self.sign_convert_token(dto)
    return dto

prepare_withdraw_token(self, subaccount, token, amount, account, destination_address, destination_endpoint, include_signature=False, nonce=None, signed_at=None) async

Prepares a token withdrawal payload for signing and submission.

Constructs an InitiateWithdrawDto containing all the data required for a withdrawal request. The payload can optionally be signed immediately if include_signature=True.

For most use cases, prefer the higher-level withdraw_token() method on AsyncRESTClient which handles preparation, signing, and submission in a single call.

Parameters:

Name	Type	Description	Default
subaccount	str	Hex-encoded (bytes32) subaccount name from which to withdraw.	required
token	str	Token contract address on the source chain.	required
amount	int	Amount to withdraw in the token's base units.	required
account	str	Wallet address that owns the subaccount.	required
destination_address	str	LayerZero destination address where funds will be sent. Can be a hex string or bytes; automatically left-padded to bytes32 format.	required
destination_endpoint	int	LayerZero endpoint ID for the destination chain. Use 0 for same-chain withdrawals.	required
include_signature	bool	If True, signs the payload immediately using the client's configured private key. Defaults to False.	False
nonce	Optional[str]	Custom nonce for the EIP-712 signature. If not provided, a random nonce is generated.	None
signed_at	Optional[int]	Unix timestamp (seconds) for the signature. If not provided, defaults to the current time.	None

Returns:

Type	Description
InitiateWithdrawDto	A prepared withdrawal payload. If include_signature=True, the
InitiateWithdrawDto	signature field will be populated; otherwise it will be empty.

Source code in meridian/rest/token.py

async def prepare_withdraw_token(
    self,
    subaccount: str,
    token: str,
    amount: int,
    account: str,
    destination_address: str,
    destination_endpoint: int,
    include_signature: bool = False,
    nonce: Optional[str] = None,
    signed_at: Optional[int] = None,
) -> InitiateWithdrawDto:
    """Prepares a token withdrawal payload for signing and submission.

    Constructs an InitiateWithdrawDto containing all the data required for a
    withdrawal request. The payload can optionally be signed immediately if
    ``include_signature=True``.

    For most use cases, prefer the higher-level ``withdraw_token()`` method
    on AsyncRESTClient which handles preparation, signing, and submission
    in a single call.

    Args:
        subaccount: Hex-encoded (bytes32) subaccount name from which to withdraw.
        token: Token contract address on the source chain.
        amount: Amount to withdraw in the token's base units.
        account: Wallet address that owns the subaccount.
        destination_address: LayerZero destination address where funds will be
            sent. Can be a hex string or bytes; automatically left-padded to
            bytes32 format.
        destination_endpoint: LayerZero endpoint ID for the destination chain.
            Use 0 for same-chain withdrawals.
        include_signature: If True, signs the payload immediately using the
            client's configured private key. Defaults to False.
        nonce: Custom nonce for the EIP-712 signature. If not provided, a
            random nonce is generated.
        signed_at: Unix timestamp (seconds) for the signature. If not provided,
            defaults to the current time.

    Returns:
        A prepared withdrawal payload. If ``include_signature=True``, the
        signature field will be populated; otherwise it will be empty.
    """
    nonce = nonce or generate_nonce()
    signed_at = signed_at or int(time.time())

    data = {
        "account": account,
        "subaccount": subaccount,
        "token": token,
        "amount": amount,
        "nonce": nonce,
        "signedAt": signed_at,
        "lzDestinationAddress": ensure_bytes32_hex(destination_address),
        "lzDestinationEid": destination_endpoint,
    }
    data_model = self._models.InitiateWithdrawDtoData.model_validate(data)
    dto = self._models.InitiateWithdrawDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = await self.sign_withdraw_token(dto)
    return dto

sign_convert_token(self, conversion_dto, private_key=None) async

Signs a token conversion payload using EIP-712 typed data signing.

Generates an EIP-712 signature for the conversion payload using either the provided private key or the client's configured chain key.

Parameters:

Name	Type	Description	Default
conversion_dto	InitiateConversionDto	A prepared conversion payload (from prepare_convert_token). The signature field will be populated in-place.	required
private_key	Optional[str]	Optional private key to use for signing. If not provided, uses the private key from the client's chain configuration.	None

Returns:

Type	Description
InitiateConversionDto	The same InitiateConversionDto with the signature field populated.

Raises:

Type	Description
ValueError	If no chain client is configured on the client.
ValueError	If no private key is available (neither provided nor configured).

Source code in meridian/rest/token.py

async def sign_convert_token(
    self,
    conversion_dto: InitiateConversionDto,
    private_key: Optional[str] = None,
) -> InitiateConversionDto:
    """Signs a token conversion payload using EIP-712 typed data signing.

    Generates an EIP-712 signature for the conversion payload using either
    the provided private key or the client's configured chain key.

    Args:
        conversion_dto: A prepared conversion payload (from ``prepare_convert_token``).
            The signature field will be populated in-place.
        private_key: Optional private key to use for signing. If not provided,
            uses the private key from the client's chain configuration.

    Returns:
        The same InitiateConversionDto with the signature field populated.

    Raises:
        ValueError: If no chain client is configured on the client.
        ValueError: If no private key is available (neither provided nor configured).
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key and not self.chain.private_key:
        raise ValueError("No private key available for signing")
    elif not private_key:
        private_key = self.chain.private_key

    message = _conversion_signing_message(conversion_dto)
    primary_type = "ConvertToken"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)
    conversion_dto.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return conversion_dto

sign_withdraw_token(self, withdraw_dto, private_key=None) async

Signs a token withdrawal payload using EIP-712 typed data signing.

Generates an EIP-712 signature for the withdrawal payload using either the provided private key or the client's configured chain key.

Parameters:

Name	Type	Description	Default
withdraw_dto	InitiateWithdrawDto	A prepared withdrawal payload (from prepare_withdraw_token). The signature field will be populated in-place.	required
private_key	Optional[str]	Optional private key to use for signing. If not provided, uses the private key from the client's chain configuration.	None

Returns:

Type	Description
InitiateWithdrawDto	The same InitiateWithdrawDto with the signature field populated.

Raises:

Type	Description
ValueError	If no chain client is configured on the client.
ValueError	If no private key is available (neither provided nor configured).

Source code in meridian/rest/token.py

async def sign_withdraw_token(
    self,
    withdraw_dto: InitiateWithdrawDto,
    private_key: Optional[str] = None,
) -> InitiateWithdrawDto:
    """Signs a token withdrawal payload using EIP-712 typed data signing.

    Generates an EIP-712 signature for the withdrawal payload using either
    the provided private key or the client's configured chain key.

    Args:
        withdraw_dto: A prepared withdrawal payload (from ``prepare_withdraw_token``).
            The signature field will be populated in-place.
        private_key: Optional private key to use for signing. If not provided,
            uses the private key from the client's chain configuration.

    Returns:
        The same InitiateWithdrawDto with the signature field populated.

    Raises:
        ValueError: If no chain client is configured on the client.
        ValueError: If no private key is available (neither provided nor configured).
    """
    if not hasattr(self, "chain") or not self.chain:
        raise ValueError("No chain client available for signing")
    if not private_key and not self.chain.private_key:
        raise ValueError("No private key available for signing")
    elif not private_key:
        private_key = self.chain.private_key

    message = _withdraw_signing_message(withdraw_dto)
    primary_type = "InitiateWithdraw"
    domain = self.rpc_config.domain.model_dump(mode="json", by_alias=True)
    types = self.chain.get_signature_types(self.rpc_config, primary_type)
    withdraw_dto.signature = self.chain.sign_message(
        private_key, domain, types, primary_type, message
    )
    return withdraw_dto

submit_convert_token(self, dto, **kwargs) async

Submits a prepared and signed token conversion request to the API.

This is the low-level submission function that sends a pre-constructed InitiateConversionDto to the conversion endpoint. For most use cases, prefer the higher-level convert_token() method on AsyncRESTClient which handles preparation, signing, and submission in a single call.

Parameters:

Name	Type	Description	Default
dto	InitiateConversionDto	A prepared and signed conversion payload containing the conversion details and EIP-712 signature.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters passed to the underlying HTTP client.

Returns:

Type	Description
ConversionDto	The conversion record created by the API, containing transaction
ConversionDto	details and status information.

Source code in meridian/rest/token.py

async def submit_convert_token(
    self,
    dto: InitiateConversionDto,
    **kwargs,
) -> ConversionDto:
    """Submits a prepared and signed token conversion request to the API.

    This is the low-level submission function that sends a pre-constructed
    InitiateConversionDto to the conversion endpoint. For most use cases,
    prefer the higher-level ``convert_token()`` method on AsyncRESTClient
    which handles preparation, signing, and submission in a single call.

    Args:
        dto: A prepared and signed conversion payload containing the conversion
            details and EIP-712 signature.

    Other Parameters:
        **kwargs: Additional request parameters passed to the underlying
            HTTP client.

    Returns:
        The conversion record created by the API, containing transaction
        details and status information.
    """
    endpoint = f"{API_PREFIX}/token/convert"
    res = await self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return self._models.ConversionDto.model_validate(res)

submit_withdraw_token(self, dto, token_id, **kwargs) async

Submits a prepared and signed token withdrawal request to the API.

This is the low-level submission function that sends a pre-constructed InitiateWithdrawDto to the withdrawal endpoint. For most use cases, prefer the higher-level withdraw_token() method on AsyncRESTClient which handles preparation, signing, and submission in a single call.

Parameters:

Name	Type	Description	Default
dto	InitiateWithdrawDto	A prepared and signed withdrawal payload containing the withdrawal details and EIP-712 signature.	required
token_id	UUID	UUID of the token being withdrawn. Used to construct the API endpoint path.	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters passed to the underlying HTTP client.

Returns:

Type	Description
WithdrawDto	The withdrawal record created by the API, containing transaction
WithdrawDto	details and status information.

Source code in meridian/rest/token.py

async def submit_withdraw_token(
    self,
    dto: InitiateWithdrawDto,
    token_id: UUID,
    **kwargs,
) -> WithdrawDto:
    """Submits a prepared and signed token withdrawal request to the API.

    This is the low-level submission function that sends a pre-constructed
    InitiateWithdrawDto to the withdrawal endpoint. For most use cases,
    prefer the higher-level ``withdraw_token()`` method on AsyncRESTClient
    which handles preparation, signing, and submission in a single call.

    Args:
        dto: A prepared and signed withdrawal payload containing the withdrawal
            details and EIP-712 signature.
        token_id: UUID of the token being withdrawn. Used to construct the
            API endpoint path.

    Other Parameters:
        **kwargs: Additional request parameters passed to the underlying
            HTTP client.

    Returns:
        The withdrawal record created by the API, containing transaction
        details and status information.
    """
    endpoint = f"{API_PREFIX}/token/{token_id}/withdraw"
    res = await self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return self._models.WithdrawDto.model_validate(res)

meridian.rest.referral

get_referral_summary(self, **kwargs) async

Gets the referral summary for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount	str	Bytes32 encoded subaccount name (0x prefix, zero padded).	required

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
ReferralDto	ReferralDto	Referral code and usage information.

Source code in meridian/rest/referral.py

async def get_referral_summary(self, **kwargs) -> "ReferralDto":
    """Gets the referral summary for a subaccount.

    Args:
        subaccount (str, optional): Bytes32 encoded subaccount name (0x prefix, zero padded).

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        ReferralDto: Referral code and usage information.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/referral/summary",
        request_model=self._models.V1ReferralSummaryGetParametersQuery,
        response_model=self._models.ReferralDto,
        **kwargs,
    )
    return res

list_referrals(self, **kwargs) async

Lists referrals for a subaccount.

Parameters:

Name	Type	Description	Default
subaccount	str	Bytes32 encoded subaccount name (0x prefix, zero padded).	required

Other Parameters:

Name	Type	Description
order	str	Sort order, 'desc'. Optional.
limit	float	Maximum number of results to return. Optional.
cursor	str	Pagination cursor for fetching the next page. Optional.
order_by	str	Field to order by, e.g., 'createdAt'. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[ReferralDto]	List[ReferralDto]: Referral objects for the subaccount.

Source code in meridian/rest/referral.py

async def list_referrals(self, **kwargs) -> List["ReferralDto"]:
    """Lists referrals for a subaccount.

    Args:
        subaccount (str, optional): Bytes32 encoded subaccount name (0x prefix, zero padded).

    Other Parameters:
        order (str, optional): Sort order, 'desc'. Optional.
        limit (float, optional): Maximum number of results to return. Optional.
        cursor (str, optional): Pagination cursor for fetching the next page. Optional.
        order_by (str, optional): Field to order by, e.g., 'createdAt'. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        List[ReferralDto]: Referral objects for the subaccount.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/referral",
        request_model=self._models.V1ReferralGetParametersQuery,
        response_model=self._models.PageOfReferralDtos,
        **kwargs,
    )
    data = [
        self._models.ReferralDto(**model.model_dump(by_alias=True))
        for model in res.data
    ]
    return data

meridian.rest.system

get_maintenance_status(self, **kwargs) async

Gets the maintenance mode status.

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
MaintenanceDto	MaintenanceDto	Whether maintenance mode is enabled.

Source code in meridian/rest/system.py

async def get_maintenance_status(self, **kwargs) -> MaintenanceDto:
    """Gets the maintenance mode status.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        MaintenanceDto: Whether maintenance mode is enabled.
    """
    endpoint = f"{API_PREFIX}/maintenance"
    res = await self.get(endpoint, **kwargs)
    return self._models.MaintenanceDto(**res)

get_rate_limit_config(self, **kwargs) async

Gets rate limit configurations for API endpoints.

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
RateLimitConfigResponseDto	RateLimitConfigResponseDto	Rate limit configurations per endpoint.

Source code in meridian/rest/system.py

async def get_rate_limit_config(self, **kwargs) -> RateLimitConfigResponseDto:
    """Gets rate limit configurations for API endpoints.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        RateLimitConfigResponseDto: Rate limit configurations per endpoint.
    """
    endpoint = f"{API_PREFIX}/rate-limit/config"
    res = await self.get(endpoint, **kwargs)
    return self._models.RateLimitConfigResponseDto(**res)

get_system_time(self, **kwargs) async

Gets the current system time.

Other Parameters:

Name	Type	Description
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
SystemTimeDto	SystemTimeDto	Current system timestamp in milliseconds since Unix epoch.

Source code in meridian/rest/system.py

async def get_system_time(self, **kwargs) -> SystemTimeDto:
    """Gets the current system time.

    Other Parameters:
        **kwargs: Additional request parameters accepted by the API. Optional.

    Returns:
        SystemTimeDto: Current system timestamp in milliseconds since Unix epoch.
    """
    endpoint = f"{API_PREFIX}/time"
    res = await self.get(endpoint, **kwargs)
    return self._models.SystemTimeDto(**res)

meridian.rest.tv

get_last_price_config(self, **kwargs) async

Gets the TradingView UDF configuration for last-price data.

Returns:

Name	Type	Description
ConfigDto	ConfigDto	UDF configuration including supported resolutions and symbol types.

Source code in meridian/rest/tv.py

async def get_last_price_config(self, **kwargs) -> "ConfigDto":
    """Gets the TradingView UDF configuration for last-price data.

    Returns:
        ConfigDto: UDF configuration including supported resolutions and symbol types.
    """
    res = await self.get(
        f"{API_PREFIX}/last-price/config",
        base_url_override=self._tv_base_url,
        **kwargs,
    )
    return self._tv_models.ConfigDto(**res)

get_last_price_history(self, **kwargs) async

Gets OHLCV bar history for a symbol using last-price data.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.	required
resolution	str	Bar resolution (e.g. '1', '5', '15', '60', '1D'). Required.	required
from_	float	Start of time range in seconds since Unix epoch. Required.	required
to	float	End of time range in seconds since Unix epoch. Required.	required

Other Parameters:

Name	Type	Description
countback	int	Maximum number of bars to return (1-4000). Optional.

Returns:

Name	Type	Description
HistoryResponseDto	HistoryResponseDto	OHLCV bar data with status.

Source code in meridian/rest/tv.py

async def get_last_price_history(self, **kwargs) -> "HistoryResponseDto":
    """Gets OHLCV bar history for a symbol using last-price data.

    Args:
        symbol (str): Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.
        resolution (str): Bar resolution (e.g. '1', '5', '15', '60', '1D'). Required.
        from_ (float): Start of time range in seconds since Unix epoch. Required.
        to (float): End of time range in seconds since Unix epoch. Required.

    Other Parameters:
        countback (int, optional): Maximum number of bars to return (1-4000). Optional.

    Returns:
        HistoryResponseDto: OHLCV bar data with status.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/last-price/history",
        request_model=self._tv_models.V1LastPriceHistoryGetParametersQuery,
        response_model=self._tv_models.HistoryResponseDto,
        base_url_override=self._tv_base_url,
        **kwargs,
    )
    return res

get_last_price_symbol(self, **kwargs) async

Resolves symbol information by ticker for last-price data.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.	required

Returns:

Name	Type	Description
SymbolInfoDto	SymbolInfoDto	Detailed symbol information.

Source code in meridian/rest/tv.py

async def get_last_price_symbol(self, **kwargs) -> "SymbolInfoDto":
    """Resolves symbol information by ticker for last-price data.

    Args:
        symbol (str): Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.

    Returns:
        SymbolInfoDto: Detailed symbol information.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/last-price/symbols",
        request_model=self._tv_models.V1LastPriceSymbolsGetParametersQuery,
        response_model=self._tv_models.SymbolInfoDto,
        base_url_override=self._tv_base_url,
        **kwargs,
    )
    return res

get_last_price_time(self, **kwargs) async

Gets the current server time from the last-price data feed.

Returns:

Name	Type	Description
dict	Dict[str, Any]	Server time response.

Source code in meridian/rest/tv.py

async def get_last_price_time(self, **kwargs) -> Dict[str, Any]:
    """Gets the current server time from the last-price data feed.

    Returns:
        dict: Server time response.
    """
    return await self.get(
        f"{API_PREFIX}/last-price/time",
        base_url_override=self._tv_base_url,
        **kwargs,
    )

get_oracle_price_config(self, **kwargs) async

Gets the TradingView UDF configuration for oracle-price data.

Returns:

Name	Type	Description
ConfigDto	ConfigDto	UDF configuration including supported resolutions and symbol types.

Source code in meridian/rest/tv.py

async def get_oracle_price_config(self, **kwargs) -> "ConfigDto":
    """Gets the TradingView UDF configuration for oracle-price data.

    Returns:
        ConfigDto: UDF configuration including supported resolutions and symbol types.
    """
    res = await self.get(
        f"{API_PREFIX}/oracle-price/config",
        base_url_override=self._tv_base_url,
        **kwargs,
    )
    return self._tv_models.ConfigDto(**res)

get_oracle_price_history(self, **kwargs) async

Gets OHLCV bar history for a symbol using oracle-price data.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.	required
resolution	str	Bar resolution (e.g. '1', '5', '15', '60', '1D'). Required.	required
from_	float	Start of time range in seconds since Unix epoch. Required.	required
to	float	End of time range in seconds since Unix epoch. Required.	required

Other Parameters:

Name	Type	Description
countback	int	Maximum number of bars to return (1-4000). Optional.

Returns:

Name	Type	Description
HistoryResponseDto	HistoryResponseDto	OHLCV bar data with status.

Source code in meridian/rest/tv.py

async def get_oracle_price_history(self, **kwargs) -> "HistoryResponseDto":
    """Gets OHLCV bar history for a symbol using oracle-price data.

    Args:
        symbol (str): Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.
        resolution (str): Bar resolution (e.g. '1', '5', '15', '60', '1D'). Required.
        from_ (float): Start of time range in seconds since Unix epoch. Required.
        to (float): End of time range in seconds since Unix epoch. Required.

    Other Parameters:
        countback (int, optional): Maximum number of bars to return (1-4000). Optional.

    Returns:
        HistoryResponseDto: OHLCV bar data with status.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/oracle-price/history",
        request_model=self._tv_models.V1OraclePriceHistoryGetParametersQuery,
        response_model=self._tv_models.HistoryResponseDto,
        base_url_override=self._tv_base_url,
        **kwargs,
    )
    return res

get_oracle_price_symbol(self, **kwargs) async

Resolves symbol information by ticker for oracle-price data.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.	required

Returns:

Name	Type	Description
SymbolInfoDto	SymbolInfoDto	Detailed symbol information.

Source code in meridian/rest/tv.py

async def get_oracle_price_symbol(self, **kwargs) -> "SymbolInfoDto":
    """Resolves symbol information by ticker for oracle-price data.

    Args:
        symbol (str): Symbol ticker in the format {ticker}-{type} (e.g. 'ETHUSD-Perp'). Required.

    Returns:
        SymbolInfoDto: Detailed symbol information.
    """
    res = await self.get_validated(
        url_path=f"{API_PREFIX}/oracle-price/symbols",
        request_model=self._tv_models.V1OraclePriceSymbolsGetParametersQuery,
        response_model=self._tv_models.SymbolInfoDto,
        base_url_override=self._tv_base_url,
        **kwargs,
    )
    return res

get_oracle_price_time(self, **kwargs) async

Gets the current server time from the oracle-price data feed.

Returns:

Name	Type	Description
dict	Dict[str, Any]	Server time response.

Source code in meridian/rest/tv.py

async def get_oracle_price_time(self, **kwargs) -> Dict[str, Any]:
    """Gets the current server time from the oracle-price data feed.

    Returns:
        dict: Server time response.
    """
    return await self.get(
        f"{API_PREFIX}/oracle-price/time",
        base_url_override=self._tv_base_url,
        **kwargs,
    )

search_last_price_symbols(self, **kwargs) async

Searches for symbols in the last-price data feed.

Parameters:

Name	Type	Description	Default
query	str	Search query string to match against symbol tickers. Required.	required

Other Parameters:

Name	Type	Description
type	str	Filter results by symbol type ('Perp' or 'Spot'). Defaults to 'Perp'.
limit	int	Maximum number of results (1-50). Optional.

Returns:

Type	Description
List[SymbolSearchResultDto]	List[SymbolSearchResultDto]: Matching symbol search results.

Source code in meridian/rest/tv.py

async def search_last_price_symbols(self, **kwargs) -> List["SymbolSearchResultDto"]:
    """Searches for symbols in the last-price data feed.

    Args:
        query (str): Search query string to match against symbol tickers. Required.

    Other Parameters:
        type (str, optional): Filter results by symbol type ('Perp' or 'Spot'). Defaults to 'Perp'.
        limit (int, optional): Maximum number of results (1-50). Optional.

    Returns:
        List[SymbolSearchResultDto]: Matching symbol search results.
    """
    validated_params = (
        self._tv_models.V1LastPriceSearchGetParametersQuery.model_validate(
            kwargs, by_name=True
        )
    )
    params = validated_params.model_dump(mode="json", exclude_none=True, by_alias=True)
    if "limit" in params:
        params["limit"] = int(params["limit"])
    res = await self.get(
        f"{API_PREFIX}/last-price/search",
        params=params,
        base_url_override=self._tv_base_url,
    )
    return [self._tv_models.SymbolSearchResultDto(**item) for item in res]

search_oracle_price_symbols(self, **kwargs) async

Searches for symbols in the oracle-price data feed.

Parameters:

Name	Type	Description	Default
query	str	Search query string to match against symbol tickers. Required.	required

Other Parameters:

Name	Type	Description
type	str	Filter results by symbol type ('Perp' or 'Spot'). Defaults to 'Perp'.
limit	int	Maximum number of results (1-50). Optional.

Returns:

Type	Description
List[SymbolSearchResultDto]	List[SymbolSearchResultDto]: Matching symbol search results.

Source code in meridian/rest/tv.py

async def search_oracle_price_symbols(self, **kwargs) -> List["SymbolSearchResultDto"]:
    """Searches for symbols in the oracle-price data feed.

    Args:
        query (str): Search query string to match against symbol tickers. Required.

    Other Parameters:
        type (str, optional): Filter results by symbol type ('Perp' or 'Spot'). Defaults to 'Perp'.
        limit (int, optional): Maximum number of results (1-50). Optional.

    Returns:
        List[SymbolSearchResultDto]: Matching symbol search results.
    """
    validated_params = (
        self._tv_models.V1OraclePriceSearchGetParametersQuery.model_validate(
            kwargs, by_name=True
        )
    )
    params = validated_params.model_dump(mode="json", exclude_none=True, by_alias=True)
    if "limit" in params:
        params["limit"] = int(params["limit"])
    res = await self.get(
        f"{API_PREFIX}/oracle-price/search",
        params=params,
        base_url_override=self._tv_base_url,
    )
    return [self._tv_models.SymbolSearchResultDto(**item) for item in res]

meridian.rest.util

client_order_id_to_bytes32(client_order_id)

Converts client_order_id to appropriate bytes32 format.

Parameters:

Name	Type	Description	Default
client_order_id	str	Client order ID to convert.	required

Returns:

Name	Type	Description
str	str	Converted client order ID in bytes32 hex format.

Raises:

Type	Description
ValueError	If string is longer than 32 characters and not a UUID, or if input is None/empty.

Source code in meridian/rest/util.py

def client_order_id_to_bytes32(client_order_id: str) -> str:
    """Converts client_order_id to appropriate bytes32 format.

    Args:
        client_order_id (str): Client order ID to convert.

    Returns:
        str: Converted client order ID in bytes32 hex format.

    Raises:
        ValueError: If string is longer than 32 characters and not a UUID, or if input is None/empty.
    """
    if client_order_id is None:
        raise ValueError("Client order ID cannot be None")

    if not client_order_id:
        raise ValueError("Client order ID cannot be empty")

    if is_uuid(client_order_id):
        return uuid_to_bytes32(client_order_id)

    if len(client_order_id) > 32:
        raise ValueError(
            f"Client order ID cannot be longer than 32 characters, got {len(client_order_id)}"
        )

    # Convert string to bytes32 hex format
    client_order_bytes = client_order_id.encode("utf-8")
    padded_bytes = client_order_bytes.ljust(32, b"\0")
    return "0x" + padded_bytes.hex()

decode_account_name(hex_name)

Converts hex-encoded subaccount name back to text.

Parameters:

Name	Type	Description	Default
hex_name	str	Hex-encoded name with '0x' prefix.	required

Returns:

Name	Type	Description
str	str	Decoded text string with null bytes stripped.

Source code in meridian/rest/util.py

def decode_account_name(hex_name: str) -> str:
    """Converts hex-encoded subaccount name back to text.

    Args:
        hex_name (str): Hex-encoded name with '0x' prefix.

    Returns:
        str: Decoded text string with null bytes stripped.
    """
    try:
        if hex_name.startswith("0x"):
            return bytes.fromhex(hex_name[2:]).decode("utf-8").rstrip("\x00")
        else:
            return bytes.fromhex(hex_name).decode("utf-8").rstrip("\x00")
    except (ValueError, UnicodeDecodeError):
        # Return original if decoding fails
        return hex_name

encode_account_name(text)

Converts text to hex-encoded subaccount name format.

Parameters:

Name	Type	Description	Default
text	str	Text to convert to hex name.	required

Returns:

Name	Type	Description
str	str	Hex-encoded name with '0x' prefix, padded to 32 bytes.

Source code in meridian/rest/util.py

def encode_account_name(text: str) -> str:
    """Converts text to hex-encoded subaccount name format.

    Args:
        text (str): Text to convert to hex name.

    Returns:
        str: Hex-encoded name with '0x' prefix, padded to 32 bytes.
    """
    hex_encoded = text.encode("utf-8").hex()
    # Pad to 64 characters (32 bytes) with zeros on the right
    padded_hex = hex_encoded.ljust(64, "0")
    return "0x" + padded_hex

ensure_bytes32(value)

Return a 32-byte representation of value (zero-padded on the left).

Source code in meridian/rest/util.py

def ensure_bytes32(value):
    """Return a 32-byte representation of ``value`` (zero-padded on the left)."""
    if isinstance(value, (bytes, bytearray)):
        data = bytes(value)
    elif isinstance(value, str):
        raw = value[2:] if value.startswith("0x") else value
        raw = raw if len(raw) % 2 == 0 else "0" + raw
        try:
            data = decode_hex(add_0x_prefix(raw))
        except ValueError as exc:
            raise Bytes32Error(f"Invalid hex string: {value!r}") from exc
    else:
        raise Bytes32Error(f"Unsupported type: {type(value).__name__}")
    if len(data) > 32:
        raise Bytes32Error(f"Expected at most 32 bytes, got {len(data)} bytes")
    return data.rjust(32, b"\x00")

ensure_bytes32_hex(value)

Return value as a 0x-prefixed 64-hex-digit string.

Source code in meridian/rest/util.py

def ensure_bytes32_hex(value):
    """Return ``value`` as a 0x-prefixed 64-hex-digit string."""
    return encode_hex(ensure_bytes32(value))

generate_nonce()

Generates a timestamp-based nonce.

Returns:

Name	Type	Description
str	str	Current timestamp in nanoseconds as string.

Source code in meridian/rest/util.py

def generate_nonce() -> str:
    """Generates a timestamp-based nonce.

    Returns:
        str: Current timestamp in nanoseconds as string.
    """
    return str(time.time_ns())

is_uuid(value)

Checks if a string is a valid UUID.

Parameters:

Name	Type	Description	Default
value	str	String to check.	required

Returns:

Name	Type	Description
bool	bool	True if string is a valid UUID, False otherwise.

Source code in meridian/rest/util.py

def is_uuid(value: str) -> bool:
    """Checks if a string is a valid UUID.

    Args:
        value (str): String to check.

    Returns:
        bool: True if string is a valid UUID, False otherwise.
    """
    try:
        return value == str(uuid.UUID(value))
    except ValueError:
        return False

uuid_to_bytes32(uuid_str)

Converts UUID string to bytes32 hex format.

Parameters:

Name	Type	Description	Default
uuid_str	str	UUID string to convert.	required

Returns:

Name	Type	Description
str	str	Bytes32 hex string prefixed with '0x'.

Source code in meridian/rest/util.py

def uuid_to_bytes32(uuid_str: str) -> str:
    """Converts UUID string to bytes32 hex format.

    Args:
        uuid_str (str): UUID string to convert.

    Returns:
        str: Bytes32 hex string prefixed with '0x'.
    """
    uuid_obj = uuid.UUID(uuid_str)

    # remove hyphens and convert to hex
    uuid_hex = uuid_obj.hex

    # pad the hex to make it 32 bytes
    padded_hex = uuid_hex.rjust(64, "0")

    return "0x" + padded_hex

meridian.async_ws_client.AsyncWSClient

Bases: AsyncWSBase

Meridian async WebSocket client.

Parameters:

Name	Type	Description	Default
config	Union[Dict[str, Any], WSConfig]	Configuration dictionary or WSConfig object. - base_url (str): Base URL for websocket requests - verbose (bool): Enables debug logging (default: False) - transport (str): Deprecated; only 'websocket' is accepted	required

Source code in meridian/async_ws_client.py

class AsyncWSClient(AsyncWSBase):
    """Meridian async WebSocket client.

    Args:
        config: Configuration dictionary or WSConfig object.
            - base_url (str): Base URL for websocket requests
            - verbose (bool): Enables debug logging (default: False)
            - transport (str): Deprecated; only 'websocket' is accepted
    """

    config: WSConfig

    def __init__(self, config: Union[Dict[str, Any], WSConfig]):
        ws_config = WSConfig.model_validate(config)
        super().__init__(ws_config)
        self.config = ws_config

    async def subscribe(
        self,
        stream_type: str,
        **params,
    ) -> Dict[str, Any]:
        """Subscribe to a stream.

        Args:
            stream_type: Type of stream (e.g., "L2Book", "Ticker", "OrderFill")
            **params: Subscription parameters (snake_case converted to camelCase)
                - symbol: Product symbol (e.g., "BTCUSD")
                - subaccount_id: Subaccount ID for account-specific streams
                - product_id: Product ID
        """
        data = {"type": stream_type, **prepare_params(**params)}
        return await self._emit("subscribe", data)

    async def unsubscribe(
        self,
        stream_type: str,
        **params,
    ) -> Dict[str, Any]:
        """Unsubscribe from a stream.

        Args:
            stream_type: Type of stream
            **params: Unsubscription parameters (snake_case converted to camelCase)
        """
        data = {"type": stream_type, **prepare_params(**params)}
        return await self._emit("unsubscribe", data)

subscribe(stream_type, **params) async

Subscribe to a stream.

Parameters:

Name	Type	Description	Default
stream_type	str	Type of stream (e.g., "L2Book", "Ticker", "OrderFill")	required
**params		Subscription parameters (snake_case converted to camelCase) - symbol: Product symbol (e.g., "BTCUSD") - subaccount_id: Subaccount ID for account-specific streams - product_id: Product ID	{}

Source code in meridian/async_ws_client.py

async def subscribe(
    self,
    stream_type: str,
    **params,
) -> Dict[str, Any]:
    """Subscribe to a stream.

    Args:
        stream_type: Type of stream (e.g., "L2Book", "Ticker", "OrderFill")
        **params: Subscription parameters (snake_case converted to camelCase)
            - symbol: Product symbol (e.g., "BTCUSD")
            - subaccount_id: Subaccount ID for account-specific streams
            - product_id: Product ID
    """
    data = {"type": stream_type, **prepare_params(**params)}
    return await self._emit("subscribe", data)

unsubscribe(stream_type, **params) async

Unsubscribe from a stream.

Parameters:

Name	Type	Description	Default
stream_type	str	Type of stream	required
**params		Unsubscription parameters (snake_case converted to camelCase)	{}

Source code in meridian/async_ws_client.py

async def unsubscribe(
    self,
    stream_type: str,
    **params,
) -> Dict[str, Any]:
    """Unsubscribe from a stream.

    Args:
        stream_type: Type of stream
        **params: Unsubscription parameters (snake_case converted to camelCase)
    """
    data = {"type": stream_type, **prepare_params(**params)}
    return await self._emit("unsubscribe", data)

meridian.ws_client.WSClient

Bases: WSBase

Meridian WebSocket client.

Parameters:

Name	Type	Description	Default
config	Union[Dict[str, Any], WSConfig]	Configuration dictionary or WSConfig object. - base_url (str): Base URL for websocket requests - verbose (bool): Enables debug logging (default: False) - transport (str): Deprecated; only 'websocket' is accepted	required

Source code in meridian/ws_client.py

class WSClient(WSBase):
    """Meridian WebSocket client.

    Args:
        config: Configuration dictionary or WSConfig object.
            - base_url (str): Base URL for websocket requests
            - verbose (bool): Enables debug logging (default: False)
            - transport (str): Deprecated; only 'websocket' is accepted
    """

    config: WSConfig

    def __init__(self, config: Union[Dict[str, Any], WSConfig]):
        ws_config = WSConfig.model_validate(config)
        super().__init__(ws_config)
        self.config = ws_config

    def subscribe(
        self,
        stream_type: str,
        **params,
    ) -> Dict[str, Any]:
        """Subscribe to a stream.

        Args:
            stream_type: Type of stream (e.g., "L2Book", "Ticker", "OrderFill")
            **params: Subscription parameters (snake_case converted to camelCase)
                - symbol: Product symbol (e.g., "BTCUSD")
                - subaccount_id: Subaccount ID for account-specific streams
                - product_id: Product ID
        """
        data = {"type": stream_type, **prepare_params(**params)}
        return self._emit("subscribe", data)

    def unsubscribe(
        self,
        stream_type: str,
        **params,
    ) -> Dict[str, Any]:
        """Unsubscribe from a stream.

        Args:
            stream_type: Type of stream
            **params: Unsubscription parameters (snake_case converted to camelCase)
        """
        data = {"type": stream_type, **prepare_params(**params)}
        return self._emit("unsubscribe", data)

subscribe(stream_type, **params)

Subscribe to a stream.

Parameters:

Name	Type	Description	Default
stream_type	str	Type of stream (e.g., "L2Book", "Ticker", "OrderFill")	required
**params		Subscription parameters (snake_case converted to camelCase) - symbol: Product symbol (e.g., "BTCUSD") - subaccount_id: Subaccount ID for account-specific streams - product_id: Product ID	{}

Source code in meridian/ws_client.py

def subscribe(
    self,
    stream_type: str,
    **params,
) -> Dict[str, Any]:
    """Subscribe to a stream.

    Args:
        stream_type: Type of stream (e.g., "L2Book", "Ticker", "OrderFill")
        **params: Subscription parameters (snake_case converted to camelCase)
            - symbol: Product symbol (e.g., "BTCUSD")
            - subaccount_id: Subaccount ID for account-specific streams
            - product_id: Product ID
    """
    data = {"type": stream_type, **prepare_params(**params)}
    return self._emit("subscribe", data)

unsubscribe(stream_type, **params)

Unsubscribe from a stream.

Parameters:

Name	Type	Description	Default
stream_type	str	Type of stream	required
**params		Unsubscription parameters (snake_case converted to camelCase)	{}

Source code in meridian/ws_client.py

def unsubscribe(
    self,
    stream_type: str,
    **params,
) -> Dict[str, Any]:
    """Unsubscribe from a stream.

    Args:
        stream_type: Type of stream
        **params: Unsubscription parameters (snake_case converted to camelCase)
    """
    data = {"type": stream_type, **prepare_params(**params)}
    return self._emit("unsubscribe", data)

meridian.chain_client.ChainClient

Bases: BaseClient

Client for interacting with the blockchain using Web3 functionality.

Parameters:

Name	Type	Description	Default
config	Union[Dict[str, Any], ChainConfig]	Chain configuration	required
rpc_config	RpcConfigDto	RPC configuration. Defaults to None.	None

Raises:

Type	Description
Exception	If RPC URL or private key is not specified in the configuration

Source code in meridian/chain_client.py

class ChainClient(BaseClient):
    """Client for interacting with the blockchain using Web3 functionality.

    Args:
        config (Union[Dict[str, Any], ChainConfig]): Chain configuration
        rpc_config (RpcConfigDto, optional): RPC configuration. Defaults to None.

    Raises:
        Exception: If RPC URL or private key is not specified in the configuration
    """

    def __init__(
        self,
        config: Union[Dict[str, Any], ChainConfig],
        rpc_config: Optional[RpcConfigDto] = None,
        tokens: Optional[List[TokenDto]] = None,
    ):
        super().__init__(config)
        self.config = ChainConfig.model_validate(config)
        self.web3 = self._setup_web3()
        self.account = self._setup_account()
        if self.account:
            self.address = self.account.address
            self.private_key = self.config.private_key
        else:
            self.address = self.config.address
            self.private_key = None

        self.chain_id = self.web3.eth.chain_id

        self.usd = None
        if tokens is not None:
            usd_token = next((t for t in tokens if t.name == "USD"), None)
            if usd_token is None:
                self.logger.warning("USD token not found in the provided tokens list.")
            else:
                usd_address = self.web3.to_checksum_address(usd_token.address)
                self.usd = self.web3.eth.contract(
                    address=usd_address,
                    abi=read_contract(self.chain_id, "MeridianUSD", common=True),
                )
        self.rpc_config = rpc_config

    @property
    def exchange_contract(self):
        return self.web3.eth.contract(
            address=self.rpc_config.domain.verifying_contract,
            abi=read_contract(self.chain_id, "ExchangeGateway"),
        )

    @property
    def usd_deposit_token(self) -> str:
        """Address of the token deposited as USD collateral.

        mUSD is a wrapper token: ``depositUsd`` pulls mUSD's underlying wrapped
        token from the sender via ``transferFrom`` and wraps it. This is the
        token to approve and fund when depositing USD, not mUSD itself.

        Returns:
            str: The checksummed address of mUSD's default wrapped token.

        Raises:
            ValueError: If no USD (mUSD) token is configured on the client.
        """
        if self.usd is None:
            raise ValueError(
                "No USD token configured; construct the client with tokens "
                "(e.g. via RESTClient) to resolve the USD deposit token"
            )
        return self.web3.to_checksum_address(
            self.usd.functions.defaultWrappedToken().call()
        )

    def _setup_web3(self):
        """Set up the Web3 instance.

        Returns:
            Web3: The Web3 instance

        Raises:
            Exception: If RPC URL is not specified in the configuration
        """
        # TODO: Support other provider types (e.g. WebSocket)
        if self.config.rpc_url is None:
            raise Exception("RPC URL must be specified in the configuration")
        return Web3(Web3.HTTPProvider(self.config.rpc_url))

    def _setup_account(self):
        """Set up the account.

        Returns:
            Account: The Web3 account instance

        Raises:
            Exception: If private key is not specified in the configuration
        """
        if self.config.private_key is None:
            self.logger.debug("Private key not specified in the configuration")
            return None

        account = self.web3.eth.account.from_key(self.config.private_key)
        if not account:
            raise Exception("Failed to create account from private key")
        if self.config.address and account.address != self.config.address:
            raise Exception(
                f"Private key does not match address specified in the config: {self.config.address}"
            )
        return account

    def _get_tx(self, value=0, to=None) -> TxParams:
        """Get default transaction parameters.

        Args:
            value (int, optional): The value to send. Defaults to 0.
            to (str, optional): The recipient address. Defaults to None.

        Returns:
            TxParams: The transaction parameters
        """
        params: TxParams = {
            "from": self.address,
            "chainId": self.chain_id,
            "value": value,
            "nonce": Nonce(self.get_nonce(self.address)),
        }
        if to is not None:
            params["to"] = to
        return params

    def _decode_error(self, error: ContractCustomError) -> str:
        abi_errors = {
            self.web3.to_hex(function_abi_to_4byte_selector(f)): f.get("name")
            for f in self.exchange_contract.abi
            if f.get("type") == "error"
        }
        data = error.data
        error_signature = data
        return abi_errors.get(error_signature, "Unknown error")

    def _get_by_alias(self, model: BaseModel, alias: str):
        """
        Get a field value by its alias from a Pydantic model.

        Args:
            model (BaseModel): The Pydantic model instance to extract the field from.
            alias (str): The alias of the field to retrieve.

        Returns:
            Any: The value of the field with the specified alias.

        Raises:
            KeyError: If the alias is not found in the model.
        """
        for field_name, field in model.__class__.model_fields.items():
            if field.alias == alias:
                return getattr(model, field_name)
        raise KeyError(f"Alias '{alias}' not found in model {model.__class__.__name__}")

    def get_signature_types(self, rpc_config: RpcConfigDto, primary_type: str):
        """Gets EIP-712 signature types.

        Args:
            rpc_config (RpcConfigDto): RPC configuration.
            primary_type (str): Primary type for the signature.

        Returns:
            dict: Dictionary containing signature type definitions.
        """
        return {
            "EIP712Domain": [
                {"name": "name", "type": "string"},
                {"name": "version", "type": "string"},
                {"name": "chainId", "type": "uint256"},
                {"name": "verifyingContract", "type": "address"},
            ],
            primary_type: self.convert_types(
                self._get_by_alias(rpc_config.signature_types, primary_type),
            ),
        }

    def convert_types(self, type_string: str) -> List[Dict[str, str]]:
        """Converts type string into EIP-712 field format.

        Args:
            type_string (str): String containing type definitions.

        Returns:
            List[Dict[str, str]]: List of field definitions.
        """
        fields = [comp.strip() for comp in type_string.split(",")]
        type_fields = []
        for field in fields:
            field_type, field_name = field.rsplit(" ", 1)
            type_fields.append({"name": field_name, "type": field_type})
        return type_fields

    def add_gas_fees(self, tx: TxParams) -> TxParams:
        """Add gas fee parameters to a transaction.

        Args:
            tx (TxParams): The transaction parameters

        Returns:
            TxParams: The transaction parameters with gas fee parameters added
        """
        if "maxFeePerGas" in tx and "maxPriorityFeePerGas" in tx:
            return tx
        try:
            gas_price = self.web3.eth.gas_price
            max_priority_fee = self.web3.eth.max_priority_fee
            tx["maxFeePerGas"] = gas_price
            tx["maxPriorityFeePerGas"] = max_priority_fee
            return tx
        except Web3Exception as e:
            self.logger.error(f"Failed to add gas: {e}")
            return tx

    def add_gas_limit(self, tx: TxParams) -> TxParams:
        """Add gas limit to a transaction.

        Args:
            tx (TxParams): The transaction parameters

        Returns:
            TxParams: The transaction parameters with gas limit added
        """
        if "gas" in tx:
            return tx
        try:
            gas = self.web3.eth.estimate_gas(tx)
            tx["gas"] = gas
            return tx
        except Web3Exception as e:
            self.logger.error(
                f"Failed to add gas limit: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
            )
            raise e

    def submit_tx(self, tx: TxParams) -> str:
        """Submit a transaction.

        Args:
            tx (TxParams): The transaction parameters

        Returns:
            str: The transaction hash
        """
        tx = self.add_gas_fees(tx)
        tx = self.add_gas_limit(tx)
        try:
            signed_tx = self.web3.eth.account.sign_transaction(
                tx, private_key=self.private_key
            )
            tx_hash = self.web3.eth.send_raw_transaction(signed_tx.raw_transaction)
            return encode_hex(tx_hash)
        except Web3Exception as e:
            self.logger.error(f"Failed to submit transaction: {e}")
            raise e

    def get_nonce(self, address: str) -> int:
        """Get the nonce for a given address.

        Args:
            address (str): The address to get the nonce for

        Returns:
            int: The nonce
        """
        try:
            return self.web3.eth.get_transaction_count(address)
        except Web3Exception as e:
            self.logger.error(f"Failed to get nonce: {e}")
            raise e

    def get_balance(self, address: str) -> int:
        """Get the balance for a given address.

        Args:
            address (str): The address to get the balance for

        Returns:
            int: The balance
        """
        try:
            return self.web3.eth.get_balance(address)
        except Web3Exception as e:
            self.logger.error(f"Failed to get balance: {e}")
            raise e

    def get_token_balance(self, address: str, token_address: str) -> int:
        """Get the token balance for a given address.

        Args:
            address (str): The address to get the token balance for
            token_address (str): The token address

        Returns:
            int: The token balance
        """
        try:
            contract = self.web3.eth.contract(
                address=token_address,
                abi=read_contract(self.chain_id, "ERC20", common=True),
            )
            return contract.functions.balanceOf(address).call()
        except Web3Exception as e:
            self.logger.error(f"Failed to get token balance: {e}")
            raise e

    def sign_message(
        self,
        private_key: str,
        domain: dict,
        types: dict,
        primary_type: str,
        message: dict,
    ):
        """Sign an EIP-712 typed data message.

        Args:
            private_key (str): private key to sign the message with
            domain (dict): domain parameters including name, version, chainId, and verifyingContract
            types (dict): type definitions for the structured data
            primary_type (str): primary type for the signature
            message (dict): message data to be signed

        Returns:
            str: the hexadecimal signature string prefixed with '0x'
        """
        # A type fix for the domain
        domain["chainId"] = int(domain["chainId"])

        # Preparing the full message as per EIP-712
        full_message = {
            "types": types,
            "primaryType": primary_type,
            "domain": domain,
            "message": message,
        }

        encoded_message = encode_typed_data(full_message=full_message)

        # Signing the message
        signed_message = Account.sign_message(encoded_message, private_key)
        return "0x" + signed_message.signature.hex()

    def get_typed_data_digest(
        self,
        domain: dict,
        types: dict,
        primary_type: str,
        message: dict,
    ) -> str:
        """Compute the EIP-712 digest of a typed data message.

        This is the digest the exchange contract uses to identify signed
        payloads, e.g. the withdraw digest passed to finalizeWithdraw.

        Args:
            domain (dict): domain parameters including name, version, chainId, and verifyingContract
            types (dict): type definitions for the structured data
            primary_type (str): primary type for the signature
            message (dict): message data

        Returns:
            str: the hexadecimal digest string prefixed with '0x'
        """
        domain["chainId"] = int(domain["chainId"])
        full_message = {
            "types": types,
            "primaryType": primary_type,
            "domain": domain,
            "message": message,
        }
        encoded_message = encode_typed_data(full_message=full_message)
        digest = keccak(
            b"\x19"
            + encoded_message.version
            + encoded_message.header
            + encoded_message.body
        )
        return "0x" + digest.hex()

    def get_token_allowance(
        self, owner: str, token_address: str, spender: Optional[str] = None
    ) -> int:
        """Get the ERC20 allowance granted by an owner to a spender.

        Args:
            owner (str): The token owner address
            token_address (str): The token address
            spender (str, optional): The spender address. Defaults to the exchange contract.

        Returns:
            int: The allowance in token base units
        """
        if spender is None:
            spender = self.exchange_contract.address
        try:
            contract = self.web3.eth.contract(
                address=self.web3.to_checksum_address(token_address),
                abi=read_contract(self.chain_id, "ERC20", common=True),
            )
            return contract.functions.allowance(
                self.web3.to_checksum_address(owner),
                self.web3.to_checksum_address(spender),
            ).call()
        except Web3Exception as e:
            self.logger.error(f"Failed to get token allowance: {e}")
            raise e

    def get_token_decimals(self, token_address: str) -> int:
        """Get the number of decimals for an ERC20 token.

        Args:
            token_address (str): The token address

        Returns:
            int: The token decimals
        """
        try:
            contract = self.web3.eth.contract(
                address=self.web3.to_checksum_address(token_address),
                abi=read_contract(self.chain_id, "ERC20", common=True),
            )
            return contract.functions.decimals().call()
        except Web3Exception as e:
            self.logger.error(f"Failed to get token decimals: {e}")
            raise e

    def approve_token(
        self,
        amount: float,
        token_address: str,
        spender: Optional[str] = None,
        submit: bool = False,
    ) -> Union[TxParams, str]:
        """Approve a spender to transfer tokens on behalf of the account.

        Deposits pull tokens from the sender via transferFrom, so the exchange
        contract must be approved before depositing.

        Args:
            amount (float): The amount to approve, in whole token units
            token_address (str): The token address
            spender (str, optional): The spender address. Defaults to the exchange contract.
            submit (bool, optional): Whether to submit the transaction. Defaults to False.

        Returns:
            Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
        """
        if spender is None:
            spender = self.exchange_contract.address
        try:
            token_address = self.web3.to_checksum_address(token_address)
            decimals = self.get_token_decimals(token_address)
            raw_amount = int(amount * 10**decimals)
            contract = self.web3.eth.contract(
                address=token_address,
                abi=read_contract(self.chain_id, "ERC20", common=True),
            )
            tx = self._get_tx(to=token_address)
            tx["data"] = contract.encode_abi(
                "approve",
                args=[self.web3.to_checksum_address(spender), raw_amount],
            )
            if submit:
                return self.submit_tx(tx)
            else:
                return tx
        except Web3Exception as e:
            self.logger.error(f"Failed to prepare approve transaction: {e}")
            raise e

    def _encode_subaccount(
        self,
        account_name: Optional[str],
        account_name_bytes: Optional[str],
    ) -> str:
        if account_name is None and account_name_bytes is None:
            account_name = "primary"
        if account_name is not None and account_name_bytes is not None:
            raise ValueError("Cannot provide both account_name and account_name_bytes")
        if account_name_bytes is not None:
            return ensure_bytes32_hex(account_name_bytes)
        return self.web3.to_hex(text=account_name).ljust(66, "0")

    def deposit_usd(
        self,
        amount: float,
        account_name: str = "primary",
        address: Optional[str] = None,
        submit: bool = False,
        account_name_bytes: Optional[str] = None,
        token_address: Optional[str] = None,
    ) -> Union[TxParams, str]:
        """Submit a deposit transaction for the default USD token.

        The exchange pulls the tokens via transferFrom, so the exchange
        contract must hold a sufficient allowance first (see approve_token).

        Args:
            amount (float): The amount to deposit, in whole token units
            address (str, optional): The address to deposit to. Defaults to None.
            submit (bool, optional): Whether to submit the transaction. Defaults to False.
            account_name (str, optional): The account name. Defaults to "primary".
            account_name_bytes (str, optional): The account name as a hex string (bytes32). Defaults to None.
            token_address (str, optional): The token to approve and pull, used to
                resolve decimals and verify the allowance. Defaults to mUSD's
                underlying wrapped token (the token depositUsd pulls via transferFrom).

        Returns:
            Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

        Raises:
            ValueError: If both account_name and account_name_bytes are provided
            ValueError: If the exchange allowance is insufficient when submitting
        """
        if address is None:
            address = self.address

        try:
            subaccount = self._encode_subaccount(account_name, account_name_bytes)
            if token_address is None and self.usd is not None:
                token_address = self.usd_deposit_token

            if token_address is not None:
                decimals = self.get_token_decimals(token_address)
            else:
                decimals = 18
            raw_amount = int(amount * 10**decimals)
            referral_code = self.web3.to_hex(0).ljust(66, "0")

            if submit and token_address is not None:
                allowance = self.get_token_allowance(address, token_address)
                if allowance < raw_amount:
                    raise ValueError(
                        f"Exchange allowance {allowance} is less than deposit amount "
                        f"{raw_amount}; call approve_token first"
                    )

            # prepare the tx
            tx = self._get_tx(to=self.exchange_contract.address)
            tx["data"] = self.exchange_contract.encode_abi(
                "depositUsd", args=[subaccount, raw_amount, referral_code]
            )

            if submit:
                return self.submit_tx(tx)
            else:
                return tx

        except Web3Exception as e:
            self.logger.error(
                f"Failed to prepare deposit transaction: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
            )
            raise e

    def deposit_token(
        self,
        token_address: str,
        amount: float,
        account_name: str = "primary",
        address: Optional[str] = None,
        submit: bool = False,
        account_name_bytes: Optional[str] = None,
    ) -> Union[TxParams, str]:
        """Submit a deposit transaction for an arbitrary supported token.

        The exchange pulls the tokens via transferFrom, so the exchange
        contract must hold a sufficient allowance first (see approve_token).

        Args:
            token_address (str): The address of the token to deposit
            amount (float): The amount to deposit, in whole token units
            address (str, optional): The address to deposit to. Defaults to None.
            submit (bool, optional): Whether to submit the transaction. Defaults to False.
            account_name (str, optional): The account name. Defaults to "primary".
            account_name_bytes (str, optional): The account name as a hex string (bytes32). Defaults to None.

        Returns:
            Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

        Raises:
            ValueError: If both account_name and account_name_bytes are provided
            ValueError: If the exchange allowance is insufficient when submitting
        """
        if address is None:
            address = self.address

        try:
            subaccount = self._encode_subaccount(account_name, account_name_bytes)
            token_address = self.web3.to_checksum_address(token_address)
            decimals = self.get_token_decimals(token_address)
            raw_amount = int(amount * 10**decimals)
            referral_code = self.web3.to_hex(0).ljust(66, "0")

            if submit:
                allowance = self.get_token_allowance(address, token_address)
                if allowance < raw_amount:
                    raise ValueError(
                        f"Exchange allowance {allowance} is less than deposit amount "
                        f"{raw_amount}; call approve_token first"
                    )

            tx = self._get_tx(to=self.exchange_contract.address)
            tx["data"] = self.exchange_contract.encode_abi(
                "deposit", args=[subaccount, token_address, raw_amount, referral_code]
            )

            if submit:
                return self.submit_tx(tx)
            else:
                return tx

        except Web3Exception as e:
            self.logger.error(
                f"Failed to prepare deposit transaction: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
            )
            raise e

    def finalize_withdraw(
        self,
        withdraw_digest: str,
        address: Optional[str] = None,
        submit: Optional[bool] = False,
        value: int = 0,
    ) -> Union[TxParams, str]:
        """Finalize a pending withdrawal.

        Args:
            withdraw_digest (str): The EIP-712 digest of the signed InitiateWithdraw
                message identifying the pending withdrawal (see
                get_typed_data_digest / get_withdraw_digest).
            address (str, optional): The account address the withdrawal belongs to. Defaults to None.
            submit (bool, optional): Whether to submit the transaction. Defaults to False.
            value (int, optional): Native value to attach for cross-chain messaging fees. Defaults to 0.

        Returns:
            Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
        """
        if address is None:
            address = self.address
        try:
            # params
            digest = ensure_bytes32_hex(withdraw_digest)

            # prepare the tx
            tx = self._get_tx(to=self.exchange_contract.address, value=value)
            tx["data"] = self.exchange_contract.encode_abi(
                "finalizeWithdraw", args=[address, digest]
            )

            if submit:
                return self.submit_tx(tx)
            else:
                return tx

        except Web3Exception as e:
            self.logger.error(
                f"Failed to prepare finalizeWithdraw transaction: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
            )
            raise e

usd_deposit_token property

Address of the token deposited as USD collateral.

mUSD is a wrapper token: depositUsd pulls mUSD's underlying wrapped token from the sender via transferFrom and wraps it. This is the token to approve and fund when depositing USD, not mUSD itself.

Returns:

Name	Type	Description
str	str	The checksummed address of mUSD's default wrapped token.

Raises:

Type	Description
ValueError	If no USD (mUSD) token is configured on the client.

add_gas_fees(tx)

Add gas fee parameters to a transaction.

Parameters:

Name	Type	Description	Default
tx	TxParams	The transaction parameters	required

Returns:

Name	Type	Description
TxParams	TxParams	The transaction parameters with gas fee parameters added

Source code in meridian/chain_client.py

def add_gas_fees(self, tx: TxParams) -> TxParams:
    """Add gas fee parameters to a transaction.

    Args:
        tx (TxParams): The transaction parameters

    Returns:
        TxParams: The transaction parameters with gas fee parameters added
    """
    if "maxFeePerGas" in tx and "maxPriorityFeePerGas" in tx:
        return tx
    try:
        gas_price = self.web3.eth.gas_price
        max_priority_fee = self.web3.eth.max_priority_fee
        tx["maxFeePerGas"] = gas_price
        tx["maxPriorityFeePerGas"] = max_priority_fee
        return tx
    except Web3Exception as e:
        self.logger.error(f"Failed to add gas: {e}")
        return tx

add_gas_limit(tx)

Add gas limit to a transaction.

Parameters:

Name	Type	Description	Default
tx	TxParams	The transaction parameters	required

Returns:

Name	Type	Description
TxParams	TxParams	The transaction parameters with gas limit added

Source code in meridian/chain_client.py

def add_gas_limit(self, tx: TxParams) -> TxParams:
    """Add gas limit to a transaction.

    Args:
        tx (TxParams): The transaction parameters

    Returns:
        TxParams: The transaction parameters with gas limit added
    """
    if "gas" in tx:
        return tx
    try:
        gas = self.web3.eth.estimate_gas(tx)
        tx["gas"] = gas
        return tx
    except Web3Exception as e:
        self.logger.error(
            f"Failed to add gas limit: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
        )
        raise e

approve_token(amount, token_address, spender=None, submit=False)

Approve a spender to transfer tokens on behalf of the account.

Deposits pull tokens from the sender via transferFrom, so the exchange contract must be approved before depositing.

Parameters:

Name	Type	Description	Default
amount	float	The amount to approve, in whole token units	required
token_address	str	The token address	required
spender	str	The spender address. Defaults to the exchange contract.	None
submit	bool	Whether to submit the transaction. Defaults to False.	False

Returns:

Type	Description
Union[TxParams, str]	Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

Source code in meridian/chain_client.py

def approve_token(
    self,
    amount: float,
    token_address: str,
    spender: Optional[str] = None,
    submit: bool = False,
) -> Union[TxParams, str]:
    """Approve a spender to transfer tokens on behalf of the account.

    Deposits pull tokens from the sender via transferFrom, so the exchange
    contract must be approved before depositing.

    Args:
        amount (float): The amount to approve, in whole token units
        token_address (str): The token address
        spender (str, optional): The spender address. Defaults to the exchange contract.
        submit (bool, optional): Whether to submit the transaction. Defaults to False.

    Returns:
        Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
    """
    if spender is None:
        spender = self.exchange_contract.address
    try:
        token_address = self.web3.to_checksum_address(token_address)
        decimals = self.get_token_decimals(token_address)
        raw_amount = int(amount * 10**decimals)
        contract = self.web3.eth.contract(
            address=token_address,
            abi=read_contract(self.chain_id, "ERC20", common=True),
        )
        tx = self._get_tx(to=token_address)
        tx["data"] = contract.encode_abi(
            "approve",
            args=[self.web3.to_checksum_address(spender), raw_amount],
        )
        if submit:
            return self.submit_tx(tx)
        else:
            return tx
    except Web3Exception as e:
        self.logger.error(f"Failed to prepare approve transaction: {e}")
        raise e

convert_types(type_string)

Converts type string into EIP-712 field format.

Parameters:

Name	Type	Description	Default
type_string	str	String containing type definitions.	required

Returns:

Type	Description
List[Dict[str, str]]	List[Dict[str, str]]: List of field definitions.

Source code in meridian/chain_client.py

def convert_types(self, type_string: str) -> List[Dict[str, str]]:
    """Converts type string into EIP-712 field format.

    Args:
        type_string (str): String containing type definitions.

    Returns:
        List[Dict[str, str]]: List of field definitions.
    """
    fields = [comp.strip() for comp in type_string.split(",")]
    type_fields = []
    for field in fields:
        field_type, field_name = field.rsplit(" ", 1)
        type_fields.append({"name": field_name, "type": field_type})
    return type_fields

deposit_token(token_address, amount, account_name='primary', address=None, submit=False, account_name_bytes=None)

Submit a deposit transaction for an arbitrary supported token.

The exchange pulls the tokens via transferFrom, so the exchange contract must hold a sufficient allowance first (see approve_token).

Parameters:

Name	Type	Description	Default
token_address	str	The address of the token to deposit	required
amount	float	The amount to deposit, in whole token units	required
address	str	The address to deposit to. Defaults to None.	None
submit	bool	Whether to submit the transaction. Defaults to False.	False
account_name	str	The account name. Defaults to "primary".	'primary'
account_name_bytes	str	The account name as a hex string (bytes32). Defaults to None.	None

Returns:

Type	Description
Union[TxParams, str]	Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

Raises:

Type	Description
ValueError	If both account_name and account_name_bytes are provided
ValueError	If the exchange allowance is insufficient when submitting

Source code in meridian/chain_client.py

def deposit_token(
    self,
    token_address: str,
    amount: float,
    account_name: str = "primary",
    address: Optional[str] = None,
    submit: bool = False,
    account_name_bytes: Optional[str] = None,
) -> Union[TxParams, str]:
    """Submit a deposit transaction for an arbitrary supported token.

    The exchange pulls the tokens via transferFrom, so the exchange
    contract must hold a sufficient allowance first (see approve_token).

    Args:
        token_address (str): The address of the token to deposit
        amount (float): The amount to deposit, in whole token units
        address (str, optional): The address to deposit to. Defaults to None.
        submit (bool, optional): Whether to submit the transaction. Defaults to False.
        account_name (str, optional): The account name. Defaults to "primary".
        account_name_bytes (str, optional): The account name as a hex string (bytes32). Defaults to None.

    Returns:
        Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

    Raises:
        ValueError: If both account_name and account_name_bytes are provided
        ValueError: If the exchange allowance is insufficient when submitting
    """
    if address is None:
        address = self.address

    try:
        subaccount = self._encode_subaccount(account_name, account_name_bytes)
        token_address = self.web3.to_checksum_address(token_address)
        decimals = self.get_token_decimals(token_address)
        raw_amount = int(amount * 10**decimals)
        referral_code = self.web3.to_hex(0).ljust(66, "0")

        if submit:
            allowance = self.get_token_allowance(address, token_address)
            if allowance < raw_amount:
                raise ValueError(
                    f"Exchange allowance {allowance} is less than deposit amount "
                    f"{raw_amount}; call approve_token first"
                )

        tx = self._get_tx(to=self.exchange_contract.address)
        tx["data"] = self.exchange_contract.encode_abi(
            "deposit", args=[subaccount, token_address, raw_amount, referral_code]
        )

        if submit:
            return self.submit_tx(tx)
        else:
            return tx

    except Web3Exception as e:
        self.logger.error(
            f"Failed to prepare deposit transaction: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
        )
        raise e

deposit_usd(amount, account_name='primary', address=None, submit=False, account_name_bytes=None, token_address=None)

Submit a deposit transaction for the default USD token.

The exchange pulls the tokens via transferFrom, so the exchange contract must hold a sufficient allowance first (see approve_token).

Parameters:

Name	Type	Description	Default
amount	float	The amount to deposit, in whole token units	required
address	str	The address to deposit to. Defaults to None.	None
submit	bool	Whether to submit the transaction. Defaults to False.	False
account_name	str	The account name. Defaults to "primary".	'primary'
account_name_bytes	str	The account name as a hex string (bytes32). Defaults to None.	None
token_address	str	The token to approve and pull, used to resolve decimals and verify the allowance. Defaults to mUSD's underlying wrapped token (the token depositUsd pulls via transferFrom).	None

Returns:

Type	Description
Union[TxParams, str]	Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

Raises:

Type	Description
ValueError	If both account_name and account_name_bytes are provided
ValueError	If the exchange allowance is insufficient when submitting

Source code in meridian/chain_client.py

def deposit_usd(
    self,
    amount: float,
    account_name: str = "primary",
    address: Optional[str] = None,
    submit: bool = False,
    account_name_bytes: Optional[str] = None,
    token_address: Optional[str] = None,
) -> Union[TxParams, str]:
    """Submit a deposit transaction for the default USD token.

    The exchange pulls the tokens via transferFrom, so the exchange
    contract must hold a sufficient allowance first (see approve_token).

    Args:
        amount (float): The amount to deposit, in whole token units
        address (str, optional): The address to deposit to. Defaults to None.
        submit (bool, optional): Whether to submit the transaction. Defaults to False.
        account_name (str, optional): The account name. Defaults to "primary".
        account_name_bytes (str, optional): The account name as a hex string (bytes32). Defaults to None.
        token_address (str, optional): The token to approve and pull, used to
            resolve decimals and verify the allowance. Defaults to mUSD's
            underlying wrapped token (the token depositUsd pulls via transferFrom).

    Returns:
        Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

    Raises:
        ValueError: If both account_name and account_name_bytes are provided
        ValueError: If the exchange allowance is insufficient when submitting
    """
    if address is None:
        address = self.address

    try:
        subaccount = self._encode_subaccount(account_name, account_name_bytes)
        if token_address is None and self.usd is not None:
            token_address = self.usd_deposit_token

        if token_address is not None:
            decimals = self.get_token_decimals(token_address)
        else:
            decimals = 18
        raw_amount = int(amount * 10**decimals)
        referral_code = self.web3.to_hex(0).ljust(66, "0")

        if submit and token_address is not None:
            allowance = self.get_token_allowance(address, token_address)
            if allowance < raw_amount:
                raise ValueError(
                    f"Exchange allowance {allowance} is less than deposit amount "
                    f"{raw_amount}; call approve_token first"
                )

        # prepare the tx
        tx = self._get_tx(to=self.exchange_contract.address)
        tx["data"] = self.exchange_contract.encode_abi(
            "depositUsd", args=[subaccount, raw_amount, referral_code]
        )

        if submit:
            return self.submit_tx(tx)
        else:
            return tx

    except Web3Exception as e:
        self.logger.error(
            f"Failed to prepare deposit transaction: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
        )
        raise e

finalize_withdraw(withdraw_digest, address=None, submit=False, value=0)

Finalize a pending withdrawal.

Parameters:

Name	Type	Description	Default
withdraw_digest	str	The EIP-712 digest of the signed InitiateWithdraw message identifying the pending withdrawal (see get_typed_data_digest / get_withdraw_digest).	required
address	str	The account address the withdrawal belongs to. Defaults to None.	None
submit	bool	Whether to submit the transaction. Defaults to False.	False
value	int	Native value to attach for cross-chain messaging fees. Defaults to 0.	0

Returns:

Type	Description
Union[TxParams, str]	Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

Source code in meridian/chain_client.py

def finalize_withdraw(
    self,
    withdraw_digest: str,
    address: Optional[str] = None,
    submit: Optional[bool] = False,
    value: int = 0,
) -> Union[TxParams, str]:
    """Finalize a pending withdrawal.

    Args:
        withdraw_digest (str): The EIP-712 digest of the signed InitiateWithdraw
            message identifying the pending withdrawal (see
            get_typed_data_digest / get_withdraw_digest).
        address (str, optional): The account address the withdrawal belongs to. Defaults to None.
        submit (bool, optional): Whether to submit the transaction. Defaults to False.
        value (int, optional): Native value to attach for cross-chain messaging fees. Defaults to 0.

    Returns:
        Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
    """
    if address is None:
        address = self.address
    try:
        # params
        digest = ensure_bytes32_hex(withdraw_digest)

        # prepare the tx
        tx = self._get_tx(to=self.exchange_contract.address, value=value)
        tx["data"] = self.exchange_contract.encode_abi(
            "finalizeWithdraw", args=[address, digest]
        )

        if submit:
            return self.submit_tx(tx)
        else:
            return tx

    except Web3Exception as e:
        self.logger.error(
            f"Failed to prepare finalizeWithdraw transaction: {self._decode_error(e) if isinstance(e, ContractCustomError) else e}"
        )
        raise e

get_balance(address)

Get the balance for a given address.

Parameters:

Name	Type	Description	Default
address	str	The address to get the balance for	required

Returns:

Name	Type	Description
int	int	The balance

Source code in meridian/chain_client.py

def get_balance(self, address: str) -> int:
    """Get the balance for a given address.

    Args:
        address (str): The address to get the balance for

    Returns:
        int: The balance
    """
    try:
        return self.web3.eth.get_balance(address)
    except Web3Exception as e:
        self.logger.error(f"Failed to get balance: {e}")
        raise e

get_nonce(address)

Get the nonce for a given address.

Parameters:

Name	Type	Description	Default
address	str	The address to get the nonce for	required

Returns:

Name	Type	Description
int	int	The nonce

Source code in meridian/chain_client.py

def get_nonce(self, address: str) -> int:
    """Get the nonce for a given address.

    Args:
        address (str): The address to get the nonce for

    Returns:
        int: The nonce
    """
    try:
        return self.web3.eth.get_transaction_count(address)
    except Web3Exception as e:
        self.logger.error(f"Failed to get nonce: {e}")
        raise e

get_signature_types(rpc_config, primary_type)

Gets EIP-712 signature types.

Parameters:

Name	Type	Description	Default
rpc_config	RpcConfigDto	RPC configuration.	required
primary_type	str	Primary type for the signature.	required

Returns:

Name	Type	Description
dict		Dictionary containing signature type definitions.

Source code in meridian/chain_client.py

def get_signature_types(self, rpc_config: RpcConfigDto, primary_type: str):
    """Gets EIP-712 signature types.

    Args:
        rpc_config (RpcConfigDto): RPC configuration.
        primary_type (str): Primary type for the signature.

    Returns:
        dict: Dictionary containing signature type definitions.
    """
    return {
        "EIP712Domain": [
            {"name": "name", "type": "string"},
            {"name": "version", "type": "string"},
            {"name": "chainId", "type": "uint256"},
            {"name": "verifyingContract", "type": "address"},
        ],
        primary_type: self.convert_types(
            self._get_by_alias(rpc_config.signature_types, primary_type),
        ),
    }

get_token_allowance(owner, token_address, spender=None)

Get the ERC20 allowance granted by an owner to a spender.

Parameters:

Name	Type	Description	Default
owner	str	The token owner address	required
token_address	str	The token address	required
spender	str	The spender address. Defaults to the exchange contract.	None

Returns:

Name	Type	Description
int	int	The allowance in token base units

Source code in meridian/chain_client.py

def get_token_allowance(
    self, owner: str, token_address: str, spender: Optional[str] = None
) -> int:
    """Get the ERC20 allowance granted by an owner to a spender.

    Args:
        owner (str): The token owner address
        token_address (str): The token address
        spender (str, optional): The spender address. Defaults to the exchange contract.

    Returns:
        int: The allowance in token base units
    """
    if spender is None:
        spender = self.exchange_contract.address
    try:
        contract = self.web3.eth.contract(
            address=self.web3.to_checksum_address(token_address),
            abi=read_contract(self.chain_id, "ERC20", common=True),
        )
        return contract.functions.allowance(
            self.web3.to_checksum_address(owner),
            self.web3.to_checksum_address(spender),
        ).call()
    except Web3Exception as e:
        self.logger.error(f"Failed to get token allowance: {e}")
        raise e

get_token_balance(address, token_address)

Get the token balance for a given address.

Parameters:

Name	Type	Description	Default
address	str	The address to get the token balance for	required
token_address	str	The token address	required

Returns:

Name	Type	Description
int	int	The token balance

Source code in meridian/chain_client.py

def get_token_balance(self, address: str, token_address: str) -> int:
    """Get the token balance for a given address.

    Args:
        address (str): The address to get the token balance for
        token_address (str): The token address

    Returns:
        int: The token balance
    """
    try:
        contract = self.web3.eth.contract(
            address=token_address,
            abi=read_contract(self.chain_id, "ERC20", common=True),
        )
        return contract.functions.balanceOf(address).call()
    except Web3Exception as e:
        self.logger.error(f"Failed to get token balance: {e}")
        raise e

get_token_decimals(token_address)

Get the number of decimals for an ERC20 token.

Parameters:

Name	Type	Description	Default
token_address	str	The token address	required

Returns:

Name	Type	Description
int	int	The token decimals

Source code in meridian/chain_client.py

def get_token_decimals(self, token_address: str) -> int:
    """Get the number of decimals for an ERC20 token.

    Args:
        token_address (str): The token address

    Returns:
        int: The token decimals
    """
    try:
        contract = self.web3.eth.contract(
            address=self.web3.to_checksum_address(token_address),
            abi=read_contract(self.chain_id, "ERC20", common=True),
        )
        return contract.functions.decimals().call()
    except Web3Exception as e:
        self.logger.error(f"Failed to get token decimals: {e}")
        raise e

get_typed_data_digest(domain, types, primary_type, message)

Compute the EIP-712 digest of a typed data message.

This is the digest the exchange contract uses to identify signed payloads, e.g. the withdraw digest passed to finalizeWithdraw.

Parameters:

Name	Type	Description	Default
domain	dict	domain parameters including name, version, chainId, and verifyingContract	required
types	dict	type definitions for the structured data	required
primary_type	str	primary type for the signature	required
message	dict	message data	required

Returns:

Name	Type	Description
str	str	the hexadecimal digest string prefixed with '0x'

Source code in meridian/chain_client.py

def get_typed_data_digest(
    self,
    domain: dict,
    types: dict,
    primary_type: str,
    message: dict,
) -> str:
    """Compute the EIP-712 digest of a typed data message.

    This is the digest the exchange contract uses to identify signed
    payloads, e.g. the withdraw digest passed to finalizeWithdraw.

    Args:
        domain (dict): domain parameters including name, version, chainId, and verifyingContract
        types (dict): type definitions for the structured data
        primary_type (str): primary type for the signature
        message (dict): message data

    Returns:
        str: the hexadecimal digest string prefixed with '0x'
    """
    domain["chainId"] = int(domain["chainId"])
    full_message = {
        "types": types,
        "primaryType": primary_type,
        "domain": domain,
        "message": message,
    }
    encoded_message = encode_typed_data(full_message=full_message)
    digest = keccak(
        b"\x19"
        + encoded_message.version
        + encoded_message.header
        + encoded_message.body
    )
    return "0x" + digest.hex()

sign_message(private_key, domain, types, primary_type, message)

Sign an EIP-712 typed data message.

Parameters:

Name	Type	Description	Default
private_key	str	private key to sign the message with	required
domain	dict	domain parameters including name, version, chainId, and verifyingContract	required
types	dict	type definitions for the structured data	required
primary_type	str	primary type for the signature	required
message	dict	message data to be signed	required

Returns:

Name	Type	Description
str		the hexadecimal signature string prefixed with '0x'

Source code in meridian/chain_client.py

def sign_message(
    self,
    private_key: str,
    domain: dict,
    types: dict,
    primary_type: str,
    message: dict,
):
    """Sign an EIP-712 typed data message.

    Args:
        private_key (str): private key to sign the message with
        domain (dict): domain parameters including name, version, chainId, and verifyingContract
        types (dict): type definitions for the structured data
        primary_type (str): primary type for the signature
        message (dict): message data to be signed

    Returns:
        str: the hexadecimal signature string prefixed with '0x'
    """
    # A type fix for the domain
    domain["chainId"] = int(domain["chainId"])

    # Preparing the full message as per EIP-712
    full_message = {
        "types": types,
        "primaryType": primary_type,
        "domain": domain,
        "message": message,
    }

    encoded_message = encode_typed_data(full_message=full_message)

    # Signing the message
    signed_message = Account.sign_message(encoded_message, private_key)
    return "0x" + signed_message.signature.hex()

submit_tx(tx)

Submit a transaction.

Parameters:

Name	Type	Description	Default
tx	TxParams	The transaction parameters	required

Returns:

Name	Type	Description
str	str	The transaction hash

Source code in meridian/chain_client.py

def submit_tx(self, tx: TxParams) -> str:
    """Submit a transaction.

    Args:
        tx (TxParams): The transaction parameters

    Returns:
        str: The transaction hash
    """
    tx = self.add_gas_fees(tx)
    tx = self.add_gas_limit(tx)
    try:
        signed_tx = self.web3.eth.account.sign_transaction(
            tx, private_key=self.private_key
        )
        tx_hash = self.web3.eth.send_raw_transaction(signed_tx.raw_transaction)
        return encode_hex(tx_hash)
    except Web3Exception as e:
        self.logger.error(f"Failed to submit transaction: {e}")
        raise e

meridian.rest.http_client.HTTPClient

Bases: BaseClient

HTTP client for making API requests.

Parameters:

Name	Type	Description	Default
config	Union[Dict[str, Any], HTTPConfig]	Client configuration.	required

Source code in meridian/rest/http_client.py

class HTTPClient(BaseClient):
    """HTTP client for making API requests.

    Args:
        config (Union[Dict[str, Any], HTTPConfig]): Client configuration.
    """

    def __init__(self, config: Union[Dict[str, Any], HTTPConfig]):
        super().__init__(config)
        self.config = HTTPConfig.model_validate(config)
        self.base_url = self.config.base_url
        self.timeout = self.config.timeout
        self.session = self._setup_session()
        self.rate_limit_headers = self.config.rate_limit_headers

    def _setup_session(self):
        """Sets up an HTTP session.

        Returns:
            requests.Session: Configured session object.
        """
        return requests.Session()

    def _handle_exception(self, response):
        """Handles HTTP exceptions.

        Args:
            response (Response): The HTTP response object.

        Raises:
            HTTPError: If response indicates an error occurred.
        """
        http_error_msg = ""
        reason = response.reason

        if 400 <= response.status_code < 500:
            if (
                response.status_code == 403
                and "'error_details':'Missing required scopes'" in response.text
            ):
                http_error_msg = f"{response.status_code} Client Error: Missing Required Scopes. Please verify your API keys include the necessary permissions."
            else:
                http_error_msg = (
                    f"{response.status_code} Client Error: {reason} {response.text}"
                )
        elif 500 <= response.status_code < 600:
            http_error_msg = (
                f"{response.status_code} Server Error: {reason} {response.text}"
            )

        if http_error_msg:
            self.logger.error(f"HTTP Error: {http_error_msg}")
            raise HTTPError(http_error_msg, response=response)

    def get(self, url_path, params: Optional[dict] = None, **kwargs) -> Dict[str, Any]:
        """Sends a GET request.

        Args:
            url_path (str): The URL path. Required.
            params (dict, optional): The query parameters. Optional.
            **kwargs: Additional arguments to pass to the request. Optional.

        Returns:
            Dict[str, Any]: The response data.
        """
        params = params or {}

        if kwargs:
            params.update(kwargs)

        return self.prepare_and_send_request("GET", url_path, params, data=None)

    def get_validated(
        self,
        url_path,
        request_model: Type[BaseModel],
        response_model: Type[BaseModel],
        **kwargs,
    ) -> BaseModel:
        """Sends a GET request including type validation of both the input and output from provided models.

        Args:
            url_path (str): The URL path. Required.
            request_model (Type[BaseModel]): Pydantic model for request validation. Required.
            response_model (Type[BaseModel]): Pydantic model for response validation. Required.
            **kwargs: Includes all arguments to pass to the request. Optional.

        Returns:
            BaseModel: The response data, validated against the response_model.
        """
        try:
            validated_params = request_model.model_validate(kwargs, by_name=True)
        except ValidationError as e:
            raise ValidationException(
                "Request", url_path, request_model, e.errors()
            ) from None
        params = validated_params.model_dump(
            mode="json", exclude_none=True, by_alias=True
        )

        response_data = self.prepare_and_send_request(
            "GET", url_path, params, data=None
        )
        validated_response = response_model.model_validate(response_data)
        return validated_response

    def post(
        self,
        url_path,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
        **kwargs,
    ) -> Dict[str, Any]:
        """Sends a POST request.

        Args:
            url_path (str): The URL path. Required.
            params (dict, optional): The query parameters. Optional.
            data (dict, optional): The request body. Optional.
            **kwargs: Additional arguments to pass to the request. Optional.

        Returns:
            Dict[str, Any]: The response data.
        """
        data = data or {}

        if kwargs:
            data.update(kwargs)

        return self.prepare_and_send_request("POST", url_path, params, data)

    def put(
        self,
        url_path,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
        **kwargs,
    ) -> Dict[str, Any]:
        """Sends a PUT request.

        Args:
            url_path (str): The URL path. Required.
            params (dict, optional): The query parameters. Optional.
            data (dict, optional): The request body. Optional.
            **kwargs: Additional arguments to pass to the request. Optional.

        Returns:
            Dict[str, Any]: The response data.
        """
        data = data or {}

        if kwargs:
            data.update(kwargs)

        return self.prepare_and_send_request("PUT", url_path, params, data)

    def delete(
        self,
        url_path,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
        **kwargs,
    ) -> Dict[str, Any]:
        """Sends a DELETE request.

        Args:
            url_path (str): The URL path. Required.
            params (dict, optional): The query parameters. Optional.
            data (dict, optional): The request body. Optional.
            **kwargs: Additional arguments to pass to the request. Optional.

        Returns:
            Dict[str, Any]: The response data.
        """
        data = data or {}

        if kwargs:
            data.update(kwargs)

        return self.prepare_and_send_request("DELETE", url_path, params, data)

    def prepare_and_send_request(
        self,
        http_method,
        url_path,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
    ):
        """Prepares and sends an HTTP request.

        Args:
            http_method (str): The HTTP method. Required.
            url_path (str): The URL path. Required.
            params (dict, optional): The query parameters. Optional.
            data (dict, optional): The request body. Optional.

        Returns:
            Dict[str, Any]: The response data.
        """
        headers = self.set_headers(http_method, url_path)

        if params is not None:
            params = {
                key: str(value).lower() if isinstance(value, bool) else value
                for key, value in params.items()
                if value is not None
            }

        if data is not None:
            data = {key: value for key, value in data.items() if value is not None}

        return self.send_request(http_method, url_path, params, headers, data=data)

    def send_request(self, http_method, url_path, params, headers, data=None):
        """Sends an HTTP request.

        Args:
            http_method (str): The HTTP method. Required.
            url_path (str): The URL path. Required.
            params (dict): The query parameters. Required.
            headers (dict): The request headers. Required.
            data (dict, optional): The request body. Optional.

        Returns:
            Dict[str, Any]: The response data.

        Raises:
            HTTPError: If the request fails.
        """
        if data is None:
            data = {}

        url = f"{self.base_url}{url_path}"

        self.logger.debug(f"Sending {http_method} request to {url}")

        response = self.session.request(
            http_method,
            url,
            params=params,
            json=data,
            headers=headers,
            timeout=self.timeout,
        )
        self._handle_exception(response)  # Raise an HTTPError for bad responses

        self.logger.debug(f"Raw response: {response.json()}")

        response_data = response.json()

        if self.rate_limit_headers:
            response_headers = dict(response.headers)
            specific_headers = {
                REST_COMMON_FIELDS.get(key, key): response_headers.get(key, None)
                for key in RATE_LIMIT_HEADERS
            }

            response_data = {**response_data, **specific_headers}

        return response_data

    def set_headers(self, method, path):
        """Sets the request headers.

        Args:
            method (str): The HTTP method. Required.
            path (str): The URL path. Required.

        Returns:
            dict: The request headers.
        """

        return {
            "User-Agent": USER_AGENT,
            "Content-Type": "application/json",
        }

delete(url_path, params=None, data=None, **kwargs)

Sends a DELETE request.

Parameters:

Name	Type	Description	Default
url_path	str	The URL path. Required.	required
params	dict	The query parameters. Optional.	None
data	dict	The request body. Optional.	None
**kwargs		Additional arguments to pass to the request. Optional.	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The response data.

Source code in meridian/rest/http_client.py

def delete(
    self,
    url_path,
    params: Optional[dict] = None,
    data: Optional[dict] = None,
    **kwargs,
) -> Dict[str, Any]:
    """Sends a DELETE request.

    Args:
        url_path (str): The URL path. Required.
        params (dict, optional): The query parameters. Optional.
        data (dict, optional): The request body. Optional.
        **kwargs: Additional arguments to pass to the request. Optional.

    Returns:
        Dict[str, Any]: The response data.
    """
    data = data or {}

    if kwargs:
        data.update(kwargs)

    return self.prepare_and_send_request("DELETE", url_path, params, data)

get(url_path, params=None, **kwargs)

Sends a GET request.

Parameters:

Name	Type	Description	Default
url_path	str	The URL path. Required.	required
params	dict	The query parameters. Optional.	None
**kwargs		Additional arguments to pass to the request. Optional.	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The response data.

Source code in meridian/rest/http_client.py

def get(self, url_path, params: Optional[dict] = None, **kwargs) -> Dict[str, Any]:
    """Sends a GET request.

    Args:
        url_path (str): The URL path. Required.
        params (dict, optional): The query parameters. Optional.
        **kwargs: Additional arguments to pass to the request. Optional.

    Returns:
        Dict[str, Any]: The response data.
    """
    params = params or {}

    if kwargs:
        params.update(kwargs)

    return self.prepare_and_send_request("GET", url_path, params, data=None)

get_validated(url_path, request_model, response_model, **kwargs)

Sends a GET request including type validation of both the input and output from provided models.

Parameters:

Name	Type	Description	Default
url_path	str	The URL path. Required.	required
request_model	Type[BaseModel]	Pydantic model for request validation. Required.	required
response_model	Type[BaseModel]	Pydantic model for response validation. Required.	required
**kwargs		Includes all arguments to pass to the request. Optional.	{}

Returns:

Name	Type	Description
BaseModel	BaseModel	The response data, validated against the response_model.

Source code in meridian/rest/http_client.py

def get_validated(
    self,
    url_path,
    request_model: Type[BaseModel],
    response_model: Type[BaseModel],
    **kwargs,
) -> BaseModel:
    """Sends a GET request including type validation of both the input and output from provided models.

    Args:
        url_path (str): The URL path. Required.
        request_model (Type[BaseModel]): Pydantic model for request validation. Required.
        response_model (Type[BaseModel]): Pydantic model for response validation. Required.
        **kwargs: Includes all arguments to pass to the request. Optional.

    Returns:
        BaseModel: The response data, validated against the response_model.
    """
    try:
        validated_params = request_model.model_validate(kwargs, by_name=True)
    except ValidationError as e:
        raise ValidationException(
            "Request", url_path, request_model, e.errors()
        ) from None
    params = validated_params.model_dump(
        mode="json", exclude_none=True, by_alias=True
    )

    response_data = self.prepare_and_send_request(
        "GET", url_path, params, data=None
    )
    validated_response = response_model.model_validate(response_data)
    return validated_response

post(url_path, params=None, data=None, **kwargs)

Sends a POST request.

Parameters:

Name	Type	Description	Default
url_path	str	The URL path. Required.	required
params	dict	The query parameters. Optional.	None
data	dict	The request body. Optional.	None
**kwargs		Additional arguments to pass to the request. Optional.	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The response data.

Source code in meridian/rest/http_client.py

def post(
    self,
    url_path,
    params: Optional[dict] = None,
    data: Optional[dict] = None,
    **kwargs,
) -> Dict[str, Any]:
    """Sends a POST request.

    Args:
        url_path (str): The URL path. Required.
        params (dict, optional): The query parameters. Optional.
        data (dict, optional): The request body. Optional.
        **kwargs: Additional arguments to pass to the request. Optional.

    Returns:
        Dict[str, Any]: The response data.
    """
    data = data or {}

    if kwargs:
        data.update(kwargs)

    return self.prepare_and_send_request("POST", url_path, params, data)

prepare_and_send_request(http_method, url_path, params=None, data=None)

Prepares and sends an HTTP request.

Parameters:

Name	Type	Description	Default
http_method	str	The HTTP method. Required.	required
url_path	str	The URL path. Required.	required
params	dict	The query parameters. Optional.	None
data	dict	The request body. Optional.	None

Returns:

Type	Description
	Dict[str, Any]: The response data.

Source code in meridian/rest/http_client.py

def prepare_and_send_request(
    self,
    http_method,
    url_path,
    params: Optional[dict] = None,
    data: Optional[dict] = None,
):
    """Prepares and sends an HTTP request.

    Args:
        http_method (str): The HTTP method. Required.
        url_path (str): The URL path. Required.
        params (dict, optional): The query parameters. Optional.
        data (dict, optional): The request body. Optional.

    Returns:
        Dict[str, Any]: The response data.
    """
    headers = self.set_headers(http_method, url_path)

    if params is not None:
        params = {
            key: str(value).lower() if isinstance(value, bool) else value
            for key, value in params.items()
            if value is not None
        }

    if data is not None:
        data = {key: value for key, value in data.items() if value is not None}

    return self.send_request(http_method, url_path, params, headers, data=data)

put(url_path, params=None, data=None, **kwargs)

Sends a PUT request.

Parameters:

Name	Type	Description	Default
url_path	str	The URL path. Required.	required
params	dict	The query parameters. Optional.	None
data	dict	The request body. Optional.	None
**kwargs		Additional arguments to pass to the request. Optional.	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The response data.

Source code in meridian/rest/http_client.py

def put(
    self,
    url_path,
    params: Optional[dict] = None,
    data: Optional[dict] = None,
    **kwargs,
) -> Dict[str, Any]:
    """Sends a PUT request.

    Args:
        url_path (str): The URL path. Required.
        params (dict, optional): The query parameters. Optional.
        data (dict, optional): The request body. Optional.
        **kwargs: Additional arguments to pass to the request. Optional.

    Returns:
        Dict[str, Any]: The response data.
    """
    data = data or {}

    if kwargs:
        data.update(kwargs)

    return self.prepare_and_send_request("PUT", url_path, params, data)

send_request(http_method, url_path, params, headers, data=None)

Sends an HTTP request.

Parameters:

Name	Type	Description	Default
http_method	str	The HTTP method. Required.	required
url_path	str	The URL path. Required.	required
params	dict	The query parameters. Required.	required
headers	dict	The request headers. Required.	required
data	dict	The request body. Optional.	None

Returns:

Type	Description
	Dict[str, Any]: The response data.

Raises:

Type	Description
HTTPError	If the request fails.

Source code in meridian/rest/http_client.py

def send_request(self, http_method, url_path, params, headers, data=None):
    """Sends an HTTP request.

    Args:
        http_method (str): The HTTP method. Required.
        url_path (str): The URL path. Required.
        params (dict): The query parameters. Required.
        headers (dict): The request headers. Required.
        data (dict, optional): The request body. Optional.

    Returns:
        Dict[str, Any]: The response data.

    Raises:
        HTTPError: If the request fails.
    """
    if data is None:
        data = {}

    url = f"{self.base_url}{url_path}"

    self.logger.debug(f"Sending {http_method} request to {url}")

    response = self.session.request(
        http_method,
        url,
        params=params,
        json=data,
        headers=headers,
        timeout=self.timeout,
    )
    self._handle_exception(response)  # Raise an HTTPError for bad responses

    self.logger.debug(f"Raw response: {response.json()}")

    response_data = response.json()

    if self.rate_limit_headers:
        response_headers = dict(response.headers)
        specific_headers = {
            REST_COMMON_FIELDS.get(key, key): response_headers.get(key, None)
            for key in RATE_LIMIT_HEADERS
        }

        response_data = {**response_data, **specific_headers}

    return response_data

set_headers(method, path)

Sets the request headers.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method. Required.	required
path	str	The URL path. Required.	required

Returns:

Name	Type	Description
dict		The request headers.

Source code in meridian/rest/http_client.py

def set_headers(self, method, path):
    """Sets the request headers.

    Args:
        method (str): The HTTP method. Required.
        path (str): The URL path. Required.

    Returns:
        dict: The request headers.
    """

    return {
        "User-Agent": USER_AGENT,
        "Content-Type": "application/json",
    }