前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。
2025年4月19日
前回までで「板情報取得→スプレッド監視→指値発注→Slack通知→約定監視」まで実装できました。今回はさらに本番環境向けの署名ロジック導入、認証トラブルシュート、証拠金不足対応を行い、運用に必要な前提条件を整えた流れをまとめます。
「約定確認」できるところまで来た。途中で色々あって疲れた。後は「注文キャンセル」「記録」を積む。そしたら、モニタリングとパラメータ最適化フェーズへ。 pic.twitter.com/fb48objaBq
— よだか(夜鷹/yodaka) (@yodakablog) April 19, 2025
🎯 今回の目的
既存の 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
- パラメータ設定
symbol
/order_id
をクエリパラメータとして/v5/order/realtime
に渡すtimeout
(秒)とinterval
(秒)で何回ポーリングするかを制御
- 署名付きヘッダーの生成
- 毎ループごとに
create_signature(config, "", params)
を呼び、timestamp + apiKey + recvWindow + クエリ文字列
を HMAC-SHA256 - これを
X-BAPI-AB-...
ヘッダーにセット
- 毎ループごとに
- 注文状況の取得
session.get(...)
で JSON レスポンスを受け取り、d["result"]["data"][0]["orderStatus"]
を参照"Filled"
であれば約定完了とみなしてTrue
を返却
- タイムアウト処理
- ループが最大回数に達しても
"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 がただ注文を出すだけでなく、実際に通ったかどうかまで自動で確認し、レポートできる」仕組みが完成します。
🔁 追加した処理
- 署名ロジックの修正
- サーバーが返す
origin_string[...]
に合わせ、timestamp + apiKey + recvWindow + (クエリ or ボディ)
を HMAC-SHA256 create_signature()
関数を全面書き換え
- サーバーが返す
- 認証チェック用スクリプト作成 (
check.py
)- GET
/v5/account/wallet-balance
のリクエスト前後でヘッダー・URL・レスポンスを全出力 - 署名エラー→
error sign! origin_string[...]
、権限エラー→Permission denied
の切り分け
- GET
- API 管理画面でのトラブルシュート
- Trade 権限(Spot/USDT Perpetual)を付与
- IP ホワイトリストにパブリック IP を登録
- 証拠金不足対応
- デモ口座の場合は
/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[通常発注サイクル]
- 署名ロジック修正 →
- 認証テスト (check.py) →
- 権限設定・ホワイトリスト登録 →
- Bot 起動 →
- 証拠金不足判定 → 必要ならデモ資金チャージ or ロット調整 →
- 発注サイクルへ
📌 ポイントまとめ
- 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:本番環境向け認証・資金チェックまで完了」をベースに、初心者向けに丁寧解説していきます。私自身の再学習にもなるよう、要所要所でポイントを整理しながら進めますので、ぜひ最後までお付き合いください。
🎙 オープニング & 目次紹介
- 前回のおさらい
- 今日の目的
- 約定確認の仕組み(
wait_for_fill
) - HMAC署名ロジック導入
- 認証トラブルシュート手順
- 証拠金不足(Insufficient Margin)対応
- 処理フロー全体の振り返り
- ポイントまとめ & 次回予告
1. 前回のおさらい
前回はBotに以下の基本機能を実装しました。
- 板情報取得(WebSocketでリアルタイムOrderBook取得)
- スプレッド監視(Ask – Bidの差をチェック)
- 指値発注(Spreadが閾値以下で買い/売り注文を送信)
- 約定監視(注文が通ったかどうかチェック)
- Slack通知(結果をSlackへレポート)
ここまでで、Botが「注文を出して終わり」ではなく、「注文がちゃんと通ったか」まで見てくれる仕組みが完成していました。
2. 今日の目的
今回のミッションは、「本番環境で安心して動かすため」の前提条件整備です。具体的には:
- Bybit v5 本番API向け HMAC 署名ロジックを組み込む
- 認証エラー(署名ミス/権限不足)を潰す
- 証拠金不足エラー(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
:監視対象のシンボルと注文IDtimeout
:最大待機時間(秒単位)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」を用いた検証で、サーバー側に「このリクエストは正当なものか?」をチェックさせる仕組みです。
署名に含める要素
timestamp
(ミリ秒)apiKey
recvWindow
(有効時間、通常は 5000ms など)- クエリ文字列 または リクエストボディ
これらを連結した文字列(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管理画面での設定
- Trade 権限の付与
- Spot/USDT Perpetual の両方にチェック
- IPホワイトリスト登録
- 実行マシンのパブリックIPを登録
6. 証拠金不足(Insufficient Margin)対応
発注時に必要な証拠金が足りないと、retCode=110007
のエラーが返ってきます。
CheckMarginRatio fail! InsufficientAB (retCode=110007)
対処法
- デモ口座の場合
→/v5/account/demo-apply-money
で資金チャージ - 本番口座の場合
- ロットサイズ(
--lot
)を小さくする - レバレッジを下げる
- 実際の資金投入
- ロットサイズ(
再学習ポイント
- 証拠金の考え方
→ 取引数量 × 価格 × 必要証拠金率 - リスク管理
→ ロットサイズやレバレッジ調整は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[通常発注サイクル]
- 署名ロジック作成
- 認証テスト(check.py)
- Trade権限&IPホワイトリスト設定
- Bot起動
- 証拠金不足判定 → 必要なら資金チャージ or ロット調整
- 発注サイクル開始
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 秒待ってから再度問い合わせ
check_order_status
を呼んでサーバーの注文状況を取得- 結果が期待通り(Filled)でなければ待機
- 待機後にまた問い合わせ
これを「ポーリング」と呼びます。
プッシュ方式との違い
- ポーリング:自分から定期的に問い合わせる
- プッシュ(WebSocket など):サーバーが状態変化を検知したら自動で通知してくる
WebSocket で約定通知を直接受け取る設計も可能ですが、API側でリアルタイム通知機能が提供されていない場合や、実装が複雑になる場合にはポーリングがシンプルで汎用的です。
ポーリングのメリット・デメリット
メリット | デメリット |
---|---|
実装がシンプル | 無駄なリクエストが増え、サーバー負荷・APIコール数が増える |
APIが通知機能を持っていなくても状態確認できる | インターバル次第で「検知遅延」が生じる |
エラーやタイムアウト処理を組み込みやすい | 短い間隔で回すほどリソースを消費 |
本Botでの使いどころ
- Bybit の本番 REST API には「約定Realtime」エンドポイントはあるものの、通知サブスクリプション機能が限られるため、ポーリングで注文状況を定期チェック
interval=1秒
、timeout=30秒
のようにパラメータを調整して、リソース消費と検知遅延のバランスを取っている
再学習ポイント
- なぜポーリングを選んだか? → 実装のシンプルさ&API制約
- リクエスト間隔(interval)は適切か? → 短すぎるとコール数増、長すぎると検知が遅れる
- 他の手法との比較 → プッシュ通知や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, ) )