Example

このページでは pybotters を利用したbotの実装例を紹介します。

Bybit インバース無期限契約

低頻度botのサンプル

ローソク足(1分足)を元にトレードを行うbot

import asyncio
import time

import pybotters


# apis
apis = {
    'bybit': ['BYBIT_API_KEY', 'BYBIT_API_SECRET']
}

async def main():
    async with pybotters.Client(apis=apis, base_url='https://api.bybit.com') as client:
        # 必要な初期処理
        ...

        # メインループ
        while True:
            # REST APIデータ並列リクエスト
            resps = await asyncio.gather(
                client.get('/v2/public/kline/list', params={
                    'symbol': 'BTCUSD', 'interval': 1, 'from': int(time.time()) - 3600
                }),
                client.get('/v2/private/order', params={'symbol': 'BTCUSD'}),
                client.get('/v2/private/position/list', params={'symbol': 'BTCUSD'}),
            )
            kline, order, position = await asyncio.gather(*[r.json() for r in resps])

            # シグナル計算
            """
            something awesome logic...
            """
            cond = 'Whether to execute...'
            side = 'Buy or Sell...'
            qty = 'Calculated value...'

            # オーダー執行
            if cond:
                await client.post('/v2/private/order/create', data={
                    'symbol': 'BTCUSD',
                    'side': side,
                    'order_type': 'Market',
                    'qty': qty,
                    # 'price': price,
                    'time_in_force': 'GoodTillCancel',
                })

            # 待機(60秒)
            await asyncio.sleep(60.0)


# 非同期メイン関数を実行(Ctrl+Cで終了)
if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

高頻度botの場合

板情報を元にトレードを行うbot

import asyncio

import pybotters


# apis
apis = {
    'bybit': ['...', '...'],
}

async def main():
    async with pybotters.Client(apis=apis, base_url='https://api.bybit.com') as client:
        # データストアのインスタンスを生成する
        store = pybotters.BybitDataStore()

        # REST API由来のデータ(オーダー・ポジション・残高)を初期データとしてデータストアに挿入する
        await store.initialize(
            client.get('/v2/private/order', params={'symbol': 'BTCUSD'}),
            client.get('/v2/private/position/list', params={'symbol': 'BTCUSD'}),
            client.get('/v2/private/wallet/balance', params={'symbol': 'BTCUSD'}),
        )

        # WebSocket接続
        wstask = await client.ws_connect(
            'wss://stream.bybit.com/realtime',
            send_json={'op': 'subscribe', 'args': [
                'orderBookL2_25.BTCUSD',
                'trade.BTCUSD',
                'instrument_info.100ms.BTCUSD',
                'position',
                'execution',
                'order',
            ]},
            hdlr_json=store.onmessage,
        )

        # WebSocketでデータを受信するまで待機
        while not all([
            len(store.orderbook),
            len(store.instrument),
        ]):
            await store.wait()

        # その他必要な初期処理
        ...

        # メインループ
        while True:
            # データ参照
            orderbook = store.orderbook.find()
            order = store.order.find()
            position = store.position_inverse.find()

            # シグナル計算
            """
            something awesome logic...
            """
            cond = 'Whether to execute...'
            side = 'Buy or Sell...'
            qty = 'Calculated value...'
            price = 'Amazing price...'

            # オーダー執行
            if cond:
                # 高頻度では重複オーダーしないようにオーダー後WebSocketでデータ受信するまで待機させる
                # RESTの応答よりWebSocketのイベントの方が速い可能性があるので先にイベント待機タスクをスケジュールする
                event = asyncio.create_task(store.order.wait())
                await client.post('/v2/private/order/create', data={
                    'symbol': 'BTCUSD',
                    'side': side,
                    'order_type': 'Limit',
                    'qty': qty,
                    'price': price,
                    'time_in_force': 'GoodTillCancel',
                })
                await event

            # 板情報のイベントまで待機
            await store.orderbook.wait()


# 非同期メイン関数を実行(Ctrl+Cで終了)
if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass