1. Get Market Info
Query Parameters :
marketId
(Required) - The ID of the market you want information about.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getMarketInfo?marketId=501
Example Response:
Copy {
"success": true,
"message": "Market information fetched successfully",
"data": [
{
"__variant__": "V1",
"base_decimals": 8,
"base_name": "APT/USDC",
"counter": "17",
"creator": "0xb61d7b57333abf8ac036e752f19d0ba0c4baa5404db1cbf868c57dac3628f2bf",
"lot_size": "100000",
"maintenance_margin": "250",
"market_address": "0xfd8d58fdd05dd9d0442a9d555a4f8ed310112eb7d1c297307038a650136e207b",
"market_id": "501",
"max_leverage": "20",
"max_lots": "150000000",
"min_lots": "500",
"quote_decimals": 6,
"quote_precision": 3,
"tick_size": "1"
}
]
}
Example Code to Fetch Market Information:
The following Python script demonstrates how to call the Get Market Info API using the requests library.
Copy import requests
import os
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getMarketInfo'
params = {
'marketId': 'your_market_id'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_market_info = response.json()
print("getMarketInfo: ", get_market_info)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
2. Get Wallet Account Balance
Query Parameters :
userAddress
(Required) - The wallet address for which the account balance is being fetched.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getWalletAccountBalance?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770
Example Response:
Copy {
"success": true,
"message": "Fetched wallet account balance successfully",
"data": "16391.60"
}
Example Code to Fetch Wallet Account Balance Information:
The following Python script demonstrates how to call the Get Wallet Account Balance API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getWalletAccountBalance'
params = {
'userAddress': 'your_wallet_address'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_wallet_account_balance = response.json()
print("getWalletAccountBalance: ", get_wallet_account_balance)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
3. Get Profile Balance Snapshot
Query Parameters :
userAddress
(Required) - The wallet address for which the account balance is being fetched.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getProfileBalanceSnpashot?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770
Example Response:
Copy {
"success": true,
"message": "Profile balance snapshot fetched successfully",
"data": "44.321972"
}
Example Code to Fetch Trading Account Balance Information:
The following Python script demonstrates how to call the Get Trading Account Balance API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getProfileBalanceSnpashot'
params = {
'userAddress': 'your_wallet_address'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_trading_account_balance = response.json()
print("getTradingAccountBalance: ", get_trading_account_balance)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
4. Deposit
Query Parameters:
amount
(Required) - The amount of the quote coin to deposit.
userAddress
(Required) - The address of the user making the deposit.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/deposit?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&amount=50000000
Example Response:
Copy {
"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.
Copy 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
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:
Copy GET https://perps-tradeapi.kanalabs.io/withdrawSpecifiMarket?marketId=501&userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&amount=50
Example Response:
Copy {
"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.
Copy 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
Query Parameters :
marketId (Required) - The ID of the market where the limit order will be placed.
tradeSide (Required) - Indicates the trade 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:
Copy GET https://perps-tradeapi.kanalabs.io/placeLimitOrder/?marketId=501&tradeSide=true&direction=false&size=10000&price=3000&leverage=3
Example Response:
Copy {
"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.
Copy 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
Query Parameters :
marketId (Required) - The ID of the market where the limit order will be placed.
tradeSide (Required) - Indicates the trade 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:
Copy GET https://perps-tradeapi.kanalabs.io/placeMarketOrder?marketId=66&tradeSide=false&direction=true&size=1000&leverage=20
Example Response:
Copy {
"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.
Copy 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
Request Body :
marketId : The ID of the market for which the orders will be canceled.
orderIds : A list of order IDs to cancel.
cancelTradeSides: The sides of the orders to cancel (true for long, false for short).
Example Request:
Copy {
"marketId": 66,
"cancelOrderIds": ["1077898597726583798162207", "1077880153515406921884445"],
"orderSides": [true, false]
}
Example Response:
Copy {
"success": true,
"message": "Cancel Multiple Orders payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::cancel_multiple_orders",
"functionArguments": [
501,
[
"1077898597726583798162207",
"1077880153515406921884445"
],
[
true,
false
]
],
"typeArguments": []
}
}
Example Code to Cancel multiple orders:
The following Python script demonstrates how to call the Post Cancel Multiple Orders API using the requests library.
Copy 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, json_data, headers):
try:
response = requests.post(api_url, json=json_data, 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.")
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]
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/cancelMultipleOrders"
BODY = {
"marketId": 501,
"cancelOrderIds": [627189579998281341784],
"orderSides": [True]
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, BODY, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
BODY["marketId"],
BODY["cancelOrderIds"],
BODY["orderSides"]
]
payload_data["argumentTypes"] = [
Serializer.u64,
Serializer.sequence_serializer(Serializer.u128),
Serializer.sequence_serializer(Serializer.bool)
]
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())
8. Place Multiple Orders
Request Body :
marketId (Required) - The ID of the market where the orders will be placed.
orderTypes (Required) - An array of order types for each order:
tradeSides (Required) - An array indicating the trade sides for each order:
false for short positions.
directions (Required) - An array indicating the direction of each trade:
false to open a position.
true to close a position.
sizes (Required) - An array of sizes for each order.
leverages (Required) - An array of leverages for each order.
prices (Required) - An array of prices at which each order is to be placed.
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.
takeProfits - The take profit value is optional. If not provided, it defaults to 0
. It can either be a number or left undefined.
stopLosses - 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:
Copy {
"marketId": 66,
"orderTypes": [true, true],
"tradeSides": [true, true],
"directions": [true, true],
"sizes": [1000, 1000],
"prices": [1000, 1000],
"leverages": [2, 2]
}
Example Response:
Copy {
"success": true,
"message": "Multiple Orders payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::place_multiple_orders",
"functionArguments": [
501,
[
true,
true
],
[
true,
true
],
[
true,
true
],
[
1000,
1000
],
[
4670,
6780
],
[
2,
2
],
[
0,
0
],
[
4600
],
[
0,
0
]
],
"typeArguments": []
}
}
Example Code to Place a multiple orders:
The following Python script demonstrates how to call the Get Place Multiple Orders API using the requests library.
Copy 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 {}
# Dynamically ensure the length matches the sizes provided
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"]))
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]
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/placeMultipleOrders"
BODY = {
"marketId": 501,
"orderTypes": [True],
"tradeSides": [True],
"directions": [False],
"sizes": [1000],
"prices": [7],
"leverages": [2]
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, BODY, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
BODY["marketId"],
BODY["orderTypes"],
BODY["tradeSides"],
BODY["directions"],
BODY["sizes"],
BODY["prices"],
BODY["leverages"],
payload_data["restrictions"],
payload_data["takeProfits"],
payload_data["stopLosses"]
]
payload_data["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)
]
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())
9. Cancel and Place Multiple Orders
Request Body :
marketId (Required) - The ID of the market where the orders will be placed.
cancelOrderIds (Required) - An array of strings representing the IDs of the orders to be canceled.
orderSides (Required) - An array indicating the sides of the orders being canceled:
orderTypes (Required) - An array of order types for each order:
tradeSides (Required) - An array indicating the trade sides for each order:
false for short positions.
directions (Required) - An array indicating the direction of each trade:
false to open a position.
true to close a position.
sizes (Required) - An array of sizes for each order.
leverages (Required) - An array of leverages for each order.
prices (Required) - An array of prices at which each order is to be placed.
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
takeProfits - The take profit value is optional. If not provided, it defaults to 0
. It can either be a number or left undefined.
stopLosses - 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:
Copy {
"marketId": 66,
"cancelOrderIds": [23434565434567", "454345665456],
"orderSides": [true, true],
"orderTypes": [true, true],
"tradeSides": [true, false],
"directions": [true, true],
"sizes": [1000, 2000],
"prices": [5000, 6000],
"leverages": [2, 2]
}
Example Response:
Copy {
"success": true,
"message": "Cancel and place multiple Orders payload has been built successfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::cancel_and_place_multiple_orders",
"functionArguments": [
501,
[
"23434565434567",
"454345665456"
],
[
"true",
"true"
],
[
true,
true
],
[
true,
false
],
[
true,
true
],
[
1000,
2000
],
[
5000,
6000
],
[
2,
2
],
[
0,
0
],
[
0,
0
],
[
0,
0
]
],
"typeArguments": []
}
}
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.
Copy 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 {}
# Dynamically ensure the length matches the sizes provided
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"]))
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]
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": [664083068162880314200],
"orderSides": [True],
"orderTypes": [True],
"tradeSides": [True],
"directions": [False],
"sizes": [1000],
"prices": [7],
"leverages": [2]
}
HEADERS = {"x-api-key": os.getenv('API_KEY')}
handler = AptosTransactionHandler(rest_client, account)
payload_data = handler.fetch_payload(API_URL, BODY, HEADERS)
if not payload_data:
print("Failed to fetch payload data.")
return
try:
payload_data["functionArguments"] = [
BODY["marketId"],
BODY["cancelOrderIds"],
BODY["orderSides"],
BODY["orderTypes"],
BODY["tradeSides"],
BODY["directions"],
BODY["sizes"],
BODY["prices"],
BODY["leverages"],
payload_data["restrictions"],
payload_data["takeProfits"],
payload_data["stopLosses"]
]
payload_data["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)
]
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
Query Parameters :
userAddress (Required) - The wallet address to retrieve open orders for.
marketId - The ID of the market to filter open orders. // Optional
Copy 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:
Copy GET https://perps-tradeapi.kanalabs.io/getOpenOrders?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&marketId=66
Example Response:
Copy {
"success": true,
"message": "Fetched open orders successfully",
"data": [
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 3,
"timestamp": "1741444213",
"price": "7",
"total_size": "2",
"remaining_size": "2",
"order_value": "14",
"order_id": "645636324063400958808",
"trade_id": "295147905179352826357",
"last_updated": "1741444214"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 3,
"timestamp": "1741445758",
"price": "7",
"total_size": "2",
"remaining_size": "2",
"order_value": "14",
"order_id": "682529812245179800408",
"trade_id": "295147905179352826357",
"last_updated": "1741445758"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 3,
"timestamp": "1741446786",
"price": "7",
"total_size": "2",
"remaining_size": "2",
"order_value": "14",
"order_id": "700976556301709482840",
"trade_id": "295147905179352826357",
"last_updated": "1741446786"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 3,
"timestamp": "1741446907",
"price": "7",
"total_size": "2",
"remaining_size": "2",
"order_value": "14",
"order_id": "719423300401188838232",
"trade_id": "295147905179352826357",
"last_updated": "1741446908"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 3,
"timestamp": "1741447619",
"price": "0.007",
"total_size": "1",
"remaining_size": "1",
"order_value": "0.007",
"order_id": "737869903703050289159",
"trade_id": "295147905179352826357",
"last_updated": "1741447620"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 3,
"timestamp": "1741447815",
"price": "0.007",
"total_size": "1",
"remaining_size": "1",
"order_value": "0.007",
"order_id": "756316647793939709959",
"trade_id": "295147905179352826357",
"last_updated": "1741447815"
}
]
}
Example Code to Fetch Open Orders:
The following Python script demonstrates how to call the Get Open Orders API using the requests library.
11. Get Order History
Query Parameters :
userAddress (Required) - The wallet address to retrieve the order history for.
marketId (optional) - The ID of the market to filter the order history.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getOrderHistory?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770&marketId=66
Example Response:
Copy {
"success": true,
"message": "Fetched order history successfully",
"data": [
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 1,
"timestamp": "1741167151",
"is_market_order": false,
"size": "1.50",
"price": "4.678",
"order_value": "7.01700",
"status": "Open",
"order_id": "18446884819787846214",
"trade_id": "18446744073709552117",
"last_updated": "1741262093"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 1,
"timestamp": "1741236277",
"is_market_order": false,
"size": "1.50",
"price": "4.678",
"order_value": "7.01700",
"status": "Cancelled",
"order_id": "18446884819787846214",
"trade_id": "18446744073709552117",
"last_updated": "1741262168"
},
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"order_type": 1,
"timestamp": "1741236468",
"is_market_order": false,
"size": "1.50",
"price": "4.50",
"order_value": "6.7500",
"status": "Open",
"order_id": "36893628893497397652",
"trade_id": "36893488147419103733",
"last_updated": "1741262168"
}
}
]
Example Code to Fetch Order History:
The following Python script demonstrates how to call the Get Order History API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getOrderHistory'
params = {
'marketId': 501,
'userAddress': "0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770"
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_order_history = response.json()
print("getOrderHistory: ", get_order_history)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
12. Get Positions
Query Parameters :
userAddress (Required) - The wallet address of the user to view positions for.
marketId (optional) - The ID of the market to filter the positions.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getPositions?marketId=66&userAddress=0x2eda5777ed2bf68cbcc67359dd00ae5fa73b1d5fa160b7c3aeb42d748d485387
Example Response:
Copy {
"success": true,
"message": "Fetched positions successfully",
"data": [
{
"address": "0x3c78886aa67752706b3502b12959edf92e68d85ae64b24226783d26ce6efc1e",
"market_id": "501",
"leverage": 2,
"is_long": true,
"size": "1",
"available_order_size": "1",
"value": "5.678",
"entry_price": "5.678",
"liq_price": "2.9117948717948717990163355253",
"margin": "2.839",
"tp": "6.012",
"sl": null,
"trade_id": "295147905179352826357",
"last_updated": "1741262185"
}
]
}
Example Code to Fetch Open Position:
The following Python script demonstrates how to call the Get Open Position API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getPositions'
params = {
'marketId': 501,
'userAddress': "0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770"
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_open_position = response.json()
print("getOpenPosition: ", get_open_position)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
12. Get All Trades
Query Parameters :
marketId (Required) - The ID of the market to retrieve all trades information.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getAllTrades?marketId=66
Example Response:
Copy {
"success": true,
"message": "Recent trades fetched successfully",
"data": [
{
"txn_version": 6297687196,
"event_idx": 5,
"emit_address": "0x2eda5777ed2bf68cbcc67359dd00ae5fa73b1d5fa160b7c3aeb42d748d485387",
"time": "2024-11-20T20:33:38.600241+00:00",
"maker_address": "0x2eda5777ed2bf68cbcc67359dd00ae5fa73b1d5fa160b7c3aeb42d748d485387",
"maker_custodian_id": 28,
"maker_order_id": "15951655917465922493754983",
"maker_side": true,
"market_id": 66,
"price": 11879,
"sequence_number_for_trade": 0,
"size": 10000,
"taker_address": "0x13eb7ea6b62c6cd791b5c4e0e2ef1dfbe59d1cf4823929343a273274c288bc8a",
"taker_custodian_id": 28,
"taker_order_id": "15951748150764039921729536",
"taker_quote_fees_paid": 59395
},
{
"txn_version": 6297687196,
"event_idx": 4,
"emit_address": "0x13eb7ea6b62c6cd791b5c4e0e2ef1dfbe59d1cf4823929343a273274c288bc8a",
"time": "2024-11-20T20:33:38.600241+00:00",
"maker_address": "0x2eda5777ed2bf68cbcc67359dd00ae5fa73b1d5fa160b7c3aeb42d748d485387",
"maker_custodian_id": 28,
"maker_order_id": "15951655917465922493754983",
"maker_side": true,
"market_id": 66,
"price": 11879,
"sequence_number_for_trade": 0,
"size": 10000,
"taker_address": "0x13eb7ea6b62c6cd791b5c4e0e2ef1dfbe59d1cf4823929343a273274c288bc8a",
"taker_custodian_id": 28,
"taker_order_id": "15951748150764039921729536",
"taker_quote_fees_paid": 59395
}
]
}
Example Code to Fetch All trades:
The following Python script demonstrates how to call the Get Open Position API using the requests library.
Copy import axios from "axios";
async function main(): Promise<void> {
const baseURL = 'https://perps-tradeapi.kanalabs.io/getAllTrades';
const params = {
marketId: 'your_market_id'
};
const res = await axios.get(baseURL, {
params,
headers: {
'x-api-key': process.env.API_KEY,
},
});
const getAllTrades = res.data;
console.log("getAllTrades : ", getAllTrades);
}
main().catch(error => {
console.error('An error occurred:', error);
});
13. Get Order Status By Order Id
Query Parameters :
marketId (Required) - The ID of the market associated with the order.
orderId (Required) - The unique identifier of the order to retrieve its status.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getOrderStatusByOrderId?marketId=66&orderId=15308325717473050761363456
Example Response:
Copy {
"status": true,
"message": "Fetched order status for a given order Id successfully",
"data": {
"market_id": 66,
"order_id": "15308325717473050761363456",
"created_at": "2024-11-20T11:54:45.957059+00:00",
"last_updated_at": "2024-11-20T11:54:45.957059+00:00",
"integrator": "0xee820ab02631dd1a195d3c53fa64f0a8f455dbb9261388e141c3bd3bd3c08363",
"total_filled": 1000,
"remaining_size": 0,
"order_status": "closed",
"order_type": "market",
"user": "0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770",
"direction": "buy",
"price": 0,
"average_execution_price": 12813,
"custodian_id": 28,
"self_match_behavior": 0,
"restriction": 0,
"last_increase_stamp": null,
"min_base": null,
"max_base": null,
"min_quote": null,
"max_quote": null,
"total_fees_paid_in_quote_subunits": 6406
}
}
Example Code to Fetch Order Status B Order Id:
The following Python script demonstrates how to call the Get Order Status By Order Id API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getOrderStatusByOrderId'
params = {
'marketId': 'your_market_id',
'orderId': 'your_order_id'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_order_status_by_order_id = response.json()
print("getOrderStatusByOrderId: ", get_order_status_by_order_id)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
14. Get Fills
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:
Copy GET https://perps-tradeapi.kanalabs.io/getFills?marketId=66&address=0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2&from=2024-11-20T00:00:00Z&to=2024-11-21T23:59:59Z
Example Response:
Copy {
"success": true,
"message": "Fetched fill trades for a given time range Successfully",
"data": [
{
"txn_version": 6296434935,
"event_idx": 1,
"emit_address": "0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2",
"time": "2024-11-20T19:23:14.67535+00:00",
"maker_address": "0x2eda5777ed2bf68cbcc67359dd00ae5fa73b1d5fa160b7c3aeb42d748d485387",
"maker_custodian_id": 28,
"maker_order_id": "15891335065048614061485282",
"maker_side": true,
"market_id": 66,
"price": 11490,
"sequence_number_for_trade": 0,
"size": 1000,
"taker_address": "0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2",
"taker_custodian_id": 28,
"taker_order_id": "15891427297643009687945216",
"taker_quote_fees_paid": 5745
},
{
"txn_version": 6295302625,
"event_idx": 4,
"emit_address": "0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2",
"time": "2024-11-20T18:19:35.70863+00:00",
"maker_address": "0x2eda5777ed2bf68cbcc67359dd00ae5fa73b1d5fa160b7c3aeb42d748d485387",
"maker_custodian_id": 28,
"maker_order_id": "15812401447297883953966476",
"maker_side": false,
"market_id": 66,
"price": 11660,
"sequence_number_for_trade": 0,
"size": 1000,
"taker_address": "0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2",
"taker_custodian_id": 28,
"taker_order_id": "15812549019983827645235200",
"taker_quote_fees_paid": 5830
}
]
}
Example Code to Fetch Order Status B Order IdFills Data:
The following Python script demonstrates how to call the Get Fills Data API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getFills'
params = {
'marketId': 'your_market_id',
'address': 'your_address',
'from': 'your_from_time',
'to': 'your_to_time'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_fills_data = response.json()
print("getFillsData: ", get_fills_data)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
14. Get Market Price
Query Parameters :
marketId (Required) - The ID of the market for which the price information is being retrieved.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getMarketPrice?marketId=66
Example Response:
Copy {
"success": true,
"message": "Fetched Market Price Successfully",
"data": {
"bestAskPrice": 6.400,
"bestBidPrice": 9.880
}
}
Example Code to Fetch Market Price:
The following Python script demonstrates how to call the Get Market Price API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getMarketPrice'
params = {
'marketId': 501
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_market_price = response.json()
print("getMarketPrice: ", get_market_price)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
15. Get Last Execution Price
Query Parameters :
marketId (Required) - The ID of the market for which the last execution price information is being retrieved.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getLastPlacedPrice?marketId=66
Example Response:
Copy {
"success": true,
"message": "Fetched last execution price Successfully",
"data": 8.989
}
Example Code to Fetch Last Execution Price:
The following Python script demonstrates how to call the Get Last Execution Price API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getLastPlacedPrice'
params = {
'marketId': '501'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_last_execution_price = response.json()
print("getLastExecutionPrice: ", get_last_execution_price)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
16. Get All Open Order Ids
Query Parameters :
userAddress(Required) - The address associated with the orders.
marketId (optional) - The ID of the market for which open order IDs are being retrieved.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getAllOpenOrderIds?marketId=66&userAddress=0x3a6b332c7889784fe57ea61f507b5056e386db8f56c6323b4d898485f468f0c2
Example Response:
Copy {
"success": true,
"message": "Open order IDs fetched successfully",
"data": [
"645636324063400958808",
"682529812245179800408",
"700976556301709482840",
"719423300401188838232",
"737869903703050289159",
"756316647793939709959"
]
}
Example Code to Fetch Open Order Ids:
The following Python script demonstrates how to call the Get Open Order Ids API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getAllOpenOrderIds'
params = {
'userAddress': 'your_wallet_address',
'marketId': 'your_market_id'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_open_order_ids = response.json()
print("getOpenOrderIds: ", get_open_order_ids)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
17. Update Take Profit
Query Parameters :
marketId (Required) - The ID of the market for which the take profit values will be updated.
tradeSide (Required) - The trade side:
false
for the short side.
newTakeProfitPrice (Required) - The new take profit price to be set for the trade.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/updateTakeProfit?marketId=66&tradeSide=true&newTakeProfitPrice=6000
Example Response:
Copy {
"success": true,
"message": "Update take profit payload built succesfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::update_take_profit",
"functionArguments": [
"501",
"true",
7590
],
"typeArguments": []
}
}
Example Code to Update take profit price:
The following Python script demonstrates how to call the Get Update Take Profit Price API using the requests library.
Copy 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, 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/updateTakeProfit"
PARAMS = {
"marketId": 501,
"tradeSide": True,
"newTakeProfitPrice": 6.89
}
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. Transaction hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())
18. Get Account Aptos Balance
userAddress (Required) - The address of the account whose balance will be
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/getAccountAptBalance?userAddress=0x4d6dc68e391e86991e58ab4d548b7e92872430d1f51bc666fe0c206bad7ff770
Example Response:
Copy {
"success": true,
"message": "Fetched Account Apt Balance Successfully",
"data": 8.2344617
}
Example Code to Fetch Wallet Account Aptos Balance Information:
The following Python script demonstrates how to call the Get Wallet Account Aptos Balance API using the requests library.
Copy import os
import requests
from dotenv import load_dotenv
load_dotenv()
def main():
try:
base_url = 'https://perps-tradeapi.kanalabs.io/getAccountAptBalance'
params = {
'userAddress': 'your_wallet_address'
}
api_key = os.getenv('API_KEY')
headers = {
'x-api-key': api_key
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
get_wallet_account_aptos_balance = response.json()
print("getWalletAccountAptosBalance: ", get_wallet_account_aptos_balance)
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
if __name__ == "__main__":
main()
19. Update Stop Loss
Query Parameters :
marketId (Required) - The ID of the market for which the take profit values will be updated.
tradeSide (Required) - The trade side:
false
for the short side.
newStopLossPrice (Required) - The new stop loss price to be set for the trade.
Example Request:
Copy GET https://perps-tradeapi.kanalabs.io/updateStopLoss?marketId=66&tradeSide=true&newStopLossPrice=6000
Example Response:
Copy {
"success": true,
"message": "Update stop loss payload built succesfully",
"data": {
"function": "0xabbf40e0c31cbedf182f3e6e9c3570f3f91629424723da876becb514bf5d9fc6::perpetual_scripts::update_stop_loss",
"functionArguments": [
"501",
"true",
7590
],
"typeArguments": []
}
}
Example Code to Update stop loss price:
The following Python script demonstrates how to call the Get Update Stop Loss Price API using the requests library.
Copy 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, 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/updateStopLoss"
PARAMS = {
"marketId": 501,
"tradeSide": True,
"newStopLossPrice": 6.89
}
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. Transaction hash: {txn_hash}")
except Exception as e:
print(f"Error during transaction process: {e}")
if __name__ == "__main__":
asyncio.run(main())