Example setup functions in python

Supported Markets

Asset	Market ID	Description
APT/USDC	301	Aptos-based trading market.
BTC/USDC	302	Bitcoin-based trading market.
ETH/USDC	303	Ethereum-based trading market.

Prerequisites

Install Required Packages

Before you begin, install the following dependencies:

pip install requests
pip install os
pip install aptos_sdk

We 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 = {
            'x-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 = {"x-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 = {
            'x-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 = {
            'x-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}')