Bot

開発記録#178(2025/4/15)「Bybit自動取引Botを最小構成で立ち上げた開発ログ【pybotters v1.8.1対応】」

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。

🧠 はじめに

本記事では、pybotters v1.8.1 を使って、Bybitの板情報に基づき条件を満たしたときだけ自動で注文を出す**最小構成のMM Bot（Market Making Bot）**を構築するまでの開発ログをまとめます。

途中、pybotters v2 への移行も検討しましたが、今回は あえて安定版の v1.8.1 を使い切る方針で進めました。

🔧 開発環境

  • OS：macOS
  • Python：3.12.4（pyenv管理）
  • 仮想環境：venv
  • ライブラリ：pybotters==1.8.1
  • 取引所：Bybit（V5 API）
  • エディタ：VSCode

🛠 ステップ1：pybottersのバージョン整理

まず当初は、pybotters v2系 を導入しようとしたものの、pip install では v2ブランチが存在せず、GitHubからのインストールも不安定だったため、v1.8.1で構築する方針に切り替えました。

pip install pybotters==1.8.1

🧩 ステップ2：WebSocket接続と板情報の取得

pybotters v1.8.1 では、DataStore などの高機能コンポーネントはv2での提供のため、WebSocketメッセージを直接処理する方式で板情報を取得しました。

await client.ws_connect(
    "wss://stream.bybit.com/v5/public/linear",
    send_json=[{"op": "subscribe", "args": [f"orderbook.1.{args.symbol}"]}],
    hdlr_json=onmessage
)

🐛 詰まったポイント1：onmessageに引数が2つ必要

初期コードでは、WebSocketハンドラ onmessage(msg) と1引数で定義していたため、以下のようなエラーが発生：

TypeError: onmessage() takes 1 positional argument but 2 were given

→ 対応：引数を (msg, ws) に修正

🐛 詰まったポイント2：Bybitのping/systemメッセージにdataがない

以下のようなKeyErrorが連発：

KeyError: 'data'

→ 対応：メッセージに "data" キーが含まれているかどうかをチェックするガード節を追加

def onmessage(msg, ws):
    if "data" not in msg:
        return  # ping や system message は無視

✅ ステップ3：スプレッドが広いときだけ発注

板情報をもとにスプレッドを計算し、設定値（例：0.0015以上）を超えたら Bidの少し下とAskの少し上に指値を出します。

if spread > args.s_entry:
    print("Spread条件成立 → 発注")
    # BuyとSellを同時に出す

📦 完成した最小構成コード

import asyncio
import pybotters
from argparse import ArgumentParser


async def main(args):
    async with pybotters.Client(apis=args.api_key_json) as client:
        orderbook = {}

        # ✅ 正しいハンドラ定義（引数2つ）
        def onmessage(msg, ws):
            a = msg["data"].get("a", [])
            b = msg["data"].get("b", [])
            if a:
                orderbook["ask"] = float(a[0][0])
            if b:
                orderbook["bid"] = float(b[0][0])

        # ✅ WebSocket接続と購読
        await client.ws_connect(
            "wss://stream.bybit.com/v5/public/linear",
            send_json=[{"op": "subscribe", "args": [f"orderbook.1.{args.symbol}"]}],
            hdlr_json=onmessage
        )

        # ✅ 板情報取得待ち
        while "bid" not in orderbook or "ask" not in orderbook:
            print("⏳ waiting for best bid/ask...")
            await asyncio.sleep(1)

        bid = orderbook["bid"]
        ask = orderbook["ask"]
        spread = (ask - bid) / bid
        print(f"✅ spread = {spread:.5f}, bid = {bid}, ask = {ask}")

        # ✅ スプレッドが条件以上なら注文実行
        if spread > args.s_entry:
            print("🚀 spread condition met, sending orders...")

            # BUY 注文
            buy = await client.post("/v5/order/create", data={
                "category": "linear",
                "symbol": args.symbol,
                "side": "Buy",
                "orderType": "Limit",
                "qty": str(args.lot),
                "price": str(int(bid - 1)),
                "timeInForce": "GTC"
            })

            # SELL 注文
            sell = await client.post("/v5/order/create", data={
                "category": "linear",
                "symbol": args.symbol,
                "side": "Sell",
                "orderType": "Limit",
                "qty": str(args.lot),
                "price": str(int(ask + 1)),
                "timeInForce": "GTC"
            })

            print("📦 Orders placed:")
            print(f"🟢 BUY @ {int(bid - 1)}")
            print(f"🔴 SELL @ {int(ask + 1)}")

        else:
            print("❌ Spread not wide enough. No orders placed.")


