Table of Contents
Base URLs
Copy 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:
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Get Order Status by Order ID
Retrieves the status of a specific order by its ID.
Endpoint : GET /getOrderStatusByOrderId
Copy 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
Copy 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
Copy 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
Copy 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
Copy 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
Copy # 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
}