Performance Optimization

Performance Optimization

Guide to optimizing trading bot performance with AxiomTradeAPI: latency, memory, concurrency, network tuning, profiling, and production configuration.

Table of Contents

• Performance Fundamentals
• Latency Optimization
• Memory Management
• Concurrent Processing
• Network Optimization
• Profiling and Monitoring
• Production Optimization
• Benchmarking Results

Performance Fundamentals

Performance optimization matters in algorithmic trading — every millisecond can affect whether a trade executes at the intended price. The AxiomTradeAPI is designed for high-performance applications; this guide covers how to get the most out of it.

Key performance metrics

import time
import asyncio
import statistics
from dataclasses import dataclass
from typing import List, Dict, Optional
from axiomtradeapi import AxiomTradeClient, AxiomTradeWebSocketClient

@dataclass
class PerformanceMetrics:
    """Comprehensive performance metrics tracking"""
    api_response_times: List[float]
    websocket_latency: List[float]
    order_execution_times: List[float]
    memory_usage: List[float]
    cpu_usage: List[float]
    network_throughput: List[float]
    error_rates: Dict[str, int]

    def get_statistics(self) -> Dict[str, Dict[str, float]]:
        """Calculate comprehensive performance statistics"""
        stats = {}

        for metric_name, values in {
            'api_response_times': self.api_response_times,
            'websocket_latency': self.websocket_latency,
            'order_execution_times': self.order_execution_times,
            'memory_usage': self.memory_usage,
            'cpu_usage': self.cpu_usage,
            'network_throughput': self.network_throughput
        }.items():
            if values:
                stats[metric_name] = {
                    'mean': statistics.mean(values),
                    'median': statistics.median(values),
                    'p95': self._percentile(values, 95),
                    'p99': self._percentile(values, 99),
                    'min': min(values),
                    'max': max(values),
                    'std_dev': statistics.stdev(values) if len(values) > 1 else 0
                }

        return stats

    def _percentile(self, values: List[float], percentile: int) -> float:
        """Calculate percentile value"""
        sorted_values = sorted(values)
        index = int(len(sorted_values) * percentile / 100)
        return sorted_values[min(index, len(sorted_values) - 1)]

class PerformanceTracker:
    """Performance tracking system for optimization work"""

    def __init__(self):
        self.metrics = PerformanceMetrics(
            api_response_times=[],
            websocket_latency=[],
            order_execution_times=[],
            memory_usage=[],
            cpu_usage=[],
            network_throughput=[],
            error_rates={}
        )
        self.start_time = time.time()

    async def track_api_call(self, func, *args, **kwargs):
        """Track API call performance"""
        start_time = time.time()
        try:
            result = await func(*args, **kwargs)
            return result
        except Exception as e:
            error_type = type(e).__name__
            self.metrics.error_rates[error_type] = self.metrics.error_rates.get(error_type, 0) + 1
            raise
        finally:
            response_time = time.time() - start_time
            self.metrics.api_response_times.append(response_time)

            if response_time > 1.0:  # > 1 second
                print(f"Slow API call detected: {func.__name__} took {response_time:.3f}s")

    def generate_performance_report(self) -> str:
        """Generate comprehensive performance report"""
        stats = self.metrics.get_statistics()
        runtime = time.time() - self.start_time

        report = f"""
AxiomTradeAPI Performance Report
{'=' * 50}
Runtime: {runtime:.2f} seconds

API Response Times:
  Mean: {stats.get('api_response_times', {}).get('mean', 0):.3f}s
  Median: {stats.get('api_response_times', {}).get('median', 0):.3f}s
  P95: {stats.get('api_response_times', {}).get('p95', 0):.3f}s
  P99: {stats.get('api_response_times', {}).get('p99', 0):.3f}s

WebSocket Latency:
  Mean: {stats.get('websocket_latency', {}).get('mean', 0):.3f}s
  P95: {stats.get('websocket_latency', {}).get('p95', 0):.3f}s
  P99: {stats.get('websocket_latency', {}).get('p99', 0):.3f}s

Order Execution:
  Mean: {stats.get('order_execution_times', {}).get('mean', 0):.3f}s
  P95: {stats.get('order_execution_times', {}).get('p95', 0):.3f}s

Memory Usage:
  Mean: {stats.get('memory_usage', {}).get('mean', 0):.1f} MB
  Peak: {stats.get('memory_usage', {}).get('max', 0):.1f} MB

CPU Usage:
  Mean: {stats.get('cpu_usage', {}).get('mean', 0):.1f}%
  Peak: {stats.get('cpu_usage', {}).get('max', 0):.1f}%

Error Rates:
"""

        for error_type, count in self.metrics.error_rates.items():
            report += f"  {error_type}: {count} occurrences\n"

        return report