if __name__ == "__main__":
    parser = ArgumentParser()
    parser.add_argument("--api_key_json", default="bybit_api.json")
    parser.add_argument("--symbol", default="BTCUSDT")
    parser.add_argument("--lot", default=0.01, type=float)
    parser.add_argument("--s_entry", default=0.0015, type=float)
    args = parser.parse_args()

    asyncio.run(main(args))

🎉 現在の成果

  • ✅ Bybit本番の板情報を取得
  • ✅ スプレッドが条件を満たしたときだけ発注
  • ✅ pybotters v1.8.1 に完全準拠したシンプル構成

🔜 次のステップ候補

  • 🔔 Slack通知：発注・約定イベントを通知
  • ♻️ 指値再注文：キャンセルと再エントリーのループ化
  • 📊 約定検知：WebSocketで約定イベントを監視
  • 📈 PnLログ保存：損益をログとして記録

✍️ まとめ

v1.8.1環境でも、WebSocketとRESTの組み合わせで十分実用的なBotが構築可能であることを再確認できました。
v2系の安定リリースまでは、このv1ベースで実戦投入しながら徐々に強化していく方針です。

ラジオで話したこと

🎙️ 仮想通貨Bot開発記録 Bybit最小構成Botの仕組みを、ちゃんと理解する回【pybotters v1.8.1】

こんにちは、よだかです。
今日は、自分でもう一度コードをかみくだきながら整理する時間として、Bybit向けの最小構成Market Making Botの仕組みについて、なるべく丁寧に解説していこうと思います。

🧠 はじめに

このBotは、Bybitの板情報をリアルタイムで監視して、スプレッド（売値と買値の差）が広がったときだけ、指値注文を出すシンプルな設計です。
使用しているライブラリは pybotters v1.8.1。DataStoreなどの抽象化は使わず、生のWebSocketメッセージを直接処理してます。

🛠 ステップ1：ライブラリの選定と構成の意図

最初はpybotters v2系も検討したんですが、v2はまだ完全なpip対応がされておらず、安定性や互換性に不安がありました。
そこで今回は、「あえて安定版のv1.8.1で固めて、仕組みをしっかり理解することに重きを置こう」という判断です。

🔧 コア部分の構造を1つずつ読み解く

✅ WebSocketで板情報を購読する部分

まず、WebSocketを開いて板情報を取得しているのが以下の部分：

await client.ws_connect(
    "wss://stream.bybit.com/v5/public/linear",
    send_json=[{"op": "subscribe", "args": [f"orderbook.1.{args.symbol}"]}],
    hdlr_json=onmessage
)

✔ ここでやっていること：

  • ws_connect(...) でWebSocket接続を確立
  • send_json に渡している内容は「どのデータを購読するか」をBybit側に伝えるもの
    • orderbook.1.BTCUSDT → 板情報を1秒間隔で購読
  • hdlr_json に渡している関数（ここでは onmessage）が、サーバーから受け取ったデータをどう処理するかを定義する処理です。

補足

「hdlr_json = onmessage」これは“ハンドラー・ジェイソン、オンメッセージ”と読みます。JSON形式で届いたメッセージをどう処理するか、その処理係に onmessage 関数を登録してる、ってことです。

hdlr_json は pybotters 独自のキーワード引数で、
**「WebSocketで受信したJSONメッセージを処理する関数を指定する場所」**です。

onmessage は自分で定義した関数で、
メッセージを受け取ったときに何をするかを記述しているんですね。

✳️ 名前の由来（推測）

  • hdlr = handler（ハンドラー） の略
  • json = JSON形式のメッセージを処理するためのもの

つまり：

hdlr_json=onmessage
→ 「JSONメッセージを処理する関数として、onmessageを使ってね」

という意味になります。

✅ onmessage関数の仕組み

def onmessage(msg, ws):
    if "data" not in msg:
        return
    a = msg["data"].get("a", [])
    b = msg["data"].get("b", [])
    if a:
        orderbook["ask"] = float(a[0][0])
    if b:
        orderbook["bid"] = float(b[0][0])

この関数が、WebSocket経由で飛んできた板情報をorderbook辞書に保存する役目をしています。

✔ それぞれの意味

  • msg はBybitから届いた1件のJSONメッセージ
  • msg["data"] に "a"（ask）と "b"（bid）のリストが入っている
  • a[0][0] は最上位のaskの価格（文字列）
  • b[0][0] は最上位のbidの価格（文字列）
  • それを float() に変換して、orderbook に保存

つまり、これでリアルタイムで「現在の最良買い／売り価格」が記録されていくんです。

⏳ 板情報の初期取得を待つ

