Looking for advice on the code

[River-Shi]

Here is my code of connecting to the binance websocket exchange. First of all, picows is an awesome package, and it's faster than websockets 13.0.1. I have few questions on the code bellow:

1. Why is on_ws_connected called when it's running, but when I end it with ctrl+C, only print("Websocket closed.") runs and I don't see Disconnected from {self._exchange_id} Websocket.?

Connected to Binance Websocket.
{'result': None, 'id': 1728739767256}
{'u': 52832030573, 's': 'BTCUSDT', 'b': '63037.99000000', 'B': '1.13459000', 'a': '63038.00000000', 'A': '3.76523000'}
{'u': 52832030581, 's': 'BTCUSDT', 'b': '63037.99000000', 'B': '1.13459000', 'a': '63038.00000000', 'A': '3.76505000'}
{'u': 52832030601, 's': 'BTCUSDT', 'b': '63037.99000000', 'B': '1.13441000', 'a': '63038.00000000', 'A': '3.76505000'}
Websocket closed.

2. auto_ping_idle_timeout What does this parameter mean? I looked at the documentation, and it says it's how long to wait before sending a ping request when there's no incoming data. So if there's always data coming in, it won't automatically ping, right?

3. Based on my understanding: there are 2 different types of ping/pong. One ping/pong is part of the websocket protocol and tests whether the websocket connection is still open and whether the other participant is still responding. I am not sure this part is handled automactically if enable_auto_ping=True, is set to True. And the other type is more like a keep alive on the security level and should be implemented separately! For example, some exchanges like bybit require the client to send heartbeat packet, they recommend to send the ping heartbeat packet every 20 seconds to maintain the connection, So if I set auto_ping_idle_timeout = 20, will I send a ping every 20 seconds?

4. I find binance websocket server would send ping to me, even if I have not enabled the enable_auto_ping, Since every ws push runs if frame.msg_type == WSMsgType.PING: once, is there any way to prevent ping from entering on_ws_frame, thus I can remove the if statement.

5. For subscribing to a large number of symbols, like 50-100, I can use two methods:

• Use the same websocket, sharing the transport, and send a send subscribe every time I subscribe to one.
• Create a separate transport for each symbol, running them concurrently through a coroutine.

Since the data arrives in order, if I put all of them into an async queue, I think there might be data congestion. The second method means a new WSListener is created for each symbol, but I'm concerned if that’s going to use too many resources. I'm wondering which one is the best practice. Do you have any suggestions?

6. I'm uncertain if this is the correct approach for re-connecting. If the transport disconnects, will it raise a WSError? Could you please provide best practices for handling reconnections?

        while True:
            transport, listener = await ws_connect(
                WSClientFactory,
                self._url,
                enable_auto_ping=True,
                auto_ping_idle_timeout=self._ping_idle_timeout, 
                auto_ping_reply_timeout=self._ping_reply_timeout,
            )
            try:
                transport.send(WSMsgType.TEXT, json.dumps(payload).encode("utf-8"))
                while True:
                    msg = await listener.msg_queue.get()
                    if asyncio.iscoroutinefunction(callback):
                        await callback(msg, *args, **kwargs)
                    else:
                        callback(msg, *args, **kwargs)
                    listener.msg_queue.task_done()
            except WSError as e:
                print(f"Connection error: {e}")
            await asyncio.sleep(1)

7. if the exchange send me a ping, do I need to handle it in on_ws_frame like the code bellow:

def on_ws_frame(self, transport: WSTransport, frame: WSFrame):
        if frame.msg_type == WSMsgType.PING:
            transport.send_pong(frame.get_payload_as_bytes())
            return
        try:
            msg = orjson.loads(frame.get_payload_as_bytes())
            self.msg_queue.put_nowait(msg)
        except orjson.JSONDecodeError:
            print("Failed to decode message")

Here is my entire code:

from picows import WSListener, WSTransport, WSFrame, WSMsgType, ws_connect, WSError
import asyncio
import json
import orjson
from collections.abc import Callable
from typing import Dict, Any
import time

def cb(msg):
    print(msg)

class WSClient(WSListener):
    def __init__(self, exchange_id: str = ""):
        self._exchange_id = exchange_id
        self.msg_queue = asyncio.Queue()

    def on_ws_connected(self, transport: WSTransport):
        print(f"Connected to {self._exchange_id} Websocket.")

    def on_ws_disconnected(self, transport: WSTransport):
        print(f"Disconnected from {self._exchange_id} Websocket.")

    def on_ws_frame(self, transport: WSTransport, frame: WSFrame):
        if frame.msg_type == WSMsgType.PING:
            return

        msg = orjson.loads(frame.get_payload_as_bytes())
        self.msg_queue.put_nowait(msg)


