Data Retrieval Guide

This document provides comprehensive information about retrieving data using the Perpetuals SDK.

Overview

The Perpetuals SDK provides both asynchronous and synchronous clients for retrieving data from the Fairground perpetuals platform. All data retrieval is performed through connectRPC calls that return strongly-typed protobuf objects.

Client Setup

Async Client (Recommended)

import asyncio
import httpx
from forte_perpetuals_sdk import PerpetualsClient

async def setup_client():
    session = httpx.AsyncClient()
    client = PerpetualsClient(
        base_url="https://api.perpetuals.forte.com",
        session=session,
        timeout_ms=30000
    )
    return client, session

# Usage
async def main():
    client, session = await setup_client()
    try:
        # Perform data retrieval operations
        order = await client.get_order(order_id=123, market_id=1)
    finally:
        await session.aclose()

Sync Client

import httpx
from forte_perpetuals_sdk import PerpetualsClientSync

def setup_sync_client():
    session = httpx.Client()
    client = PerpetualsClientSync(
        base_url="https://api.perpetuals.forte.com",
        session=session,
        timeout_ms=30000
    )
    return client, session

# Usage
def main():
    client, session = setup_sync_client()
    try:
        # Perform data retrieval operations
        order = client.get_order(order_id=123, market_id=1)
    finally:
        session.close()

Available Data Retrieval Methods

Get Order

Retrieve detailed information about a specific order.

Method Signature

async def get_order(
    self,
    order_id: int,
    market_id: int,
    reduce_only: bool = False,
    timeout_ms: Optional[int] = None,
) -> Order

Parameters

order_id (int): The unique identifier for the order
market_id (int): The market identifier where the order was placed
reduce_only (bool): Whether to filter for reduce-only orders (default: False)
timeout_ms (Optional[int]): Request timeout in milliseconds (overrides client default)

Returns

Returns an Order protobuf object containing:

Order details (size, price, leverage, etc.)
Current status and fill information
Market and trader information
Timestamps and metadata

Get Orders (Batch)

Retrieve multiple orders with filtering options for efficient batch operations.

Method Signature

async def get_orders(
    self,
    address: str,
    timeout_ms: Optional[int] = None,
) -> List[Order]

Parameters

address (str): The wallet address to get orders for
timeout_ms (Optional[int]): Request timeout in milliseconds

Returns

Returns a list of Order protobuf objects for the specified address.

Get Open Positions

Retrieve open trading positions with filtering options.

Method Signature

async def get_open_positions(
    self,
    address: str,
    timeout_ms: Optional[int] = None,
) -> List[Position]

Parameters

address (str): The wallet address to get positions for
timeout_ms (Optional[int]): Request timeout in milliseconds

Returns

Returns a list of Position protobuf objects containing:

Position details (size, entry price, leverage, etc.)
Profit and loss information
Margin and liquidation data
Market and trader information

Example Usage

# Async version - Single Order
async def retrieve_order_data():
    async with httpx.AsyncClient() as session:
        client = PerpetualsClient(
            base_url="https://api.perpetuals.forte.com",
            session=session
        )

        try:
            # Get order information
            order = await client.get_order(
                order_id=123,
                market_id=1,
                reduce_only=False,
                timeout_ms=10000  # 10 second timeout
            )

            # Access order data
            print(f"Order Status: {order.status}")
            print(f"Market: {order.market}")
            print(f"Size: {order.size}")
            print(f"Filled: {order.filled_size}")
            print(f"Price: {order.threshold_price}")

            return order

        except Exception as e:
            print(f"Failed to retrieve order: {e}")
            return None

# Async version - Batch Orders
async def retrieve_orders_batch():
    async with httpx.AsyncClient() as session:
        client = PerpetualsClient(
            base_url="https://api.perpetuals.forte.com",
            session=session
        )

        try:
            # Get all orders for an address
            orders = await client.get_orders(address="0x1234567890abcdef")

            print(f"Retrieved {len(orders)} orders")
            for order in orders:
                print(f"Order #{order.order_id}: {order.size} @ {order.threshold_price}")

            return orders

        except Exception as e:
            print(f"Failed to retrieve orders: {e}")
            return []

