Bot mmbot 開発ログ

🛠️開発記録#185(2025/4/19)「Bybit自動取引Bot：本番環境向け認証・資金チェックまで完了【Python×pybotters】」

2025年4月19日

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

2025年4月19日

前回までで「板情報取得→スプレッド監視→指値発注→Slack通知→約定監視」まで実装できました。今回はさらに本番環境向けの署名ロジック導入、認証トラブルシュート、証拠金不足対応を行い、運用に必要な前提条件を整えた流れをまとめます。

🎯 今回の目的

既存の Bot（WS 板監視＋スプレッド監視＋指値発注＋約定検知＋Slack通知）に対し、

  • Bybit v5 本番 API の正しい HMAC 署名ロジックを組み込む
  • 認証エラー（署名ミス／権限不足）の原因究明
  • 証拠金不足エラーの対処（デモ資金追加 or ロット調整）

を行い、「本番環境で動かす前提」を整えること。

🔎 約定確認のコード解説

以下の wait_for_fill 関数とその呼び出し部分で、「出した注文が実際に通った（Filled になった）か」をポーリングでチェックしています。

async def wait_for_fill(
    session: aiohttp.ClientSession,
    config: dict,
    symbol: str,
    order_id: str,
    timeout: int = 30,
    interval: float = 1.0
) -> bool:
    """
    注文の約定を一定時間ポーリングで待つ。
    Filled になれば True、タイムアウトしたら False を返す。
    """
    path = "/v5/order/realtime"

    for i in range(int(timeout / interval)):
        params = {"symbol": symbol, "orderId": order_id}
        # 署名ヘッダーを生成
        headers = create_signature(config, "", params)
        url = config["rest_base"] + path

        # REST GET で注文状況を取得
        async with session.get(url, headers=headers, params=params) as res:
            d = await res.json()
            status = d["result"]["data"][0]["orderStatus"]
            print(f"🔍 [{order_id}] status check {i+1}: {status}")

            if status == "Filled":
                # Filled になったら即座に True を返して抜ける
                return True

        # まだ約定していなければ interval 秒待って再試行
        await asyncio.sleep(interval)

    # 一定時間待っても約定しなかった場合
    print(f"⚠️ [{order_id}] did not fill within {timeout}s")
    return False
  1. パラメータ設定
    • symbol／order_id をクエリパラメータとして /v5/order/realtime に渡す
    • timeout（秒）と interval（秒）で何回ポーリングするかを制御
  2. 署名付きヘッダーの生成
    • 毎ループごとに create_signature(config, "", params) を呼び、
      timestamp + apiKey + recvWindow + クエリ文字列 を HMAC-SHA256
    • これを X-BAPI-AB-... ヘッダーにセット
  3. 注文状況の取得
    • session.get(...) で JSON レスポンスを受け取り、
      d["result"]["data"][0]["orderStatus"] を参照
    • "Filled" であれば約定完了とみなして True を返却
  4. タイムアウト処理
    • ループが最大回数に達しても "Filled" にならなければ
      ログに警告を出して False を返す

run_loop 内での呼び出し

# 発注直後に約定監視をスタート
filled_buy  = await wait_for_fill(http_session, config, symbol, buy_id)
filled_sell = await wait_for_fill(http_session, config, symbol, sell_id)
  • 買い (buy_id)／売り (sell_id) をそれぞれ監視
  • 両方の結果 (True/False) を組み合わせて Slack に最終報告

これで「Bot がただ注文を出すだけでなく、実際に通ったかどうかまで自動で確認し、レポートできる」仕組みが完成します。

🔁 追加した処理

  1. 署名ロジックの修正
    • サーバーが返す origin_string[...] に合わせ、timestamp + apiKey + recvWindow + (クエリ or ボディ) を HMAC-SHA256
    • create_signature() 関数を全面書き換え
  2. 認証チェック用スクリプト作成 (check.py)
    • GET /v5/account/wallet-balance のリクエスト前後でヘッダー・URL・レスポンスを全出力
    • 署名エラー→error sign! origin_string[...]、権限エラー→Permission denied の切り分け
  3. API 管理画面でのトラブルシュート
    • Trade 権限（Spot／USDT Perpetual）を付与
    • IP ホワイトリストにパブリック IP を登録
  4. 証拠金不足対応
    • デモ口座の場合は /v5/account/demo-apply-money で資金チャージ
    • 本番／デモともに --lot を小さくする、あるいはレバレッジ設定

