ccxt: How to restart a coroutine after a websocket stream stops receiving data? (1006 error code)

Hello,

I’m writing an asyncio application to monitor prices and trade/order events, but for an unknown reason some streams stop receiving data after few hours. I’m not familiar with the asyncio package and I would appreciate help in finding a solution.

Basically, the code below establishs websocket connections with Binance to listen streams of six symbols (ETH/USD, BTC/USD, BNB/USD for both spot and future) and trades events from two accounts (user1, user2). The application uses the library ccxtpro. The public method watch_ohlcv get price steams, while private methods watchMyTrades and watchOrders get new orders and trades events at account level.

The problem is that one or several streams are interrupted after few hours, and the object response get empty or None. I would like to detect and restart these streams after they stops working, how can I do that ?

# tasks.py
@app.task(bind=True, name='Start websocket loops')
def start_ws_loops(self):
    ws_loops()

# methods.py
def ws_loops():

    async def method_loop(client, exid, wallet, method, private, args):

        exchange = Exchange.objects.get(exid=exid)

        if private:
            account = args['account']
        else:
            symbol = args['symbol']

        while True:
            try:

                if private:
                    response = await getattr(client, method)()
                    if method == 'watchMyTrades':
                        do_stuff(response)

                    elif method == 'watchOrders':
                        do_stuff(response)

                else:
                    response = await getattr(client, method)(**args)  # <-------- Empty object after some times !?
                    if method == 'watch_ohlcv':
                        do_stuff(response)

                # await asyncio.sleep(3)

            except Exception as e:
                print(str(e))
                break
        
        await client.close()

    async def clients_loop(loop, dic):

        exid = dic['exid']
        wallet = dic['wallet']
        method = dic['method']
        private = dic['private']
        args = dic['args']

        exchange = Exchange.objects.get(exid=exid)
        parameters = {'enableRateLimit': True, 'asyncio_loop': loop, 'newUpdates': True}

        if private:
            log.info('Initialize private instance')
            account = args['account']
            client = exchange.get_ccxt_client_pro(parameters, wallet=wallet, account=account)

        else:
            log.info('Initialize public instance')
            client = exchange.get_ccxt_client_pro(parameters, wallet=wallet)

        mloop = method_loop(client, exid, wallet, method, private, args)
        await gather(mloop)
        await client.close()

    async def main(loop):

        lst = []
        private = ['watchMyTrades', 'watchOrders']
        public = ['watch_ohlcv']

        for exid in ['binance']:
            for wallet in ['spot', 'future']:
                
                # Private
                for method in private:
                    for account in ['user1', 'user2']:
                        lst.append(dict(exid=exid,
                                        wallet=wallet,
                                        method=method,
                                        private=True,
                                        args=dict(account=account)
                                        ))
                
                # Public
                for method in public:
                    for symbol in ['ETH/USD', 'BTC/USD', 'BNB/USD']:
                        lst.append(dict(exid=exid,
                                        wallet=wallet,
                                        method=method,
                                        private=False,
                                        args=dict(symbol=symbol,
                                                  timeframe='5m',
                                                  limit=1
                                                  )
                                        ))

        loops = [clients_loop(loop, dic) for dic in lst]
        await gather(*loops)

    loop = asyncio.new_event_loop()
    loop.run_until_complete(main(loop))