This comprehensive reference covers all classes, methods, and functions available in the AxiomTradeAPI-py library. Use this as your complete guide for integrating Solana trading functionality into your applications.
The main client class for interacting with the Axiom Trade API.
from axiomtradeapi import AxiomTradeClient
client = AxiomTradeClient(
auth_token=None,
refresh_token=None,
log_level=logging.INFO,
timeout=30,
max_retries=3
)
Parameter | Type | Default | Description |
---|---|---|---|
auth_token |
str |
None |
Authentication token for API access |
refresh_token |
str |
None |
Refresh token for token renewal |
log_level |
int |
logging.INFO |
Logging level for client operations |
timeout |
int |
30 |
Request timeout in seconds |
max_retries |
int |
3 |
Maximum number of retry attempts |
Retrieve the SOL balance for a specific wallet address.
balance = client.GetBalance("BJBgjyDZx5FSsyJf6bFKVXuJV7DZY9PCSMSi5d9tcEVh")
Parameters:
wallet_address
(str): Valid Solana wallet address (44 characters)Returns:
{
"sol": 1.234567890, # Balance in SOL (float)
"lamports": 1234567890, # Balance in lamports (int)
"slot": 344031778 # Blockchain slot number (int)
}
Raises:
ValueError
: Invalid wallet address formatAPIError
: API request failedNetworkError
: Network connectivity issuesRetrieve balances for multiple wallet addresses in a single request.
addresses = [
"BJBgjyDZx5FSsyJf6bFKVXuJV7DZY9PCSMSi5d9tcEVh",
"Cpxu7gFhu3fDX1eG5ZVyiFoPmgxpLWiu5LhByNenVbPb"
]
balances = client.GetBatchedBalance(addresses)
Parameters:
wallet_addresses
(List[str]): List of wallet addresses (max 1000)Returns:
{
"wallet_address_1": {
"sol": 1.234567890,
"lamports": 1234567890,
"slot": 344031778
},
"wallet_address_2": {
"sol": 0.567890123,
"lamports": 567890123,
"slot": 344031778
},
"invalid_address": None # Failed queries return None
}
Performance:
Subscribe to real-time new token launches via WebSocket.
async def handle_new_tokens(tokens):
for token in tokens:
print(f"New token: {token['tokenName']}")
await client.subscribe_new_tokens(handle_new_tokens)
Parameters:
callback
(async callable): Function to handle incoming token dataCallback Parameters:
async def callback(tokens: List[dict]):
# tokens is a list of token objects
pass
Token Object Structure:
{
"tokenName": "Example Token",
"tokenTicker": "EXAMPLE",
"tokenAddress": "token_address_here",
"marketCapSol": 100.0,
"volumeSol": 50.0,
"liquiditySol": 200.0,
"protocol": "Raydium",
"createdAt": "2024-01-01T00:00:00Z",
"website": "https://example.com",
"twitter": "https://twitter.com/example",
"telegram": "https://t.me/example"
}
Dedicated WebSocket client for real-time data streaming.
from axiomtradeapi.websocket import AxiomTradeWebSocketClient
ws_client = AxiomTradeWebSocketClient(
auth_token="your-auth-token",
reconnect_delay=5,
max_reconnects=10
)
Parameter | Type | Default | Description |
---|---|---|---|
auth_token |
str |
Required | Authentication token |
reconnect_delay |
int |
5 |
Delay between reconnection attempts |
max_reconnects |
int |
10 |
Maximum reconnection attempts |
Establish WebSocket connection to Axiom Trade servers.
await ws_client.connect()
Returns:
bool
: True if connection successful, False otherwiseListen for incoming WebSocket messages.
async for message in ws_client.listen():
print(f"Received: {message}")
Yields:
dict
: Parsed WebSocket message dataSend message through WebSocket connection.
await ws_client.send({
"type": "subscribe",
"channel": "new_tokens"
})
Parameters:
message
(dict): Message to sendAuthentication helper class for managing API tokens.
from axiomtradeapi.auth import AxiomAuth
auth = AxiomAuth()
Authenticate with Axiom Trade using email and password.
credentials = await auth.login("user@example.com", "password123")
Parameters:
email
(str): User email addresspassword
(str): User passwordReturns:
{
"auth_token": "eyJhbGciOiJIUzI1NiIs...",
"refresh_token": "eyJhbGciOiJIUzI1NiIs...",
"expires_at": "2024-01-01T12:00:00Z",
"user_id": "user_123",
"success": True
}
Refresh authentication tokens using refresh token.
new_tokens = await auth.refresh_tokens("current_refresh_token")
Parameters:
refresh_token
(str): Current refresh tokenReturns:
{
"auth_token": "new_auth_token",
"refresh_token": "new_refresh_token",
"expires_at": "2024-01-01T13:00:00Z",
"success": True
}
Validate if an authentication token is still valid.
is_valid = await auth.validate_token("auth_token_here")
Parameters:
auth_token
(str): Token to validateReturns:
bool
: True if token is valid, False otherwiseQuick utility function to login and get trending tokens.
from axiomtradeapi import quick_login_and_get_trending
trending = await quick_login_and_get_trending(
email="user@example.com",
password="password123"
)
Parameters:
email
(str): User emailpassword
(str): User passwordReturns:
{
"trending_tokens": [
{
"tokenName": "Trending Token 1",
"tokenAddress": "address1",
"volumeSol": 1000.0,
"priceChange24h": 15.5
}
],
"auth_token": "token_for_future_use"
}
Get trending tokens with existing authentication token.
from axiomtradeapi import get_trending_with_token
trending = await get_trending_with_token("your_auth_token")
Parameters:
auth_token
(str): Valid authentication tokenReturns:
Base exception class for API-related errors.
from axiomtradeapi.exceptions import APIError
try:
balance = client.GetBalance("invalid_address")
except APIError as e:
print(f"API Error: {e.message}")
print(f"Status Code: {e.status_code}")
print(f"Error Code: {e.error_code}")
Attributes:
message
(str): Human-readable error messagestatus_code
(int): HTTP status codeerror_code
(str): Specific error identifierException raised for network connectivity issues.
from axiomtradeapi.exceptions import NetworkError
try:
balance = client.GetBalance("address")
except NetworkError as e:
print(f"Network Error: {e.message}")
print(f"Retry After: {e.retry_after}")
Attributes:
message
(str): Error descriptionretry_after
(int): Suggested retry delay in secondsException raised for authentication failures.
from axiomtradeapi.exceptions import AuthenticationError
try:
await client.subscribe_new_tokens(callback)
except AuthenticationError as e:
print(f"Auth Error: {e.message}")
print(f"Expired: {e.token_expired}")
Attributes:
message
(str): Error descriptiontoken_expired
(bool): Whether token has expiredException raised for input validation errors.
from axiomtradeapi.exceptions import ValidationError
try:
balance = client.GetBalance("invalid_format")
except ValidationError as e:
print(f"Validation Error: {e.message}")
print(f"Field: {e.field}")
print(f"Value: {e.value}")
Attributes:
message
(str): Error descriptionfield
(str): Field that failed validationvalue
(any): Invalid value providedConfiguration class for customizing client behavior.
from axiomtradeapi.config import ClientConfig
config = ClientConfig(
api_base_url="https://api.axiom.trade",
websocket_url="wss://ws.axiom.trade",
timeout=30,
max_retries=3,
rate_limit={
"requests_per_minute": 100,
"burst_limit": 10
}
)
client = AxiomTradeClient(config=config)
Attributes:
api_base_url
(str): Base URL for API requestswebsocket_url
(str): WebSocket endpoint URLtimeout
(int): Request timeout in secondsmax_retries
(int): Maximum retry attemptsrate_limit
(dict): Rate limiting configurationFor high-volume operations, use batch processing utilities:
from axiomtradeapi.batch import BatchProcessor
processor = BatchProcessor(client)
# Process 10,000 wallets efficiently
large_wallet_list = [...] # Your wallet addresses
results = await processor.process_wallets(
large_wallet_list,
batch_size=100,
concurrent_batches=5
)
Optimize performance with connection pooling:
from axiomtradeapi.pool import ConnectionPool
pool = ConnectionPool(
max_connections=20,
max_keepalive_connections=5,
keepalive_expiry=30
)
client = AxiomTradeClient(connection_pool=pool)
Customize retry behavior:
from axiomtradeapi.retry import ExponentialBackoffRetry
retry_strategy = ExponentialBackoffRetry(
max_retries=5,
base_delay=1,
max_delay=60,
exponential_base=2
)
client = AxiomTradeClient(retry_strategy=retry_strategy)
The API enforces the following rate limits:
Endpoint | Limit | Window |
---|---|---|
Balance queries | 100 requests | 1 minute |
Batch operations | 10 requests | 1 minute |
WebSocket connections | 5 connections | Per account |
Authentication | 10 requests | 1 minute |
Typical memory usage patterns:
import asyncio
import logging
from axiomtradeapi import AxiomTradeClient
class SimpleTradingBot:
def __init__(self, auth_token, refresh_token):
self.client = AxiomTradeClient(
auth_token=auth_token,
refresh_token=refresh_token,
log_level=logging.INFO
)
self.monitored_wallets = []
async def start_monitoring(self):
"""Start monitoring for new tokens and portfolio changes"""
# Subscribe to new tokens
await self.client.subscribe_new_tokens(self.handle_new_token)
# Start portfolio monitoring
asyncio.create_task(self.monitor_portfolio())
# Start WebSocket listener
await self.client.ws.start()
async def handle_new_token(self, tokens):
"""Handle new token announcements"""
for token in tokens:
if self.should_trade_token(token):
await self.execute_trade(token)
def should_trade_token(self, token):
"""Determine if we should trade this token"""
return (
token['marketCapSol'] > 10.0 and
token['liquiditySol'] > 50.0 and
token.get('verified_contract', False)
)
async def execute_trade(self, token):
"""Execute trading logic"""
logging.info(f"Trading signal for {token['tokenName']}")
# Implement your trading logic here
async def monitor_portfolio(self):
"""Monitor portfolio performance"""
while True:
try:
balances = self.client.GetBatchedBalance(self.monitored_wallets)
total_value = sum(b['sol'] for b in balances.values() if b)
logging.info(f"Portfolio value: {total_value:.6f} SOL")
await asyncio.sleep(60) # Check every minute
except Exception as e:
logging.error(f"Portfolio monitoring error: {e}")
# Usage
async def main():
bot = SimpleTradingBot(
auth_token="your-auth-token",
refresh_token="your-refresh-token"
)
await bot.start_monitoring()
if __name__ == "__main__":
asyncio.run(main())
from axiomtradeapi import AxiomTradeClient
import pandas as pd
from datetime import datetime, timedelta
class PortfolioAnalyzer:
def __init__(self, client):
self.client = client
self.historical_data = []
def analyze_portfolio(self, wallet_addresses):
"""Perform comprehensive portfolio analysis"""
# Get current balances
balances = self.client.GetBatchedBalance(wallet_addresses)
# Calculate metrics
total_sol = sum(b['sol'] for b in balances.values() if b)
wallet_count = len([b for b in balances.values() if b])
avg_balance = total_sol / wallet_count if wallet_count > 0 else 0
# Generate report
report = {
'timestamp': datetime.now().isoformat(),
'total_value_sol': total_sol,
'wallet_count': wallet_count,
'average_balance': avg_balance,
'largest_wallet': max(balances.values(), key=lambda x: x['sol'] if x else 0),
'smallest_wallet': min(balances.values(), key=lambda x: x['sol'] if x else float('inf')),
'distribution': self.calculate_distribution(balances)
}
return report
def calculate_distribution(self, balances):
"""Calculate balance distribution statistics"""
valid_balances = [b['sol'] for b in balances.values() if b]
if not valid_balances:
return {}
df = pd.Series(valid_balances)
return {
'mean': df.mean(),
'median': df.median(),
'std': df.std(),
'percentiles': {
'25th': df.quantile(0.25),
'75th': df.quantile(0.75),
'90th': df.quantile(0.90),
'95th': df.quantile(0.95)
}
}
# Usage
client = AxiomTradeClient()
analyzer = PortfolioAnalyzer(client)
wallets = ["wallet1", "wallet2", "wallet3"]
report = analyzer.analyze_portfolio(wallets)
print(f"Portfolio Report: {report}")
Key changes in the latest version:
# Old way (v0.x)
client = AxiomTradeClient("auth_token")
# New way (v1.x)
client = AxiomTradeClient(auth_token="auth_token")
# Old way
client.subscribe_tokens(callback)
# New way
await client.subscribe_new_tokens(callback)
# Old way
try:
balance = client.get_balance(address)
except Exception as e:
print(e)
# New way
try:
balance = client.GetBalance(address)
except APIError as e:
print(f"API Error: {e.message}")
except NetworkError as e:
print(f"Network Error: {e.message}")
We welcome contributions! See our Contributing Guide for details.
This API reference is part of the AxiomTradeAPI-py library, licensed under the MIT License.
Last updated: 2025-07-29