🛠️ 処理の流れ

flowchart LR
  A[署名ロジック作成] --> B[check.py で認証テスト]
  B --> C{署名 OK?}
  C -- No --> A
  C -- Yes --> D[権限チェック＆IP登録]
  D --> E{権限 OK?}
  E -- No --> D
  E -- Yes --> F[Botを起動]
  F --> G[証拠金不足?]
  G -- Yes --> H[デモ資金追加／lot調整]
  G -- No --> I[通常発注サイクル]
  1. 署名ロジック修正 →
  2. 認証テスト (check.py) →
  3. 権限設定・ホワイトリスト登録 →
  4. Bot 起動 →
  5. 証拠金不足判定 → 必要ならデモ資金チャージ or ロット調整 →
  6. 発注サイクルへ

📌 ポイントまとめ

  • Bybit v5 本番は「timestamp+apiKey+recvWindow+payload」が署名対象
  • API 管理画面で Trade 権限（Derivatives→USDT Perpetual）を忘れずに設定
  • IP ホワイトリストは 実行マシンのパブリックIP
  • 証拠金不足は retCode=110007 で判別 → デモ口座なら /demo-apply-money でチャージ

🔍 実行結果ログ抜粋

# 署名ミス時
error sign! origin_string[1745055965705pzJwJou…5000coin=USDT]

# 権限不足時
Permission denied, please check your API key permissions. (retCode=10005)

# 証拠金不足時
CheckMarginRatio fail! InsufficientAB (retCode=110007)

✅ 状況分析

  • 署名ミス → create_signature がサーバー仕様に未対応
  • Permission denied → Trade 権限 or IP 設定漏れ
  • InsufficientAB → 資金不足 or ロット過大

この三つを順に解消することで、ようやく「🆔 Buy id: …」の正常な発注ログが得られるようになりました。

🧠 なぜ証拠金不足エラーが出たのか？

  • デフォルト --lot 0.01 では必要証拠金を満たせず
  • デモ口座残高も不足 → /v5/account/demo-apply-money でチャージする必要あり
  • 本番ならリアル残高投入、あるいはロット・レバレッジ調整

✅ ここまでで実現できたこと

機能状態
WS 板監視✅
スプレッド判定✅
指値発注✅
約定ポーリング監視✅
Slack 通知✅
本番署名ロジック導入✅
認証・権限トラブルシュート✅
証拠金不足対応✅

🔜 次に取り組む予定の強化ポイント

フェーズ内容
✅ リスク管理約定しない注文の自動キャンセル（/v5/order/cancel）
✅ 記録loguru などでファイルログ出力＆PnL 集計
🆕 モニタリングPrometheus/Grafana 連携
🆕 パラメータ最適化スプレッド閾値／ロットサイズの動的調整

✍️ まとめ


本番環境向けの署名・認証周りと証拠金チェックをクリアし、Bot が「市場監視→発注→約定確認→Slack報告」を安定して回せる状態になりました。
次回は「自動キャンセル」「詳細ログ」「運用監視ツール連携」を加え、さらに実運用に耐えうる Bot を目指します。

👇ラジオで話したこと

こんにちは、よだかです。『仮想通貨Bot開発ラジオ』へようこそ。今日は開発記録#185「Bybit自動取引Bot：本番環境向け認証・資金チェックまで完了」をベースに、初心者向けに丁寧解説していきます。私自身の再学習にもなるよう、要所要所でポイントを整理しながら進めますので、ぜひ最後までお付き合いください。

🎙 オープニング & 目次紹介

  1. 前回のおさらい
  2. 今日の目的
  3. 約定確認の仕組み（wait_for_fill）
  4. HMAC署名ロジック導入
  5. 認証トラブルシュート手順
  6. 証拠金不足（Insufficient Margin）対応
  7. 処理フロー全体の振り返り
  8. ポイントまとめ & 次回予告

