Asyncio function inside a thread

Question

The cryptofeed is a python library that uses asyncio library to get real-time prices of different Crypto Exchanges. In this short program, we try to run the cryptofeed FeedHandler in an independent thread. An example of the code is shown below:

import functools as fct
from cryptofeed import FeedHandler
from cryptofeed.defines import BID, ASK, L2_BOOK
from cryptofeed.exchanges import Kraken
from datetime import datetime
import threading


async def bookfunc(params , orderbooks, feed, symbol, book, timestamp, receipt_timestamp):

    print(f'Timestamp: {timestamp} Cryptofeed Receipt: {receipt_timestamp} Feed: {feed} Symbol: {symbol}'
          f' Book Bid Size is {len(book[BID])} Ask Size is {len(book[ASK])}')
    orderbooks = filter_orderbook(orderbooks, book, symbol, params['orderbook']['depth'])


def func():

    # Parameters
    params = {'orderbook': {'depth': 2}, 'price_model':{}, 'trade_model': {}}
    config = {'log': {'filename': 'logs/demo.log', 'level': 'INFO'}}

    orderbooks = {}

    f = FeedHandler(config=config)
    f.add_feed(Kraken(checksum_validation=True, subscription={L2_BOOK: ['BTC-USD', 'ETH-USD', 'LINK-USD', 'LTC-USD', 'ADA-USD']},
                      callbacks={L2_BOOK: fct.partial(bookfunc, params, orderbooks)})) # This way passes the orderbooks inside the callback

    f.run()


if __name__ == '__main__':
    thread = threading.Thread(target=func, args=())
    thread.start()

When the code is executed, the get the following error:

raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'Thread-1'.

Any idea how to solve this problem?

EDIT: This is the solution for different questions in stackoverflow. An example is the following Question:

Running cryptofeed (asyncio library) in a new thread

user_2072035 · Answer

(As of cryptofeed==1.9.2) there are two important things to make it run from a thread:

FeedHandler.run sets up signal handlers by default, which must be done from the main thread. To avoid that, there's install_signal_handlers argument on the method.
Initialiser of FeedHandler set's uvloop's policy, but doesn't call uvloop.install() (assuming it's the application's responsibility I guess). Without it asyncio.set_event_loop doesn't have effect. Alternatively you can set 'uvloop': False in the feed handler config (as shown below), or just uninstall uvloop.

import asyncio
import threading

from cryptofeed import FeedHandler
from cryptofeed.defines import BID, ASK, L2_BOOK
from cryptofeed.exchanges import Kraken


async def bookfunc(**kwargs):
    print('bookfunc', kwargs)
    

def run_feed_handler_forever():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    config = {
        'uvloop': False,
        'log': {'filename': 'log.log', 'level': 'DEBUG'},
    }
    l2_book = ['BTC-USD', 'ETH-USD', 'LINK-USD', 'LTC-USD', 'ADA-USD']
    feed = Kraken(
        subscription={L2_BOOK: l2_book}, callbacks={L2_BOOK: bookfunc}
    )
    fh = FeedHandler(config)
    fh.add_feed(feed)
    fh.run(install_signal_handlers=False)


if __name__ == '__main__':
    thread = threading.Thread(target=run_feed_handler_forever)
    thread.start()
    thread.join()

user_15337735 · Answer

You have to manually set an event loop for your thread through asyncio. You can do it by wrapping your function in a sort of loop setter:

def create_loop(func):
   loop = asyncio.new_event_loop()
   asyncio.set_event_loop(loop)
   loop.run_until_complete(func())
   
if __name__ == "__main__":
    thread = threading.Thread(target=create_loop, args=(func,))
    thread.start()

Asyncio function inside a thread

2 Answers