API Reference

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

Client Classes

RESTClient

The primary client for interacting with the Ethereal API via REST endpoints.

from ethereal import RESTClient

client = RESTClient({
    "private_key": "your_private_key",  # optional
    "base_url": "https://api.etherealtest.net"
})

ethereal.rest_client.RESTClient

Bases: HTTPClient

REST client for interacting with the Ethereal API.

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 "https://api.etherealtest.net". 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.	{}

Source code in ethereal/rest_client.py

class RESTClient(HTTPClient):
    """
    REST client for interacting with the Ethereal API.

    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 "https://api.etherealtest.net".
            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.
    """

    list_funding = list_funding
    get_projected_funding = get_projected_funding
    get_order = get_order
    list_fills = list_fills
    list_orders = list_orders
    list_trades = list_trades
    prepare_order = prepare_order
    sign_order = sign_order
    submit_order = submit_order
    dry_run_order = dry_run_order
    prepare_cancel_order = prepare_cancel_order
    sign_cancel_order = sign_cancel_order
    cancel_order = cancel_order
    get_signer = get_signer
    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
    list_positions = list_positions
    get_position = get_position
    get_market_liquidity = get_market_liquidity
    list_market_prices = list_market_prices
    list_products = list_products
    get_rpc_config = get_rpc_config
    list_subaccounts = list_subaccounts
    get_subaccount = get_subaccount
    get_subaccount_balances = get_subaccount_balances
    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
    withdraw_token = withdraw_token

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

        # fetch RPC configuration
        self.chain: Optional[ChainClient] = None
        self.rpc_config = self.get_rpc_config()
        if self.config.chain_config:
            self._init_chain_client(
                self.config.chain_config, self.rpc_config, self.tokens
            )
        self.private_key = self.chain.private_key if self.chain else None
        self.provider = self.chain.provider if self.chain else None

        self.default_time_in_force = self.config.default_time_in_force
        self.default_post_only = self.config.default_post_only

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

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

    def _get_pages(
        self,
        endpoint: str,
        request_model: Type[BaseModel],
        response_model: Type[BaseModel],
        paginate: bool = False,
        **kwargs,
    ) -> Any:
        """Make a GET request with validated parameters and response and handling for 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 (BaseModel): Pydantic model for request validation
            response_model (BaseModel): Pydantic model for response validation
            paginate (bool): Whether to fetch additional pages of data
            **kwargs: Parameters to validate and include in the request

        Returns:
            response (BaseModel): Validated response object

        Example:
            orders = client.validated_get(
                endpoint="order",
                request_model=V1OrderGetParametersQuery,
                response_model=PageOfOrderDtos,
                subaccount_id="abc123",
                limit=50
            )
        """
        result = self.get_validated(
            url_path=f"{API_PREFIX}/{endpoint}",
            request_model=request_model,
            response_model=response_model,
            **kwargs,
        )

        # If pagination is requested, fetch additional pages
        try:
            page_response = response_model.model_validate(result)
        except ValidationError as e:
            raise e
        if paginate:
            all_data = list(page_response.data)  # type: ignore

            # Continue fetching while there are more pages
            current_result = page_response
            while current_result.has_next and current_result.next_cursor:  # type: ignore
                new_result = self.get_validated(
                    url_path=f"{API_PREFIX}/{endpoint}",
                    request_model=request_model,
                    response_model=response_model,
                    cursor=current_result.next_cursor,  # type: ignore
                    **kwargs,
                )
                # Add data from this page
                current_result = response_model.model_validate(new_result)
                all_data.extend(current_result.data)  # type: ignore

            # Update the result with the combined data
            page_response.data = all_data  # type: ignore
            page_response.has_next = False  # type: ignore
            page_response.next_cursor = None  # type: ignore
        return page_response.data  # type: ignore

    @cached_property
    def subaccounts(self):
        """Get the list of subaccounts.

        Returns:
            subaccounts (List): List of subaccount objects.
        """
        return self.list_subaccounts(sender=self.chain.address)

    @cached_property
    def products(self):
        """Get the list of products.

        Returns:
            products (List): List of product objects.
        """
        return self.list_products()

    @cached_property
    def tokens(self):
        """Get the list of tokens.

        Returns:
            tokens (List): List of token objects.
        """
        return self.list_tokens()

    @cached_property
    def products_by_ticker(self):
        """Get the products indexed by ticker.

        Returns:
            products_by_ticker (Dict[str, ProductDto]): Dictionary of products keyed by ticker.
        """
        return {p.ticker: p for p in self.products}

    @cached_property
    def products_by_id(self):
        """Get the products indexed by ID.

        Returns:
            products_by_id (Dict[str, ProductDto]): Dictionary of products keyed by ID.
        """
        return {p.id: p for p in self.products}

    def create_order(
        self,
        order_type: str,
        quantity: float,
        side: int,
        price: Optional[float] = None,
        ticker: Optional[str] = None,
        product_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,
        otoco_trigger: Optional[bool] = None,
        otoco_group_id: Optional[str] = None,
        sign: bool = True,
        dry_run: bool = False,
        submit: bool = True,
    ) -> Union[SubmitOrderCreatedDto, OrderDryRunDto, SubmitOrderDto]:
        """Create and submit an order.

        Args:
            order_type (str): The type of order (market or limit)
            quantity (float): The quantity of the order
            side (int): The side of the order (0 = BUY, 1 = SELL)
            price (float, optional): The price of the order (for limit orders)
            ticker (str, optional): The ticker of the product
            product_id (str, optional): The ID of the product
            sender (str, optional): The sender address
            subaccount (str, optional): The subaccount name
            time_in_force (str, optional): The time in force for limit orders
            post_only (bool, optional): Whether the order is post-only (for limit orders)
            reduce_only (bool, optional): Whether the order is reduce only
            close (bool, optional): Whether the order is a close order
            stop_price (float, optional): The stop price for stop orders
            stop_type (int, optional): The stop type (0 = STOP, 1 = STOP_LIMIT)
            otoco_trigger (bool, optional): Whether the order is an OCO trigger
            otoco_group_id (str, optional): The OCO group ID
            dry_run (bool, optional): Whether to perform a dry run (no actual order submission)

        Returns:
            Union[OrderDto, OrderDryRunDto, SubmitOrderDto]: The created order object, dry run result, or result from order submission.

        Raises:
            ValueError: If neither product_id nor ticker is provided or if order type is invalid
        """
        # get the sender and account info
        if sender is None and self.chain:
            sender = self.chain.address
        if subaccount is None:
            subaccount = self.subaccounts[0].name

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

        # prepare the order params
        if order_type == "MARKET":
            order_params = {
                "sender": sender,
                "subaccount": subaccount,
                "side": side,
                "price": "0",
                "quantity": quantity,
                "onchain_id": onchain_id,
                "order_type": order_type,
                "reduce_only": reduce_only,
                "dryrun": dry_run,
                "close": close,
                "stop_price": stop_price,
                "stop_type": stop_type,
                "otoco_trigger": otoco_trigger,
                "otoco_group_id": otoco_group_id,
            }
        elif order_type == "LIMIT":
            order_params = {
                "sender": sender,
                "subaccount": subaccount,
                "side": side,
                "price": price,
                "quantity": quantity,
                "onchain_id": onchain_id,
                "order_type": order_type,
                "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,
                "close": close,
                "stop_price": stop_price,
                "stop_type": stop_type,
                "otoco_trigger": otoco_trigger,
                "otoco_group_id": otoco_group_id,
                "dryrun": dry_run,
            }
        else:
            raise ValueError("Invalid order type")

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

    def cancel_orders(
        self,
        order_ids: List[str],
        sender: str,
        subaccount: 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]): List of order UUIDs to cancel
            sender (str): Address of the sender
            subaccount (str): Subaccount address
            sign (bool, optional): Whether to sign the request. Defaults to True.
            submit (bool, optional): Whether to submit the request to the API. Defaults to True.

        Returns:
            Union[List[CancelOrderResultDto], CancelOrderDto]: List of cancellation results or the prepared cancel order data.
        """
        if len(order_ids) == 0:
            raise ValueError("No order IDs provided for cancellation")
        try:
            prepared_cancel = self.prepare_cancel_order(
                order_ids=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

        # Submit the cancellation request
        endpoint = f"{API_PREFIX}/order/cancel"
        response = self.post(
            endpoint,
            data=prepared_cancel.model_dump(
                mode="json", by_alias=True, exclude_none=True
            ),
            **kwargs,
        )
        return [
            CancelOrderResultDto.model_validate(item)
            for item in response.get("data", [])
        ]

    def cancel_all_orders(
        self,
        subaccount_id: str,
        product_ids: Optional[List[str]] = None,
        **kwargs,
    ) -> List[CancelOrderResultDto]:
        """
        Cancel all orders for a given subaccount.

        Args:
            subaccount_id (str): The ID of the subaccount.
            product_ids (List[str], optional): The IDs of the products to filter orders. If not provided, all orders will be canceled.
            **kwargs: Additional parameters for the request.

        Returns:
            List[CancelOrderResultDto]: The results of the cancel operations.
        """
        subaccount = self.get_subaccount(id=subaccount_id)
        query_params = {
            "subaccount_id": subaccount_id,
            "statuses": ["FILLED_PARTIAL", "NEW", "PENDING", "SUBMITTED"],
            **kwargs,
        }
        if product_ids:
            query_params["product_ids"] = product_ids

        orders = self._get_pages(
            endpoint="order",
            request_model=V1OrderGetParametersQuery,
            response_model=PageOfOrderDtos,
            paginate=True,
            **query_params,
        )
        order_ids = [order.id for order in orders]

        # cancel all orders
        if len(order_ids) == 0:
            raise ValueError("No order IDs provided for cancellation")
        cancel_results = self.cancel_orders(
            order_ids=order_ids,
            sender=subaccount.account,
            subaccount=subaccount.name,
            sign=True,
            submit=True,
        )
        if not isinstance(cancel_results, list):
            raise ValueError("Failed to cancel orders")
        return cancel_results

    def replace_order(
        self,
        order: Optional[OrderDto] = None,
        order_id: Optional[str] = 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.

        Args:
            order (OrderDto, optional): The order to replace.
            order_id (str, optional): The ID of the order to replace.
            quantity (float, optional): The new quantity of the order.
            price (float, optional): The new price of the order (for limit orders).
            time_in_force (str, optional): The time in force for limit orders.
            post_only (bool, optional): Whether the order is post-only (for limit orders).
            reduce_only (bool, optional): Whether the order is reduce only.

        Returns:
            Tuple[OrderDto, bool]: The response data from the API and a success flag for the cancel operation.

        Raises:
            ValueError: If neither order nor order_id is provided, or if both are provided.
        """
        # get the order info
        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 = self.get_order(id=order_id)
        subaccount = self.get_subaccount(id=old_order.subaccount_id)

        # set default values
        if quantity is None:
            quantity = float(old_order.quantity)
        if price is None:
            price = float(old_order.price)
        if time_in_force is None:
            time_in_force = (
                old_order.time_in_force.value if old_order.time_in_force else None
            )
        if post_only is None:
            post_only = old_order.post_only
        if reduce_only is None:
            reduce_only = old_order.reduce_only

        # cancel the old order
        cancel_result = 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}"
            )

        # create the new order params
        new_order = 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 SubmitOrderCreatedDto.model_validate(
            new_order
        ), canceled_order.result.value == "Ok"