1. 前回のおさらい

前回はBotに以下の基本機能を実装しました。

  • 板情報取得（WebSocketでリアルタイムOrderBook取得）
  • スプレッド監視（Ask – Bidの差をチェック）
  • 指値発注（Spreadが閾値以下で買い/売り注文を送信）
  • 約定監視（注文が通ったかどうかチェック）
  • Slack通知（結果をSlackへレポート）

ここまでで、Botが「注文を出して終わり」ではなく、「注文がちゃんと通ったか」まで見てくれる仕組みが完成していました。

2. 今日の目的

今回のミッションは、「本番環境で安心して動かすため」の前提条件整備です。具体的には：

  1. Bybit v5 本番API向け HMAC 署名ロジックを組み込む
  2. 認証エラー（署名ミス／権限不足）を潰す
  3. 証拠金不足エラー（Insufficient Margin）への対応策を準備

これらをクリアして、はじめて「本番環境で動かす準備が整った」と言えます。

3. 約定確認の仕組み解説：wait_for_fill

まず、Botが「出した注文が通っているか」をどうチェックしているかを見ていきましょう。以下の関数を使ったポーリング方法です。

async def wait_for_fill(
    session: aiohttp.ClientSession,
    config: dict,
    symbol: str,
    order_id: str,
    timeout: int = 30,
    interval: float = 1.0
) -> bool:
    """
    注文の約定を一定時間ポーリングで待つ。
    Filled になれば True、タイムアウトしたら False。
    """
    path = "/v5/order/realtime"
    for i in range(int(timeout / interval)):
        params = {"symbol": symbol, "orderId": order_id}
        headers = create_signature(config, "", params)
        url = config["rest_base"] + path
        async with session.get(url, headers=headers, params=params) as res:
            d = await res.json()
            status = d["result"]["data"][0]["orderStatus"]
            print(f"🔍 [{order_id}] status check {i+1}: {status}")
            if status == "Filled":
                return True
        await asyncio.sleep(interval)
    print(f"⚠️ [{order_id}] did not fill within {timeout}s")
    return False

ポイント解説

  • パラメータ設定
    • symbol／order_id：監視対象のシンボルと注文ID
    • timeout：最大待機時間（秒単位）
    • interval：ポーリング間隔（秒単位）
  • 署名付きヘッダー生成
    • 毎回 create_signature(config, "", params) を呼び、HMAC-SHA256 署名ヘッダーを作成
  • REST GET リクエスト
    • /v5/order/realtime エンドポイントへクエリパラメータ付きでリクエスト
    • レスポンス JSON から orderStatus を取得
  • Filled 判定
    • orderStatus == "Filled" になったら即時 True を返してループ脱出
  • タイムアウト処理
    • 最大回数リトライ後も未約定なら警告をログに出して False を返却

実際の発注サイクルでは、

filled_buy  = await wait_for_fill(http_session, config, symbol, buy_id)
filled_sell = await wait_for_fill(http_session, config, symbol, sell_id)

と呼び出し、買いと売りの両方を監視。その結果をまとめてSlackにレポートします。

4. HMAC署名ロジック導入

Bybit本番環境のAPIは、リクエストごとに署名付きヘッダーを要求します。これは「HMAC-SHA256」を用いた検証で、サーバー側に「このリクエストは正当なものか？」をチェックさせる仕組みです。

署名に含める要素

  1. timestamp（ミリ秒）
  2. apiKey
  3. recvWindow（有効時間、通常は 5000ms など）
  4. クエリ文字列 または リクエストボディ

これらを連結した文字列（origin_string）を HMAC-SHA256 でハッシュ化し、生成された署名を X-BAPI-SIGN などのヘッダーにセットします。

def create_signature(config, body: str, params: dict) -> dict:
    # 1. timestamp + apiKey + recvWindow + (query or body)
    origin = f"{timestamp}{config['apiKey']}{config['recvWindow']}{query_string_or_body}"
    # 2. HMAC-SHA256
    signature = hmac_sha256(config['apiSecret'], origin)
    # 3. ヘッダーにセット
    return {
        "X-BAPI-API-KEY": config["apiKey"],
        "X-BAPI-SIGN": signature,
        "X-BAPI-TIMESTAMP": timestamp,
        "X-BAPI-RECV-WINDOW": config["recvWindow"],
    }

