Kana Labs
  • Getting Started
    • Welcome to Kana Labs
  • CROSS CHAIN SWAPS
    • AMM DEX Aggregator
  • INTEGRATE KANA WIDGET
    • Kana Widget
      • Install Widget
      • Configure Widget
      • Configure Aptos Keyless
  • Web3 Aggregator SDK
    • Web3 Aggregator SDK
      • Installation
      • SameChain
      • Cross Chain Swap
      • Aggregator API's
  • SPOT TRADING PLATFORM
    • Kana Trade
      • API Docs
  • PERPETUAL FUTURES
    • Kana Perps
      • Getting Started
        • Mint Tokens on Testnet
      • Breaking Down Kana Perps
        • Assets Supported
        • Order Types
        • Orderbook
        • Funding Rate
        • Leverage
        • Margin and Liquidation
        • Hedge Mode
          • Hedging a Short-Term 2-3% Price Decline
          • Dual Positioning for Flexible Profit-Taking
        • Trading Fees
      • API Docs
        • Installation Setup
        • Kana Perps Typescript REST API
        • Kana Perps Python Websocket API
        • Kana Perps Python REST API
        • Steps to place an order
        • Perps Contract Error Codes
        • Websocket Connection
        • Supported Markets
    • Perpetual Exchange(2)
      • Technical Architecture
      • Features
        • 1-Click Trading in Kana Perps
        • Delegation
      • Tutorials
        • Getting Started
        • Connecting Wallet & Enabling “One Click Transaction”
        • Deposit & Withdraw Tokens
        • Placing a Market Trade Order
        • Placing a Limit Trade Order
        • Partially & Fully Closing a Live Trade Order
        • Adding Margin to an Open Position
        • Defining Take Profit & Stop Loss
      • For Developers & MMs
        • Installation setup
        • Available Markets
        • Integrating REST APIs
          • Typescript
          • Python
          • Walkthrough
        • Integrating WebSockets
        • FAQ
          • Faucets
  • SPOT & PERP APIs
    • Trading APIs
      • Kana Trade API
      • Kana Perps API
        • Installation Setup
        • Example setup functions
        • Kana Perps Typescript REST API
        • Kana Perps Websocket API
        • Kana Perps Python Websocket API
        • Kana Perps Python REST API
        • Steps to place an order
  • PAYMASTER SERVICE
    • Kana Paymaster For Aptos and Supra
      • How it works?
      • How to Register?
      • Deposit Allowance
      • Manage Users
      • Paymaster SDK Tutorial (Typescript)
      • Paymaster API
      • Module & Function Whitelist
      • Subscription - Coming soon
      • FAQS
  • PERPETUAL OPTIONS
    • OPerps
  • Tokenomics & Governance
    • Kana Labs Tokenomics
  • REWARDS & REFERRAL PROGRAM
    • Rewards Program
      • Reward Program Season 1
      • Reward Program Season 2
      • How to Keep Track of Your Points?
      • Where to find the Missions Dashboard?
  • Referral Program
    • How to Generate Referral Link? (For the Referrer)
    • How to map your wallet to the invite IDs? (For the invited users)
Powered by GitBook
On this page
  • Table of Contents
  • Base URLs
  • Authentication
  • Core Concepts
  • Account Management
  • Market Data
  • Order Management
  • Position Management
  • Trade History
  • Utility Endpoints
  1. PERPETUAL FUTURES
  2. Perpetual Exchange(2)
  3. For Developers & MMs
  4. Integrating REST APIs

Python

PreviousTypescriptNextWalkthrough

Last updated 4 days ago

Table of Contents

Base URLs

BASE_URLS = {
    "TESTNET": "https://perps-tradeapi.kanalabs.io",
    "MAINNET": "https://perps-tradeapi.kana.trade"
}

NODE_URLS = {
    "TESTNET": "https://api.testnet.aptoslabs.com/v1",
    "MAINNET": "https://api.mainnet.aptoslabs.com/v1"
}

Authentication

All endpoints require an API key in the request headers:

import os
from dotenv import load_dotenv

load_dotenv()

headers = {"x-api-key": os.getenv("API_KEY")}

Core Concepts

Profile Address

On-chain trades use a profile address rather than a wallet address. You can fetch it using the /getProfileAddress endpoint.

Endpoint: GET /getProfileAddress

import requests