# Async version - Open Positions
async def retrieve_positions():
    async with httpx.AsyncClient() as session:
        client = PerpetualsClient(
            base_url="https://api.perpetuals.forte.com",
            session=session
        )

        try:
            # Get all open positions for an address
            positions = await client.get_open_positions(address="0x1234567890abcdef")

            print(f"Found {len(positions)} open positions")
            for position in positions:
                pnl_status = "+" if position.unrealized_pnl >= 0 else ""
                print(f"Position ({position.market}): "
                      f"{position.side} {position.size} @ ${position.entry_price:,.2f}, "
                      f"PnL: {pnl_status}${position.unrealized_pnl:,.2f}")

            return positions

        except Exception as e:
            print(f"Failed to retrieve positions: {e}")
            return []

# Sync version
def retrieve_order_data_sync():
    with httpx.Client() as session:
        client = PerpetualsClientSync(
            base_url="https://api.perpetuals.forte.com",
            session=session
        )

        try:
            order = client.get_order(order_id=123, market_id=1)
            orders = client.get_orders(address="0x1234567890abcdef")
            positions = client.get_open_positions(address="0x1234567890abcdef")

            return order, orders, positions
        except Exception as e:
            print(f"Failed to retrieve data: {e}")
            return None, [], []

Error Handling

ConnectRPC Errors

The SDK uses connectRPC’s error handling system. All API calls can raise ConnectError exceptions:

from connectrpc.errors import ConnectError
from connectrpc.code import Code

async def safe_order_retrieval(client, order_id, market_id):
    try:
        order = await client.get_order(order_id=order_id, market_id=market_id)
        return order
    except ConnectError as e:
        if e.code == Code.NOT_FOUND:
            print(f"Order {order_id} not found in market {market_id}")
        elif e.code == Code.DEADLINE_EXCEEDED:
            print("Request timed out")
        elif e.code == Code.PERMISSION_DENIED:
            print("Access denied - check authentication")
        elif e.code == Code.UNAVAILABLE:
            print("Service temporarily unavailable")
        else:
            print(f"API error [{e.code}]: {e.message}")
        return None
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None

Common Error Codes

Code	Description	Typical Causes
`NOT_FOUND`	Resource not found	Invalid order_id or market_id
`DEADLINE_EXCEEDED`	Request timeout	Network issues or server overload
`PERMISSION_DENIED`	Access denied	Authentication or authorization failure
`UNAVAILABLE`	Service unavailable	Server maintenance or outage
`INVALID_ARGUMENT`	Invalid request	Malformed parameters

Data Processing

Converting to Python Dictionaries

from forte_perpetuals_sdk.utils import order_to_dict, position_to_dict

async def process_data():
    # Get data
    order = await client.get_order(order_id=123, market_id=1)
    positions = await client.get_open_positions(owner="trader_123")

    # Convert to dictionary with human-readable values
    order_dict = order_to_dict(order)

    # Access data with familiar Python dictionary syntax
    print(f"Status: {order_dict['status']}")  # "FULLY_FILLED"
    print(f"Side: {order_dict['side']}")      # "LONG"
    print(f"Type: {order_dict['type']}")      # "MARKET"

    # Process positions
    for position in positions:
        position_dict = position_to_dict(position)
        print(f"Position: {position_dict['market']} - {position_dict['margin_type']}")

    return order_dict

Enum Processing

from forte_perpetuals_sdk.utils import OrderStatus, Side, OrderType, MarginType

async def analyze_data(client, address):
    # Get orders and positions
    orders = await client.get_orders(address=address)
    positions = await client.get_open_positions(address=address)

    # Analyze orders
    for order in orders:
        status = OrderStatus.from_proto(order.status)
        side = Side.from_proto(order.side)
        order_type = OrderType.from_proto(order.type)

        if status == OrderStatus.FULLY_FILLED:
            print(f"Order #{order.order_id} completed successfully")
        elif status == OrderStatus.PARTIALLY_FILLED:
            fill_percentage = (order.filled_size / order.size) * 100
            print(f"Order #{order.order_id} {fill_percentage:.1f}% filled")

    # Analyze positions
    for position in positions:
        side = Side.from_proto(position.side)
        margin_type = MarginType.from_proto(position.margin_type)

        print(f"Position: {side.value} {position.size} ({margin_type.value} margin)")

        if position.unrealized_pnl < -position.margin_amount * 0.8:
            print(f"Warning: Position {position.market} near liquidation!")

