> ## Documentation Index
> Fetch the complete documentation index at: https://docs.fairground.fi/llms.txt
> Use this file to discover all available pages before exploring further.
# PerpetualsClient
> Python SDK reference for PerpetualsClient.
Asynchronous client for the Forte perpetuals platform.
class
PerpetualsClient:
```python theme={null}
def __init__(
self,
base_url: str,
session: httpx.AsyncClient,
timeout_ms: Optional[int] = 30000,
):
self.base_url = base_url
self.session = session
self.timeout_ms = timeout_ms
# api_key = os.getenv("SERVER_API_KEY", "")
# if api_key:
# self.session.headers["authorization"] = f"Bearer {api_key}"
# self.session.headers["x-sdk-version"] = "perpetuals-sdk-py/0.1.0"
# # TODO: remove bypass once we are ready to implement api key enforcement.
# self.session.headers["x-sdk-bypass"] = "true"
self._order_client = OrderServiceClient(address=base_url, session=session)
self._permissions_client = PermissionServiceClient(
address=base_url, session=session
)
self._price_client = PriceServiceClient(address=base_url, session=session)
self._position_client = PositionServiceClient(address=base_url, session=session)
self._market_client = MarketServiceClient(address=base_url, session=session)
self._orderbook_client = OrderBookServiceClient(
address=base_url, session=session
)
self._liquidation_client = LiquidationServiceClient(
address=base_url, session=session
)
self._portfolio_client = PortfolioServiceClient(
address=base_url, session=session
)
self._adl_client = ADLServiceClient(address=base_url, session=session)
self._fill_history_client = FillHistoryServiceClient(
address=base_url, session=session
)
```
base\_url
session
timeout\_ms
```python theme={null}
async def get_order(
self,
order_id: int,
market: Union[str, int],
reduce_only: bool = False,
timeout_ms: Optional[int] = None,
) -> Order:
request = _create_get_order_request(order_id, market, reduce_only)
response = await self._order_client.get_order(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.order
```
```python theme={null}
async def get_orders(
self,
address: str,
market_id: Union[str, int, None] = None,
timeout_ms: Optional[int] = None,
) -> list[Order]:
request = _create_get_orders_request(address, market_id=market_id)
response = await self._order_client.get_orders(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.orders)
```
Tuple of (orders\_list, total\_count)
```python theme={null}
async def get_all_orders(self, page_size_limit: int, offset_amount: int, address: Optional[str]=None, only_open: Optional[bool]=None, timeout_ms: Optional[int]=None) -> tuple[list[Order], int]:
request = _create_all_orders_request(page_size_limit, offset_amount, address=address, only_open=only_open)
response = await self._order_client.get_all_orders(request, timeout_ms=timeout_ms or self.timeout_ms)
orders_list = [order for order in response.orders]
total_count = response.pagination.total_count
return (orders_list, total_count)
```
The market identifier (string symbol or integer ID)
Price threshold for executable orders
Optional side filter (from Side enum: 0=LONG, 1=SHORT)
Optional limit on number of results
Optional timeout in milliseconds
Tuple of (orders\_list, total\_count)
```python theme={null}
async def get_executable_orders(self, market: Union[str, int], price: float, side: Optional[int]=None, limit: Optional[int]=None, timeout_ms: Optional[int]=None) -> tuple[list[Order], int]:
request = _create_get_executable_orders_request(market, price, side, limit)
response = await self._order_client.get_executable_orders(request, timeout_ms=timeout_ms or self.timeout_ms)
orders_list = [order for order in response.orders]
total_count = response.total_count
return (orders_list, total_count)
```
```python theme={null}
async def get_user_fills(
self,
address: Optional[str] = None,
start_time: Optional[Timestamp] = None,
end_time: Optional[Timestamp] = None,
limit: Optional[int] = None,
offset: Optional[int] = None,
timeout_ms: Optional[int] = None,
) -> list[FillInstance]:
request = _create_get_user_fills_request(
address=address,
start_time=start_time,
end_time=end_time,
limit=limit,
offset=offset,
)
response = await self._fill_history_client.get_user_fills(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.fills)
```
```python theme={null}
async def get_historical_prices(
self,
market: Union[str, int],
startTime: Timestamp,
endTime: Timestamp,
limit: int,
interval: str = "60",
timeout_ms: Optional[int] = None,
) -> CandleStick:
request = _create_get_historical_position(
market, startTime, endTime, limit, interval
)
response = await self._price_client.get_historical_prices(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.priceCandleSticks
```
```python theme={null}
async def get_price(
self, market: Union[str, int], timeout_ms: Optional[int] = None
):
request = _create_get_price_request(market)
response = await self._price_client.get_price(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response
```
```python theme={null}
async def get_adl(
self,
timeout_ms: Optional[int] = None,
) -> list[ADLPerMarket]:
request = _create_get_adl()
response = await self._adl_client.get_a_d_l(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.adl_per_market)
```
The market identifier (string symbol or integer ID).
The position size.
The leverage used for the position.
Whether the position is long (True) or short (False).
The entry price of the position.
Optional timeout in milliseconds.
The liquidation price as a float.
```python theme={null}
async def get_liquidation_price_preview(self, market: Union[str, int], amount: int, leverage: int, is_long: bool, price: int, timeout_ms: Optional[int]=None) -> float:
request = _create_get_liquidation_price_preview_request(market, amount, leverage, is_long, price)
response = await self._liquidation_client.get_liquidation_price_preview(request, timeout_ms=timeout_ms or self.timeout_ms)
return response.liquidation_price
```
Tuple of (positions\_list, total\_count)
```python theme={null}
async def get_all_positions(self, page_size_limit: int, offset_amount: int, timeout_ms: Optional[int]=None) -> tuple[list[Position], int]:
request = _create_all_positions_request(page_size_limit, offset_amount)
response = await self._position_client.get_all_positions(request, timeout_ms=timeout_ms or self.timeout_ms)
positions_list = [position for position in response.positions]
total_count = response.pagination.total_count
return (positions_list, total_count)
```
```python theme={null}
async def get_open_positions(
self, address: str, timeout_ms: Optional[int] = None
) -> list[Position]:
request = _create_get_open_positions_request(address)
response = await self._position_client.get_open_positions(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.positions)
```
```python theme={null}
async def get_market_config(
self, market: Union[str, int], timeout_ms: Optional[int] = None
) -> MarketConfig:
print("market", market)
request = _create_get_market_config_request(market)
response = await self._market_client.get_market_configs(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.market_config
```
```python theme={null}
async def get_markets(self, timeout_ms: Optional[int] = None) -> list[MarketConfig]:
request = _create_get_markets_request()
response = await self._market_client.get_markets(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.markets)
```
```python theme={null}
async def get_market_id_by_name(
self, market: str, timeout_ms: Optional[int] = None
) -> int:
request = _create_get_market_id_by_name_request(market)
response = await self._market_client.get_market_id_by_name(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.market_id
```
```python theme={null}
async def get_market_summary(
self, market: Union[str, int], timeout_ms: Optional[int] = None
) -> MarketSummary:
request = _create_get_market_summary_request(market)
response = await self._market_client.get_market_summary(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.market_summary
```
The market identifier (string symbol or integer ID)
Optional timeout in milliseconds
MarketSummary objects as they are received from the stream
```python theme={null}
async def stream_market_summary(self, market: Union[str, int], timeout_ms: Optional[int]=None) -> AsyncIterator[MarketSummary]:
request = _create_get_market_summary_streaming_request(market)
async for response in self._market_client.get_market_summary_streaming(request, timeout_ms=timeout_ms or self.timeout_ms):
yield response.market_summary
```
Optional timeout in milliseconds
Lists of MarketSummary objects as they are received from the stream
```python theme={null}
async def stream_all_market_summaries(self, timeout_ms: Optional[int]=None) -> AsyncIterator[list[MarketSummary]]:
request = _create_get_all_market_summaries_streaming_request()
async for response in self._market_client.get_all_market_summaries_streaming(request, timeout_ms=timeout_ms or self.timeout_ms):
yield list(response.market_summaries)
```
```python theme={null}
async def get_order_book(
self,
market: Union[str, int],
tick_size: float,
usd_price: bool,
levels: int = None,
limit: int = 50,
offset: int = 0,
timeout_ms: Optional[int] = None,
) -> OrderBook:
request = _create_get_order_book_request(
market, tick_size, usd_price, levels, limit, offset
)
response = await self._orderbook_client.get_order_book(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.order_book
```
The trader's wallet address
Optional request timeout in milliseconds
Portfolio object containing all metrics, positions, and orders
```python theme={null}
async def get_portfolio(self, owner: str, timeout_ms: Optional[int]=None) -> Portfolio:
request = _create_get_portfolio_request(owner)
response = await self._portfolio_client.get_portfolio(request, timeout_ms=timeout_ms or self.timeout_ms)
return response.portfolio
```
```python theme={null}
async def get_position_rebate(
self,
owner: str,
position_id: int,
market_id: int,
timeout_ms: Optional[int] = None,
) -> int:
request = _create_get_position_fee_info(owner, position_id, market_id)
response = await self._position_client.get_position_fee_info(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.estimatedRebate
```
```python theme={null}
async def get_position_fees(
self,
owner: str,
position_id: int,
market_id: int,
timeout_ms: Optional[int] = None,
) -> int:
request = _create_get_position_fee_info(owner, position_id, market_id)
response = await self._position_client.get_position_fee_info(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.fees
```
```python theme={null}
async def get_open_order_fees(
self,
market_id: int,
margin: float,
leverage: float,
amount: float,
price: float,
timeout_ms: Optional[int] = None,
) -> int:
request = _create_get_open_order_fees(
market_id, margin, leverage, amount, price
)
response = await self._order_client.get_estimated_fees(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.fees
```