products cached property

Get the list of products.

Returns:

Name	Type	Description
products	List	List of product objects.

products_by_id cached property

Get the products indexed by ID.

Returns:

Name	Type	Description
products_by_id	Dict[str, ProductDto]	Dictionary of products keyed by ID.

products_by_ticker cached property

Get the products indexed by ticker.

Returns:

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

subaccounts cached property

Get the list of subaccounts.

Returns:

Name	Type	Description
subaccounts	List	List of subaccount objects.

tokens cached property

Get the list of tokens.

Returns:

Name	Type	Description
tokens	List	List of token objects.

cancel_all_orders(subaccount_id, product_ids=None, **kwargs)

Cancel all orders for a given subaccount.

Parameters:

Name	Type	Description	Default
subaccount_id	str	The ID of the subaccount.	required
product_ids	List[str]	The IDs of the products to filter orders. If not provided, all orders will be canceled.	None
**kwargs		Additional parameters for the request.	{}

Returns:

Type	Description
List[CancelOrderResultDto]	List[CancelOrderResultDto]: The results of the cancel operations.

Source code in ethereal/rest_client.py

def cancel_all_orders(
    self,
    subaccount_id: str,
    product_ids: Optional[List[str]] = None,
    **kwargs,
) -> List[CancelOrderResultDto]:
    """
    Cancel all orders for a given subaccount.

    Args:
        subaccount_id (str): The ID of the subaccount.
        product_ids (List[str], optional): The IDs of the products to filter orders. If not provided, all orders will be canceled.
        **kwargs: Additional parameters for the request.

    Returns:
        List[CancelOrderResultDto]: The results of the cancel operations.
    """
    subaccount = self.get_subaccount(id=subaccount_id)
    query_params = {
        "subaccount_id": subaccount_id,
        "statuses": ["FILLED_PARTIAL", "NEW", "PENDING", "SUBMITTED"],
        **kwargs,
    }
    if product_ids:
        query_params["product_ids"] = product_ids

    orders = self._get_pages(
        endpoint="order",
        request_model=V1OrderGetParametersQuery,
        response_model=PageOfOrderDtos,
        paginate=True,
        **query_params,
    )
    order_ids = [order.id for order in orders]

    # cancel all orders
    if len(order_ids) == 0:
        raise ValueError("No order IDs provided for cancellation")
    cancel_results = self.cancel_orders(
        order_ids=order_ids,
        sender=subaccount.account,
        subaccount=subaccount.name,
        sign=True,
        submit=True,
    )
    if not isinstance(cancel_results, list):
        raise ValueError("Failed to cancel orders")
    return cancel_results

cancel_orders(order_ids, sender, subaccount, sign=True, submit=True, **kwargs)

Prepares and optionally submits a request to cancel multiple orders.

Parameters:

Name	Type	Description	Default
order_ids	List[str]	List of order UUIDs to cancel	required
sender	str	Address of the sender	required
subaccount	str	Subaccount address	required
sign	bool	Whether to sign the request. Defaults to True.	True
submit	bool	Whether to submit the request to the API. Defaults to True.	True

Returns:

Type	Description
Union[List[CancelOrderResultDto], CancelOrderDto]	Union[List[CancelOrderResultDto], CancelOrderDto]: List of cancellation results or the prepared cancel order data.

Source code in ethereal/rest_client.py