Latency Optimization

Connection pool optimization

import aiohttp
import asyncio
from typing import Optional

class OptimizedAxiomClient:
    """High-performance AxiomTradeAPI client with connection pooling"""

    def __init__(self, auth_token: str, max_connections: int = 100):
        self.auth_token = auth_token
        self.session: Optional[aiohttp.ClientSession] = None
        self.max_connections = max_connections
        self.connection_pool_initialized = False

    async def initialize_connection_pool(self):
        """Initialize optimized connection pool"""
        if self.connection_pool_initialized:
            return

        connector = aiohttp.TCPConnector(
            limit=self.max_connections,           # Total connection pool size
            limit_per_host=50,                   # Max connections per host
            keepalive_timeout=60,                # Keep connections alive longer
            enable_cleanup_closed=True,          # Clean up closed connections
            use_dns_cache=True,                  # Cache DNS lookups
            ttl_dns_cache=300,                   # DNS cache TTL (5 minutes)
            family=0,                            # Use both IPv4 and IPv6
            force_close=False                    # Reuse connections
        )

        timeout = aiohttp.ClientTimeout(
            total=30,      # Total timeout
            connect=5,     # Connection timeout
            sock_read=10   # Socket read timeout
        )

        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={
                'User-Agent': 'AxiomTradeAPI-OptimizedClient/1.0',
                'Accept': 'application/json',
                'Accept-Encoding': 'gzip, deflate',
                'Connection': 'keep-alive'
            }
        )

        await self._warmup_connections()
        self.connection_pool_initialized = True

    async def _warmup_connections(self):
        """Pre-warm connection pool to reduce first-request latency"""
        warmup_tasks = []
        base_url = "https://api.axiomtrade.com"

        for i in range(5):
            task = self._warmup_request(f"{base_url}/health")
            warmup_tasks.append(task)

        await asyncio.gather(*warmup_tasks, return_exceptions=True)

    async def _warmup_request(self, url: str):
        """Single warmup request"""
        try:
            async with self.session.get(url) as response:
                await response.read()
        except Exception:
            pass  # Ignore warmup errors

    async def optimized_api_call(self, endpoint: str, method: str = 'GET', **kwargs):
        """Make optimized API call with performance tracking"""
        if not self.connection_pool_initialized:
            await self.initialize_connection_pool()

        start_time = time.time()

        try:
            url = f"https://api.axiomtrade.com{endpoint}"
            headers = kwargs.pop('headers', {})
            headers.update({
                'Authorization': f'Bearer {self.auth_token}',
                'Accept': 'application/json'
            })

            async with self.session.request(method, url, headers=headers, **kwargs) as response:
                # Stream response for large payloads
                if response.content_length and response.content_length > 1024 * 1024:  # > 1MB
                    data = b''
                    async for chunk in response.content.iter_chunked(8192):
                        data += chunk
                    result = data.decode()
                else:
                    result = await response.text()

                response_time = time.time() - start_time

                if response_time > 0.5:  # > 500ms
                    print(f"Slow API response: {endpoint} took {response_time:.3f}s")

                return {
                    'data': result,
                    'status_code': response.status,
                    'response_time': response_time,
                    'headers': dict(response.headers)
                }

        except Exception as e:
            response_time = time.time() - start_time
            print(f"API call failed: {endpoint} - {str(e)} - Time: {response_time:.3f}s")
            raise

    async def close(self):
        """Properly close the session"""
        if self.session:
            await self.session.close()

WebSocket optimization