while "bid" not in orderbook or "ask" not in orderbook:
    print("⏳ waiting for best bid/ask...")
    await asyncio.sleep(1)

最初にWebSocketを開いても、すぐに価格が来るわけじゃありません。
だからこの部分で、必要なデータが入ってくるまで待つループを回しています。

📏 スプレッドの計算と条件判定

spread = (ask - bid) / bid
if spread > args.s_entry:
    print("🚀 spread condition met, sending orders...")

スプレッドとは：

  • ask（売値）− bid（買値）
  • それを bid で割って割合として計算

つまり、「今どれくらい価格の差が開いているか」を%で判定していて、設定した閾値（args.s_entry）を超えたときにだけ注文を出す、という判断になります。

補足

この '閾値（しきいち）' というのは、発注するかどうかを決めるボーダーラインみたいなもので、スプレッドがこの値を超えたら“いける！”と判断して注文を出す設計になっています。

一言で言えば、

「ある条件を判定するための境界となる数値」
です。

たとえば今回のBotでいうと：

if spread > args.s_entry:

この args.s_entry が スプレッドの閾値（しきいち）。
つまり、「この数値を超えたら発注するよ」という判断の基準点です。

🛒 発注処理

await client.post("/v5/order/create", data={
    "category": "linear",
    "symbol": args.symbol,
    ...
})

BybitのV5 REST APIを叩いて注文を出しています。

  • BUYは Bidより1ドル下
  • SELLは Askより1ドル上

という風に、若干保守的な価格で板に指値を出す設計ですね。

これにより、あまりにもスプレッドが狭い場面では注文を避け、「値幅が取れるかも」という場面だけ狙い撃ちできるようになってます。

🎉 現時点でできていること

  • ✅ Bybitから板情報を取得
  • ✅ スプレッド計算に基づいて発注判断
  • ✅ REST APIによる指値注文の送信
  • ✅ pybotters v1.8.1のみで完結する構成

🔜 今後の展望と学び

ここまで来たら、今後のステップは以下の通り：

  • Slack通知 → いつ発注されたかをすぐに知る
  • 再注文ループ → キャンセルと再発注を自動化
  • 約定検知 → 約定したらSlackに通知／ログ記録
  • PnLログ保存 → 損益の推移を追跡するための基盤整備

そして個人的な気づきとして、今回あらためて感じたのは：

「理解してないまま動くもの」は、いざトラブルが起きたときに、何が壊れたかがわからない。

動かすだけならChatGPTでもできるけど、“育てていくため”には、自分の理解が必須なんですよね。
今回のコード解読も、Cursorで関数の関係を追いながら少しずつ丁寧に進めました。

🎙️ ということで今回は、Bybit向けMM Botの最小構成を、自分の理解もかねて解説する放送でした。

次回はSlack通知や約定検知など、「Botと人間の対話インターフェース」について踏み込んでいきます。
それではまた、よだかでした！

おまけ：ハッカー的にエレガント化してみた

🎯 ハッカー的エレガンスの視点

視点内容
✅ 冗長性の排除同じような処理をまとめて関数化・短縮
✅ 意味を込めた関数名・変数名読みやすさと一貫性
✅ 即時利用の値はインラインで定義bid, ask, spreadの定義を整理
✅ 構造の分離と抽象化place_order() の共通化など
✅ CLIとロジックの完全分離本体 main() と引数パースの切り分け

✨ エレガントリファクタ版（フルコード）

import asyncio
import pybotters
from argparse import ArgumentParser


async def place_limit_order(client, symbol, side, qty, price):
    return await client.post("/v5/order/create", data={
        "category": "linear",
        "symbol": symbol,
        "side": side,
        "orderType": "Limit",
        "qty": str(qty),
        "price": str(price),
        "timeInForce": "GTC"
    })


async def get_orderbook(symbol, api_key_json):
    orderbook = {}

    def onmessage(msg, ws):
        if "data" not in msg:
            return
        data = msg["data"]
        if data.get("a"):
            orderbook["ask"] = float(data["a"][0][0])
        if data.get("b"):
            orderbook["bid"] = float(data["b"][0][0])

    async with pybotters.Client(apis=api_key_json) as client:
        await client.ws_connect(
            "wss://stream.bybit.com/v5/public/linear",
            send_json=[{"op": "subscribe", "args": [f"orderbook.1.{symbol}"]}],
            hdlr_json=onmessage
        )

        while "bid" not in orderbook or "ask" not in orderbook:
            print("⏳ waiting for best bid/ask...")
            await asyncio.sleep(1)

        return orderbook["bid"], orderbook["ask"], client


