Example setup functions in python
Supported Markets
Asset
Market ID
Description
APT/USDC
66
Aptos-based trading market.
BTC/USDC
67
Bitcoin-based trading market.
ETH/USDC
68
Ethereum-based trading market.
Prerequisites
Install Required Packages
Before you begin, install the following dependencies:
pip install requests
pip install os
pip install aptos_sdkWe will see how to implement the view functions : (Example get market info function)
1. Get Market Info:
# 1. Importing Required Libraries
import requests
import os
# 2. Main Function Setup
def main():
try:
# 3. API Endpoint Setup
base_url = 'https://perps-tradeapi.kanalabs.io/getMarketInfo'
# 4. Parameters Setup
params = {
'marketId': 'your_marketId' # Replace with the actual marketId Eg: --> 66, 67, 68 --> (type number)
}
# 5. Headers Setup
headers = {
'api_key': os.getenv('API_KEY') # API key passed in the headers from environment variables
}
# 6. API Request
response = requests.get(base_url, params=params, headers=headers)
# Ensure the response is successful
response.raise_for_status()
# 7. Extract and Log Data
get_market_info = response.json()
print("getMarketInfo: ", get_market_info)
# 8. Error Handling
except requests.exceptions.RequestException as error:
print('An error occurred:', error)
# 9. Call the Main Function
if __name__ == "__main__":
try:
main()
except Exception as error:
print('An error occurred:', error)We will see how to implement the run functions : (Example deposit function)
2. Depsoit Function:
import asyncio
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 typing import Any, List
# 1. Define the AptosTransactionHandler class to handle transaction-related operations
class AptosTransactionHandler:
# 2. Initialize the class with the RestClient and Account
def __init__(self, rest_client: RestClient, account: Account):
self.rest_client = rest_client
self.account = account
# 3. Fetch payload data from an API
def fetch_payload(self, api_url, params, headers):
try:
response = requests.get(api_url, params=params, headers=headers)
response.raise_for_status() # Check if the request was successful
return response.json().get("data") # Return the 'data' part of the response JSON
except requests.exceptions.RequestException as e:
print(f"Error fetching payload: {e}")
return None
# 4. Create the transaction function arguments for the deposit transaction
def create_deposit_transaction_function_arguments(self, arguments: List[Any]) -> List[TransactionArgument]:
# Check if there is exactly one argument
if len(arguments) != 1:
raise ValueError("Invalid number of arguments for deposit transaction")
return [TransactionArgument(arguments[0], Serializer.u64)] # Convert to transaction argument format
# 5. Create the transaction payload from the provided payload data
def create_transaction_payload(self, payload: dict) -> TransactionPayload:
# Split the function name to get the module and function name
function_information = payload["function"].split("::")
module = "::".join(function_information[:-1])
function_id = function_information[-1]
# Create function arguments and type arguments from the payload
function_arguments = self.create_deposit_transaction_function_arguments(payload["functionArguments"])
type_arguments = [TypeTag(StructTag.from_str(argument)) for argument in payload["typeArguments"]]
# Create the entry function and return the transaction payload
entry_function = EntryFunction.natural(
module=module,
function=function_id,
ty_args=type_arguments,
args=function_arguments,
)
return TransactionPayload(payload=entry_function)
# 6. Submit the transaction to the Aptos blockchain
async def submit_transaction(self, transaction_payload: TransactionPayload) -> str:
# Create a signed transaction
signed_transaction_request = await self.rest_client.create_bcs_signed_transaction(
sender=self.account, payload=transaction_payload
)
# Submit the signed transaction to the blockchain
txn_hash = await self.rest_client.submit_bcs_transaction(
signed_transaction=signed_transaction_request
)
# Await transaction confirmation
await self.rest_client.wait_for_transaction(txn_hash=txn_hash)
# Return the transaction hash
return txn_hash
# 7. Main async function that orchestrates the transaction process
async def main():
# 8. Set up the Aptos node URL and initialize the RestClient
NODE_URL = "https://api.testnet.aptoslabs.com/v1"
rest_client = RestClient(NODE_URL)
# 9. Load the account using the private key
private_key_hex = "your_private_key"
private_key_bytes = bytes.fromhex(private_key_hex)
account = Account.load_key(private_key_bytes)
# 10. Print account address for verification
print(f"Address: {account.address()}\n")
# 11. Define the API URL and parameters for the deposit API
API_URL = "https://perps-tradeapi.kanalabs.io/deposit"
PARAMS = {"marketId": 66, "amount": 100}
HEADERS = {"api_key": "your_api_key"} # Replace with actual API key
# 12. Initialize the AptosTransactionHandler with RestClient and Account
handler = AptosTransactionHandler(rest_client, account)
# 13. Fetch the payload data using the handler
payload_data = handler.fetch_payload(API_URL, PARAMS, HEADERS)
# 14. Check if the payload data was fetched successfully
if not payload_data:
print("Failed to fetch payload data.")
return
# 15. Print fetched payload data for debugging
print(f"Fetched payload data: {payload_data}")
try:
# 16. Create the transaction payload based on the fetched data
transaction_payload = handler.create_transaction_payload(payload_data)
print("Transaction payload created successfully.")
# 17. Submit the transaction to the blockchain
txn_hash = await handler.submit_transaction(transaction_payload)
print(f"Transaction submitted successfully. Transaction hash: {txn_hash}")
# 18. Error handling in case something goes wrong during the transaction process
except Exception as e:
print(f"Error during transaction process: {e}")
# 19. Execute the main async function
if __name__ == "__main__":
asyncio.run(main())We will see how to implement the websocket client initialization code: (Example order book stream) Example code to fetch the Order Book Client Initialization Code:
3. Order Book Stream:
# 1. Importing Required Libraries
import websocket
import json
# 2. WebSocket Event Handlers
def on_open(ws):
print('[client] Connected to the server')
# Sending a message once connected
ws.send('Request data from endpoint')
def on_message(ws, message):
print(f'Received a message from the server: {message}')
def on_close(ws, close_status_code, close_msg):
print('[client] Disconnected from the server')
# 3. WebSocket Setup and Connection
def main():
market_id = 'your_market_id' # Replace with the actual marketId Eg: --> 66, 67, 68 --> (type number)
ws_url = f"wss://perps-sdk-ws.kanalabs.io/wsOrderBook?marketId={market_id}"
# 4. Create WebSocket connection
ws = websocket.WebSocketApp(ws_url,
on_open=on_open,
on_message=on_message,
on_close=on_close)
# 5. Run the WebSocket connection
ws.run_forever()
# 6. Call the Main Function
if __name__ == "__main__":
main()
Example code to fetch the Fill Events Client Initialization Code:
4. Fill Events Stream: (Recent Trdes)
# 1. Importing Required Libraries
import websocket
import json
# 2. WebSocket Event Handlers
def on_open(ws):
print('[client] Connected to the server')
# Sending a message once connected
ws.send('Request data from endpoint')
def on_message(ws, message):
print(f'Received a message from the server: {message}')
def on_close(ws, close_status_code, close_msg):
print('[client] Disconnected from the server')
# 3. WebSocket Setup and Connection
def main():
market_id = 'your_market_id' # Replace with the actual marketId Eg: --> 66, 67, 68 --> (type number)
ws_url = f"wss://perps-sdk-ws.kanalabs.io/wsRecentTrades?marketId={market_id}"
# 4. Create WebSocket connection
ws = websocket.WebSocketApp(ws_url,
on_open=on_open,
on_message=on_message,
on_close=on_close)
# 5. Run the WebSocket connection
ws.run_forever()
# 6. Call the Main Function
if __name__ == "__main__":
main()We will see how to implement the Fill Events Data :
5. Get Fill Events Data:
# 1. Importing Required Libraries
import requests
import json
import os # To get environment variables
# 2. Main Function Setup
def main():
try:
# 3. API Endpoint Setup
base_url = 'https://aptos-testnet-econia.nodeinfra.com/fill_events'
# 4. Market ID and Parameters Setup
market_id = 66 # Replace with the actual marketId
params = {
market_id = 'your_marketId' # Replace with the actual marketId Eg: --> 66, 67, 68 --> (type number)
# 'offset': 0,
# 'limit': 30,
# 'order': 'txn_version.desc' # 'txn_version.asc',
}
# 5. Headers Setup (Fetching API key from environment variables)
headers = {
'api_key': os.getenv('API_KEY'), # Ensure you have set the API_KEY in your environment variables
}
# 6. API Request
response = requests.get(base_url, params=params, headers=headers)
# Ensure the response is successful
response.raise_for_status()
# 7. Process and Clean the Response Data
json_string = response.text
# Replacing numeric IDs with string values
json_string = json_string.replace('"maker_order_id":(\\d+)', r'"maker_order_id":"\1"') \
.replace('"taker_order_id":(\\d+)', r'"taker_order_id":"\1"')
# 8. Parsing JSON Response
get_fill_events_data = json.loads(json_string)
print("getFillEventsData: ", get_fill_events_data)
# 9. Error Handling
except requests.exceptions.RequestException as error:
print(f'An error occurred: {error}')
# 10. Call the Main Function
if __name__ == "__main__":
try:
main()
except Exception as error:
print(f'An error occurred: {error}')
We will see how to implement the Order Status Data:
6. Get Order Status Data:
# 1. Importing Required Libraries
import requests
import json
import os # To get environment variables
# 2. Main Function Setup
def main():
try:
# 3. API Endpoint Setup
base_url = 'https://aptos-testnet-econia.nodeinfra.com/orders'
# 4. Market ID and Parameters Setup
market_id = 'your_marketId' # Replace with the actual marketId Eg: --> 66, 67, 68 --> (type number)
user = 'your_wallet_address' # Replace with the actual user address
params = {
'market_id': f'eq.{market_id}',
'user': f'eq.{user}'
# 'offset': 0,
# 'limit': 30,
# 'order': 'txn_version.desc' # 'txn_version.asc',
}
# 5. Headers Setup (Fetching API key from environment variables)
headers = {
'api_key': os.getenv('API_KEY'), # Ensure you have set the API_KEY in your environment variables
}
# 6. API Request
response = requests.get(base_url, params=params, headers=headers)
# Ensure the response is successful
response.raise_for_status()
# 7. Process and Clean the Response Data
json_string = response.text
# Replacing numeric order IDs with string values
json_string = json_string.replace('"order_id":(\\d+)', r'"order_id":"\1"')
# 8. Parsing JSON Response
order_history = json.loads(json_string)
print("orderHistory: ", order_history)
# 9. Error Handling
except requests.exceptions.RequestException as error:
print(f'An error occurred: {error}')
# 10. Call the Main Function
if __name__ == "__main__":
try:
main()
except Exception as error:
print(f'An error occurred: {error}')
Last updated