再学習ポイント

  • なぜ署名が必要？
    → 不正な注文送信やなりすましを防ぐ。
  • HMAC-SHA256 の流れ
    → 「秘密鍵 + メッセージ」をハッシュ化し、同じ秘密鍵を持つサーバーだけが検証可能。

5. 認証トラブルシュート手順

署名ロジックを組み込んでも、以下のようなエラーが出ることがあります。

  • 署名ミス 
error sign! origin_string[...]
  • 権限不足 
Permission denied, please check your API key permissions. (retCode=10005)

手順①：認証チェック用スクリプト check.py を作成

  • /v5/account/wallet-balance でヘッダー・URL・レスポンスを全出力
  • 署名ミスか権限不足かを切り分け
# 標準出力にヘッダー・origin_string・レスポンスを全部見る
python check.py

手順②：API管理画面での設定

  1. Trade 権限の付与
    • Spot／USDT Perpetual の両方にチェック
  2. IPホワイトリスト登録
    • 実行マシンのパブリックIPを登録

6. 証拠金不足（Insufficient Margin）対応

発注時に必要な証拠金が足りないと、retCode=110007 のエラーが返ってきます。

CheckMarginRatio fail! InsufficientAB (retCode=110007)

対処法

  • デモ口座の場合
    → /v5/account/demo-apply-money で資金チャージ
  • 本番口座の場合
    1. ロットサイズ（--lot）を小さくする
    2. レバレッジを下げる
    3. 実際の資金投入

再学習ポイント

  • 証拠金の考え方
    → 取引数量 × 価格 × 必要証拠金率
  • リスク管理
    → ロットサイズやレバレッジ調整はBotの根幹

7. 処理フロー全体の振り返り

最後に、今回整備した処理フローを改めて確認しましょう。

flowchart LR
  A[署名ロジック作成] --> B[check.py で認証テスト]
  B --> C{署名 OK?}
  C -- No --> A
  C -- Yes --> D[権限チェック＆IP登録]
  D --> E{権限 OK?}
  E -- No --> D
  E -- Yes --> F[Botを起動]
  F --> G[証拠金不足?]
  G -- Yes --> H[デモ資金追加／lot調整]
  G -- No --> I[通常発注サイクル]
  1. 署名ロジック作成
  2. 認証テスト（check.py）
  3. Trade権限＆IPホワイトリスト設定
  4. Bot起動
  5. 証拠金不足判定 → 必要なら資金チャージ or ロット調整
  6. 発注サイクル開始

8. 📌 ポイントまとめ & 次回予告

  • HMAC署名 は「timestamp + apiKey + recvWindow + payload」を正しく組み立てること
  • API権限設定 を忘れずに（Spot／Perp 両方）
  • IPホワイトリスト は実行環境のパブリックIP
  • Insufficient Margin は retCode=110007 で判別 → デモ資金 or ロット調整

✅ これで「市場監視→発注→約定確認→Slack報告」を本番APIで安定して回せる土台が整いました。

次回予告

次回は、さらに 自動キャンセル 機能（/v5/order/cancel）の実装、詳細ログ出力（loguru 連携＆PnL集計）、そして Prometheus/Grafana による運用監視ツール連携 に挑戦します。お楽しみに！

最後まで聴いてくださってありがとうございました。引き続き一緒にBot開発を学んでいきましょう。ではまた次回！

おまけ：「ポーリング」（polling）って何？

「ポーリング」（polling）とは、プログラムが一定間隔で繰り返し「ある状態になっているか」を自分でチェックしに行く方式のことです。サーバーからの通知を待つのではなく、自分で定期的に問い合わせを投げて確認します。

具体例でイメージすると

