def __init__(
self, base_url: str, session: httpx.Client, 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 = OrderServiceClientSync(address=base_url, session=session)
self._position_client = PositionServiceClientSync(
address=base_url, session=session
)
self._permissions_client = PermissionServiceClientSync(
address=base_url, session=session
)
self._market_client = MarketServiceClientSync(address=base_url, session=session)
self._orderbook_client = OrderBookServiceClientSync(
address=base_url, session=session
)
self._liquidation_client = LiquidationServiceClientSync(
address=base_url, session=session
)
self._portfolio_client = PortfolioServiceClientSync(
address=base_url, session=session
)
self._adl_client = ADLServiceClientSync(address=base_url, session=session)
self._fill_history_client = FillHistoryServiceClientSync(
address=base_url, session=session
)
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 = self._order_client.get_order(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.order
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)
response = self._order_client.get_orders(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.orders)
def get_open_positions(
self, address: str, timeout_ms: Optional[int] = None
) -> list[Position]:
request = _create_get_open_positions_request(address)
response = self._position_client.get_open_positions(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.positions)
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 = self._liquidation_client.get_liquidation_price_preview(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.liquidation_price
def get_market_config(
self, market: Union[str, int], timeout_ms: Optional[int] = None
) -> MarketConfig:
request = _create_get_market_config_request(market)
response = self._market_client.get_market_configs(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.market_config
def get_markets(self, timeout_ms: Optional[int] = None) -> list[MarketConfig]:
request = _create_get_markets_request()
response = self._market_client.get_markets(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.markets)
def get_market_summary(
self, market: Union[str, int], timeout_ms: Optional[int] = None
) -> MarketSummary:
request = _create_get_market_summary_request(market)
response = self._market_client.get_market_summary(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.market_summary
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 = self._market_client.get_market_id_by_name(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.market_id
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
def stream_market_summary(self, market: Union[str, int], timeout_ms: Optional[int]=None) -> Iterator[MarketSummary]:
request = _create_get_market_summary_streaming_request(market)
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
def stream_all_market_summaries(self, timeout_ms: Optional[int]=None) -> Iterator[list[MarketSummary]]:
request = _create_get_all_market_summaries_streaming_request()
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)
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 = 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
def get_portfolio(self, owner: str, timeout_ms: Optional[int]=None) -> Portfolio:
request = _create_get_portfolio_request(owner)
response = self._portfolio_client.get_portfolio(request, timeout_ms=timeout_ms or self.timeout_ms)
return response.portfolio
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 = self._position_client.get_position_fee_info(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.estimatedRebate
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 = self._position_client.get_position_fee_info(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.fees
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 = self._order_client.get_estimated_fees(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return response.fees
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 = self._fill_history_client.get_user_fills(
request, timeout_ms=timeout_ms or self.timeout_ms
)
return list(response.fills)