def cancel_orders(
    self,
    order_ids: List[str],
    sender: str,
    subaccount: 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]): List of order UUIDs to cancel
        sender (str): Address of the sender
        subaccount (str): Subaccount address
        sign (bool, optional): Whether to sign the request. Defaults to True.
        submit (bool, optional): Whether to submit the request to the API. Defaults to True.

    Returns:
        Union[List[CancelOrderResultDto], CancelOrderDto]: List of cancellation results or the prepared cancel order data.
    """
    if len(order_ids) == 0:
        raise ValueError("No order IDs provided for cancellation")
    try:
        prepared_cancel = self.prepare_cancel_order(
            order_ids=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

    # Submit the cancellation request
    endpoint = f"{API_PREFIX}/order/cancel"
    response = self.post(
        endpoint,
        data=prepared_cancel.model_dump(
            mode="json", by_alias=True, exclude_none=True
        ),
        **kwargs,
    )
    return [
        CancelOrderResultDto.model_validate(item)
        for item in response.get("data", [])
    ]

create_order(order_type, quantity, side, price=None, ticker=None, product_id=None, sender=None, subaccount=None, time_in_force=None, post_only=None, reduce_only=False, close=None, stop_price=None, stop_type=None, otoco_trigger=None, otoco_group_id=None, sign=True, dry_run=False, submit=True)

Create and submit an order.

Parameters:

Name	Type	Description	Default
order_type	str	The type of order (market or limit)	required
quantity	float	The quantity of the order	required
side	int	The side of the order (0 = BUY, 1 = SELL)	required
price	float	The price of the order (for limit orders)	None
ticker	str	The ticker of the product	None
product_id	str	The ID of the product	None
sender	str	The sender address	None
subaccount	str	The subaccount name	None
time_in_force	str	The time in force for limit orders	None
post_only	bool	Whether the order is post-only (for limit orders)	None
reduce_only	bool	Whether the order is reduce only	False
close	bool	Whether the order is a close order	None
stop_price	float	The stop price for stop orders	None
stop_type	int	The stop type (0 = STOP, 1 = STOP_LIMIT)	None
otoco_trigger	bool	Whether the order is an OCO trigger	None
otoco_group_id	str	The OCO group ID	None
dry_run	bool	Whether to perform a dry run (no actual order submission)	False

Returns:

Type	Description
Union[SubmitOrderCreatedDto, OrderDryRunDto, SubmitOrderDto]	Union[OrderDto, OrderDryRunDto, SubmitOrderDto]: The created order object, dry run result, or result from order submission.

Raises:

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

Source code in ethereal/rest_client.py

def create_order(
    self,
    order_type: str,
    quantity: float,
    side: int,
    price: Optional[float] = None,
    ticker: Optional[str] = None,
    product_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,
    otoco_trigger: Optional[bool] = None,
    otoco_group_id: Optional[str] = None,
    sign: bool = True,
    dry_run: bool = False,
    submit: bool = True,
) -> Union[SubmitOrderCreatedDto, OrderDryRunDto, SubmitOrderDto]:
    """Create and submit an order.

    Args:
        order_type (str): The type of order (market or limit)
        quantity (float): The quantity of the order
        side (int): The side of the order (0 = BUY, 1 = SELL)
        price (float, optional): The price of the order (for limit orders)
        ticker (str, optional): The ticker of the product
        product_id (str, optional): The ID of the product
        sender (str, optional): The sender address
        subaccount (str, optional): The subaccount name
        time_in_force (str, optional): The time in force for limit orders
        post_only (bool, optional): Whether the order is post-only (for limit orders)
        reduce_only (bool, optional): Whether the order is reduce only
        close (bool, optional): Whether the order is a close order
        stop_price (float, optional): The stop price for stop orders
        stop_type (int, optional): The stop type (0 = STOP, 1 = STOP_LIMIT)
        otoco_trigger (bool, optional): Whether the order is an OCO trigger
        otoco_group_id (str, optional): The OCO group ID
        dry_run (bool, optional): Whether to perform a dry run (no actual order submission)

    Returns:
        Union[OrderDto, OrderDryRunDto, SubmitOrderDto]: The created order object, dry run result, or result from order submission.

    Raises:
        ValueError: If neither product_id nor ticker is provided or if order type is invalid
    """
    # get the sender and account info
    if sender is None and self.chain:
        sender = self.chain.address
    if subaccount is None:
        subaccount = self.subaccounts[0].name

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

    # prepare the order params
    if order_type == "MARKET":
        order_params = {
            "sender": sender,
            "subaccount": subaccount,
            "side": side,
            "price": "0",
            "quantity": quantity,
            "onchain_id": onchain_id,
            "order_type": order_type,
            "reduce_only": reduce_only,
            "dryrun": dry_run,
            "close": close,
            "stop_price": stop_price,
            "stop_type": stop_type,
            "otoco_trigger": otoco_trigger,
            "otoco_group_id": otoco_group_id,
        }
    elif order_type == "LIMIT":
        order_params = {
            "sender": sender,
            "subaccount": subaccount,
            "side": side,
            "price": price,
            "quantity": quantity,
            "onchain_id": onchain_id,
            "order_type": order_type,
            "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,
            "close": close,
            "stop_price": stop_price,
            "stop_type": stop_type,
            "otoco_trigger": otoco_trigger,
            "otoco_group_id": otoco_group_id,
            "dryrun": dry_run,
        }
    else:
        raise ValueError("Invalid order type")

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

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

Replace an existing order.

Parameters:

Name	Type	Description	Default
order	OrderDto	The order to replace.	None
order_id	str	The ID of the order to replace.	None
quantity	float	The new quantity of the order.	None
price	float	The new price of the order (for limit orders).	None
time_in_force	str	The time in force for limit orders.	None
post_only	bool	Whether the order is post-only (for limit orders).	None
reduce_only	bool	Whether the order is reduce only.	False

Returns:

Type	Description
Tuple[SubmitOrderCreatedDto, bool]	Tuple[OrderDto, bool]: The response data from the API and a success flag for the cancel operation.

Raises:

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

Source code in ethereal/rest_client.py

def replace_order(
    self,
    order: Optional[OrderDto] = None,
    order_id: Optional[str] = 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.

    Args:
        order (OrderDto, optional): The order to replace.
        order_id (str, optional): The ID of the order to replace.
        quantity (float, optional): The new quantity of the order.
        price (float, optional): The new price of the order (for limit orders).
        time_in_force (str, optional): The time in force for limit orders.
        post_only (bool, optional): Whether the order is post-only (for limit orders).
        reduce_only (bool, optional): Whether the order is reduce only.

    Returns:
        Tuple[OrderDto, bool]: The response data from the API and a success flag for the cancel operation.

    Raises:
        ValueError: If neither order nor order_id is provided, or if both are provided.
    """
    # get the order info
    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 = self.get_order(id=order_id)
    subaccount = self.get_subaccount(id=old_order.subaccount_id)

    # set default values
    if quantity is None:
        quantity = float(old_order.quantity)
    if price is None:
        price = float(old_order.price)
    if time_in_force is None:
        time_in_force = (
            old_order.time_in_force.value if old_order.time_in_force else None
        )
    if post_only is None:
        post_only = old_order.post_only
    if reduce_only is None:
        reduce_only = old_order.reduce_only

    # cancel the old order
    cancel_result = 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}"
        )

    # create the new order params
    new_order = 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 SubmitOrderCreatedDto.model_validate(
        new_order
    ), canceled_order.result.value == "Ok"

ethereal.rest

funding

get_projected_funding(self, **kwargs)

Gets the projected funding rate for a product for the next hour.

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
ProjectedFundingDto	ProjectedFundingDto	Projected funding rate for the next hour for the product.

Source code in ethereal/rest/funding.py

def get_projected_funding(self, **kwargs) -> ProjectedFundingDto:
    """Gets the projected funding rate for a product for the next hour.

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

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

    Returns:
        ProjectedFundingDto: Projected funding rate for the next hour for the product.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/funding/projected",
        request_model=V1FundingProjectedGetParametersQuery,
        response_model=ProjectedFundingDto,
        **kwargs,
    )
    return res

list_funding(self, **kwargs)

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	The range of time of funding rates to retrieve. 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]: List of funding rate history objects for the product.

Source code in ethereal/rest/funding.py

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): The range of time of funding rates to retrieve. 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]: List of funding rate history objects for the product.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/funding",
        request_model=V1FundingGetParametersQuery,
        response_model=PageOfFundingDtos,
        **kwargs,
    )
    data = [FundingDto(**model.model_dump(by_alias=True)) for model in res.data]
    return data

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 ethereal/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 ethereal/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 ethereal/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 ethereal/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 ethereal/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 ethereal/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 ethereal/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 ethereal/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 ethereal/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",
    }

linked_signer

get_signer(self, id, **kwargs)

Gets a specific linked signer by ID.

Parameters:

Name	Type	Description	Default
id	str	UUID of the 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 information.

Source code in ethereal/rest/linked_signer.py

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

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

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

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

get_signer_quota(self, **kwargs)

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	Account signer quota information.

Source code in ethereal/rest/linked_signer.py

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: Account signer quota information.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/linked-signer/quota",
        request_model=V1LinkedSignerQuotaGetParametersQuery,
        response_model=AccountSignerQuotaDto,
        **kwargs,
    )
    return res

link_linked_signer(self, dto, **kwargs)

Submits a prepared and signed LinkSignerDto to link a signer.

Source code in ethereal/rest/linked_signer.py

def link_linked_signer(
    self,
    dto: LinkSignerDto,
    **kwargs,
) -> SignerDto:
    """Submits a prepared and signed LinkSignerDto to link a signer."""
    endpoint = f"{API_PREFIX}/linked-signer/link"
    res = self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return SignerDto.model_validate(res)

list_signers(self, **kwargs)

Lists all 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.
active	bool	Filter for active signers. 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]: List of linked signers for the subaccount.

Source code in ethereal/rest/linked_signer.py

def list_signers(
    self,
    **kwargs,
) -> List[SignerDto]:
    """Lists all 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.
        active (bool, optional): Filter for active signers. 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]: List of linked signers for the subaccount.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/linked-signer",
        request_model=V1LinkedSignerGetParametersQuery,
        response_model=PageOfSignersDto,
        **kwargs,
    )
    data = [SignerDto(**model.model_dump(by_alias=True)) for model in res.data]
    return data

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

