pythonのリアルタイム通信の接続が切れた

初心者です。
pythonで取引所のデータをリアルタイムに、非同期通信で取得する
プログラムを実行していると、最初はうまく取得してくれるのですが、そのうち
２.３分するとエラーは出ずに止まってしまいます。

取引所のリファレンスには、

[ネットワークやプログラムの問題を回避するためにping、WebSocket接続を維持するために30秒ごとにハートビートパケットを送信することをお勧めします。ws.send('{"op":"ping"}');]

とかかれていたのですがpingを送信する方法や再接続の仕方など
ご教示いただけたら幸いです。
よろしくお願いいたします。

python
1import asyncio
2import aiohttp
3import time
4
5async def run_forever(url, send_json):
6    
7    async with session.ws_connect(url) as ws:
8        
9        await ws.send_json(send_json)
10        async for msg in ws:
11            if msg.type == aiohttp.WSMsgType.TEXT:
12                
13                a = msg.json()
14                
15                print("depth {}".format(a))
16                
17            elif msg.type == aiohttp.WSMsgType.ERROR:
18                
19                print('error')
20                break
21
22async def main():
23    global session
24    async with aiohttp.ClientSession() as session:
25        wstask1 = asyncio.create_task(run_forever("wss://stream.bybit.com/spot/quote/ws/v2", {"topic": "depth","event": "sub","params": {"symbol": "BITUSDT"}}))
26        wstask2 = asyncio.create_task(run_forever("wss://stream.bybit.com/spot/quote/ws/v2", {"topic": "depth","event": "sub","params": {"symbol": "BITBTC"}}))
27        wstask3 = asyncio.create_task(run_forever("wss://stream.bybit.com/spot/quote/ws/v2", {"topic": "depth","event": "sub","params": {"symbol": "BTCUSDT"}}))
28    
29        await wstask1
30        await wstask2
31        await wstask3
32        
33
34await main()

行動規範の内容に同意します

回答1件

ベストアンサー

定期的にその{"op":"ping"}のデータを送らないといけないですが、async for msg in ws:とするとメッセージを受信するまで待機されてしまい送る機会がなくなる可能性があります。
ですので、async for msg in ws:ではなくws.receiveを使って取得を行うようにして、標準ライブラリasyncioにある指定されただけ実行終了を待機するwait_forを使えば良いです。
ws.receiveはメッセージを受信するものでasync for msg in ws:の内部で呼ばれているものです。
それとそのwait_forは非同期関数を実行して指定された時間だけ待機して、その時間分待機した場合はエラー(asyncio.TimeoutError)を発生させて実行を止めるということをしてくれます。
そのwait_forで1秒程メッセージ受信待機をするようにして、メッセージ受信で一秒経っても受信できなかったかつ前回のping送信または接続時から三十秒経っていたら、{"op":"ping"}を送信するというプログラムを組めば良いです。

例（動作未確認）：

python
1import asyncio
2import aiohttp
3import time
4
5async def run_forever(url, send_json):
6    
7    async with session.ws_connect(url) as ws:
8        
9        await ws.send_json(send_json)
10        before = time.time() # 三十秒が経過したかを確認するために時間を記録する。(エポック秒)
11        while not ws.closed: # 接続が切れるまでループさせる。
12            try:
13                # メッセージの受信を一秒だけ待機する。
14                msg = await asyncio.wait_for(ws.receive(), timeout=1)
15            except asyncio.TimeoutError:
16                # wait_forは待機し終わったらエラーを発生させるのでハンドリングする。
17                # 待機し終わったかつ三十秒が経過をしている場合はハートビートパケットを送る。
18                if time.time() - before > 30:
19                    await ws.send_str('{"op":"ping"'})
20                    before = time.time()
21                    continue
22
23            if msg.type == aiohttp.WSMsgType.TEXT:
24                
25                a = msg.json()
26                
27                print("depth {}".format(a))
28                
29            elif msg.type == aiohttp.WSMsgType.ERROR:
30                
31                print('error')
32                break
33
34async def main():
35    global session
36    async with aiohttp.ClientSession() as session:
37        wstask1 = asyncio.create_task(run_forever("wss://stream.bybit.com/spot/quote/ws/v2", {"topic": "depth","event": "sub","params": {"symbol": "BITUSDT"}}))
38        wstask2 = asyncio.create_task(run_forever("wss://stream.bybit.com/spot/quote/ws/v2", {"topic": "depth","event": "sub","params": {"symbol": "BITBTC"}}))
39        wstask3 = asyncio.create_task(run_forever("wss://stream.bybit.com/spot/quote/ws/v2", {"topic": "depth","event": "sub","params": {"symbol": "BTCUSDT"}}))
40    
41        await wstask1
42        await wstask2
43        await wstask3
44        
45
46await main()