while True:
    status = check_order_status(order_id)   # ← サーバーに「どうなった？」と問い合わせ
    if status == "Filled":
        break                               # 約定していればループを抜ける
    sleep(interval)                         # まだなら interval 秒待ってから再度問い合わせ
  1. check_order_status を呼んでサーバーの注文状況を取得
  2. 結果が期待通り（Filled）でなければ待機
  3. 待機後にまた問い合わせ

これを「ポーリング」と呼びます。

プッシュ方式との違い

  • ポーリング：自分から定期的に問い合わせる
  • プッシュ（WebSocket など）：サーバーが状態変化を検知したら自動で通知してくる

WebSocket で約定通知を直接受け取る設計も可能ですが、API側でリアルタイム通知機能が提供されていない場合や、実装が複雑になる場合にはポーリングがシンプルで汎用的です。

ポーリングのメリット・デメリット

メリットデメリット
実装がシンプル無駄なリクエストが増え、サーバー負荷・APIコール数が増える
APIが通知機能を持っていなくても状態確認できるインターバル次第で「検知遅延」が生じる
エラーやタイムアウト処理を組み込みやすい短い間隔で回すほどリソースを消費

本Botでの使いどころ

  • Bybit の本番 REST API には「約定Realtime」エンドポイントはあるものの、通知サブスクリプション機能が限られるため、ポーリングで注文状況を定期チェック
  • interval=1秒、timeout=30秒 のようにパラメータを調整して、リソース消費と検知遅延のバランスを取っている

再学習ポイント

  1. なぜポーリングを選んだか？ → 実装のシンプルさ＆API制約
  2. リクエスト間隔（interval）は適切か？ → 短すぎるとコール数増、長すぎると検知が遅れる
  3. 他の手法との比較 → プッシュ通知やWebhookとの組み合わせ検討も今後の強化ポイント

もし「サーバーからのリアルタイム通知が欲しい」「APIコールを減らしたい」と感じたら、WebSocket や Webhook を使った設計も視野に入れてみてください。

現在のコード：自分用ログ

import asyncio
import json
import time
import hmac
import hashlib
import aiohttp
import pybotters
from argparse import ArgumentParser
from urllib.parse import urlencode

# --- Signature for Bybit v5 (align with server's origin_string logic) ---
def create_signature(config: dict, body: str = "", params: dict | None = None) -> dict:
    """
    Bybit v5 signature: HMAC_SHA256 of timestamp + apiKey + recvWindow + (queryString or body)
    Returns headers dict to include in request.
    """
    timestamp = str(int(time.time() * 1000))
    api_key = config["bybit"]["apiKey"]
    secret = config["bybit"]["secret"]
    recv_window = config.get("recv_window", "5000")

    if params:
        payload = urlencode(params)
    else:
        payload = body or ""

    # origin_string: timestamp + apiKey + recvWindow + payload
    to_sign = timestamp + api_key + recv_window + payload
    signature = hmac.new(
        secret.encode(),
        to_sign.encode(),
        hashlib.sha256
    ).hexdigest()

    return {
        "Content-Type": "application/json",
        "X-BAPI-API-KEY":    api_key,
        "X-BAPI-TIMESTAMP":  timestamp,
        "X-BAPI-SIGN":       signature,
        "X-BAPI-RECV-WINDOW": recv_window,
    }

async def place_limit_order(session: aiohttp.ClientSession, config: dict, symbol: str, side: str, qty: float, price: int) -> str:
    """Place a limit order and return orderId."""
    path = "/v5/order/create"
    body_dict = {
        "category": "linear",
        "symbol": symbol,
        "side": side,
        "orderType": "Limit",
        "qty": str(qty),
        "price": str(price),
        "timeInForce": "GTC"
    }
    body_str = json.dumps(body_dict)
    headers = create_signature(config, body_str, None)
    url = config["rest_base"] + path
    async with session.post(url, headers=headers, data=body_str) as res:
        data = await res.json()
        if data.get("retCode") != 0:
            raise RuntimeError(f"Order failed: {data}")
        return data["result"]["orderId"]

