Ws Test Load

import asyncio
import os
import sys
import psutil
import time
from collections import defaultdict
from datetime import datetime

import ccxt.pro as ccxt

class PerformanceMetrics:
    def __init__(self):
        self.message_count = defaultdict(int)
        self.error_count = defaultdict(int)
        self.start_time = time.time()
        self.process = psutil.Process()
        self.last_print_time = time.time()
        self.print_interval = 5  # Print metrics every 5 seconds

    def increment_message_count(self, symbol):
        self.message_count[symbol] += 1

    def increment_error_count(self, symbol):
        self.error_count[symbol] += 1

    def get_metrics(self):
        current_time = time.time()
        elapsed = current_time - self.start_time
        
        # Calculate messages per second
        total_messages = sum(self.message_count.values())
        total_errors = sum(self.error_count.values())
        messages_per_second = total_messages / elapsed if elapsed > 0 else 0
        
        # Get system metrics
        memory_info = self.process.memory_info()
        cpu_percent = self.process.cpu_percent()
        
        return {
            'elapsed_time': elapsed,
            'total_messages': total_messages,
            'total_errors': total_errors,
            'messages_per_second': messages_per_second,
            'memory_usage_mb': memory_info.rss / (1024 * 1024),  # Convert to MB
            'cpu_percent': cpu_percent,
            'symbols_subscribed': len(self.message_count)
        }

    def should_print_metrics(self):
        current_time = time.time()
        if current_time - self.last_print_time >= self.print_interval:
            self.last_print_time = current_time
            return True
        return False

async def watch_orderbook(binance, symbol, metrics):
    while True:
        try:
            await binance.watch_order_book(symbol)
            metrics.increment_message_count(symbol)
            
            if metrics.should_print_metrics():
                current_metrics = metrics.get_metrics()
                print(f"\nPerformance Metrics at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}:")
                print(f"Symbols subscribed: {current_metrics['symbols_subscribed']}")
                print(f"Messages per second: {current_metrics['messages_per_second']:.2f}")
                print(f"Total messages: {current_metrics['total_messages']}")
                print(f"Total errors: {current_metrics['total_errors']}")
                print(f"Memory usage: {current_metrics['memory_usage_mb']:.2f} MB")
                print(f"CPU usage: {current_metrics['cpu_percent']:.1f}%")
                print(f"Elapsed time: {current_metrics['elapsed_time']:.1f} seconds")
                print("-" * 50)
        except Exception as e:
            print(f"Error in {symbol}: {str(e)}")
            metrics.increment_error_count(symbol)
            await asyncio.sleep(1)  # Wait before retrying

async def main():
    binance = ccxt.binance({})
    await binance.load_markets()
    symbols = binance.symbols
    metrics = PerformanceMetrics()
    
    print(f"Starting to monitor {len(symbols)} symbols...")
    
    tasks = []
    for symbol in symbols[:500]:
        await asyncio.sleep(0.1)
        task = asyncio.create_task(watch_orderbook(binance, symbol, metrics))
        tasks.append(task)
    

    await asyncio.gather(*tasks)

if __name__ == '__main__':
    asyncio.run(main())

# Results Using asyncio 3.12^
# Performance Metrics at 2025-05-18 20:24:04:
# Symbols subscribed: 100
# Messages per second: 358.01
# Total messages: 174240
# Memory usage: 73.94 MB
# CPU usage: 6.6%
# Elapsed time: 486.7 seconds

# Results using current version using asyncio 3.10
# Performance Metrics at 2025-05-18 20:41:31:
# Symbols subscribed: 100
# Messages per second: 297.41
# Total messages: 119353
# Memory usage: 79.78 MB
# CPU usage: 7.5%
# Elapsed time: 401.3 seconds