Prepares the data for linking a signer without signing or submitting.

Parameters:

Name	Type	Description	Default
sender	str	Address of the sender. Required.	required
signer	str	Address of the signer to be linked. Required.	required
subaccount	str	Address of the subaccount. Required.	required
subaccount_id	str	UUID of the subaccount. Required.	required

Returns:

Name	Type	Description
LinkSignerDto	LinkSignerDto	DTO containing the data model and signatures.

Source code in ethereal/rest/linked_signer.py

def prepare_linked_signer(
    self,
    sender: str,
    signer: str,
    subaccount: str,
    subaccount_id: str,
    signer_signature: str = "",
    include_signature: bool = False,
    **kwargs,
) -> LinkSignerDto:
    """Prepares the data for linking a signer without signing or submitting.

    Args:
        sender (str): Address of the sender. Required.
        signer (str): Address of the signer to be linked. Required.
        subaccount (str): Address of the subaccount. Required.
        subaccount_id (str): UUID of the subaccount. Required.

    Returns:
        LinkSignerDto: DTO containing the data model and signatures.
    """
    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 = LinkSignerDtoData.model_validate(data)

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

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

Prepares the data for revoking a linked signer without signing or submitting.

Parameters:

Name	Type	Description	Default
sender	str	Address of the sender. Required.	required
signer	str	Address of the signer to be revoked. Required.	required
subaccount	str	Address of the subaccount. Required.	required
subaccount_id	str	UUID of the subaccount. Required.	required

Returns:

Name	Type	Description
RevokeLinkedSignerDto	RevokeLinkedSignerDto	DTO containing the data model and signatures.

Source code in ethereal/rest/linked_signer.py

def prepare_revoke_linked_signer(
    self,
    sender: str,
    signer: str,
    subaccount: str,
    subaccount_id: str,
    include_signature: bool = False,
    **kwargs,
) -> RevokeLinkedSignerDto:
    """Prepares the data for revoking a linked signer without signing or submitting.

    Args:
        sender (str): Address of the sender. Required.
        signer (str): Address of the signer to be revoked. Required.
        subaccount (str): Address of the subaccount. Required.
        subaccount_id (str): UUID of the subaccount. Required.

    Returns:
        RevokeLinkedSignerDto: DTO containing the data model and signatures.
    """
    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 = RevokeLinkedSignerDtoData.model_validate(data)
    dto = RevokeLinkedSignerDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = self.sign_revoke_linked_signer(dto)
    return dto

revoke_linked_signer(self, dto, **kwargs)

Submits a prepared and signed RevokeLinkedSignerDto to revoke a linked signer.

Source code in ethereal/rest/linked_signer.py

def revoke_linked_signer(
    self,
    dto: RevokeLinkedSignerDto,
    **kwargs,
) -> SignerDto:
    """Submits a prepared and signed RevokeLinkedSignerDto to revoke a linked signer."""
    endpoint = f"{API_PREFIX}/linked-signer/revoke"
    res = self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return SignerDto.model_validate(res)

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

Signs the data for linking a signer without submitting.

This function signs the prepared data for linking a signer. The message is prepared and signed with the private keys provided. If no private key is provided for either the signer or the sender, the signature will remain empty. If both private keys are provided, or the sender's private key is available in the chain client, the message will be signed and the signature will be included in the returned DTO.

Parameters:

Name	Type	Description	Default
link_to_sign	LinkSignerDto	The prepared LinkSignerDto from prepare_link_signer	required
signer_private_key	Optional[str]	The private key of the signer being linked. Optional.	None
private_key	Optional[str]	The private key of the sender. Optional.	None

Returns:

Name	Type	Description
LinkSignerDto	LinkSignerDto	DTO containing the data model and signature

Raises:

Type	Description
ValueError	If the chain client or private key is not available

Source code in ethereal/rest/linked_signer.py

def sign_linked_signer(
    self,
    link_to_sign: LinkSignerDto,
    signer_private_key: Optional[str] = None,
    private_key: Optional[str] = None,
) -> LinkSignerDto:
    """Signs the data for linking a signer without submitting.

    This function signs the prepared data for linking a signer. The message is prepared
    and signed with the private keys provided. If no private key is provided for either
    the signer or the sender, the signature will remain empty. If both private keys are
    provided, or the sender's private key is available in the chain client, the message will be
    signed and the signature will be included in the returned DTO.

    Args:
        link_to_sign: The prepared LinkSignerDto from prepare_link_signer
        signer_private_key: The private key of the signer being linked. Optional.
        private_key: The private key of the sender. Optional.

    Returns:
        LinkSignerDto: DTO containing the data model and signature

    Raises:
        ValueError: If the chain client or private key is not 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 self.chain.private_key:
        signature = self.chain.sign_message(
            private_key, domain, types, primary_type, message
        )
        link_to_sign.signature = signature
    return link_to_sign

sign_revoke_linked_signer(self, revoke_to_sign)

Signs the data for revoking a linked signer without submitting.

Parameters:

Name	Type	Description	Default
revoke_to_sign	RevokeLinkedSignerDto	The prepared RevokeLinkedSignerDto from prepare_revoke_linked_signer	required

Returns:

Name	Type	Description
RevokeLinkedSignerDto	RevokeLinkedSignerDto	DTO containing the data model and signature

Raises:

Type	Description
ValueError	If the chain client or private key is not available

Source code in ethereal/rest/linked_signer.py

def sign_revoke_linked_signer(
    self,
    revoke_to_sign: RevokeLinkedSignerDto,
) -> RevokeLinkedSignerDto:
    """Signs the data for revoking a linked signer without submitting.

    Args:
        revoke_to_sign: The prepared RevokeLinkedSignerDto from prepare_revoke_linked_signer

    Returns:
        RevokeLinkedSignerDto: DTO containing the data model and signature

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

    # 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(
        self.chain.private_key, domain, types, primary_type, message
    )
    return revoke_to_sign

order

cancel_order(self, order_to_cancel, **kwargs)

Submits a prepared and signed cancel order.

Parameters:

Name	Type	Description	Default
order_to_cancel	CancelOrderDto	Prepared and signed cancel order data.	required

Returns:

Name	Type	Description
CancelOrderResultDto	CancelOrderResultDto	Response containing the cancellation result.

Source code in ethereal/rest/order.py

