Test all UBDCC DepthCaches.

ubdcc_test_all_depthcaches.py

from unicorn_binance_local_depth_cache import BinanceLocalDepthCacheManager, DepthCacheClusterNotReachableError
import asyncio
import datetime

limit_count: int = 1
ubdcc_address: str = "YOUR_UBDCC_IP"
ubdcc_port: int = 80
  

async def main():
    while True:
        dc = await ubldc.cluster.get_depthcache_list_async()
        errors = {}
        non_working_caches = []
        working_caches = []
        for dcl_exchange in dc['depthcache_list']:
            print(f"Testing {len(dc['depthcache_list'][dcl_exchange])} DepthCaches for exchange '{dcl_exchange}' on "
                  f"UBDCC '{ubdcc_address}' at {datetime.datetime.now()}!")
            for market in dc['depthcache_list'][dcl_exchange]:
                asks = await ubldc.cluster.get_bids_async(exchange=dcl_exchange, market=market, limit_count=limit_count)
                if asks.get('error_id') is not None:
                    errors[asks.get('error_id')] = 1 if errors.get(asks.get('error_id')) is None else errors.get(asks.get('error_id')) + 1
                    non_working_caches.append(market)
                else:
                    working_caches.append(market)
        print(f"Successful working caches: {len(working_caches)}")
        for error in errors:
            print(f"ERROR: {error}: {errors[error]}")
        await asyncio.sleep(1)

try:
    with BinanceLocalDepthCacheManager(ubdcc_address=ubdcc_address, ubdcc_port=ubdcc_port) as ubldc:
        try:
            asyncio.run(main())
        except KeyboardInterrupt:
            print("\r\nGracefully stopping ...")
        except Exception as error_msg:
            print(f"ERROR: {error_msg}")
except DepthCacheClusterNotReachableError as error_msg:
    print(f"ERROR: {error_msg}")