import websockets
import json
import asyncio
from typing import Callable, Dict, Any

class HighPerformanceWebSocketClient:
    """Low-latency WebSocket client for real-time trading"""

    def __init__(self, auth_token: str):
        self.auth_token = auth_token
        self.websocket = None
        self.is_connected = False
        self.message_handlers: Dict[str, Callable] = {}
        self.latency_tracker = []
        self.compression_enabled = True

    async def connect(self, url: str = "wss://ws.axiomtrade.com"):
        """Connect with settings tuned for minimum latency"""
        try:
            extra_headers = {
                'Authorization': f'Bearer {self.auth_token}',
                'User-Agent': 'AxiomTradeAPI-HighPerf/1.0'
            }

            compression = 'deflate' if self.compression_enabled else None

            self.websocket = await websockets.connect(
                url,
                extra_headers=extra_headers,
                compression=compression,
                ping_interval=20,      # Ping every 20 seconds
                ping_timeout=10,       # Ping timeout
                close_timeout=10,      # Close timeout
                max_size=10 * 1024 * 1024,  # 10MB max message size
                read_limit=2**16,      # 64KB read buffer
                write_limit=2**16      # 64KB write buffer
            )

            self.is_connected = True

            asyncio.create_task(self._ping_monitor())
            asyncio.create_task(self._latency_monitor())

        except Exception as e:
            print(f"WebSocket connection failed: {e}")
            raise

    async def _ping_monitor(self):
        """Monitor connection health with custom ping/pong"""
        while self.is_connected:
            try:
                if self.websocket:
                    ping_time = time.time()
                    await self.websocket.send(json.dumps({
                        'type': 'ping',
                        'timestamp': ping_time
                    }))

                await asyncio.sleep(10)  # Ping every 10 seconds
            except Exception as e:
                print(f"Ping monitor error: {e}")
                break

    async def _latency_monitor(self):
        """Monitor and track WebSocket latency"""
        while self.is_connected:
            try:
                if len(self.latency_tracker) > 100:
                    self.latency_tracker = self.latency_tracker[-100:]

                if self.latency_tracker:
                    avg_latency = sum(self.latency_tracker) / len(self.latency_tracker)
                    if avg_latency > 0.1:  # > 100ms average
                        print(f"High WebSocket latency detected: {avg_latency:.3f}s")

                await asyncio.sleep(30)  # Check every 30 seconds
            except Exception as e:
                print(f"Latency monitor error: {e}")
                break

    async def subscribe_to_tokens(self, token_addresses: List[str]):
        """Subscribe to token updates with batch optimization"""
        if not self.is_connected:
            await self.connect()

        subscription_message = {
            'type': 'subscribe',
            'tokens': token_addresses,
            'timestamp': time.time()
        }

        await self.websocket.send(json.dumps(subscription_message))

    async def listen_for_messages(self):
        """High-performance message processing loop"""
        while self.is_connected:
            try:
                if not self.websocket:
                    break

                message = await asyncio.wait_for(
                    self.websocket.recv(),
                    timeout=1.0
                )

                receive_time = time.time()

                try:
                    data = json.loads(message)

                    if data.get('type') == 'pong':
                        latency = receive_time - data.get('timestamp', receive_time)
                        self.latency_tracker.append(latency)
                        continue

                    data['_receive_time'] = receive_time

                    message_type = data.get('type', 'unknown')
                    if message_type in self.message_handlers:
                        asyncio.create_task(
                            self.message_handlers[message_type](data)
                        )

                except json.JSONDecodeError:
                    continue

            except asyncio.TimeoutError:
                continue  # Normal timeout, keep listening
            except websockets.exceptions.ConnectionClosed:
                self.is_connected = False
                break
            except Exception as e:
                print(f"WebSocket error: {e}")
                break

    def register_handler(self, message_type: str, handler: Callable):
        """Register message handler for specific message types"""
        self.message_handlers[message_type] = handler

    async def close(self):
        """Gracefully close WebSocket connection"""
        self.is_connected = False
        if self.websocket:
            await self.websocket.close()

Memory Management

Memory-efficient data structures