def cancel_order(
    self,
    order_to_cancel: CancelOrderDto,
    **kwargs,
) -> CancelOrderResultDto:
    """Submits a prepared and signed cancel order.

    Args:
        order_to_cancel (CancelOrderDto): Prepared and signed cancel order data.

    Returns:
        CancelOrderResultDto: Response containing the cancellation result.
    """
    endpoint = f"{API_PREFIX}/order/cancel"
    res = self.post(
        endpoint,
        data=order_to_cancel.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return CancelOrderResultDto.model_validate(res)

dry_run_order(self, order, **kwargs)

Submits a prepared order for a dry run.

Parameters:

Name	Type	Description	Default
order	SubmitOrderDto	Prepared and signed order data.	required

Returns:

Name	Type	Description
OrderDryRunDto	OrderDryRunDto	Response containing the dry run order information.

Source code in ethereal/rest/order.py

def dry_run_order(
    self,
    order: SubmitOrderDto,
    **kwargs,
) -> OrderDryRunDto:
    """Submits a prepared order for a dry run.

    Args:
        order (SubmitOrderDto): Prepared and signed order data.

    Returns:
        OrderDryRunDto: Response containing the dry run order information.
    """
    submit_payload = SubmitDryOrderDto.model_validate(
        {"data": order.data.model_dump(mode="json", by_alias=True)}
    )
    endpoint = f"{API_PREFIX}/order/dry-run"
    res = self.post(
        endpoint,
        data=submit_payload.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return OrderDryRunDto.model_validate(res)

get_order(self, id, **kwargs)

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 information.

Source code in ethereal/rest/order.py

def get_order(self, id: str, **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 information.
    """
    endpoint = f"{API_PREFIX}/order/{id}"
    response = self.get(endpoint, **kwargs)
    return OrderDto(**response)

list_fills(self, **kwargs)

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]	List of product UUIDs to filter. Optional.
created_after	float	Filter for fills created after this timestamp. Optional.
created_before	float	Filter for fills created before this timestamp. Optional.
side	int	Fill side (BUY = 0, SELL = 1). Optional.
statuses	List[str]	List of fill statuses. Optional.
order_by	str	Field to order by, e.g., 'productId'. 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.
include_self_trades	bool	Whether to include self trades. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
PageOfOrderFillDtos	List[OrderFillDto]	List of order fill information.

Source code in ethereal/rest/order.py

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): List of product UUIDs to filter. Optional.
        created_after (float, optional): Filter for fills created after this timestamp. Optional.
        created_before (float, optional): Filter for fills created before this timestamp. Optional.
        side (int, optional): Fill side (BUY = 0, SELL = 1). Optional.
        statuses (List[str], optional): List of fill statuses. Optional.
        order_by (str, optional): Field to order by, e.g., 'productId'. 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.
        include_self_trades (bool, optional): Whether to include self trades. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

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

list_orders(self, **kwargs)

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]	List of product UUIDs to filter. Optional.
created_after	float	Filter for orders created after this timestamp. Optional.
created_before	float	Filter for orders created before this timestamp. Optional.
side	int	Order side (BUY = 0, SELL = 1). Optional.
close	bool	Filter for close orders. Optional.
stop_types	List[int]	List of stop types. Optional.
statuses	List[str]	List of order statuses. Optional.
order_by	str	Field to order by, e.g., 'type'. 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.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
PageOfOrderDtos	List[OrderDto]	List of order information.

Source code in ethereal/rest/order.py

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): List of product UUIDs to filter. Optional.
        created_after (float, optional): Filter for orders created after this timestamp. Optional.
        created_before (float, optional): Filter for orders created before this timestamp. Optional.
        side (int, optional): Order side (BUY = 0, SELL = 1). Optional.
        close (bool, optional): Filter for close orders. Optional.
        stop_types (List[int], optional): List of stop types. Optional.
        statuses (List[str], optional): List of order statuses. Optional.
        order_by (str, optional): Field to order by, e.g., 'type'. 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.
        **kwargs: Additional request parameters accepted by the API. Optional.

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

list_trades(self, **kwargs)

Lists trades for a product if specified, otherwise lists trades for all products.

Other Parameters:

Name	Type	Description
product_id	str	UUID of the product. Optional.
order_by	str	Field to order by, e.g., 'createdAt'. 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.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Name	Type	Description
PageOfTradeDtos	List[TradeDto]	List of trade information.

Source code in ethereal/rest/order.py

def list_trades(
    self,
    **kwargs,
) -> List[TradeDto]:
    """Lists trades for a product if specified, otherwise lists trades for all products.

    Other Parameters:
        product_id (str, optional): UUID of the product. Optional.
        order_by (str, optional): Field to order by, e.g., 'createdAt'. 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.
        **kwargs: Additional request parameters accepted by the API. Optional.

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

prepare_cancel_order(self, order_ids, sender, subaccount, include_signature=False, **kwargs)

Prepares the data model and optionally the signature for cancelling orders.

Parameters:

Name	Type	Description	Default
order_ids	List[str]	List of order UUIDs to cancel	required
sender	str	Address of the sender	required
subaccount	str	Subaccount address	required
include_signature	bool	Whether to include the signature. Defaults to False.	False

Returns:

Type	Description
CancelOrderDto	Dict[str, Any]: Dictionary containing the data model and optionally the signature.

Raises:

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

Source code in ethereal/rest/order.py

def prepare_cancel_order(
    self,
    order_ids: List[str],
    sender: str,
    subaccount: str,
    include_signature: bool = False,
    **kwargs,
) -> CancelOrderDto:
    """Prepares the data model and optionally the signature for cancelling orders.

    Args:
        order_ids (List[str]): List of order UUIDs to cancel
        sender (str): Address of the sender
        subaccount (str): Subaccount address
        include_signature (bool, optional): Whether to include the signature. Defaults to False.

    Returns:
        Dict[str, Any]: Dictionary containing the data model and optionally the signature.

    Raises:
        ValueError: If include_signature is True and no chain client or private key is available.
    """
    nonce = kwargs.get("nonce", None) or generate_nonce()

    # Create the data model
    data_model = CancelOrderDtoData(
        sender=sender,
        subaccount=subaccount,
        nonce=nonce,
        orderIds=order_ids,
    )
    result = CancelOrderDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        return self.sign_cancel_order(result)
    return result

prepare_order(self, sender, price, quantity, side, subaccount, onchain_id, order_type, time_in_force=None, post_only=False, reduce_only=False, close=None, stop_price=None, stop_type=None, otoco_trigger=None, otoco_group_id=None, include_signature=False, **kwargs)

Prepares the data model and optionally the signature for an order submission.

Parameters:

Name	Type	Description	Default
sender	str	Address of the sender	required
price	str	Order price	required
quantity	str	Order quantity	required
side	int	Order side (BUY = 0, SELL = 1)	required
subaccount	str	Subaccount address	required
onchain_id	float	On-chain ID of the product	required
order_type	str	Order type (LIMIT or MARKET)	required
time_in_force	str	Time in force for limit orders. Defaults to None.	None
post_only	bool	Whether the order is post-only. Defaults to False.	False
reduce_only	bool	Whether the order is reduce-only. Defaults to False.	False
close	bool	Whether the order closes a position. Defaults to None.	None
stop_price	float	Stop price for stop orders. Defaults to None.	None
stop_type	int	Stop type for stop orders. Defaults to None.	None
otoco_trigger	bool	Whether this is an OTOCO trigger order. Defaults to None.	None
otoco_group_id	str	OTOCO group ID. Defaults to None.	None
include_signature	bool	Whether to include the signature. Defaults to False.	False

Returns:

Type	Description
SubmitOrderDto	Dict[str, Any]: Dictionary containing the data model and optionally the signature.

Raises:

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

Source code in ethereal/rest/order.py

def prepare_order(
    self,
    sender: str,
    price: Union[str, float, Decimal],
    quantity: Union[str, float, Decimal],
    side: int,
    subaccount: str,
    onchain_id: float,
    order_type: str,
    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,
    otoco_trigger: Optional[bool] = None,
    otoco_group_id: Optional[str] = None,
    include_signature: bool = False,
    **kwargs,
) -> SubmitOrderDto:
    """Prepares the data model and optionally the signature for an order submission.

    Args:
        sender (str): Address of the sender
        price (str): Order price
        quantity (str): Order quantity
        side (int): Order side (BUY = 0, SELL = 1)
        subaccount (str): Subaccount address
        onchain_id (float): On-chain ID of the product
        order_type (str): Order type (LIMIT or MARKET)
        time_in_force (str, optional): Time in force for limit orders. Defaults to None.
        post_only (bool, optional): Whether the order is post-only. Defaults to False.
        reduce_only (bool, optional): Whether the order is reduce-only. Defaults to False.
        close (bool, optional): Whether the order closes a position. Defaults to None.
        stop_price (float, optional): Stop price for stop orders. Defaults to None.
        stop_type (int, optional): Stop type for stop orders. Defaults to None.
        otoco_trigger (bool, optional): Whether this is an OTOCO trigger order. Defaults to None.
        otoco_group_id (str, optional): OTOCO group ID. Defaults to None.
        include_signature (bool, optional): Whether to include the signature. Defaults to False.

    Returns:
        Dict[str, Any]: Dictionary containing the data model and optionally the signature.

    Raises:
        ValueError: If include_signature is True and no chain client or private key is available.
    """
    # 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
    order_data = {
        "sender": sender,
        "subaccount": subaccount,
        "quantity": quantity,
        "price": price,
        "side": side,
        "engineType": 0,
        "onchainId": onchain_id,
        "nonce": nonce,
        "type": order_type,
        "reduceOnly": reduce_only,
        "signedAt": signed_at,
        "close": close,
        "stopPrice": stop_price,
        "stopType": stop_type,
        "otocoTrigger": otoco_trigger,
        "otocoGroupId": otoco_group_id,
    }

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

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

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

    if include_signature:
        result = self.sign_order(result)

    return result

sign_cancel_order(self, order_to_cancel, private_key=None)

Signs the cancel order data model.

Parameters:

Name	Type	Description	Default
order_to_cancel	CancelOrderDto	The order data model to sign.	required
private_key	Optional[str]	Private key for signing. If None, uses the instance's private key.	None

Returns:

Name	Type	Description
CancelOrderDto	CancelOrderDto	The signed order data model.

Source code in ethereal/rest/order.py

def sign_cancel_order(
    self,
    order_to_cancel: CancelOrderDto,
    private_key: Optional[str] = None,
) -> CancelOrderDto:
    """Signs the cancel order data model.

    Args:
        order_to_cancel (CancelOrderDto): The order data model to sign.
        private_key (Optional[str]): Private key for signing. If None, uses the instance's private key.

    Returns:
        CancelOrderDto: The signed order data model.
    """
    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(order_id) for order_id in order_to_cancel.data.order_ids
    ]
    message["orderIds"] = 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(
        self.chain.private_key, domain, types, primary_type, message
    )
    return order_to_cancel

sign_order(self, order, private_key=None)

Signs the order data using the chain client.

Parameters:

Name	Type	Description	Default
order	SubmitOrderDto	Order data to sign.	required
private_key	Optional[str]	Private key for signing. If None, uses the instance's private key.	None

Returns:

Name	Type	Description
str	SubmitOrderDto	Signature of the order data.

Raises:

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

Source code in ethereal/rest/order.py

def sign_order(
    self, order: SubmitOrderDto, private_key: Optional[str] = None
) -> SubmitOrderDto:
    """Signs the order data using the chain client.

    Args:
        order (SubmitOrderDto): Order data to sign.
        private_key (Optional[str]): Private key for signing. If None, uses the instance's private key.

    Returns:
        str: Signature of the order data.

    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

    # Update message signedAt
    order.data.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"])

    # 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(
        self.chain.private_key, domain, types, primary_type, message
    )
    return order

submit_order(self, order, **kwargs)

Submits a prepared and signed order.

Parameters:

Name	Type	Description	Default
order	SubmitOrderDto	Prepared and signed order data.	required

Returns:

Name	Type	Description
SubmitOrderCreatedDto	SubmitOrderCreatedDto	Response containing the order information.

Source code in ethereal/rest/order.py

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

    Args:
        order (SubmitOrderDto): Prepared and signed order data.

    Returns:
        SubmitOrderCreatedDto: Response containing the order information.
    """
    endpoint = f"{API_PREFIX}/order"
    res = self.post(
        endpoint,
        data=order.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return SubmitOrderCreatedDto.model_validate(res)

position

get_position(self, id, **kwargs)

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 ethereal/rest/position.py

def get_position(
    self,
    id: str,
    **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 = self.get(endpoint, **kwargs)
    return PositionDto(**res)

list_positions(self, **kwargs)

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_id	str	UUID of the product to filter by. Optional.
open	bool	Filter for open 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', or 'updatedAt'. 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 ethereal/rest/position.py

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

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

    Other Parameters:
        product_id (str, optional): UUID of the product to filter by. Optional.
        open (bool, optional): Filter for open 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', or 'updatedAt'. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

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

product

get_market_liquidity(self, **kwargs)

Gets market liquidity for a product.

Parameters:

Name	Type	Description	Default
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
MarketLiquidityDto	MarketLiquidityDto	Market liquidity information.

Source code in ethereal/rest/product.py

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

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

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

    Returns:
        MarketLiquidityDto: Market liquidity information.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/product/market-liquidity",
        request_model=V1ProductMarketLiquidityGetParametersQuery,
        response_model=MarketLiquidityDto,
        **kwargs,
    )
    return res

list_market_prices(self, **kwargs)

Gets market prices for multiple products.

Parameters:

Name	Type	Description	Default
product_ids	List[str]	List of product UUIDs. Required.	required

Other Parameters:

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

Returns:

Type	Description
List[MarketPriceDto]	List[MarketPriceDto]: List of market prices.

Source code in ethereal/rest/product.py

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

    Args:
        product_ids (List[str]): List of product UUIDs. Required.

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

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

list_products(self, **kwargs)

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	Product ticker to filter by. Optional.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

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

Source code in ethereal/rest/product.py

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): Product ticker to filter by. Optional.
        **kwargs: Additional request parameters accepted by the API. Optional.

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

rpc

get_rpc_config(self, **kwargs)

Gets RPC configuration.

Other Parameters:

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

Returns:

Name	Type	Description
RpcConfigDto	RpcConfigDto	EIP-712 Domain Data necessary for message signing.

Source code in ethereal/rest/rpc.py

def get_rpc_config(self, **kwargs) -> RpcConfigDto:
    """Gets RPC configuration.

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

    Returns:
        RpcConfigDto: EIP-712 Domain Data necessary for message signing.
    """
    endpoint = f"{API_PREFIX}/rpc/config"

    res = self.get(endpoint, **kwargs)
    domain = Domain(**res["domain"])
    signature_types = SignatureTypes(**res["signatureTypes"])
    return RpcConfigDto(domain=domain, signatureTypes=signature_types)

subaccount

get_subaccount(self, id, **kwargs)

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 information.

Source code in ethereal/rest/subaccount.py

def get_subaccount(self, id: str, **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 information.
    """
    endpoint = f"{API_PREFIX}/subaccount/{id}"
    res = self.get(endpoint, **kwargs)
    return SubaccountDto(**res)

get_subaccount_balances(self, **kwargs)

Gets 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.
**kwargs		Additional request parameters accepted by the API. Optional.

Returns:

Type	Description
List[SubaccountBalanceDto]	List[SubaccountBalanceDto]: List of balance information.

Source code in ethereal/rest/subaccount.py

def get_subaccount_balances(self, **kwargs) -> List[SubaccountBalanceDto]:
    """Gets 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.
        **kwargs: Additional request parameters accepted by the API. Optional.

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

list_subaccounts(self, **kwargs)

Lists subaccounts for a sender.

Parameters:

Name	Type	Description	Default
sender	str	Address of the account which registered 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.
name	str	Filter by subaccount name. 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]: List of subaccount information.

Source code in ethereal/rest/subaccount.py

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

    Args:
        sender (str): Address of the account which registered 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.
        name (str, optional): Filter by subaccount name. 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]: List of subaccount information.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/subaccount/",
        request_model=V1SubaccountGetParametersQuery,
        response_model=PageOfSubaccountDtos,
        **kwargs,
    )
    data = [SubaccountDto(**model.model_dump(by_alias=True)) for model in res.data]
    return data

