For all the API endpoints below, pass your wallet address as the userAddress parameter.
To track your transactions on-chain (e.g., via Aptos Explorer or Aptos Scan), you must use your profile address, as all on-chain trades are executed using this profile address.
To obtain your profile address, use the /getProfileAddress endpoint and provide your wallet address as the userAddress parameter.
How to Fetch the Profile Address:
Example endpoint:Example code:
get-profile-address
✅ Testnet API Endpoint URL
https://perps-tradeapi.kanalabs.io
✅ Mainnet API Endpoint URL
https://perps-tradeapi.kana.trade
Please ensure that the correct network is used: for Testnet, set NODE_URL = "https://api.testnet.aptoslabs.com/v1" and for Mainnet, set NODE_URL = "https://api.mainnet.aptoslabs.com/v1"
1. Get Market Info
Endpoint URL:
Method:
GET
Query Parameters:
marketId (Required) - The ID of the market you want information about.
Example Request:
GET https://perps-tradeapi.kanalabs.io/getMarketInfo?marketId=501
amount (Required) - The amount of the quote coin to deposit.
userAddress (Required) - The address of the user making the deposit.
Example Request:
GET https://perps-tradeapi.kanalabs.io/deposit?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&amount=50000000
Example Response:
Status Code: 200 OK
Response Body:
{
"success": true,
"message": "Deposit payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::deposit",
"functionArguments": [
"0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770",
50000000
],
"typeArguments": []
}
}
Example Code to Depsoit a quote coin:
The following Python script demonstrates how to call the Get Deposit API using the requests library.
import asyncio
import os
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.account_address import AccountAddress # Import this
from aptos_sdk.type_tag import TypeTag, StructTag
import requests
from dotenv import load_dotenv
from typing import Any, List
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url, params, headers):
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status()
return response.json().get("data")
except requests.exceptions.RequestException as e:
print(f"Error fetching payload: {e}")
return None
def create_transaction_function_arguments(self, arguments: List[Any], types: List[Serializer]) -> List[TransactionArgument]:
if len(arguments) != len(types):
raise ValueError("Arguments and types length mismatch.")
return [TransactionArgument(arg, serializer) for arg, serializer in zip(arguments, types)]
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
function_information = payload["function"].split("::")
module = "::".join(function_information[:-1])
function_id = function_information[-1]
function_arguments = self.create_transaction_function_arguments(
payload["functionArguments"],
payload["argumentTypes"]
)
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
entry_function = EntryFunction.natural(
module=module,
function=function_id,
ty_args=type_arguments,
args=function_arguments,
)
return TransactionPayload(payload=entry_function)
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_transaction_request = await self.rest_client.create_bcs_signed_transaction(
sender=self.account, payload=transaction_payload
)
txn_hash = await self.rest_client.submit_bcs_transaction(
signed_transaction=signed_transaction_request
)
await self.rest_client.wait_for_transaction(txn_hash=txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
rest_client = RestClient(NODE_URL)
private_key_hex = os.getenv('APTOS_PRIVATEKEY')
if private_key_hex.startswith("0x"):
private_key_hex = private_key_hex[2:]
private_key_bytes = bytes.fromhex(private_key_hex)
account = Account.load_key(private_key_bytes)
API_URL = "https://perps-tradeapi.kanalabs.io/deposit"
PARAMS = {
"userAddress":"0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770",
"amount": 50
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
AccountAddress.from_str(payload_data["functionArguments"][0]),
int(payload_data["functionArguments"][1]),
]
payload_data["argumentTypes"] = [Serializer.struct, Serializer.u64]
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Transaction hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())
5. Withdraw Specific Market
Method:
GET
Query Parameters:
marketId (Required) - The ID of the market you want to withdraw.
amount (Required) - The amount of the quote coin to withdraw.
userAddress (Required) - The address of the user making the withdraw.
Example Request:
GET https://perps-tradeapi.kanalabs.io/withdrawSpecifiMarket?marketId=501&userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&amount=50
Example Response:
Status Code: 200 OK
Response Body:
{
"success": true,
"message": "Withdraw specific market payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::withdraw_specific_market",
"functionArguments": [
"0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770",
"501",
50000000
],
"typeArguments": []
}
}
Example Code to Withdraw a quote coin:
The following Python script demonstrates how to call the Get Withdraw API using the requests library.
import asyncio
import os
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.account_address import AccountAddress # Import this
from aptos_sdk.type_tag import TypeTag, StructTag
import requests
from dotenv import load_dotenv
from typing import Any, List
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url, params, headers):
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status()
return response.json().get("data")
except requests.exceptions.RequestException as e:
print(f"Error fetching payload: {e}")
return None
def create_transaction_function_arguments(self, arguments: List[Any], types: List[Serializer]) -> List[TransactionArgument]:
if len(arguments) != len(types):
raise ValueError("Arguments and types length mismatch.")
return [TransactionArgument(arg, serializer) for arg, serializer in zip(arguments, types)]
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
function_information = payload["function"].split("::")
module = "::".join(function_information[:-1])
function_id = function_information[-1]
function_arguments = self.create_transaction_function_arguments(
payload["functionArguments"],
payload["argumentTypes"]
)
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
entry_function = EntryFunction.natural(
module=module,
function=function_id,
ty_args=type_arguments,
args=function_arguments,
)
return TransactionPayload(payload=entry_function)
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_transaction_request = await self.rest_client.create_bcs_signed_transaction(
sender=self.account, payload=transaction_payload
)
txn_hash = await self.rest_client.submit_bcs_transaction(
signed_transaction=signed_transaction_request
)
await self.rest_client.wait_for_transaction(txn_hash=txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
rest_client = RestClient(NODE_URL)
private_key_hex = os.getenv('APTOS_PRIVATEKEY')
if private_key_hex.startswith("0x"):
private_key_hex = private_key_hex[2:]
private_key_bytes = bytes.fromhex(private_key_hex)
account = Account.load_key(private_key_bytes)
API_URL = "https://perps-tradeapi.kanalabs.io/withdrawSpecifiMarket"
PARAMS = {
"userAddress": "0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770",
"marketId": 501,
"amount": 50
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
AccountAddress.from_str(payload_data["functionArguments"][0]),
int(payload_data["functionArguments"][1]),
int(payload_data["functionArguments"][1]),
]
payload_data["argumentTypes"] = [Serializer.struct, Serializer.u64, Serializer.u64]
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Transaction hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())
6. Place Limit Order
Method:
GET
Query Parameters:
marketId (Required) - The ID of the market where the limit order will be placed.
tradeSide (Required) - Indicates the trade side:
true for the long side.
false for the short side.
direction (Required) - Indicates the direction of the trade:
false to open a position.
true to close a position.
size (Required) - The size of the order.
price (Required) - The price at which the order is to be placed.
leverage (Required) - The leverage to be used for the order.
restriction - Specifies the type of order restriction. It is an optional parameter with a default value of 0.
Possible values are:
0 - NO_RESTRICTION: Optionally fill as a taker, then post to the book as a maker.
1 - FILL_OR_ABORT: Abort if any size posts as a maker (only fill).
3 - POST_OR_ABORT: Abort if any size fills as a taker (only post).
If restriction is not provided, it defaults to 0. It can either be a number or left undefined.
takeProfit - The take profit value is optional. If not provided, it defaults to 0. It can either be a number or left undefined.
stopLoss - The stop loss value is also optional. If not provided, it defaults to 0. It can either be a number or left undefined.
Example Request:
GET https://perps-tradeapi.kanalabs.io/placeLimitOrder/?marketId=501&tradeSide=true&direction=false&size=10000&price=3000&leverage=3
Example Response:
Status Code: 200 OK
Response Body:
{
"success": true,
"message": "Limit Order payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::place_limit_order",
"functionArguments": [
"501",
"true",
"false",
1000000,
8000000,
"5",
0,
0,
0
],
"typeArguments": []
}
}
Example Code to Place a limit order:
The following Python script demonstrates how to call the Get Place Limit Order API using the requests library.
import asyncio
import os
import requests
from dotenv import load_dotenv
from typing import List
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.type_tag import TypeTag, StructTag
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url, params, headers):
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status()
return response.json().get("data")
except requests.RequestException as e:
print(f"Error fetching payload: {e}")
return None
def create_transaction_function_arguments(self, arguments: List, types: List[Serializer]) -> List[TransactionArgument]:
if len(arguments) != len(types):
raise ValueError("Arguments and types length mismatch.")
return [TransactionArgument(arg, serializer) for arg, serializer in zip(arguments, types)]
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
module, function_id = "::".join(payload["function"].split("::")[:-1]), payload["function"].split("::")[-1]
function_arguments = self.create_transaction_function_arguments(payload["functionArguments"], payload["argumentTypes"])
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
entry_function = EntryFunction.natural(module=module, function=function_id, ty_args=type_arguments, args=function_arguments)
return TransactionPayload(payload=entry_function)
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_transaction = await self.rest_client.create_bcs_signed_transaction(self.account, transaction_payload)
txn_hash = await self.rest_client.submit_bcs_transaction(signed_transaction)
await self.rest_client.wait_for_transaction(txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
private_key_hex = os.getenv('APTOS_PRIVATEKEY')
if not private_key_hex:
print("Error: APTOS_PRIVATEKEY is missing in .env file.")
return
account = Account.load_key(bytes.fromhex(private_key_hex[2:] if private_key_hex.startswith("0x") else private_key_hex))
rest_client = RestClient(NODE_URL)
API_URL = "https://perps-tradeapi.kanalabs.io/placeLimitOrder"
PARAMS = {"marketId": 501, "tradeSide": True, "direction": False, "size": 2, "price": 7, "leverage": 2}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
int(payload_data["functionArguments"][0]),
payload_data["functionArguments"][1].lower() == "true",
payload_data["functionArguments"][2].lower() == "true",
*map(int, payload_data["functionArguments"][3:])
]
payload_data["argumentTypes"] = [
Serializer.u64, Serializer.bool, Serializer.bool, Serializer.u64,
Serializer.u64, Serializer.u64, Serializer.u8, Serializer.u64, Serializer.u64
]
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Hash: {txn_hash}")
except Exception as e:
print(f"Transaction process error: {e}")
if __name__ == "__main__":
asyncio.run(main())
7. Place Market Order
Method:
GET
Query Parameters:
marketId (Required) - The ID of the market where the limit order will be placed.
tradeSide (Required) - Indicates the trade side:
true for the long side.
false for the short side.
direction (Required) - Indicates the direction of the trade:
false to open a position.
true to close a position.
size (Required) - The size of the order.
leverage (Required) - The leverage to be used for the order.
takeProfit - The take profit value is optional. If not provided, it defaults to 0. It can either be a number or left undefined.
stopLoss - The stop loss value is also optional. If not provided, it defaults to 0. It can either be a number or left undefined.
Example Request:
GET https://perps-tradeapi.kanalabs.io/placeMarketOrder?marketId=66&tradeSide=false&direction=true&size=1000&leverage=20
Example Response:
Status Code: 200 OK
Response Body:
{
"success": true,
"message": "Market Order payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::place_market_order",
"functionArguments": [
"501",
"true",
"false",
1000000,
"5",
0,
0
],
"typeArguments": []
}
}
Example Code to Place a market order:
The following Python script demonstrates how to call the Get Place Market Order API using the requests library.
import asyncio
import os
import requests
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.type_tag import TypeTag, StructTag
from dotenv import load_dotenv
from typing import List
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url: str, params: dict, headers: dict) -> dict | None:
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status()
return response.json().get("data")
except requests.RequestException as e:
print(f"Error fetching payload: {e}")
return None
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
module, function_id = "::".join(payload["function"].split("::")[:-1]), payload["function"].split("::")[-1]
type_arguments = [TypeTag(StructTag.from_str(arg)) for arg in payload["typeArguments"]]
function_arguments = [TransactionArgument(arg, serializer) for arg, serializer in zip(payload["functionArguments"], payload["argumentTypes"])]
return TransactionPayload(payload=EntryFunction.natural(module, function_id, type_arguments, function_arguments))
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_txn = await self.rest_client.create_bcs_signed_transaction(sender=self.account, payload=transaction_payload)
txn_hash = await self.rest_client.submit_bcs_transaction(signed_transaction=signed_txn)
await self.rest_client.wait_for_transaction(txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
API_URL = "https://perps-tradeapi.kanalabs.io/placeMarketOrder"
HEADERS = {"x-api-key": os.getenv('API_KEY')}
PARAMS = {"marketId": 501, "tradeSide": True, "direction": False, "size": 2, "leverage": 2}
rest_client = RestClient(NODE_URL)
private_key = os.getenv('APTOS_PRIVATEKEY', "").lstrip("0x")
if not private_key:
print("Error: APTOS_PRIVATEKEY is missing in .env file.")
return
account = Account.load_key(bytes.fromhex(private_key))
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
int(payload_data["functionArguments"][0]),
payload_data["functionArguments"][1].lower() == "true",
payload_data["functionArguments"][2].lower() == "true",
*map(int, payload_data["functionArguments"][3:])
]
payload_data["argumentTypes"] = [Serializer.u64, Serializer.bool, Serializer.bool] + [Serializer.u64] * (len(payload_data["functionArguments"]) - 3)
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())
8. Cancel Multiple Orders
Method:
POST
Request Body:
marketId: The ID of the market for which the orders will be canceled.
cancelOrderIds: A list of order IDs to cancel.
orderSides: The sides of the orders to cancel (true for long, false for short).
Example Code to cancel and Place a multiple orders:
The following TypeScript/Node.js script demonstrates how to call the Get Cancel And Place Multiple Orders API using the axios library.
import asyncio
import os
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.type_tag import TypeTag, StructTag
import requests
from dotenv import load_dotenv
from typing import Any, List
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url: str, json_data: dict, headers: dict) -> dict:
try:
response = requests.post(api_url, json=json_data, headers=headers)
response.raise_for_status()
payload_data = response.json().get("data")
if payload_data is None:
print("No data returned from API.")
return {}
# Ensure correct integer representation for order IDs
payload_data["takeProfits"] = payload_data.get("takeProfits", [0] * len(json_data["sizes"]))
payload_data["stopLosses"] = payload_data.get("stopLosses", [0] * len(json_data["sizes"]))
payload_data["restrictions"] = payload_data.get("restrictions", [0] * len(json_data["sizes"]))
# Convert order ID lists from float to integer
payload_data["functionArguments"][1] = [int(x) for x in payload_data["functionArguments"][1]]
return payload_data
except requests.exceptions.RequestException as e:
print(f"Error fetching payload: {e}")
return {}
def create_transaction_function_arguments(self, arguments: List[Any], types: List[Serializer]) -> List[TransactionArgument]:
if len(arguments) != len(types):
raise ValueError("Arguments and types length mismatch.")
try:
return [TransactionArgument(arg, serializer) for arg, serializer in zip(arguments, types)]
except Exception as e:
print(f"Error creating transaction arguments: {e}")
raise
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
function_information = payload["function"].split("::")
module = "::".join(function_information[:-1])
function_id = function_information[-1]
if "argumentTypes" not in payload:
raise KeyError("'argumentTypes' missing in payload.")
function_arguments = self.create_transaction_function_arguments(
payload["functionArguments"],
payload["argumentTypes"]
)
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
entry_function = EntryFunction.natural(
module=module,
function=function_id,
ty_args=type_arguments,
args=function_arguments,
)
return TransactionPayload(payload=entry_function)
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_transaction_request = await self.rest_client.create_bcs_signed_transaction(
sender=self.account, payload=transaction_payload
)
txn_hash = await self.rest_client.submit_bcs_transaction(
signed_transaction=signed_transaction_request
)
await self.rest_client.wait_for_transaction(txn_hash=txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
rest_client = RestClient(NODE_URL)
private_key_hex = os.getenv('APTOS_PRIVATEKEY')
if private_key_hex.startswith("0x"):
private_key_hex = private_key_hex[2:]
private_key_bytes = bytes.fromhex(private_key_hex)
account = Account.load_key(private_key_bytes)
API_URL = "https://perps-tradeapi.kanalabs.io/cancelAndPlaceMultipleOrders"
BODY = {
"marketId": 501,
"cancelOrderIds": ["45877054622524009092208"], # Ensure orderId given as string
"orderSides": [True],
"orderTypes": [True],
"tradeSides": [True],
"directions": [False],
"sizes": [1.8],
"prices": [7.6868],
"leverages": [2]
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, BODY, HEADERS)
print("Before payload:", payload_data)
if not payload_data:
print("Failed to fetch payload data.")
return
# Ensure argumentTypes is included in payload_data
payload_data["argumentTypes"] = [
Serializer.u64, # marketId
Serializer.sequence_serializer(Serializer.u128),
Serializer.sequence_serializer(Serializer.bool),
Serializer.sequence_serializer(Serializer.bool), # orderTypes
Serializer.sequence_serializer(Serializer.bool), # tradeSides
Serializer.sequence_serializer(Serializer.bool), # directions
Serializer.sequence_serializer(Serializer.u64), # sizes
Serializer.sequence_serializer(Serializer.u64), # prices
Serializer.sequence_serializer(Serializer.u64), # leverages
Serializer.sequence_serializer(Serializer.u8), # restrictions
Serializer.sequence_serializer(Serializer.u64), # takeProfits
Serializer.sequence_serializer(Serializer.u64), # stopLosses
]
try:
print("After payload:", payload_data)
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Transaction hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())
10. Get Open Orders
Method:
GET
Query Parameters:
userAddress (Required) - The wallet address to retrieve open orders for.
marketId - The ID of the market to filter open orders. // Optional
Note : order_type Explanation:
OPEN_LONG (1): Opens a new long position.
OPEN_SHORT (2): Opens a new short position.
INCREASE_LONG (3): Increases the size of an existing long position.
INCREASE_SHORT (4): Increases the size of an existing short position.
DECREASE_LONG (5): Reduces the size of an existing long position.
DECREASE_SHORT (6): Reduces the size of an existing short position.
CLOSE_LONG (7): Closes an existing long position.
CLOSE_SHORT (8): Closes an existing short position.
Example Request:
GET https://perps-tradeapi.kanalabs.io/getOpenOrders?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&marketId=66
Note:
For the getFills endpoint, you must pass the profile address, not the wallet address.
Please make sure to first call the Get Profile Address endpoint and use that returned address as the address parameter in the getFills request.
Method:
GET
Query Parameters:
marketId (Required) - The ID of the market for which fills are retrieved.
address (Optional) - The address of the user to filter fills (if applicable).
from (Required) - The start time of the range in ISO 8601 format (e.g., 2024-11-21T00:00:00Z).
to (Required) - The end time of the range in ISO 8601 format (e.g., 2024-11-21T23:59:59Z).
orderId (Required) - The unique identifier of the order to retrieve its status.
Example Request:
GET https://perps-tradeapi.kanalabs.io/getFills?marketId=66&address=0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2&from=2024-11-20T00:00:00Z&to=2024-11-21T23:59:59Z
The following python demonstrates how to call the Get Add Margin API using the axios library.
import asyncio
import os
import requests
from dotenv import load_dotenv
from typing import List
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.type_tag import TypeTag, StructTag
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url, params, headers):
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status()
return response.json().get("data")
except requests.RequestException as e:
print(f"Error fetching payload: {e}")
return None
def create_transaction_function_arguments(self, arguments: List, types: List[Serializer]) -> List[TransactionArgument]:
if len(arguments) != len(types):
raise ValueError("Arguments and types length mismatch.")
return [TransactionArgument(arg, serializer) for arg, serializer in zip(arguments, types)]
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
module, function_id = "::".join(payload["function"].split("::")[:-1]), payload["function"].split("::")[-1]
function_arguments = self.create_transaction_function_arguments(payload["functionArguments"], payload["argumentTypes"])
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
entry_function = EntryFunction.natural(module=module, function=function_id, ty_args=type_arguments, args=function_arguments)
return TransactionPayload(payload=entry_function)
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_transaction = await self.rest_client.create_bcs_signed_transaction(self.account, transaction_payload)
txn_hash = await self.rest_client.submit_bcs_transaction(signed_transaction)
await self.rest_client.wait_for_transaction(txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
private_key_hex = os.getenv('APTOS_PRIVATEKEY')
if not private_key_hex:
print("Error: APTOS_PRIVATEKEY is missing in .env file.")
return
account = Account.load_key(bytes.fromhex(private_key_hex[2:] if private_key_hex.startswith("0x") else private_key_hex))
rest_client = RestClient(NODE_URL)
API_URL = "https://perps-tradeapi.kanalabs.io/addMargin"
PARAMS = {"marketId": 1338, "tradeSide": True, "amount": 10}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
int(payload_data["functionArguments"][0]),
payload_data["functionArguments"][1].lower() == "true",
int(payload_data["functionArguments"][2]),
]
payload_data["argumentTypes"] = [
Serializer.u64, Serializer.bool, Serializer.u64
]
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Hash: {txn_hash}")
except Exception as e:
print(f"Transaction process error: {e}")
if __name__ == "__main__":
asyncio.run(main())
27. 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.
Method:
GET
Query Parameters:
marketId (Required) - A numeric identifier representing the specific perpetual market.
userAddress (Required) - The wallet address of the user whose PnL is to be settled.
Example Request:
GET https://perps-tradeapi.kanalabs.io/settlePnl?userAddress=0xd0ff27c8411015f386ffc464e3baaa12fca10f1bef1ec6af36c27ed96b65f6c6&marketId=1338=6000
The following TypeScript/Node.js script demonstrates how to call the Get Settle pnl API using the axios library.
import asyncio
import os
from aptos_sdk.account import Account
from aptos_sdk.async_client import RestClient
from aptos_sdk.transactions import EntryFunction, TransactionArgument, TransactionPayload
from aptos_sdk.bcs import Serializer
from aptos_sdk.account_address import AccountAddress # Import this
from aptos_sdk.type_tag import TypeTag, StructTag
import requests
from dotenv import load_dotenv
from typing import Any, List
load_dotenv()
class AptosTransactionHandler:
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
def fetch_payload(self, api_url, params, headers):
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status()
return response.json().get("data")
except requests.exceptions.RequestException as e:
print(f"Error fetching payload: {e}")
return None
def create_transaction_function_arguments(self, arguments: List[Any], types: List[Serializer]) -> List[TransactionArgument]:
if len(arguments) != len(types):
raise ValueError("Arguments and types length mismatch.")
return [TransactionArgument(arg, serializer) for arg, serializer in zip(arguments, types)]
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
try:
function_information = payload["function"].split("::")
module = "::".join(function_information[:-1])
function_id = function_information[-1]
function_arguments = self.create_transaction_function_arguments(
payload["functionArguments"],
payload["argumentTypes"]
)
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
entry_function = EntryFunction.natural(
module=module,
function=function_id,
ty_args=type_arguments,
args=function_arguments,
)
return TransactionPayload(payload=entry_function)
except Exception as e:
print(f"Error creating transaction payload: {e}")
raise
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
try:
signed_transaction_request = await self.rest_client.create_bcs_signed_transaction(
sender=self.account, payload=transaction_payload
)
txn_hash = await self.rest_client.submit_bcs_transaction(
signed_transaction=signed_transaction_request
)
await self.rest_client.wait_for_transaction(txn_hash=txn_hash)
return txn_hash
except Exception as e:
print(f"Error during transaction submission: {e}")
raise
async def main():
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
rest_client = RestClient(NODE_URL)
private_key_hex = os.getenv('APTOS_PRIVATEKEY')
if private_key_hex.startswith("0x"):
private_key_hex = private_key_hex[2:]
private_key_bytes = bytes.fromhex(private_key_hex)
account = Account.load_key(private_key_bytes)
API_URL = "https://perps-tradeapi.kanalabs.io/settlePnl"
PARAMS = {
"userAddress":"0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770",
"marketId": 1338
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
AccountAddress.from_str(payload_data["functionArguments"][0]),
int(payload_data["functionArguments"][1]),
]
payload_data["argumentTypes"] = [Serializer.struct, Serializer.u64]
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Transaction hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())
26. 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.
Method: GET
Query Parameters:
userAddress (Required) - The wallet address for which the account balance is being fetched.
Example Request:
GET https://perps-tradeapi.kanalabs.io/getNetProfileBalance?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770
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.
Method: GET
Query Parameters:
marketId (Required) -
userAddress (Required) -
Example Request:
GET https://perps-tradeapi.kanalabs.io/settlePnl?userAddress=0xd0ff27c8411015f386ffc464e3baaa12fca10f1bef1ec6af36c27ed96b65f6c6&marketId=1338=6000