async def run_bot(symbol, qty, s_entry, api_key_json):
    bid, ask, client = await get_orderbook(symbol, api_key_json)
    spread = (ask - bid) / bid
    print(f"✅ spread = {spread:.5f}, bid = {bid}, ask = {ask}")

    if spread > s_entry:
        print("🚀 spread condition met, sending orders...")

        buy_price, sell_price = int(bid - 1), int(ask + 1)
        await asyncio.gather(
            place_limit_order(client, symbol, "Buy", qty, buy_price),
            place_limit_order(client, symbol, "Sell", qty, sell_price)
        )

        print(f"📦 Orders placed: 🟢 BUY @ {buy_price}, 🔴 SELL @ {sell_price}")
    else:
        print("❌ Spread not wide enough. No orders placed.")


def cli():
    parser = ArgumentParser()
    parser.add_argument("--api_key_json", default="bybit_api.json")
    parser.add_argument("--symbol", default="BTCUSDT")
    parser.add_argument("--lot", default=0.01, type=float)
    parser.add_argument("--s_entry", default=0.0015, type=float)
    return parser.parse_args()


if __name__ == "__main__":
    args = cli()
    asyncio.run(run_bot(args.symbol, args.lot, args.s_entry, args.api_key_json))

💡 主なリファクタポイント

変更箇所改善内容
place_limit_order()同じ注文処理を1関数に抽象化
get_orderbook()WebSocketと板取得を1関数にまとめて、クリーンなインタフェースに
run_bot()注文判断と発注ロジックだけに集中
cli()CLIパーサーをmainロジックから分離して明瞭化
asyncio.gather()BUY/SELLを並列に実行

🧘‍♂️ 一言でまとめると：

このコードは動きの本質を際立たせ、周辺の雑音を極限まで削った構成になっています。
→ 見る人が「構造と意図」を一目で把握できる。これが“ハッカーの美学”の一つです。

初心者向け解説

以下に、変更ポイントを初心者にも分かるようにやさしく解説していきます。
元のコードからの差分が「なぜ必要なのか」「何が良くなったのか」を順を追って説明します。

🧠 1. 共通処理は関数にまとめた

✨変更前：

BUYとSELLの注文をそれぞれ手動で書いていた。

await client.post("/v5/order/create", data={...})  # Buy
await client.post("/v5/order/create", data={...})  # Sell

✅変更後：

1つの place_limit_order() 関数にまとめました。

async def place_limit_order(client, symbol, side, qty, price):
    return await client.post("/v5/order/create", data={...})

💡解説：

「やることが同じなら関数にまとめよう！」が基本の考え方。
こうすることで、間違いも減り、あとで一括で修正しやすくなります。

🧠 2. 板情報の取得も関数に切り出した

✨変更前：

WebSocket接続と板取得が main() にベタ書きされていた。

✅変更後：

get_orderbook() という関数にして、BidとAskを取得して返すように。

async def get_orderbook(symbol, api_key_json):
    ...
    return bid, ask, client

💡解説：

「役割を分けて、関数ごとに責任を明確にする」のがコツ。
この関数は「板データを取るだけ」の役割に集中していて、中の処理を知らなくても使えるようになっています。

🧠 3. Botの本体ロジックを run_bot() にまとめた

✅変更後：

async def run_bot(symbol, qty, s_entry, api_key_json):
    bid, ask, client = await get_orderbook(...)
    spread = (ask - bid) / bid
    ...

💡解説：

main() にあった処理を「1つの意味あるまとまり（＝Botの動作）」としてくくっただけです。
これにより、CLIの処理とBotのロジックが分離されて、読みやすくなっています。

🧠 4. コマンドライン引数の読み取りを cli() に分離

✅変更後：

def cli():
    parser = ArgumentParser()
    ...
    return parser.parse_args()

そして最後に呼び出すときに：

args = cli()
asyncio.run(run_bot(...))

💡解説：

「引数のパース処理」と「Botの実行処理」を別々にすることで、
run_bot() は 引数を渡せばいつでも再利用できる関数になります。

✅ まとめ：このリファクタの本質

改善理由
✅ 処理を関数に分けた役割を明確にし、可読性と再利用性をUP
✅ 冗長なコードを削減間違いを減らし、保守しやすく
✅ main関数の中をスリムに見る人が全体像を一瞬で把握できるように

📌 こんなイメージで捉えよう

Before:
    - 料理の材料、手順、盛り付けが全部1ページにベタ書き

After:
    - 材料リスト関数、調理手順関数、盛り付け関数に分けたレシピ帳
Yodaka

以上、ハッカー的視点でエレガントな構成にする工夫でした！

-Bot