token

get_token(self, id, **kwargs)

Gets a specific token by ID.

Parameters:

Name	Type	Description	Default
id	str	The token identifier. Required.	required

Other Parameters:

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

Returns:

Name	Type	Description
TokenDto	TokenDto	The requested token information.

Source code in ethereal/rest/token.py

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

    Args:
        id (str): The token identifier. Required.

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

    Returns:
        TokenDto: The requested token information.
    """
    endpoint = f"{API_PREFIX}/token/{id}"
    res = self.get(endpoint, **kwargs)
    return TokenDto(**res)

list_token_transfers(self, **kwargs)

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
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.
statuses	List[str]	List of transfer statuses. 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]: A list of transfer information.

Source code in ethereal/rest/token.py

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:
        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.
        statuses (List[str], optional): List of transfer statuses. 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]: A list of transfer information.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/token/transfer",
        request_model=V1TokenTransferGetParametersQuery,
        response_model=PageOfTransfersDtos,
        **kwargs,
    )
    data = [TransferDto(**model.model_dump(by_alias=True)) for model in res.data]
    return data

list_token_withdraws(self, **kwargs)

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
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.
active	bool	Filter for active withdrawals. 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]: A list of withdrawal information.

Source code in ethereal/rest/token.py

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:
        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.
        active (bool, optional): Filter for active withdrawals. 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]: A list of withdrawal information.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/token/withdraw",
        request_model=V1TokenWithdrawGetParametersQuery,
        response_model=PageOfWithdrawDtos,
        **kwargs,
    )
    data = [WithdrawDto(**model.model_dump(by_alias=True)) for model in res.data]
    return data

list_tokens(self, **kwargs)

Lists all tokens.

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.
deposit_enabled	bool	Filter for deposit-enabled tokens. Optional.
withdraw_enabled	bool	Filter for withdraw-enabled tokens. 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]: A list containing all token information.

Source code in ethereal/rest/token.py

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

    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.
        deposit_enabled (bool, optional): Filter for deposit-enabled tokens. Optional.
        withdraw_enabled (bool, optional): Filter for withdraw-enabled tokens. 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]: A list containing all token information.
    """
    res = self.get_validated(
        url_path=f"{API_PREFIX}/token",
        request_model=V1TokenGetParametersQuery,
        response_model=PageOfTokensDtos,
        **kwargs,
    )
    data = [TokenDto(**model.model_dump(by_alias=True)) for model in res.data]
    return data