async def wait_for_fill(session: aiohttp.ClientSession, config: dict, symbol: str, order_id: str, timeout: int = 30, interval: float = 1.0) -> bool:
    """Poll order status until Filled or timeout."""
    path = "/v5/order/realtime"
    for i in range(int(timeout / interval)):
        params = {"symbol": symbol, "orderId": order_id}
        headers = create_signature(config, "", params)
        url = config["rest_base"] + path
        async with session.get(url, headers=headers, params=params) as res:
            d = await res.json()
            status = d["result"]["data"][0]["orderStatus"]
            print(f"🔍 [{order_id}] status check {i+1}: {status}")
            if status == "Filled":
                return True
        await asyncio.sleep(interval)
    print(f"⚠️ [{order_id}] did not fill within {timeout}s")
    return False

async def notify_slack(webhook_url: str, message: str):
    """Send message to Slack via Incoming Webhook."""
    async with aiohttp.ClientSession() as session:
        await session.post(webhook_url, json={"text": message})

async def run_loop(symbol: str, qty: float, s_entry: float, config: dict, interval: int = 10):
    # Setup production endpoints
    orderbook: dict[str, float] = {}
    config["rest_base"] = "https://api.bybit.com"

    # Open WS client and HTTP session in nested async with
    async with pybotters.Client(base_url="wss://stream.bybit.com/v5/public/linear") as ws_client:
        async with aiohttp.ClientSession() as http_session:
            # Subscribe to public orderbook
            await ws_client.ws_connect(
                "wss://stream.bybit.com/v5/public/linear",
                send_json=[{"op": "subscribe", "args": [f"orderbook.1.{symbol}"]}],
                hdlr_json=lambda msg, ws: (
                    orderbook.update({"ask": float(msg["data"]["a"][0][0])}) if "data" in msg and msg["data"].get("a") else None,
                    orderbook.update({"bid": float(msg["data"]["b"][0][0])}) if "data" in msg and msg["data"].get("b") else None,
                )
            )
            # Wait for initial book
            while "bid" not in orderbook or "ask" not in orderbook:
                await asyncio.sleep(1)

            # Main trading loop
            while True:
                bid, ask = orderbook["bid"], orderbook["ask"]
                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)
                    try:
                        buy_id = await place_limit_order(http_session, config, symbol, "Buy", qty, buy_price)
                        sell_id = await place_limit_order(http_session, config, symbol, "Sell", qty, sell_price)
                        print(f"🆔 Buy id: {buy_id}, Sell id: {sell_id}")
                    except Exception as e:
                        err = f"❌ Order error: {e}"
                        print(err)
                        await notify_slack(config["slack_webhook_url"], err)
                        await asyncio.sleep(interval)
                        continue

                    # Execution monitoring
                    filled_buy = await wait_for_fill(http_session, config, symbol, buy_id)
                    filled_sell = await wait_for_fill(http_session, config, symbol, sell_id)

                    # Slack notification
                    msg = (
                        f"📈 *MMBot Execution Report*\n"
                        f"> Pair: `{symbol}`\n"
                        f"> Spread: `{spread:.5f}`\n"
                        f"> 🟢 Buy @{buy_price} ➡️ **{'Filled' if filled_buy else 'Pending'}**\n"
                        f"> 🔴 Sell @{sell_price} ➡️ **{'Filled' if filled_sell else 'Pending'}**\n"
                        f"> Lot: `{qty}`"
                    )
                    await notify_slack(config["slack_webhook_url"], msg)
                    print("✅ Cycle complete, sleeping...\n")
                    await asyncio.sleep(interval)
                else:
                    print("❌ Spread not wide enough. Waiting...\n")
                    await asyncio.sleep(1)


def cli():
    p = ArgumentParser()
    p.add_argument("--config", default="config.json")
    p.add_argument("--symbol", default="BTCUSDT")
    p.add_argument("--lot", default=0.01, type=float)
    p.add_argument("--s_entry", default=0.000001, type=float)
    p.add_argument("--interval", default=10, type=int)
    return p.parse_args()

if __name__ == "__main__":
    args = cli()
    with open(args.config) as f:
        cfg = json.load(f)
    asyncio.run(
        run_loop(
            args.symbol,
            args.lot,
            args.s_entry,
            cfg,
            args.interval,
        )
    )

-Bot, mmbot, 開発ログ