Better way of outputting variable from websockets with asyncio?-CodePudding

I have been playing around with websockets and asyncio to get price data streamed from Bitstamp. After connecting to the websockets feed I am struggling to return the data I'm receiving for use in another function. I'm currently using the function store_data() as a placeholder to check the price data being returned.

I have come up with a fairly crude way of storing the price, which involves declaring the global variable bitstamp_asks and then overwriting it within the websockets while loop.

I feel there must be a better way, so any help would be really appreciated. My code is below:

import asyncio
import websockets
import json

bitstamp_asks = []

async def bitstamp_connect():
    
    global bitstamp_asks

    uri = "wss://ws.bitstamp.net/"
    subscription = {
        "event": "bts:subscribe",
        "data": {
            "channel": "order_book_btcusd"
        }
    }

    async with websockets.connect(uri) as ws:
        await ws.send(json.dumps(subscription))

        subscription_response = await ws.recv()
        print(json.loads(subscription_response)['event'])

        while True:
            response = await ws.recv()
            bitstamp_data = json.loads(response)['data']
            bitstamp_asks = float(bitstamp_data['asks'][0][0])
            print(bitstamp_asks)
            await asyncio.sleep(0.0001)

async def store_data():
    while True:
        print(f'The stored value is: {bitstamp_asks}')
        await asyncio.sleep(0.1)

asyncio.get_event_loop().run_until_complete(asyncio.wait([bitstamp_connect(),store_data()]))

CodePudding user response：

You can use asyncio.Queue to share values between the tasks. For example:

import asyncio
import websockets
import json


q = asyncio.Queue()


async def bitstamp_connect():

    uri = "wss://ws.bitstamp.net/"
    subscription = {
        "event": "bts:subscribe",
        "data": {"channel": "order_book_btcusd"},
    }

    async with websockets.connect(uri) as ws:
        await ws.send(json.dumps(subscription))

        subscription_response = await ws.recv()
        print(json.loads(subscription_response)["event"])

        last_value = None
        while True:
            response = await ws.recv()
            bitstamp_data = json.loads(response)["data"]
            bitstamp_asks = float(bitstamp_data["asks"][0][0])

            if last_value is None or bitstamp_asks != last_value:
                q.put_nowait(bitstamp_asks)
                last_value = bitstamp_asks


async def store_data():
    while True:
        data = await q.get()
        print(f"The stored value is: {data}")
        q.task_done()


asyncio.get_event_loop().run_until_complete(
    asyncio.wait([bitstamp_connect(), store_data()])
)