Formatting for Display

from forte_perpetuals_sdk.utils import format_order_summary, format_position_summary

async def display_portfolio_info(client, address):
    orders = await client.get_orders(address=address)
    positions = await client.get_open_positions(address=address)

    print("=== Recent Orders ===")
    for order in orders:
        summary = format_order_summary(order)
        print(summary)
        # Output: "Order #123 (BTC-PERP): LONG 0.5 @ $75,000.00 - Status: FULLY_FILLED"

    print("\n=== Open Positions ===")
    for position in positions:
        summary = format_position_summary(position)
        print(summary)
        # Output: "Position (BTC-PERP): LONG 0.5 @ $75,000.00, PnL: +$500.00 (CROSS margin)"

    # Calculate total portfolio PnL
    total_pnl = sum(pos.unrealized_pnl for pos in positions)
    print(f"\nTotal Portfolio PnL: ${total_pnl:,.2f}")

Performance Considerations

Connection Pooling

Reuse HTTP sessions for better performance:

class DataRetriever:
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.session = None
        self.client = None

    async def __aenter__(self):
        self.session = httpx.AsyncClient()
        self.client = PerpetualsClient(
            base_url=self.base_url,
            session=self.session
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.aclose()

    async def get_order(self, order_id: int, market_id: int):
        return await self.client.get_order(order_id=order_id, market_id=market_id)

# Usage
async def efficient_data_retrieval():
    async with DataRetriever("https://api.perpetuals.forte.com") as retriever:
        # Reuse connection for multiple requests
        order1 = await retriever.get_order(123, 1)
        order2 = await retriever.get_order(124, 1)
        order3 = await retriever.get_order(125, 2)

        return [order1, order2, order3]

Timeout Configuration

Configure appropriate timeouts for different use cases:

# For real-time applications - shorter timeout
real_time_client = PerpetualsClient(
    base_url="https://api.perpetuals.forte.com",
    session=session,
    timeout_ms=5000  # 5 seconds
)

# For batch processing - longer timeout
batch_client = PerpetualsClient(
    base_url="https://api.perpetuals.forte.com",
    session=session,
    timeout_ms=30000  # 30 seconds
)

# Per-request timeout override
order = await client.get_order(
    order_id=123,
    market_id=1,
    timeout_ms=10000  # Override default timeout
)

Concurrent Requests

Use asyncio for concurrent data retrieval and prefer batch operations when possible:

import asyncio

async def get_multiple_orders(client, order_requests):
    """Retrieve multiple orders concurrently using individual calls."""
    tasks = [
        client.get_order(
            order_id=req['order_id'],
            market_id=req['market_id']
        )
        for req in order_requests
    ]

    # Execute all requests concurrently
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Process results
    orders = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Failed to retrieve order {order_requests[i]['order_id']}: {result}")
        else:
            orders.append(result)

    return orders

async def get_portfolio_data(client, addresses):
    """Efficiently retrieve all portfolio data using batch operations."""

    # Use batch operations for better performance
    tasks = []
    for address in addresses:
        tasks.append(client.get_orders(address=address))
        tasks.append(client.get_open_positions(address=address))

    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Split results into orders and positions
    all_orders = []
    all_positions = []

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            address_idx = i // 2
            operation = "orders" if i % 2 == 0 else "positions"
            print(f"Failed to retrieve {operation} for address {addresses[address_idx]}: {result}")
        else:
            if i % 2 == 0:  # orders
                all_orders.extend(result)
            else:  # positions
                all_positions.extend(result)

    return all_orders, all_positions

# Usage - Prefer batch operations for efficiency
async def main():
    async with httpx.AsyncClient() as session:
        client = PerpetualsClient(
            base_url="https://api.perpetuals.forte.com",
            session=session
        )

        # Efficient: Use batch operations
        all_orders = await client.get_orders(address="0x1234567890abcdef")
        all_positions = await client.get_open_positions(address="0x1234567890abcdef")

        print(f"Retrieved {len(all_orders)} orders and {len(all_positions)} positions")

        # For multiple addresses, use concurrent batch calls
        addresses = ["0x1234567890abcdef", "0xabcdef1234567890", "0x567890abcdef1234"]
        orders, positions = await get_portfolio_data(client, addresses)

        print(f"Total across addresses: {len(orders)} orders, {len(positions)} positions")

Best Practices

Use async clients for better performance in I/O-bound applications
Prefer batch operations - Use get_orders() and get_open_positions() instead of multiple individual calls
Reuse HTTP sessions to avoid connection overhead
Implement proper error handling for network failures and API errors
Set appropriate timeouts based on your application’s requirements
Use connection pooling for high-throughput applications
Cache frequently accessed data to reduce API calls
Implement retry logic for transient failures
Monitor API rate limits and implement backoff strategies
Filter data server-side using method parameters to reduce bandwidth and processing
Use concurrent requests wisely - batch operations are usually more efficient than parallel individual calls

Testing Data Retrieval

Mocking for Unit Tests

import pytest
from unittest.mock import AsyncMock, patch
from forte_perpetuals_sdk import PerpetualsClient
from forte_perpetuals_sdk.generated.base_objects.v1.order_pb2 import Order

@pytest.mark.asyncio
async def test_order_retrieval():
    # Create mock order
    mock_order = Order(
        owner="test_user",
        market_id=1,
        size=1.0,
        status=OrderStatus.ORDER_STATUS_PENDING
    )

    # Mock the client method
    with patch.object(PerpetualsClient, 'get_order', new_callable=AsyncMock) as mock_get:
        mock_get.return_value = mock_order

        # Test your data retrieval logic
        client = PerpetualsClient("http://test", AsyncMock())
        order = await client.get_order(123, 1)

        assert order.market_id == 1
        assert order.size == 1.0
        mock_get.assert_called_once_with(123, 1)

Future Extensions

As the API expands, additional data retrieval methods will be added following the same patterns:

get_market_data() - Retrieve market information and pricing data
get_account_info() - Retrieve account details and balances
get_trade_history() - Retrieve historical trade records
get_funding_payments() - Retrieve funding payment history

Each new method will follow the established patterns for:

Async/sync support with consistent signatures
Comprehensive error handling and timeout configuration
Type safety with strongly-typed protobuf objects
Utility function support for formatting and conversion
Efficient batch operations where applicable

Developer Docs

API Reference

SDK Reference

Data Retrieval Guide

Overview

Client Setup

Async Client (Recommended)

Sync Client

Available Data Retrieval Methods

Get Order

Method Signature

Parameters

Returns

Get Orders (Batch)

Method Signature

Parameters

Returns

Get Open Positions

Method Signature

Parameters

Returns

Example Usage

Error Handling

ConnectRPC Errors

Common Error Codes

Data Processing

Converting to Python Dictionaries

Enum Processing

Formatting for Display

Performance Considerations

Connection Pooling

Timeout Configuration

Concurrent Requests

Best Practices

Testing Data Retrieval

Mocking for Unit Tests

Future Extensions

Developer Docs

API Reference

SDK Reference

​Overview

​Client Setup

​Async Client (Recommended)

​Sync Client

​Available Data Retrieval Methods

​Get Order

​Method Signature

​Parameters

​Returns

​Get Orders (Batch)

​Method Signature

​Parameters

​Returns

​Get Open Positions

​Method Signature

​Parameters

​Returns

​Example Usage

​Error Handling

​ConnectRPC Errors

​Common Error Codes

​Data Processing

​Converting to Python Dictionaries

​Enum Processing

​Formatting for Display

​Performance Considerations

​Connection Pooling

​Timeout Configuration

​Concurrent Requests

​Best Practices

​Testing Data Retrieval

​Mocking for Unit Tests

​Future Extensions

Overview

Client Setup

Async Client (Recommended)

Sync Client

Available Data Retrieval Methods

Get Order

Method Signature

Parameters

Returns

Get Orders (Batch)

Method Signature

Parameters

Returns

Get Open Positions

Method Signature

Parameters

Returns

Example Usage

Error Handling

ConnectRPC Errors

Common Error Codes

Data Processing

Converting to Python Dictionaries

Enum Processing

Formatting for Display

Performance Considerations

Connection Pooling

Timeout Configuration

Concurrent Requests

Best Practices

Testing Data Retrieval

Mocking for Unit Tests

Future Extensions