def get_profile_address(base_url, user_address):
    response = requests.get(
        f"{base_url}/getProfileAddress", 
        params={"userAddress": user_address}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Account Management

Get Wallet Balance

Retrieves the wallet balance for a given user address.

Endpoint: GET /getWalletAccountBalance

def get_wallet_balance(base_url, user_address):
    response = requests.get(
        f"{base_url}/getWalletAccountBalance", 
        params={"userAddress": user_address}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Profile Balance

Retrieves the profile balance snapshot for a given user address.

Endpoint: GET /getProfileBalanceSnapshot

def get_profile_balance(base_url, user_address):
    response = requests.get(
        f"{base_url}/getProfileBalanceSnapshot", 
        params={"userAddress": user_address}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Net Profile Balance

This endpoint returns the net profile balance, which includes both the available balance in trading account and any pending balances from closed positions on all markets.

Endpoint: GET /getNetProfileBalance

def get_net_profile_balance(base_url, user_address):
    response = requests.get(
        f"{base_url}/getNetProfileBalance", 
        params={"userAddress": user_address}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Deposit Funds

Deposits funds from wallet to trading profile.

Endpoint: GET /deposit

import asyncio
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionPayload
from aptos_sdk.bcs import Serializer

async def deposit_funds(base_url, node_url, user_address, amount, private_key):
    response = requests.get(
        f"{base_url}/deposit", 
        params={"userAddress": user_address, "amount": amount}, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [user_address, int(payload["functionArguments"][1])]
    payload["argumentTypes"] = [Serializer.struct, Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Withdraw Funds

Withdraws funds from trading profile to wallet for a specific market.

Endpoint: GET /withdrawSpecifiMarket

async def withdraw_funds(base_url, node_url, user_address, market_id, amount, private_key):
    response = requests.get(
        f"{base_url}/withdrawSpecifiMarket", 
        params={"userAddress": user_address, "marketId": market_id, "amount": amount}, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [user_address, int(payload["functionArguments"][1]), int(payload["functionArguments"][2])]
    payload["argumentTypes"] = [Serializer.struct, Serializer.u64, Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Market Data

Get Market Info

Retrieves information about a specific market.

Endpoint: GET /getMarketInfo

def get_market_info(base_url, market_id):
    response = requests.get(
        f"{base_url}/getMarketInfo", 
        params={"marketId": market_id}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"][0]

Get Market Price

Gets the current price for a specific market.

Endpoint: GET /getMarketPrice

def get_market_price(base_url, market_id):
    response = requests.get(
        f"{base_url}/getMarketPrice", 
        params={"marketId": market_id}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Last Execution Price

Retrieves the last execution price for a market.

Endpoint: GET /getLastPlacedPrice

def get_last_execution_price(base_url, market_id):
    response = requests.get(
        f"{base_url}/getLastPlacedPrice", 
        params={"marketId": market_id}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Perpetual Market Info

Retrieves information about perpetual assets, optionally filtered by market ID or base name.

Endpoint: GET /getPerpetualAssetsInfo

def get_perpetual_assets_info(base_url, market_id=None, base_name=None):
    params = {}
    if market_id:
        params["marketId"] = market_id
    if base_name:
        params["baseName"] = base_name
    response = requests.get(
        f"{base_url}/getPerpetualAssetsInfo", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get All Markets

Retrieves information about all available markets.

Endpoint: GET /getPerpetualAssetsInfo/allMarkets

def get_all_markets(base_url):
    response = requests.get(
        f"{base_url}/getPerpetualAssetsInfo/allMarkets", 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Order Management

Place Limit Order

Places a limit order on the exchange.

Endpoint: GET /placeLimitOrder

async def place_limit_order(base_url, node_url, params, private_key):
    response = requests.get(
        f"{base_url}/placeLimitOrder", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [
        int(payload["functionArguments"][0]),
        payload["functionArguments"][1].lower() == "true",
        payload["functionArguments"][2].lower() == "true",
        *map(int, payload["functionArguments"][3:])
    ]
    payload["argumentTypes"] = [
        Serializer.u64, Serializer.bool, Serializer.bool, 
        Serializer.u64, Serializer.u64, Serializer.u64, 
        Serializer.u8, Serializer.u64, Serializer.u64
    ]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Place Market Order

Places a market order on the exchange.

Endpoint: GET /placeMarketOrder

async def place_market_order(base_url, node_url, params, private_key):
    response = requests.get(
        f"{base_url}/placeMarketOrder", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [
        int(payload["functionArguments"][0]),
        payload["functionArguments"][1].lower() == "true",
        payload["functionArguments"][2].lower() == "true",
        *map(int, payload["functionArguments"][3:])
    ]
    payload["argumentTypes"] = [Serializer.u64, Serializer.bool, Serializer.bool] + [Serializer.u64] * (len(payload["functionArguments"]) - 3)
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Place Multiple Orders

Places multiple orders in a single transaction.

Endpoint: POST /placeMultipleOrders

async def place_multiple_orders(base_url, node_url, body, private_key):
    response = requests.post(
        f"{base_url}/placeMultipleOrders", 
        json=body, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["argumentTypes"] = [
        Serializer.u64,
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u8),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u64)
    ]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Cancel Multiple Orders

Cancels multiple orders in a single transaction.

Endpoint: POST /cancelMultipleOrders

async def cancel_multiple_orders(base_url, node_url, body, private_key):
    response = requests.post(
        f"{base_url}/cancelMultipleOrders", 
        json=body, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]
    payload["functionArguments"][1] = [int(x) for x in payload["functionArguments"][1]]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["argumentTypes"] = [
        Serializer.u64,
        Serializer.sequence_serializer(Serializer.u128),
        Serializer.sequence_serializer(Serializer.bool)
    ]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Cancel and Place Multiple Orders

Cancels existing orders and places new ones in a single transaction.

Endpoint: POST /cancelAndPlaceMultipleOrders

async def cancel_and_place_multiple_orders(base_url, node_url, body, private_key):
    response = requests.post(
        f"{base_url}/cancelAndPlaceMultipleOrders", 
        json=body, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]
    payload["functionArguments"][1] = [int(x) for x in payload["functionArguments"][1]]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["argumentTypes"] = [
        Serializer.u64,
        Serializer.sequence_serializer(Serializer.u128),
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.bool),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u8),
        Serializer.sequence_serializer(Serializer.u64),
        Serializer.sequence_serializer(Serializer.u64)
    ]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Position Management

Get Positions

Retrieves current positions for a user, optionally filtered by market ID.

Endpoint: GET /getPositions

def get_positions(base_url, user_address, market_id=None):
    params = {"userAddress": user_address}
    if market_id:
        params["marketId"] = market_id
    response = requests.get(
        f"{base_url}/getPositions", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Update Take Profit

Updates the take profit price for a position.

Endpoint: GET /updateTakeProfit

async def update_take_profit(base_url, node_url, market_id, trade_side, new_take_profit_price, private_key):
    params = {
        "marketId": market_id, 
        "tradeSide": trade_side, 
        "newTakeProfitPrice": new_take_profit_price
    }
    response = requests.get(
        f"{base_url}/updateTakeProfit", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [
        int(payload["functionArguments"][0]),
        payload["functionArguments"][1].lower() == "true",
        int(payload["functionArguments"][2])
    ]
    payload["argumentTypes"] = [Serializer.u64, Serializer.bool, Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Update Stop Loss

Updates the stop loss price for a position.

Endpoint: GET /updateStopLoss

async def update_stop_loss(base_url, node_url, market_id, trade_side, new_stop_loss_price, private_key):
    params = {
        "marketId": market_id, 
        "tradeSide": trade_side, 
        "newStopLossPrice": new_stop_loss_price
    }
    response = requests.get(
        f"{base_url}/updateStopLoss", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [
        int(payload["functionArguments"][0]),
        payload["functionArguments"][1].lower() == "true",
        int(payload["functionArguments"][2])
    ]
    payload["argumentTypes"] = [Serializer.u64, Serializer.bool, Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Collapse Position

Closes a position immediately at market price.

Endpoint: GET /collapsePosition

async def collapse_position(base_url, node_url, market_id, private_key):
    response = requests.get(
        f"{base_url}/collapsePosition", 
        params={"marketId": market_id}, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [int(payload["functionArguments"][0])]
    payload["argumentTypes"] = [Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Trade History

Get Open Orders

Retrieves all open orders for a user, optionally filtered by market ID.

Endpoint: GET /getOpenOrders

def get_open_orders(base_url, user_address, market_id=None):
    params = {"userAddress": user_address}
    if market_id:
        params["marketId"] = market_id
    response = requests.get(
        f"{base_url}/getOpenOrders", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get All Open Order Ids

Endpoint: GET /getAllOpenOrderIds

def get_all_open_order_ids(base_url, user_address, market_id=None):
    params = {"userAddress": user_address}
    if market_id:
        params["marketId"] = market_id
    response = requests.get(
        f"{base_url}/getAllOpenOrderIds", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Order Status By Order Id

Endpoint: GET /getOrderStatusByOrderId

def get_order_status_by_order_id(base_url, market_id, order_id):
    params = {"marketId": market_id, "orderId": order_id}
    response = requests.get(
        f"{base_url}/getOrderStatusByOrderId", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Order History

Retrieves order history for a user, optionally filtered by market ID.

Endpoint: GET /getOrderHistory

def get_order_history(base_url, user_address, market_id=None):
    params = {"userAddress": user_address}
    if market_id:
        params["marketId"] = market_id
    response = requests.get(
        f"{base_url}/getOrderHistory", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get All Trades

Retrieves all trades for a specific market.

Endpoint: GET /getAllTrades

def get_all_trades(base_url, market_id):
    response = requests.get(
        f"{base_url}/getAllTrades", 
        params={"marketId": market_id}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Fills

Retrieves fill history for a specific market and address within a time range.

Endpoint: GET /getFills

def get_fills(base_url, market_id, address, from_time, to_time, order_id=None):
    params = {
        "marketId": market_id, 
        "address": address, 
        "from": from_time, 
        "to": to_time
    }
    if order_id:
        params["orderId"] = order_id
    response = requests.get(
        f"{base_url}/getFills", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get Order Status by Order ID

Retrieves the status of a specific order by its ID.

Endpoint: GET /getOrderStatusByOrderId

def get_order_status_by_order_id(base_url, market_id, order_id):
    response = requests.get(
        f"{base_url}/getOrderStatusByOrderId", 
        params={"marketId": market_id, "orderId": order_id}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Get All Open Order IDs

Endpoint: GET /getAllOpenOrderIds

def get_all_open_order_ids(base_url, user_address, market_id=None):
    params = {"userAddress": user_address}
    if market_id:
        params["marketId"] = market_id
    response = requests.get(
        f"{base_url}/getAllOpenOrderIds", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Retrieves all open order IDs for a user, optionally filtered by market ID.

Add Margin

Endpoint: GET /addMargin

async def add_margin(base_url, node_url, market_id, trade_side, amount, private_key):
    params = {
        "marketId": market_id, 
        "tradeSide": trade_side, 
        "amount": amount
    }
    response = requests.get(
        f"{base_url}/addMargin", 
        params=params, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [
        int(payload["functionArguments"][0]),
        payload["functionArguments"][1].lower() == "true",
        int(payload["functionArguments"][2])
    ]
    payload["argumentTypes"] = [Serializer.u64, Serializer.bool, Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Settle Pnl

This API returns a transaction payload used to manually (or automatically, on user interactions) refresh the user's position and settle the Pnl for a specific market, allowing the user to claim any unreleased Pnl.

Endpoint: GET /settlePnl

async def settle_pnl(base_url, node_url, user_address, market_id, private_key):
    response = requests.get(
        f"{base_url}/settlePnl", 
        params={"userAddress": user_address, "marketId": market_id}, 
        headers=headers
    )
    response.raise_for_status()
    payload = response.json()["data"]

    rest_client = RestClient(node_url)
    account = Account.load_key(bytes.fromhex(private_key.lstrip("0x")))
    
    payload["functionArguments"] = [user_address, int(payload["functionArguments"][1])]
    payload["argumentTypes"] = [Serializer.struct, Serializer.u64]
    
    module, function_id = payload["function"].split("::")[:-1], payload["function"].split("::")[-1]
    entry_function = EntryFunction.natural(
        "::".join(module), 
        function_id, 
        [], 
        [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
    )
    
    transaction = await rest_client.create_bcs_signed_transaction(account, TransactionPayload(entry_function))
    txn_hash = await rest_client.submit_bcs_transaction(transaction)
    await rest_client.wait_for_transaction(txn_hash)
    return txn_hash

Utility Endpoints

Get Account APT Balance

Retrieves the APT balance for a user's account.

Endpoint: GET /getAccountAptBalance

def get_account_apt_balance(base_url, user_address):
    response = requests.get(
        f"{base_url}/getAccountAptBalance", 
        params={"userAddress": user_address}, 
        headers=headers
    )
    response.raise_for_status()
    return response.json()["data"]

Helper Types

# Order type constants
ORDER_TYPES = {
    "OPEN_LONG": 1,
    "OPEN_SHORT": 2,
    "INCREASE_LONG": 3,
    "INCREASE_SHORT": 4,
    "DECREASE_LONG": 5,
    "DECREASE_SHORT": 6,
    "CLOSE_LONG": 7,
    "CLOSE_SHORT": 8
}

# Order restriction types
RESTRICTION_TYPES = {
    "NO_RESTRICTION": 0,
    "FILL_OR_ABORT": 1,
    "POST_OR_ABORT": 3
}

Base URLs
Authentication
Core Concepts
Account Management
Market Data
Order Management
Position Management
Trade History
Utility Endpoints