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._adl_client = ADLServiceClient(address=base_url, session=session)
self._portfolio_client = PortfolioServiceClient(
address=base_url, session=session
)
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
async def get_orders(
self,
address: str,
market: Union[str, int, None] = None,
timeout_ms: Optional[int] = None,
) -> list[Order]:
request = _create_get_orders_request(address, market=market)
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)
async def get_all_orders(self, page_size_limit: int, offset_amount: int, timeout_ms: Optional[int]=None) -> tuple[list[Order], int]:
request = _create_all_orders_request(page_size_limit, offset_amount)
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)
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)
async def get_trade_history(
self,
address: str,
startTime: Timestamp,
endTime: Timestamp,
timeout_ms: Optional[int] = None,
) -> list[Order]:
request = _create_get_trade_history(address, startTime, endTime)
response = await self._order_client.get_trade_history(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.orders)
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
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
async def get_adl(
self,
amountToSeize: int,
timeout_ms: Optional[int] = None,
) -> list[ADLPerMarket]:
request = _create_get_adl(amountToSeize)
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.
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)
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)
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)
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
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)
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
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
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
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)
async def get_order_book(
self,
market: Union[str, int],
tick_size: float,
usd_price: bool,
timeout_ms: Optional[int] = None,
) -> OrderBook:
request = _create_get_order_book_request(market, tick_size, usd_price)
response = await self._orderbook_client.get_order_book(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.order_book
async def get_liquidation(
self,
market: Union[str, int],
price_offset: int,
timeout_ms: Optional[int] = None,
) -> LiquidationPerMarket:
request = _create_get_liquidation_request(market, price_offset)
response = await self._liquidation_client.get_liquidations(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.liquidations_per_market
The trader’s wallet address
Optional request timeout in milliseconds
Portfolio object containing all metrics, positions, and orders
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
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
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
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