class BinanceWsManager:
    def __init__(self, url: str):
        self._url = url
        self._ping_idle_timeout = 2
        self._ping_reply_timeout = 1
        self._tasks = []
    
    async def _subscribe(
        self, payload: Dict[str, Any], callback: Callable[..., Any], *args, **kwargs
    ):
        WSClientFactory = lambda: WSClient("Binance")  # noqa: E731
        while True:
            transport, listener = await ws_connect(
                WSClientFactory,
                self._url,
                enable_auto_ping=True,
                auto_ping_idle_timeout=self._ping_idle_timeout, 
                auto_ping_reply_timeout=self._ping_reply_timeout,
            )
            try:
                transport.send(WSMsgType.TEXT, json.dumps(payload).encode("utf-8"))
                while True:
                    msg = await listener.msg_queue.get()
                    if asyncio.iscoroutinefunction(callback):
                        await callback(msg, *args, **kwargs)
                    else:
                        callback(msg, *args, **kwargs)
                    listener.msg_queue.task_done()
            except WSError as e:
                print(f"Connection error: {e}")
            await asyncio.sleep(1)
        
    async def subscribe_book_ticker(
        self, symbol, callback: Callable[..., Any] = None, *args, **kwargs
    ):
        id = int(time.time() * 1000)
        payload = {
            "method": "SUBSCRIBE",
            "params": [f"{symbol.lower()}@bookTicker"],
            "id": id,
        }
        self._tasks.append(
            asyncio.create_task(self._subscribe(payload, callback=callback, *args, **kwargs))
        )

async def main():
    try:
        url = "wss://stream.binance.com:9443/ws"
        ws_manager = BinanceWsManager(url)
        await ws_manager.subscribe_book_ticker("BTCUSDT", callback=cb)
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Websocket closed.")


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

[Tapanhaz]

The auto ping mechanism should work .. I already prepared two clients : https://github.com/Tapanhaz/Shoonya-Websocket with manual ping management in python and https://github.com/Tapanhaz/Kite-Ticker with auto ping in cython . almost all broker websocket requires ping in idle time.

[tarasko]

1. Cancellation in asyncio is a quite complex topic. In theory asyncio.run should cancel all outstanding tasks (CancelledError is emitted for both main task and all background tasks, that were created by create_task). Then asyncio.run should give control to all the tasks to handle CancelledError and finish cleanly.

When it comes to picows, it relies fully on original transport/protocol behavior of asyncio.
asyncio doesn't initiate disconnect and call Protocol.connection_lost on cancellation. In asyncio protocol/transport you are supposed to handle this explicitly, i.e. call disconnect explicitly and wait, if necessary, until disconnect is complete.

Something like this:

    async def _subscribe(
            self, payload: Dict[str, Any], callback: Callable[..., Any], *args,
            **kwargs
    ):
        WSClientFactory = lambda: WSClient("Binance")  # noqa: E731
        while True:
            transport = None
            try:
                transport, listener = await ws_connect(
                    WSClientFactory,
                    self._url,
                    enable_auto_ping=True,
                    auto_ping_idle_timeout=self._ping_idle_timeout,
                    auto_ping_reply_timeout=self._ping_reply_timeout,
                )
                transport.send(WSMsgType.TEXT,
                               json.dumps(payload).encode("utf-8"))

                while True:
                    msg = await listener.msg_queue.get()
                    if asyncio.iscoroutinefunction(callback):
                        await callback(msg, *args, **kwargs)
                    else:
                        callback(msg, *args, **kwargs)
                    listener.msg_queue.task_done()
            except asyncio.CancelledError:
                # No need to handle as asyncio will handle it itself
                raise
            except Exception as e:
                # For all other exceptions, reconnect
                print(f"Connection failure: {str(e)}, retry in 1 second")
            finally:
                # Initiate disconnect and wait until it is finished
                # It is safe to call this if socket is already disconnecting/disconnected
                transport.disconnect()
                await transport.wait_disconnected()

            # Sleep a little bit before reconnect
            await asyncio.sleep(1)