import sys
import gc
import psutil
import weakref
from collections import deque
from typing import Dict, List, Optional
import numpy as np

class MemoryOptimizedDataStore:
    """Memory-efficient data storage for trading applications"""

    def __init__(self, max_history_size: int = 10000):
        self.max_history_size = max_history_size

        # Use deque for O(1) append/pop operations
        self.price_history = deque(maxlen=max_history_size)
        self.volume_history = deque(maxlen=max_history_size)
        self.timestamp_history = deque(maxlen=max_history_size)

        # Use numpy arrays for numerical calculations (more memory efficient)
        self._price_array: Optional[np.ndarray] = None
        self._volume_array: Optional[np.ndarray] = None

        # Weak references to avoid memory leaks
        self._observers = weakref.WeakSet()

        self.process = psutil.Process()
        self.initial_memory = self.process.memory_info().rss

    def add_price_data(self, price: float, volume: float, timestamp: float):
        """Add price data with automatic memory management"""
        self.price_history.append(price)
        self.volume_history.append(volume)
        self.timestamp_history.append(timestamp)

        # Invalidate cached arrays
        self._price_array = None
        self._volume_array = None

        if len(self.price_history) % 1000 == 0:
            self._cleanup_memory()

    def get_price_array(self) -> np.ndarray:
        """Get price data as numpy array (cached for efficiency)"""
        if self._price_array is None:
            self._price_array = np.array(list(self.price_history), dtype=np.float32)
        return self._price_array

    def get_volume_array(self) -> np.ndarray:
        """Get volume data as numpy array (cached for efficiency)"""
        if self._volume_array is None:
            self._volume_array = np.array(list(self.volume_history), dtype=np.float32)
        return self._volume_array

    def _cleanup_memory(self):
        """Perform memory cleanup and optimization"""
        gc.collect()

        self._price_array = None
        self._volume_array = None

        current_memory = self.process.memory_info().rss
        memory_growth = (current_memory - self.initial_memory) / 1024 / 1024  # MB

        if memory_growth > 100:  # > 100MB growth
            print(f"Memory usage increased by {memory_growth:.1f}MB")

    def get_memory_stats(self) -> Dict[str, float]:
        """Get detailed memory usage statistics"""
        memory_info = self.process.memory_info()

        return {
            'rss_mb': memory_info.rss / 1024 / 1024,
            'vms_mb': memory_info.vms / 1024 / 1024,
            'percent': self.process.memory_percent(),
            'data_points': len(self.price_history),
            'estimated_data_size_mb': (
                len(self.price_history) * (sys.getsizeof(0.0) * 3) / 1024 / 1024
            )
        }

class MemoryEfficientBot:
    """Trading bot optimized for memory efficiency"""

    def __init__(self, auth_token: str):
        self.client = OptimizedAxiomClient(auth_token)
        self.data_store = MemoryOptimizedDataStore()
        self.memory_monitor_task: Optional[asyncio.Task] = None

    async def start_memory_monitoring(self):
        """Start background memory monitoring"""
        self.memory_monitor_task = asyncio.create_task(self._memory_monitor_loop())

    async def _memory_monitor_loop(self):
        """Background task to monitor memory usage"""
        while True:
            try:
                stats = self.data_store.get_memory_stats()

                print(f"Memory: {stats['rss_mb']:.1f}MB RSS, "
                      f"{stats['percent']:.1f}% usage, "
                      f"{stats['data_points']} data points")

                if stats['percent'] > 80:
                    print(f"High memory usage: {stats['percent']:.1f}%")
                    await self._emergency_memory_cleanup()

                await asyncio.sleep(60)  # Check every minute

            except Exception as e:
                print(f"Memory monitoring error: {e}")
                await asyncio.sleep(60)

    async def _emergency_memory_cleanup(self):
        """Emergency memory cleanup when usage is high"""
        if len(self.data_store.price_history) > 5000:
            # Keep only last 5000 data points
            recent_prices = list(self.data_store.price_history)[-5000:]
            recent_volumes = list(self.data_store.volume_history)[-5000:]
            recent_timestamps = list(self.data_store.timestamp_history)[-5000:]

            self.data_store.price_history.clear()
            self.data_store.volume_history.clear()
            self.data_store.timestamp_history.clear()

            for p, v, t in zip(recent_prices, recent_volumes, recent_timestamps):
                self.data_store.add_price_data(p, v, t)

        gc.collect()

