pybottersでbybitの約定履歴と板情報をMongodbに保存するスクリプト

仮想通貨トレードbotのオープンソースライブラリpybottersを利用して、バックテスト用のデータを保存するスクリプトです。本記事の内容はpybotters開発者のまちゅけんさんのDiscordでのレクチャーを、私が勝手にまとめたものになります。まちゅけんさんがすでに同様の記事を書いているので、まずはこちらを参照してもらったほうがいいかもしれません。（参考）BybitのWebSocketから約定履歴をデータベースに溜め込むスクリプト【pybotters】

スクリプト本体

以下のスクリプトを実行すると、bybitの約定履歴と5秒ごとの板情報がMongodbに保存されます。

bybit_mongodb.py

import asyncio
import pybotters
import motor.motor_asyncio
import json
import time
from rich import print
import datetime


async def main():
    async with pybotters.Client() as client:
        store = pybotters.BybitDataStore()
        await client.ws_connect(
            'wss://stream.bybit.com/realtime',
            send_json={
                'op': 'subscribe',
                'args': ['orderBookL2_25.BTCUSD', 'trade.BTCUSD'],
            },
            hdlr_json=store.onmessage,
        )

        # MongoDBクライアント
        mongo = motor.motor_asyncio.AsyncIOMotorClient()
        db = mongo['bybit']
        collection_orderbook = db['orderbook']
        collection_trade = db['trade']

        # 最初のsnapshotが送られてくるまで待つ
        while not len(store.orderbook) >= 50:
            await store.orderbook.wait()

        while True:
            # trade保存
            trade = store.trade.find()
            if trade:
                # DBに保存する分はDataStoreから削除
                store.trade._clear()
                await collection_trade.insert_many(trade)
                print(trade)

            # orderbook 保存
            symbol = 'BTCUSD'
            orderbook = store.orderbook.sorted({'symbol': symbol})
            text = json.dumps(orderbook)
            doc = {'time': time.time(), 'symbol': symbol, 'orderbook': text}
            await collection_orderbook.insert_one(doc)

            # best_ask, best_bid 表示
            # この部分は動作確認用のprint機能なのでなくても動きます
            best_ask = float(orderbook['Sell'][0]['price'])
            best_bid = float(orderbook['Buy'][0]['price'])
            spread = best_ask - best_bid
            spread_ratio = spread / best_ask
            time_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print(
                [
                    time_str,
                    {'best_ask': best_ask, 'best_bid': best_bid},
                    {'spread': spread, 'spread_ratio': '{:%}'.format(spread_ratio)},
                ]
            )

            # データを保存は5秒ごとに行う
            await asyncio.sleep(5)


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

実行結果

実行するとターミナルに5秒ごとにbest ask, best bidと約定履歴がprintされます。

Mongodbには以下のように約定履歴が保存されます(GUIツールはMongoDB Compassを使用しています)。

板情報についても5秒ごとに{"Sell": [], "Buy": []}の形でask, bidの価格、サイズがリストでソートされたものが保存されていきます。

スクリプトの概要

bybitのpublic websocketに接続し、約定履歴と板情報を取得しています。データの保存に関しては、pybottersのDataStoreクラスを用いて、約定履歴(trade)、板情報(orderbook)を5秒ごとにDBに書き出しています。スクリプト上でstoreという名前で定義されているのがそれです。

        store = pybotters.BybitDataStore()

DataStoreクラスは、Websocketから受け取った情報をリアルタイムで格納しておくためのもので、実際のbotトレードではstoreに格納されている情報を参照し売買ロジックの計算を行います。DataStoreクラスはbotトレード用の機能であり、今回のようなデータ収集のための機能ではないのですが、それを転用してデータ収集にも使えますという感じです。

DataStoreクラスを使わないでデータを保存する方法は、まちゅけんさんの記事が参考になります。BybitのWebSocketから約定履歴をデータベースに溜め込むスクリプト【pybotters】

スクリプトの流れ

Websocket接続のコード

今回は板情報と約定履歴が欲しいので、orderbook2_25とtradeのチャンネルをsubscribeします。ハンドラ関数の部分は、store.onmessageとし、storeにWebsocketから流れてきたデータをリアルタイムで格納するようにします。

    async with pybotters.Client() as client:
        store = pybotters.BybitDataStore()
        await client.ws_connect(
            'wss://stream.bybit.com/realtime',
            send_json={
                'op': 'subscribe',
                'args': ['orderBookL2_25.BTCUSD', 'trade.BTCUSD'],
            },
            hdlr_json=store.onmessage,
        )

約定履歴の保存

store.trade.find()でstoreに格納されたtradeの情報を参照しています。store.tradeは直近10万件の約定履歴を保存する仕様になっていますが、今回は5秒ごとにstoreからDBにデータを移し替えて、移し替えた分のstore.tradeの値は削除したいので、保存処理（insert_many)の前にclear()処理を入れています。

（補足）insertする前にclearするようにしてください。今回はinsertが非同期処理のため、insert処理実行中にもstoreには約定履歴が溜め込まれています。insertしてからclearしてしまうとinsert実行中に溜め込めれたデータがあった場合はDBに保存されずにclearされてしまいます。insertする前にclearしておけば、このような事態を避けられます。

            # trade保存
            trade = store.trade.find()
            if trade:
                # DBに保存する分はDataStoreから削除
                store.trade._clear()
                await collection_trade.insert_many(trade)
                print(trade)

板情報の保存

orderbookは、直近の板情報がstore.orderbookに格納されています。sorted()でask, bidをソートしてリストとして出力できるので、これをDBに保存します。

            # orderbook 保存
            symbol = 'BTCUSD'
            orderbook = store.orderbook.sorted({'symbol': symbol})
            text = json.dumps(orderbook)
            doc = {'time': time.time(), 'symbol': symbol, 'orderbook': text}
            await collection_orderbook.insert_one(doc)

ループ処理の頻度

今回は5秒おきに板情報を保存するのでasyncio.sleep(5)として、5秒ごとのループを実行します。こちらの数字を変更すれば、何秒おきに保存するかを設定できます。またawait store.trade.wait()のように書き換えれば、約定があったタイミングで板情報を保存できます。

            # データを保存は5秒ごとに行う
            await asyncio.sleep(5)

応用

bybit以外でもFTX, Binanceなどの取引所でDataStoreクラスが整備されているため、今回のようなスクリプトでデータ収集できます。またWebsocket接続部分で、複数通貨をsubscribeすれば、複数通貨のデータを同時に収集できます。板情報を書き出す際はorderbook = store.orderbook.sorted({'symbol': symbol})のsymbolを、特定の通貨に書き換えてデータを保存してください。

私の実行環境

上記のスクリプトを24時間流すことでバックテスト用のデータを自炊することができます。私はなるべくコストを抑えて自炊したいので、ラズパイにHDDを接続してスクリプトを流しています。ラズパイを用いるときの注意ですが、MongoDBは64bitのOSしかサポートしていないため、ラズパイが32bitだと上記のスクリプトを実行することができません。ラズパイを用いる場合は64bitのOSをインストールしてください。私はUbuntuの64bitをインストールしています。

最後に

pybottersのDataStoreクラスを用いて、バックテスト用のデータをMongodbに保存することができました。このような素晴らしいライブラリをオープンソースで提供してくれたまちゅけんさんに感謝。