2. Correct. Currently it won't automatically ping if there is data.
   Indeed it is a good idea to just send ping periodically regardless of incoming data.
   Somehow I just wasn't doing it (I was only pinging when there is no data) and it always worked in all of my cases.

I'm thinking about adding a new parameter to ws_connect: auto_ping_strategy with 2 possible values PING_PERIODICALLY and PING_WHEN_IDLE.
With PING_PERIODICALLY, auto_ping_idle_timeout will mean just how often to ping.
Will this work for you?

3. There are 3 different problems:
   a) we need to check if remote peer is alive. We always want to check if peer's application level is alive, not just that low-level websocket library is replying to PING frames.
   b) remote peer want's to check if we're alive
   c) we need to make sure that http connection stays alive, proxies and servers don't disconnect us. Proxies and servers don't care what data is send/received. There just must be some data withing keep-alive timeout (around 30 seconds)

picows auto_ping solves a) and c) by sending either websocket PING/PONG or user defined messages (which could be application level messages)

To achieve a) and c) it doesn't really matter what is send and received and, imho it is overkill to send both websockets PING/PONG and user-level heartbeats.

You can just choose one (application level preferrably), and it will work for all practical cases. I don't think any remote peer would ever have logic to check if you indeed send websocket PING or heartbeat every N seconds. They all just immediately reply and, maybe, log but nobody would set any timer to check that you're being regular with your pings or heartbeats.

4. Replying to PING is solving problem b. Currently picows doesn't have a code that automatically replies to websocket PING but I'm fine with adding this to picows.
   But for now you have to do it:

        if frame.msg_type == WSMsgType.PING:
            transport.send_pong(frame.get_payload_as_bytes())
            return

5. From memory usage/locality perspective it is better to use one transport because then you would re-use the same incoming data buffer. Perhaps there would be less L1/L2/L3 cache misses. But without measuring it, it is really difficult to say.

What I really don't recommend to do is to use async when handling data. I would keep the whole data path non-async or at least do async at the very last step, only if really necessary.

So, this whole asyncio.Queue() thing, I would definitely not do it! :)

I would pass all market events to autotrader directly from on_ws_frame.
Autotrader logic would process them and decide whether it wants to send an order to exchange (maybe 1 out of 1000 events). Then right before sending order, maybe go async with create_task. But only if really necessary.

After ws_connect has succeed it is only necessary to wait_disconnected().

• wait_disconnected() doesn't throw.
• ws_connect may throw any exception that asyncio.Loop.create_connection can throw.
• ws_connect may also throw WSError when there is an error during websocket negotiation phase.

I will clarify this in the docs.

I personally just do ws_connect and wait_disconnected in the loop. The data is passed to a non-async callback directly from on_ws_frame.
If you're using queue and async callbacks, I think it is good to make a separate task for it.

But again, for performance, I would strongly advise not to use async callbacks and queues in the data path.

    async def _subscribe(
            self, payload: Dict[str, Any], callback: Callable[..., Any], *args,
            **kwargs
    ):
        WSClientFactory = lambda: WSClient("Binance")  # noqa: E731
        while True:
            drain_queue_task = None
            transport = None
            try:
                transport, listener = await ws_connect(
                    WSClientFactory,
                    self._url,
                    enable_auto_ping=True,
                    auto_ping_idle_timeout=self._ping_idle_timeout,
                    auto_ping_reply_timeout=self._ping_reply_timeout,
                )
                transport.send(WSMsgType.TEXT,
                               json.dumps(payload).encode("utf-8"))

                drain_queue_task = asyncio.get_running_loop().create_task(self._drain_queue(listener, callback, *args, **kwargs))
                await transport.wait_disconnected()
            except asyncio.CancelledError:
                raise
            except Exception as e:
                print(f"Connection failure: {str(e)}, retry in 1 second")
            finally:
                if transport is not None:
                    # It is ok to call transport.disconnect/wait_disconnected multiple times
                    transport.disconnect()
                    await transport.wait_disconnected()

                if drain_queue_task is not None:
                    drain_queue_task.cancel()

            await asyncio.sleep(1)

    async def _drain_queue(self, listener: asyncio.Queue, callback, *args, **kwargs):
        while True:
            try:
                msg = await listener.msg_queue.get()
                if asyncio.iscoroutinefunction(callback):
                    await callback(msg, *args, **kwargs)
                else:
                    callback(msg, *args, **kwargs)
                listener.msg_queue.task_done()
            except asyncio.CancelledError:
                break

7. Yes, you need for now. But I'll move it to the picows in the next release