prepare_withdraw_token(self, subaccount, token, amount, account, include_signature=False, **kwargs)

Prepares the data model and optionally the signature for a token withdrawal.

Parameters:

Name	Type	Description	Default
subaccount	str	Subaccount name as a bytes string	required
token	str	Address of the token to withdraw	required
amount	int	Amount to withdraw	required
account	str	Address to withdraw to	required
include_signature	bool	Whether to include the signature. Defaults to False.	False

Other Parameters:

Name	Type	Description
nonce	str	Custom nonce for the withdraw. If not provided, one will be generated.
signed_at	int	Custom timestamp for the withdraw. If not provided, current time will be used.
**kwargs		Additional parameters for the withdraw.

Returns:

Name	Type	Description
InitiateWithdrawDto	InitiateWithdrawDto	DTO containing the data model and optionally the signature.

Raises:

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

Source code in ethereal/rest/token.py

def prepare_withdraw_token(
    self,
    subaccount: str,
    token: str,
    amount: int,
    account: str,
    include_signature: bool = False,
    **kwargs,
) -> InitiateWithdrawDto:
    """Prepares the data model and optionally the signature for a token withdrawal.

    Args:
        subaccount (str): Subaccount name as a bytes string
        token (str): Address of the token to withdraw
        amount (int): Amount to withdraw
        account (str): Address to withdraw to
        include_signature (bool, optional): Whether to include the signature. Defaults to False.

    Other Parameters:
        nonce (str, optional): Custom nonce for the withdraw. If not provided, one will be generated.
        signed_at (int, optional): Custom timestamp for the withdraw. If not provided, current time will be used.
        **kwargs: Additional parameters for the withdraw.

    Returns:
        InitiateWithdrawDto: DTO containing the data model and optionally the signature.

    Raises:
        ValueError: If include_signature is True and no chain client or private key is available.
    """
    nonce = kwargs.get("nonce") or generate_nonce()
    signed_at = kwargs.get("signed_at") or int(time.time())
    data = {
        "account": account,
        "subaccount": subaccount,
        "token": token,
        "amount": amount,
        "nonce": nonce,
        "signedAt": signed_at,
    }
    data_model = InitiateWithdrawDtoData.model_validate(data)
    dto = InitiateWithdrawDto.model_validate(
        {"data": data_model.model_dump(mode="json", by_alias=True), "signature": ""}
    )
    if include_signature:
        dto = self.sign_withdraw_token(dto)
    return dto

sign_withdraw_token(self, withdraw_dto, private_key=None)

Signs the token withdrawal data using the chain client.

Parameters:

Name	Type	Description	Default
withdraw_dto	InitiateWithdrawDto	Withdrawal data to sign.	required
private_key	Optional[str]	Private key for signing. If None, uses the instance's private key.	None

Returns:

Name	Type	Description
InitiateWithdrawDto	InitiateWithdrawDto	The signed withdrawal DTO with signature included.

Raises:

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

Source code in ethereal/rest/token.py

