Pybottersを使おう WebSocket編
前回の続きで、Pybottersを使ってWebSocketに接続してみます。
1.インストール
PyPIに登録してあるので、pip installだけでインストール可能です。
pip install pybotters2.使用方法
2-1.RestAPI
2-2.WebSocket
bybitで試してみましょう。
import pybotters
import asyncio
from rich import print
import pandas as pd
#windowsのみ
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
apis = {
'bybit' :['', ''],
'bybit_testnet' :['', ''],
'binance' :['', ''],
'binance_testnet':['', ''],
'ftx' :['', ''],
'bitflyer' :['', ''],
'gmocoin' :['', ''],
'liquid' :['', ''],
'bitbank' :['', '']
}
RestAPI_url = {
'bybit' :'https://api.bybit.com',
'bybit_testnet' :'https://api-testnet.bybit.com',
'binance' :'https://fapi.binance.com',
'binance_testnet':'https://testnet.binancefuture.com',
'ftx' :'https://ftx.com/api/',
'bitflyer' :'https://api.bitflyer.com/v1/',
'gmocoin_public' :'https://api.coin.z.com/public',
'gmocoin_private':'https://api.coin.z.com/private',
'liquid' :'https://api.liquid.com',
'bitbank_public' :'https：//public.bitbank.cc',
'bitbank_private':'https：//api.bitbank.cc/v1'
}
wss_url = {
'bybit' :'wss://stream.bybit.com/realtime',
'bybit_testnet' :'wss://stream-testnet.bybit.com/realtime',
'binance' :'wss://fstream.binance.com',
'binance_testnet':'wss://stream.binancefuture.com',
'ftx' :'wss://ftx.com/ws/',
'gmocoin' :''
}
async def bybit_ws():
async with pybotters.Client(apis=apis) as client:
# データストアのインスタンスを生成する
store = pybotters.BybitDataStore()
# WebSocket接続
wstask = await client.ws_connect(
wss_url['bybit'],
send_json={'op': 'subscribe',
'args': [
"orderBook_200.100ms.BTCUSD",
'trade.BTCUSD',
'instrument_info.100ms.BTCUSD',
]
},
hdlr_json=store.onmessage,
)
# WebSocketでデータを受信するまで待機
while not all([
len(store.orderbook),
len(store.instrument)
]):
await store.wait()
# メインループ
while True:
# データ参照
data = dict(
orderbook =store.orderbook.sorted(),
instrument=store.instrument.get({"symbol":"BTCUSD"}),
position =store.position_inverse.find(),
trade =store.trade.find(),
execution =store.execution.find(),
order =store.order.find()
)
print(data['instrument'])
await asyncio.sleep(1)
if __name__ == '__main__':
try:
asyncio.run(bybit_ws())
except KeyboardInterrupt:
passメイン関数の解説をしていきます。
メイン関数前の辞書リストは前回の記事で説明しています。
# データストアのインスタンスを生成する
store = pybotters.BybitDataStore()まずはBybit専用のデータストアを作成します。
データストアはWSで取得したデータを保存しておく場所で、取り出す際に必要に応じてソートや最新のものを取り出す等ができます。
# WebSocket接続
wstask = await client.ws_connect(
wss_url['bybit'],
send_json={'op': 'subscribe',
'args': [
"orderBook_200.100ms.BTCUSD",
'trade.BTCUSD',
'instrument_info.100ms.BTCUSD',
]
},
hdlr_json=store.onmessage,
)WebSocketに接続します。
send_json()の中に、購読したいトピックを記入して、WS接続を行います。
hdlr_jsonは受け取ったjson形式のデータをどう処理するか決める部分で、store.onmessageを指定するとデータストアにどんどんデータが突っ込まれていきます。
pybotters.print_handlerにすると文字列でコンソールに出力されます。
一度WSに接続したら、再接続処理等は不要です。
pybottersの仕様で自動でやってくれるらしいです（神か、、、）
# WebSocketでデータを受信するまで待機
while not all(
[
len(store.orderbook),
len(store.instrument)
]
):
await store.wait()地味に重要な部分です。
WS接続を行うと、トピックごとに最初のデータを受信する時間が違います。
ここでほしいデータを受信したことを確認してからメインの処理に移ります。
# メインループ
while True:
# データ参照
data = dict(
orderbook =store.orderbook.sorted(),
instrument=store.instrument.get({"symbol":"BTCUSD"}),
position =store.position_inverse.find(),
trade =store.trade.find(),
execution =store.execution.find(),
order =store.order.find()
)
print(data['instrument'])
await asyncio.sleep(1)続いてメインループです。
dataに辞書形式でデータストアから取り出した値を格納しています。
※dict(a=b)で{'a':'b'}と同じ扱いになります。
ループの待機時間にはtime.sleep()ではなく、await asyncio.sleep()を使います。こうしないとsleepが同期処理として扱われるので、全体の処理がスリープしていしまいます。
if __name__ == '__main__':
try:
asyncio.run(bybit_ws())
except KeyboardInterrupt:
pass最後にメイン関数の実行部分です。
前回説明したように、呼び出し時はasyncio.run()を使います。
・・・・・
とりあえずここまでにします。
冒頭のコードを動かすと、1秒ごとにコンソールにティッカー情報が表示されますので、ぜひ動かしてみてください。