[River-Shi]

@tarasko @Tapanhaz Thanks for the reply.

1. Yes, PING_PERIODICALLY would work for me.

2. You suggest removing async.Queue from the WSListener class and adding data.emit in on_ws_frame, which triggers on_event_data in the autotrader each time data is received. I use async.Queue because if on_event_data has extensive calculations (taking 0.2s) while the push speed is 0.01s, on_ws_frame would lag. The async.Queue allows on_ws_frame to function smoothly while enabling the consumer to process data at its own pace.

3. If I use the same transport across subscriptions, the first call to _subscription will assign self._transport and self._listener, while subsequent calls will reuse the existing transport without running ws_connect. However, if the connection is disconnected, ws_connect will not execute as expected since self._transport is not None. How to solve the problem?

    async def _subscribe(
        self, payload: Dict[str, Any], callback: Callable[..., Any], *args, **kwargs
    ):
        WSClientFactory = lambda: WSClient("Binance")  # noqa: E731
        while True:
            if not self._listener and self._transport:
                transport, listener = await ws_connect(
                    WSClientFactory,
                    self._url,
                    enable_auto_ping=True,
                    auto_ping_idle_timeout=self._ping_idle_timeout, 
                    auto_ping_reply_timeout=self._ping_reply_timeout,
                )
            try:
                transport.send(WSMsgType.TEXT, json.dumps(payload).encode("utf-8"))
                while True:
                    msg = await listener.msg_queue.get()
                    if asyncio.iscoroutinefunction(callback):
                        await callback(msg, *args, **kwargs)
                    else:
                        callback(msg, *args, **kwargs)
                    listener.msg_queue.task_done()
            except WSError as e:
                print(f"Connection error: {e}")
            await asyncio.sleep(1)

4. Do I need to include await transport.wait_disconnected()? I believed this method blocks the thread, so if I have a while True loop after transport.send(), is it necessary to add await transport.wait_disconnected()?

            while True:
                    msg = await listener.msg_queue.get()
                    if asyncio.iscoroutinefunction(callback):
                        await callback(msg, *args, **kwargs)
                    else:
                        callback(msg, *args, **kwargs)
                    listener.msg_queue.task_done()

[River-Shi]

I have solved the problem 3

[River-Shi]

I rewrote my code a bit, and I'm not sure if this is the best practice. Since I needed to use the producer-consumer pattern, I still went with async.Queue.

I also have several questions:

1. After reconnecting, do I need to resubscribe all the symbols?
2. Is this the best way to handle reconnecting?

async def _handle_connection(self):
        reconnect = False
        while True:
            try:
                await self._connect(reconnect)
                # TODO: when reconnecting, need to resubscribe to the channels
                await self._transport.wait_disconnected()
            except WSError as e:
                print(f"Connection error: {e}")
            reconnect = True
            await asyncio.sleep(1)

Here’s my code:

from picows import WSListener, WSTransport, WSFrame, WSMsgType, ws_connect, WSError
import asyncio
import json
import orjson
import time
from asynciolimiter import Limiter


class WSClient(WSListener):
    def __init__(self, exchange_id: str = ""):
        self._exchange_id = exchange_id
        self.msg_queue = asyncio.Queue()

    def on_ws_connected(self, transport: WSTransport):
        print(f"Connected to {self._exchange_id} Websocket.")

    def on_ws_disconnected(self, transport: WSTransport):
        print(f"Disconnected from {self._exchange_id} Websocket.")

    def on_ws_frame(self, transport: WSTransport, frame: WSFrame):
        if frame.msg_type == WSMsgType.PING:
            transport.send_pong(frame.get_payload_as_bytes())
            return
        try:
            msg = orjson.loads(frame.get_payload_as_bytes())
            self.msg_queue.put_nowait(msg)
        except Exception as e:
            print(frame.get_payload_as_bytes())
            print(f"Error parsing message: {e}")


