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
)
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_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)
object
required
Tuple of (orders_list, total_count)
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)
object
required
The market identifier (string symbol or integer ID)
float
required
Price threshold for executable orders
object
Optional side filter (from Side enum: 0=LONG, 1=SHORT)
object
Optional limit on number of results
object
Optional timeout in milliseconds
object
required
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_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)
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,
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)
object
required
The market identifier (string symbol or integer ID).
int
required
The position size.
int
required
The leverage used for the position.
bool
required
Whether the position is long (True) or short (False).
int
required
The entry price of the position.
object
Optional timeout in milliseconds.
float
required
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
object
required
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
object
required
The market identifier (string symbol or integer ID)
object
Optional timeout in milliseconds
object
required
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
object
Optional timeout in milliseconds
object
required
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,
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
str
required
The trader’s wallet address
object
Optional request timeout in milliseconds
Portfolio
required
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