Advanced Usage

This guide covers advanced usage patterns for the Ethereal Python SDK. It's designed for users who need deeper control over the SDK's behavior or want to extend its functionality.

API Naming Conventions

The SDK uses generated types from an OpenAPI spec, however values are accessed using "snake case", with the original "camel case" values being stored as an alias. This SDK will always favor the snake_case naming convention for consistency and readability, but when interacting directly with the API it will expose the original camelCase values. These can be accessed using the model_dump method with by_alias=True.

# Method names use snake_case
client.list_orders()

# API parameters use snake_case
client.list_orders(
    subaccount_id="your-subaccount-id",  # snake_case
    product_id="product-id",             # snake_case
    statuses=["NEW", "FILLED_PARTIAL", "SUBMITTED"]
)

# Responses will have snake_case fields as defined in the SDK
order = client.get_order(id="order-id")
print(order.model_dump(mode='json'))
# {'id': '9945941f-9ad7-45ee-a3d0-1393a5e45ef2', 'type': 'MARKET', 'available_quantity': '0.05', 'quantity': '0.05', 'side': 0, 'product_id': '49a814e4-0004-4829-9902-9cd6f4a846ac', 'subaccount_id': '5ab6801c-3be5-4610-ad01-8d1881b28744', 'status': 'FILLED', 'reduce_only': False, 'updated_at': 1743536868709.0, 'created_at': 1743536868129.0, 'sender': '0x80606C5b3602b7cbf8FDD06e12308504A941400b', 'price': '0', 'filled': '0.05', 'stop_price': '1', 'stop_type': None, 'stop_price_type': 1, 'time_in_force': None, 'post_only': None}

# Access the original camelCase values
print(order.model_dump(mode='json', by_alias=True))
# {'id': '9945941f-9ad7-45ee-a3d0-1393a5e45ef2', 'type': 'MARKET', 'availableQuantity': '0.05', 'quantity': '0.05', 'side': 0, 'productId': '49a814e4-0004-4829-9902-9cd6f4a846ac', 'subaccountId': '5ab6801c-3be5-4610-ad01-8d1881b28744', 'status': 'FILLED', 'reduceOnly': False, 'updatedAt': 1743536868709.0, 'createdAt': 1743536868129.0, 'sender': '0x80606C5b3602b7cbf8FDD06e12308504A941400b', 'price': '0', 'filled': '0.05', 'stopPrice': '1', 'stopType': None, 'stopPriceType': 1, 'timeInForce': None, 'postOnly': None}

The exception is for methods like create_order, where the API endpoints have been wrapped in Python methods. These methods use snake_case for consistency.

# Place a limit order
order = client.create_order(
    order_type="LIMIT",
    quantity=1.0,
    side=0,
    price=50000.0,
    ticker="BTCUSD"
)

Understanding the Type System

The Ethereal SDK uses Pydantic for type validation and serialization. All API models are defined as Pydantic models, which are generated from the Ethereal API's OpenAPI specification.

from ethereal.models.rest import OrderDto, ProductDto
from ethereal import RESTClient

client = RESTClient()

# Access strongly-typed models
products: list[ProductDto] = client.list_products()
for product in products:
    # All fields are properly typed with appropriate validation
    print(f"Product {product.ticker}: {product.id}, Min size: {product.min_size}")

Working with Model Schemas

You can inspect the schema of any model to understand its structure:

from ethereal.models.rest import OrderDto
import json

# Print the JSON schema of an order
schema = OrderDto.model_json_schema()
print(json.dumps(schema, indent=2))

# Validate a dictionary against the schema
data = {
    "id": "order-id",
    "quantity": 1.0,
    "side": 0,
    "price": 50000.0,
    "ticker": "BTCUSD"
}
OrderDto.model_validate(data)

# Convert a model to a dictionary
order = OrderDto(**data)
order_dict = order.model_dump()

Running Validated Queries

For scenarios where the users wants to provide their own types, the SDK provides a get_validated method that allows you to pass in Pydantic models for request and response validation. This could be useful if the API has been updated and you need to add new fields to the request or response, but still want to maintain type safety.

from ethereal import RESTClient
from ethereal.constants import API_PREFIX
from pydantic import BaseModel
from typing import List, Optional

# Define a custom model for a new/changed endpoint
class CustomEndpointParams(BaseModel):
    subaccount_id: str
    include_history: Optional[bool] = None

class CustomEndpointResponse(BaseModel):
    data: List[dict]
    timestamp: str

client = RESTClient()

# Use get_validated for type-safe requests
response = client.get_validated(
    url_path=f"{API_PREFIX}/custom/endpoint",
    request_model=CustomEndpointParams,
    response_model=CustomEndpointResponse,
    subaccount_id="your-subaccount-id",
    include_history=True
)

# The response is validated against the CustomEndpointResponse model
print(f"Response timestamp: {response.timestamp}")
for item in response.data:
    print(item)

Working with Paginated Responses

Many Ethereal API endpoints return paginated results. The SDK provides multiple ways to handle pagination:

Basic Pagination

For manual pagination control:

from ethereal import RESTClient

client = RESTClient()

# First page of orders
orders_page = client.list_orders(
    subaccount_id="your-subaccount-id",
    limit=100  # Request 100 items per page
)

all_orders = orders_page.data
print(f"Retrieved {len(all_orders)} orders")

# Continue fetching pages while there are more results
cursor = orders_page.next_cursor
while orders_page.has_next and cursor:
    # Use the cursor to fetch the next page
    orders_page = client.list_orders(
        subaccountId="your-subaccount-id",
        limit=100,
        cursor=cursor
    )

    # Add these results to our collection
    all_orders.extend(orders_page.data)
    cursor = orders_page.next_cursor
    print(f"Retrieved {len(all_orders)} orders total")

Automated Pagination with _get_pages

For endpoints that return paginated data, the SDK provides a convenience method _get_pages that automatically fetches all pages:

from ethereal import RESTClient
from ethereal.constants import API_PREFIX
from ethereal.models.rest import PageOfOrderDtos, V1OrderGetParametersQuery

client = RESTClient()

# Fetch all orders automatically with pagination
all_orders = client._get_pages(
    endpoint="order",
    request_model=V1OrderGetParametersQuery,
    response_model=PageOfOrderDtos,
    subaccount_id="your-subaccount-id",
    limit=100,  # Items per page (will fetch multiple pages)
    paginate=True,  # Enable automatic pagination
    order_by="createdAt",  # Optional sorting
    order="desc"  # Optional sorting direction
)

print(f"Retrieved {len(all_orders)} orders total")

# Similarly for fills or other paginated endpoints
from ethereal.models.rest import PageOfOrderFillDtos, V1OrderFillGetParametersQuery

all_fills = client._get_pages(
    endpoint="order/fill",
    request_model=V1OrderFillGetParametersQuery,
    response_model=PageOfOrderFillDtos,
    subaccount_id="your-subaccount-id",
    limit=100,
    paginate=True
)

print(f"Retrieved {len(all_fills)} fills total")

The _get_pages method handles all the cursor management and data concatenation for you, making it easy to retrieve large datasets without writing boilerplate pagination code.

Using Standalone Clients

You can use the SDK's components independently for more specialized applications.

Standalone HTTP Client

For applications that only need HTTP functionality:

from ethereal.rest.http_client import HTTPClient

# Create a standalone HTTP client
http_client = HTTPClient({
    "base_url": "https://api.example.com",
    "timeout": 10,
    "verbose": True
})

# Make HTTP requests directly
response = http_client.get("/endpoint", param1="value1")
print(response)

# POST with JSON data
data = {"key": "value", "nested": {"field": 123}}
response = http_client.post("/other/endpoint", data=data)

Standalone Chain Client

For applications that need Ethereum signing capabilities:

from ethereal.chain_client import ChainClient
from ethereal.models.config import ChainConfig

# Create a chain configuration
chain_config = ChainConfig(
    rpc_url="https://rpc.etherealtest.net",
    private_key="your_private_key"
)

# Initialize the chain client
chain_client = ChainClient(chain_config)

WebSocket Advanced Usage

WebSockets allow for realtime communication with the Ethereal API. The SDK provides a WebSocket client that can be used to subscribe to market data, order updates, and other streams. In order to use them, you need to have a REST client to fetch necessary data and a WebSocket client to subscribe to streams and set up callbacks.

For applications requiring real-time data with complex processing:

import asyncio
import threading
from ethereal import WSClient, RESTClient

# Start with a REST client to get necessary data
rest_client = RESTClient()
products = rest_client.list_products()
product_id = products[0].id

# Initialize the WebSocket client
ws_client = WSClient({
    "base_url": "wss://ws.etherealtest.net",
    "verbose": True
})

# Create a message processor for complex logic
class MessageProcessor:
    def __init__(self):
        self.book = {}  # Order book state

    def process_book_depth(self, data):
        if "bids" in data:
            for bid in data["bids"]:
                price, size = bid
                if float(size) > 0:
                    self.book[price] = {"side": "buy", "size": size}
                else:
                    self.book.pop(price, None)

        # Process and output current book state
        print(f"Order book has {len(self.book)} active levels")

# Create processor instance
processor = MessageProcessor()

# Set up callback
def on_book_depth(data):
    processor.process_book_depth(data)

# Connect and subscribe
ws_client.open()
ws_client.subscribe(
    stream_type="BookDepth",
    product_id=product_id,
    callback=on_book_depth
)

# Keep the connection alive in a background thread
def keep_alive():
    while True:
        try:
            asyncio.run(asyncio.sleep(30))  # Check connection every 30 seconds
            if not ws_client.sio.connected:
                print("Reconnecting...")
                ws_client.open()
        except Exception as e:
            print(f"Error in keep_alive: {e}")

# Start the keep-alive thread
keep_alive_thread = threading.Thread(target=keep_alive, daemon=True)
keep_alive_thread.start()

# To disconnect (in your main application logic)
# ws_client.close()