class BinanceWsManager:
    def __init__(self, url: str):
        self._url = url
        self._ping_idle_timeout = 2
        self._ping_reply_timeout = 1
        self._listener = None
        self._transport = None
        self._tasks = []
        self._limiter = Limiter(5/1) # 5 requests per second

    async def _connect(self, reconnect: bool = False):
        if not self._transport and not self._listener or reconnect:
            WSClientFactory = lambda: WSClient("Binance")  # noqa: E731
            self._transport, self._listener = await ws_connect(
                WSClientFactory,
                self._url,
                enable_auto_ping=True,
                auto_ping_idle_timeout=self._ping_idle_timeout,
                auto_ping_reply_timeout=self._ping_reply_timeout,
            )

    async def _handle_connection(self):
        reconnect = False
        while True:
            try:
                await self._connect(reconnect)
                # TODO: when reconnecting, need to resubscribe to the channels
                await self._transport.wait_disconnected()
            except WSError as e:
                print(f"Connection error: {e}")
            reconnect = True
            await asyncio.sleep(1)

    async def subscribe_book_ticker(self, symbol):
        await self._connect()
        await self._limiter.wait()
        id = int(time.time() * 1000)
        payload = {
            "method": "SUBSCRIBE",
            "params": [f"{symbol.lower()}@bookTicker"],
            "id": id,
        }
        self._transport.send(WSMsgType.TEXT, json.dumps(payload).encode("utf-8"))
    
    async def subscribe_trade(self, symbol):
        await self._connect()
        await self._limiter.wait()
        id = int(time.time() * 1000)
        payload = {
            "method": "SUBSCRIBE",
            "params": [f"{symbol.lower()}@trade"],
            "id": id,
        }
        self._transport.send(WSMsgType.TEXT, json.dumps(payload).encode("utf-8"))
        
    async def _msg_handler(self):
        while True:
            msg = await self._listener.msg_queue.get()
            # TODO: handle different event types of messages
            print(msg)
            self._listener.msg_queue.task_done()

    async def start(self):
        asyncio.create_task(self._msg_handler())
        await self._handle_connection()


async def main():
    try:
        url = "wss://stream.binance.com:9443/ws"
        ws_manager = BinanceWsManager(url)
        await ws_manager.subscribe_book_ticker("BTCUSDT")
        await ws_manager.subscribe_book_ticker("ETHUSDT")
        await ws_manager.subscribe_book_ticker("SOLUSDT")
        await ws_manager.subscribe_book_ticker("BNBUSDT")
        await ws_manager.subscribe_trade("BTCUSDT")
        await ws_manager.subscribe_trade("ETHUSDT")
        await ws_manager.subscribe_trade("BNBUSDT")
        await ws_manager.subscribe_trade("SOLUSDT")
        await ws_manager.start()

    except asyncio.CancelledError:
        print("Websocket closed.")


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

[tarasko]

You're welcome!

I'll release PING_PERIODICALLY this weeks, perhaps even today.

[tarasko]

On a side note and maybe I'll put this in the docs:

A lot of complexity around handling cancellation/disconnects comes from low-level asyncio.Loop.create_connection.
If you ever use asyncio.Loop.create_connection, you would be facing the exactly same problems.

I was a bit scared to wrap it with some friendlier interface, because there were reasons why asyncio.Loop.create_connection is designed like this. It is difficult to satisfy all possible use cases, I certainly can't reason about all possible use-cases, and a wrapper can easily reduce flexibility.

That's why:
ws_create_server returns asyncio.Server (same as asyncio.Loop.create_server)
ws_connect returns tuple (transport, listener) (same as asyncio.Loop.create_connection)

When dealing with trade-off between convenient interface vs performance/flexibility, picows goes for the latter because convenience is already provide by aiohttp/websockets.

[tarasko]

Hi @River-Shi
1.6.0 is released

[River-Shi]

Hi @River-Shi 1.6.0 is released

Thank you so much! I’m not sure if I can ask you a question about Cython, and if you think it’s unrelated to Picows, feel free to skip it. I’ve created two classes using Cython and Python dataclass, but the accessing speed of the Cython class is slower than that of the Python dataclass. Here’s a detailed description of my issue cdef-test. I would really appreciate your help.

[tarasko]

I agree with the stackoverflow's answer.
When accessing Cython type double attribute from Python there will be an extra call to PyFloat_FromDouble.

The following will replicate the original Python's behavior with Cython and should have, I believe, simular access times. But it will loose a lot of other cool features offered by cython

cdef class BookL1:
    cdef readonly str exchange
    cdef readonly str symbol
    cdef readonly object bid
    cdef readonly object ask
    cdef readonly object bid_size
    cdef readonly object ask_size

    def __init__(self, str exchange, str symbol, bid, ask, bid_size, ask_size):
        self.exchange = exchange
        self.symbol = symbol
        self.bid = bid
        self.ask = ask
        self.bid_size = bid_size
        self.ask_size = ask_size