def sign_withdraw_token(
    self,
    withdraw_dto: InitiateWithdrawDto,
    private_key: Optional[str] = None,
) -> InitiateWithdrawDto:
    """Signs the token withdrawal data using the chain client.

    Args:
        withdraw_dto (InitiateWithdrawDto): Withdrawal data to sign.
        private_key (Optional[str], optional): Private key for signing. If None, uses the instance's private key.

    Returns:
        InitiateWithdrawDto: The signed withdrawal DTO with signature included.

    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 the message for signing
    message = withdraw_dto.data.model_dump(mode="json", by_alias=True)
    message["amount"] = int(Decimal(message["amount"]) * Decimal(1e9))
    message["signedAt"] = int(message["signedAt"])

    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

withdraw_token(self, dto, token_id, **kwargs)

Submits a prepared and signed token withdrawal request.

Parameters:

Name	Type	Description	Default
dto	InitiateWithdrawDto	Prepared and signed withdrawal data.	required
token_id	str	ID of the token to withdraw.	required

Other Parameters:

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

Returns:

Name	Type	Description
WithdrawDto	WithdrawDto	Response containing the withdrawal information.

Source code in ethereal/rest/token.py

def withdraw_token(
    self,
    dto: InitiateWithdrawDto,
    token_id: str,
    **kwargs,
) -> WithdrawDto:
    """Submits a prepared and signed token withdrawal request.

    Args:
        dto (InitiateWithdrawDto): Prepared and signed withdrawal data.
        token_id (str): ID of the token to withdraw.

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

    Returns:
        WithdrawDto: Response containing the withdrawal information.
    """
    endpoint = f"{API_PREFIX}/token/{token_id}/withdraw"
    res = self.post(
        endpoint,
        data=dto.model_dump(mode="json", by_alias=True, exclude_none=True),
        **kwargs,
    )
    return WithdrawDto.model_validate(res)

util

generate_nonce()

Generates a timestamp-based nonce.

Returns:

Name	Type	Description
str	str	Current timestamp in nanoseconds as string.

Source code in ethereal/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())

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 ethereal/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

ethereal.ws_client.WSClient

Bases: WSBase

Ethereal websocket client.

Parameters:

Name	Type	Description	Default
config	Union[Dict[str, Any], WSConfig]	Configuration dictionary or WSConfig object. Required fields include: - base_url (str): Base URL for websocket requests Optional fields include: - verbose (bool): Enables debug logging, defaults to False	required

Source code in ethereal/ws_client.py

class WSClient(WSBase):
    """Ethereal websocket client.

    Args:
        config (Union[Dict[str, Any], WSConfig]): Configuration dictionary or WSConfig object.
            Required fields include:
            - base_url (str): Base URL for websocket requests
            Optional fields include:
            - verbose (bool): Enables debug logging, defaults to False
    """

    def __init__(self, config: Union[Dict[str, Any], WSConfig]):
        super().__init__(config)

    def subscribe(
        self,
        stream_type: str,
        product_id: Optional[str] = None,
        subaccount_id: Optional[str] = None,
        callback: Optional[Callable[[Dict[str, Any]], None]] = None,
        namespace: Optional[str] = "/v1/stream",
    ) -> Dict[str, Any]:
        """Subscribe to a specific stream.

        Args:
            stream_type (str): Type of stream to subscribe to
            product_id (Optional[str]): Product ID to subscribe to
            subaccount_id (Optional[str]): Subaccount ID, optional
            callback (Optional[Callable]): Callback function to handle incoming messages

        Returns:
            Dict[str, Any]: Subscription response
        """
        subscription_data = {"type": stream_type}

        if subaccount_id:
            subscription_data["subaccountId"] = subaccount_id
        if product_id:
            subscription_data["productId"] = product_id

        # Register callback if provided
        if callback:
            event_key = stream_type
            if event_key not in self.callbacks:
                self.callbacks[event_key] = []

            self.callbacks[event_key].append(callback)

        # Send subscription request
        return self._emit("subscribe", subscription_data, namespace=namespace)

    def unsubscribe(
        self,
        stream_type: str,
        product_id: Optional[str] = None,
        subaccount_id: Optional[str] = None,
        namespace: Optional[str] = "/v1/stream",
    ) -> Dict[str, Any]:
        """Unsubscribe from a specific stream.

        Args:
            stream_type (str): Type of stream to unsubscribe from
            product_id (str): Product ID to unsubscribe from
            subaccount_id (Optional[str]): Subaccount ID, optional

        Returns:
            Dict[str, Any]: Unsubscription response
        """
        unsubscription_data = {"type": stream_type}

        if subaccount_id:
            unsubscription_data["subaccountId"] = subaccount_id
        if product_id:
            unsubscription_data["productId"] = product_id

        # Send unsubscription request
        return self._emit("unsubscribe", unsubscription_data, namespace=namespace)

subscribe(stream_type, product_id=None, subaccount_id=None, callback=None, namespace='/v1/stream')

Subscribe to a specific stream.

Parameters:

Name	Type	Description	Default
stream_type	str	Type of stream to subscribe to	required
product_id	Optional[str]	Product ID to subscribe to	None
subaccount_id	Optional[str]	Subaccount ID, optional	None
callback	Optional[Callable]	Callback function to handle incoming messages	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Subscription response

Source code in ethereal/ws_client.py

def subscribe(
    self,
    stream_type: str,
    product_id: Optional[str] = None,
    subaccount_id: Optional[str] = None,
    callback: Optional[Callable[[Dict[str, Any]], None]] = None,
    namespace: Optional[str] = "/v1/stream",
) -> Dict[str, Any]:
    """Subscribe to a specific stream.

    Args:
        stream_type (str): Type of stream to subscribe to
        product_id (Optional[str]): Product ID to subscribe to
        subaccount_id (Optional[str]): Subaccount ID, optional
        callback (Optional[Callable]): Callback function to handle incoming messages

    Returns:
        Dict[str, Any]: Subscription response
    """
    subscription_data = {"type": stream_type}

    if subaccount_id:
        subscription_data["subaccountId"] = subaccount_id
    if product_id:
        subscription_data["productId"] = product_id

    # Register callback if provided
    if callback:
        event_key = stream_type
        if event_key not in self.callbacks:
            self.callbacks[event_key] = []

        self.callbacks[event_key].append(callback)

    # Send subscription request
    return self._emit("subscribe", subscription_data, namespace=namespace)

unsubscribe(stream_type, product_id=None, subaccount_id=None, namespace='/v1/stream')

Unsubscribe from a specific stream.

Parameters:

Name	Type	Description	Default
stream_type	str	Type of stream to unsubscribe from	required
product_id	str	Product ID to unsubscribe from	None
subaccount_id	Optional[str]	Subaccount ID, optional	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Unsubscription response

Source code in ethereal/ws_client.py

def unsubscribe(
    self,
    stream_type: str,
    product_id: Optional[str] = None,
    subaccount_id: Optional[str] = None,
    namespace: Optional[str] = "/v1/stream",
) -> Dict[str, Any]:
    """Unsubscribe from a specific stream.

    Args:
        stream_type (str): Type of stream to unsubscribe from
        product_id (str): Product ID to unsubscribe from
        subaccount_id (Optional[str]): Subaccount ID, optional

    Returns:
        Dict[str, Any]: Unsubscription response
    """
    unsubscription_data = {"type": stream_type}

    if subaccount_id:
        unsubscription_data["subaccountId"] = subaccount_id
    if product_id:
        unsubscription_data["productId"] = product_id

    # Send unsubscription request
    return self._emit("unsubscribe", unsubscription_data, namespace=namespace)

ethereal.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 ethereal/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.provider = self._setup_provider()
        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.provider.eth.chain_id

        if tokens is not None:
            usde_token = next((t for t in tokens if t.name == "USD"), None)
            if usde_token is None:
                self.logger.warning("USD token not found in the provided tokens list.")
            else:
                usde_address = self.provider.to_checksum_address(usde_token.address)
                usde_abi = (
                    read_contract(self.chain_id, "WUSDe", common=True)
                    if usde_token.erc20_name and "Wrapped" in usde_token.erc20_name
                    else read_contract(self.chain_id, "ERC20", common=True)
                )

                self.usde = self.provider.eth.contract(
                    address=usde_address,
                    abi=usde_abi,
                )
        self.rpc_config = rpc_config

    @property
    def exchange_contract(self):
        return self.provider.eth.contract(
            address=self.rpc_config.domain.verifying_contract,
            abi=read_contract(self.chain_id, "ExchangeGateway"),
        )

    def _setup_provider(self):
        """Set up the Web3 provider.

        Returns:
            Web3: The Web3 provider 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.info("Private key not specified in the configuration")
            return None

        account = self.provider.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.provider.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.provider.eth.gas_price
            max_priority_fee = self.provider.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.provider.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.provider.eth.account.sign_transaction(
                tx, private_key=self.private_key
            )
            tx_hash = self.provider.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, or -1 if failed
        """
        try:
            return self.provider.eth.get_transaction_count(address)
        except Web3Exception as e:
            self.logger.error(f"Failed to get nonce: {e}")
            return -1

    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, or -1 if failed
        """
        try:
            return self.provider.eth.get_balance(address)
        except Web3Exception as e:
            self.logger.error(f"Failed to get balance: {e}")
            return -1

    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, or -1 if failed
        """
        try:
            contract = self.provider.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}")
            return -1

    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 deposit_usde(
        self,
        amount: float,
        address: Optional[str] = None,
        submit: Optional[bool] = False,
        account_name: Optional[str] = "primary",
    ) -> Union[TxParams, str]:
        """Submit a deposit transaction.

        Args:
            amount (float): The amount to deposit
            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".

        Returns:
            Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
        """
        if address is None:
            address = self.address
        try:
            # params
            subaccount = self.provider.to_hex(text=account_name).ljust(66, "0")
            amount = self.provider.to_wei(amount, "ether")
            referral_code = self.provider.to_hex(0).ljust(66, "0")

            # prepare the tx
            tx = self._get_tx(to=self.exchange_contract.address, value=amount)
            tx["data"] = self.exchange_contract.encode_abi(
                "depositUsd", args=[subaccount, 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,
        address: Optional[str] = None,
        submit: Optional[bool] = False,
        account_name: Optional[str] = "primary",
    ) -> Union[TxParams, str]:
        """Finalize a withdrawal.

        Args:
            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 name of the account. Defaults to "primary".

        Returns:
            Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
        """
        if address is None:
            address = self.address
        try:
            # params
            subaccount = self.provider.to_hex(text=account_name).ljust(66, "0")

            # prepare the tx
            tx = self._get_tx(to=self.exchange_contract.address)
            tx["data"] = self.exchange_contract.encode_abi(
                "finalizeWithdraw", args=[address, subaccount]
            )

            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

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 ethereal/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.provider.eth.gas_price
        max_priority_fee = self.provider.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 ethereal/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.provider.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

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 ethereal/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_usde(amount, address=None, submit=False, account_name='primary')

Submit a deposit transaction.

Parameters:

Name	Type	Description	Default
amount	float	The amount to deposit	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'

Returns:

Type	Description
Union[TxParams, str]	Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

Source code in ethereal/chain_client.py

def deposit_usde(
    self,
    amount: float,
    address: Optional[str] = None,
    submit: Optional[bool] = False,
    account_name: Optional[str] = "primary",
) -> Union[TxParams, str]:
    """Submit a deposit transaction.

    Args:
        amount (float): The amount to deposit
        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".

    Returns:
        Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
    """
    if address is None:
        address = self.address
    try:
        # params
        subaccount = self.provider.to_hex(text=account_name).ljust(66, "0")
        amount = self.provider.to_wei(amount, "ether")
        referral_code = self.provider.to_hex(0).ljust(66, "0")

        # prepare the tx
        tx = self._get_tx(to=self.exchange_contract.address, value=amount)
        tx["data"] = self.exchange_contract.encode_abi(
            "depositUsd", args=[subaccount, 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(address=None, submit=False, account_name='primary')

Finalize a withdrawal.

Parameters:

Name	Type	Description	Default
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 name of the account. Defaults to "primary".	'primary'

Returns:

Type	Description
Union[TxParams, str]	Union[TxParams, str]: The transaction parameters or transaction hash if submit=True

Source code in ethereal/chain_client.py

def finalize_withdraw(
    self,
    address: Optional[str] = None,
    submit: Optional[bool] = False,
    account_name: Optional[str] = "primary",
) -> Union[TxParams, str]:
    """Finalize a withdrawal.

    Args:
        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 name of the account. Defaults to "primary".

    Returns:
        Union[TxParams, str]: The transaction parameters or transaction hash if submit=True
    """
    if address is None:
        address = self.address
    try:
        # params
        subaccount = self.provider.to_hex(text=account_name).ljust(66, "0")

        # prepare the tx
        tx = self._get_tx(to=self.exchange_contract.address)
        tx["data"] = self.exchange_contract.encode_abi(
            "finalizeWithdraw", args=[address, subaccount]
        )

        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, or -1 if failed

Source code in ethereal/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, or -1 if failed
    """
    try:
        return self.provider.eth.get_balance(address)
    except Web3Exception as e:
        self.logger.error(f"Failed to get balance: {e}")
        return -1

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, or -1 if failed

Source code in ethereal/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, or -1 if failed
    """
    try:
        return self.provider.eth.get_transaction_count(address)
    except Web3Exception as e:
        self.logger.error(f"Failed to get nonce: {e}")
        return -1

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 ethereal/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_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, or -1 if failed

Source code in ethereal/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, or -1 if failed
    """
    try:
        contract = self.provider.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}")
        return -1

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 ethereal/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 ethereal/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.provider.eth.account.sign_transaction(
            tx, private_key=self.private_key
        )
        tx_hash = self.provider.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