Concurrent Processing

Advanced async patterns

import asyncio
from asyncio import Queue, Semaphore
from typing import List, Callable, Any
import concurrent.futures

class ConcurrentTradingEngine:
    """High-concurrency trading engine for AxiomTradeAPI"""

    def __init__(self, auth_token: str, max_concurrent_requests: int = 50):
        self.client = OptimizedAxiomClient(auth_token)
        self.max_concurrent_requests = max_concurrent_requests
        self.request_semaphore = Semaphore(max_concurrent_requests)

        # Task queues for different priorities
        self.high_priority_queue: Queue = Queue(maxsize=1000)
        self.normal_priority_queue: Queue = Queue(maxsize=5000)
        self.low_priority_queue: Queue = Queue(maxsize=10000)

        self.api_workers: List[asyncio.Task] = []
        self.processing_workers: List[asyncio.Task] = []

        # Thread pool for CPU-intensive tasks
        self.thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=4)

        self.completed_tasks = 0
        self.failed_tasks = 0
        self.start_time = time.time()

    async def start_workers(self, num_workers: int = 10):
        """Start concurrent worker tasks"""
        for i in range(num_workers):
            worker = asyncio.create_task(self._api_worker(f"api-worker-{i}"))
            self.api_workers.append(worker)

        for i in range(num_workers // 2):
            worker = asyncio.create_task(self._processing_worker(f"proc-worker-{i}"))
            self.processing_workers.append(worker)

    async def _api_worker(self, worker_id: str):
        """Worker for handling API requests with priority queuing"""
        while True:
            try:
                # Check high priority queue first
                try:
                    task = self.high_priority_queue.get_nowait()
                    priority = "HIGH"
                except asyncio.QueueEmpty:
                    try:
                        task = self.normal_priority_queue.get_nowait()
                        priority = "NORMAL"
                    except asyncio.QueueEmpty:
                        try:
                            task = await asyncio.wait_for(
                                self.low_priority_queue.get(),
                                timeout=1.0
                            )
                            priority = "LOW"
                        except asyncio.TimeoutError:
                            continue

                async with self.request_semaphore:
                    start_time = time.time()
                    try:
                        result = await task['func'](*task['args'], **task['kwargs'])
                        task['callback'](result, None)
                        self.completed_tasks += 1

                        execution_time = time.time() - start_time
                        if execution_time > 1.0:  # Log slow tasks
                            print(f"Slow task ({priority}): {task['name']} took {execution_time:.3f}s")

                    except Exception as e:
                        task['callback'](None, e)
                        self.failed_tasks += 1

            except Exception as e:
                print(f"Worker {worker_id} error: {e}")
                await asyncio.sleep(1)

    async def _processing_worker(self, worker_id: str):
        """Worker for CPU-intensive processing tasks"""
        while True:
            try:
                # Implementation depends on your specific processing needs
                await asyncio.sleep(0.1)  # Placeholder
            except Exception as e:
                print(f"Processing worker {worker_id} error: {e}")
                await asyncio.sleep(1)

    async def submit_task(self, func: Callable, callback: Callable,
                         priority: str = "NORMAL", name: str = "task",
                         *args, **kwargs):
        """Submit task to appropriate priority queue"""
        task = {
            'func': func,
            'args': args,
            'kwargs': kwargs,
            'callback': callback,
            'name': name,
            'submitted_at': time.time()
        }

        if priority == "HIGH":
            await self.high_priority_queue.put(task)
        elif priority == "LOW":
            await self.low_priority_queue.put(task)
        else:
            await self.normal_priority_queue.put(task)

    async def batch_balance_requests(self, wallet_addresses: List[str]) -> Dict[str, Any]:
        """Efficiently handle batch balance requests"""
        results = {}
        batch_semaphore = Semaphore(10)  # Max 10 concurrent requests

        async def fetch_balance(address: str):
            async with batch_semaphore:
                try:
                    balance = await self.client.optimized_api_call(
                        f'/balance/{address}',
                        method='GET'
                    )
                    return address, balance
                except Exception as e:
                    return address, {'error': str(e)}

        tasks = [fetch_balance(addr) for addr in wallet_addresses]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        for response in responses:
            if isinstance(response, Exception):
                continue

            address, balance_data = response
            results[address] = balance_data

        return results

    def get_performance_stats(self) -> Dict[str, Any]:
        """Get engine performance statistics"""
        runtime = time.time() - self.start_time

        return {
            'runtime_seconds': runtime,
            'completed_tasks': self.completed_tasks,
            'failed_tasks': self.failed_tasks,
            'success_rate': self.completed_tasks / max(1, self.completed_tasks + self.failed_tasks),
            'tasks_per_second': self.completed_tasks / max(1, runtime),
            'queue_sizes': {
                'high_priority': self.high_priority_queue.qsize(),
                'normal_priority': self.normal_priority_queue.qsize(),
                'low_priority': self.low_priority_queue.qsize()
            },
            'active_workers': len([w for w in self.api_workers if not w.done()])
        }

Network Optimization

Advanced network tuning

import socket
from typing import Optional

class NetworkOptimizedClient:
    """Network-optimized client for AxiomTradeAPI"""

    def __init__(self, auth_token: str):
        self.auth_token = auth_token
        self.tcp_socket_options = self._get_optimized_socket_options()

    def _get_optimized_socket_options(self) -> List[tuple]:
        """Get optimized TCP socket options"""
        return [
            (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),    # Enable keepalive
            (socket.SOL_TCP, socket.TCP_NODELAY, 1),        # Disable Nagle's algorithm
            (socket.SOL_TCP, socket.TCP_QUICKACK, 1),       # Enable quick ACK
            (socket.SOL_SOCKET, socket.SO_REUSEADDR, 1),    # Reuse addresses
        ]

    async def create_optimized_connection(self, host: str, port: int = 443):
        """Create optimized network connection"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        for level, optname, value in self.tcp_socket_options:
            try:
                sock.setsockopt(level, optname, value)
            except OSError:
                pass  # Some options may not be available on all systems

        # Set socket buffer sizes for high throughput
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)  # 64KB send buffer
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)  # 64KB receive buffer

        return sock

Profiling and Monitoring

Production profiling system

import cProfile
import pstats
import tracemalloc
import os
from typing import Dict, List, Any
import functools

class ProductionProfiler:
    """Production-ready profiling system for trading bots"""

    def __init__(self, enable_memory_profiling: bool = True):
        self.enable_memory_profiling = enable_memory_profiling
        self.profilers = {}
        self.memory_snapshots = []

        if enable_memory_profiling:
            tracemalloc.start(10)  # Track up to 10 frames

    def profile_function(self, func_name: str = None):
        """Decorator for profiling individual functions"""
        def decorator(func):
            @functools.wraps(func)
            async def async_wrapper(*args, **kwargs):
                profiler = cProfile.Profile()
                profiler.enable()

                start_time = time.time()
                try:
                    result = await func(*args, **kwargs)
                    return result
                finally:
                    profiler.disable()
                    execution_time = time.time() - start_time

                    profile_name = func_name or func.__name__
                    self.profilers[profile_name] = {
                        'profiler': profiler,
                        'execution_time': execution_time,
                        'timestamp': time.time()
                    }

                    if execution_time > 0.1:  # > 100ms
                        print(f"Slow function: {profile_name} took {execution_time:.3f}s")

            @functools.wraps(func)
            def sync_wrapper(*args, **kwargs):
                profiler = cProfile.Profile()
                profiler.enable()

                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    return result
                finally:
                    profiler.disable()
                    execution_time = time.time() - start_time

                    profile_name = func_name or func.__name__
                    self.profilers[profile_name] = {
                        'profiler': profiler,
                        'execution_time': execution_time,
                        'timestamp': time.time()
                    }

            return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
        return decorator

    def take_memory_snapshot(self, description: str = ""):
        """Take memory snapshot for leak detection"""
        if not self.enable_memory_profiling:
            return

        snapshot = tracemalloc.take_snapshot()
        self.memory_snapshots.append({
            'snapshot': snapshot,
            'description': description,
            'timestamp': time.time()
        })

        # Keep only last 10 snapshots
        if len(self.memory_snapshots) > 10:
            self.memory_snapshots.pop(0)

    def analyze_memory_growth(self) -> Dict[str, Any]:
        """Analyze memory growth between snapshots"""
        if len(self.memory_snapshots) < 2:
            return {'error': 'Need at least 2 snapshots for analysis'}

        current = self.memory_snapshots[-1]['snapshot']
        previous = self.memory_snapshots[-2]['snapshot']

        top_stats = current.compare_to(previous, 'lineno')

        analysis = {
            'top_memory_growth': [],
            'total_growth_mb': 0
        }

        total_growth = 0
        for stat in top_stats[:10]:  # Top 10 memory growth areas
            growth_mb = stat.size_diff / 1024 / 1024
            total_growth += stat.size_diff

            analysis['top_memory_growth'].append({
                'file': stat.traceback.format()[-1] if stat.traceback else 'unknown',
                'growth_mb': growth_mb,
                'current_mb': stat.size / 1024 / 1024,
                'count_diff': stat.count_diff
            })

        analysis['total_growth_mb'] = total_growth / 1024 / 1024
        return analysis

    def generate_performance_report(self) -> str:
        """Generate comprehensive performance report"""
        report = "AxiomTradeAPI Performance Analysis Report\n"
        report += "=" * 60 + "\n\n"

        if self.profilers:
            report += "Function Performance Analysis:\n"
            report += "-" * 40 + "\n"

            for func_name, data in sorted(
                self.profilers.items(),
                key=lambda x: x[1]['execution_time'],
                reverse=True
            )[:10]:  # Top 10 slowest functions
                stats = pstats.Stats(data['profiler'])
                stats.sort_stats('cumulative')

                report += f"\n{func_name}:\n"
                report += f"   Total Time: {data['execution_time']:.3f}s\n"

        if self.enable_memory_profiling and len(self.memory_snapshots) >= 2:
            memory_analysis = self.analyze_memory_growth()
            report += f"\nMemory Growth Analysis:\n"
            report += "-" * 40 + "\n"
            report += f"Total Growth: {memory_analysis['total_growth_mb']:.2f} MB\n\n"

            for i, growth in enumerate(memory_analysis['top_memory_growth'][:5]):
                report += f"{i+1}. {growth['file']}\n"
                report += f"   Growth: {growth['growth_mb']:.2f} MB\n"
                report += f"   Current: {growth['current_mb']:.2f} MB\n\n"

        return report

    def save_profile_data(self, directory: str = "profiles"):
        """Save profile data to files for detailed analysis"""
        os.makedirs(directory, exist_ok=True)
        timestamp = int(time.time())

        for func_name, data in self.profilers.items():
            filename = f"{directory}/profile_{func_name}_{timestamp}.prof"
            data['profiler'].dump_stats(filename)

        if self.memory_snapshots:
            latest_snapshot = self.memory_snapshots[-1]['snapshot']
            snapshot_file = f"{directory}/memory_snapshot_{timestamp}.txt"

            with open(snapshot_file, 'w') as f:
                top_stats = latest_snapshot.statistics('lineno')
                f.write("Top 50 Memory Usage Locations:\n")
                f.write("=" * 50 + "\n")

                for stat in top_stats[:50]:
                    f.write(f"{stat}\n")

# Usage example for production profiling
class ProfiledTradingBot:
    """Example bot with comprehensive profiling"""

    def __init__(self, auth_token: str):
        self.client = OptimizedAxiomClient(auth_token)
        self.profiler = ProductionProfiler(enable_memory_profiling=True)

    @property
    def profile_function(self):
        return self.profiler.profile_function

    @profile_function("get_balance")
    async def get_balance(self, wallet_address: str):
        """Profiled balance retrieval"""
        return await self.client.optimized_api_call(f'/balance/{wallet_address}')

    @profile_function("process_market_data")
    async def process_market_data(self, data: Dict[str, Any]):
        """Profiled market data processing"""
        await asyncio.sleep(0.01)  # Simulate processing

        if hash(str(data)) % 100 == 0:  # Every ~100 calls
            self.profiler.take_memory_snapshot("market_data_processing")

    async def generate_performance_report(self):
        """Generate and save performance report"""
        report = self.profiler.generate_performance_report()
        self.profiler.save_profile_data()
        return report

Production Optimization

Production configuration

import os
from typing import Dict, Any

class ProductionOptimizer:
    """Production optimization configuration"""

    @staticmethod
    def get_optimized_config() -> Dict[str, Any]:
        """Get optimized configuration for production"""
        return {
            'api_client': {
                'max_connections': int(os.getenv('MAX_CONNECTIONS', '100')),
                'connection_timeout': 5,
                'read_timeout': 30,
                'keepalive_timeout': 60,
                'retry_attempts': 3,
                'retry_delay': 1.0,
                'enable_compression': True
            },
            'websocket': {
                'ping_interval': 20,
                'ping_timeout': 10,
                'max_message_size': 10 * 1024 * 1024,  # 10MB
                'compression': 'deflate',
                'auto_reconnect': True,
                'reconnect_delay': 5.0
            },
            'performance': {
                'enable_profiling': os.getenv('ENABLE_PROFILING', 'false').lower() == 'true',
                'profile_memory': os.getenv('PROFILE_MEMORY', 'false').lower() == 'true',
                'max_memory_mb': int(os.getenv('MAX_MEMORY_MB', '1024')),
                'gc_threshold': (700, 10, 10),  # Aggressive garbage collection
                'log_slow_calls': True,
                'slow_call_threshold': 1.0  # seconds
            },
            'concurrency': {
                'max_concurrent_requests': int(os.getenv('MAX_CONCURRENT_REQUESTS', '50')),
                'worker_count': int(os.getenv('WORKER_COUNT', '10')),
                'queue_size': int(os.getenv('QUEUE_SIZE', '5000')),
                'thread_pool_size': int(os.getenv('THREAD_POOL_SIZE', '4'))
            },
            'monitoring': {
                'metrics_enabled': True,
                'metrics_port': int(os.getenv('METRICS_PORT', '8080')),
                'health_check_interval': 30,
                'performance_report_interval': 300  # 5 minutes
            }
        }

    @staticmethod
    def apply_system_optimizations():
        """Apply system-level optimizations"""
        import gc

        # Configure garbage collection for performance
        gc.set_threshold(700, 10, 10)  # More aggressive GC

        os.environ['PYTHONOPTIMIZE'] = '1'    # Enable optimizations
        os.environ['PYTHONUNBUFFERED'] = '1'  # Unbuffered output

Benchmarking Results

Representative benchmark results comparing optimization levels (Python 3.11, Ubuntu 22.04, 16GB RAM):

API response times (1000 requests):

WebSocket latency (10 minutes monitoring):

Memory usage (24 hour test):

Throughput (requests/second):

Best Practices Summary

Key optimization principles

1. Connection management - use connection pooling with proper limits, enable keepalive, optimize timeouts, and pre-warm connections during startup.
2. Memory efficiency - use appropriate data structures (deque, numpy arrays), implement proper cleanup and garbage collection, and monitor memory usage continuously.
3. Concurrency design - leverage async/await effectively, use semaphores to limit concurrent operations, and implement priority queuing for critical tasks.
4. Network optimization - tune TCP socket options, use compression when beneficial, and implement proper retry logic.
5. Monitoring and profiling - profile regularly to identify bottlenecks, monitor key performance metrics, and set up alerts for performance degradation.

Conclusion

Performance optimization is an ongoing process that requires careful measurement, analysis, and iteration. Key takeaways:

• Measure before optimizing
• Focus on the biggest bottlenecks first
• Test optimizations thoroughly
• Monitor performance in production

Start with the basic optimizations and gradually implement more advanced techniques as your system scales. Premature optimization can be counterproductive — always profile and measure the impact of your changes.

Related

• Building Trading Bots
• WebSocket Integration
• Error Handling