Bybit API × Python 自動売買の始め方
pybit入門・APIキー発行から
テストネット接続まで完全ガイド

mkdir bybit-bot
cd bybit-bot

# 仮想環境の作成
python -m venv .venv

# 有効化（Windows PowerShell）
.venv\Scripts\Activate.ps1
# 有効化（macOS / Linux）
source .venv/bin/activate

python --version        # 3.10 以上であることを確認
pip install -U pip
pip install -r requirements.txt

# Bybit公式Python SDK（HTTP + WebSocketの両方を含む）
pybit>=5.17.0

# データ処理
pandas>=2.2
pyarrow>=17.0          # Parquet保存用（記事07で使用）

# 設定・通信
python-dotenv>=1.0     # .envからAPIキーを読む
requests>=2.32         # 署名を自作する場合（サンプル10）

# 開発補助
rich>=13.7             # CLIでの表示を見やすくする

# 本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
from pybit.unified_trading import HTTP

session = HTTP(testnet=True)          # Market系エンドポイントはAPIキー不要
res = session.get_tickers(category="linear", symbol="BTCUSDT")

t = res["result"]["list"][0]
print(f"シンボル    : {t['symbol']}")
print(f"最終価格    : {t['lastPrice']}")
print(f"24h変化率   : {float(t['price24hPcnt']) * 100:+.2f} %")
print(f"資金調達率  : {float(t['fundingRate']) * 100:+.4f} %")
print(f"未決済建玉  : {t['openInterest']}")

シンボル    : BTCUSDT
最終価格    : 118342.50
24h変化率   : +1.83 %
資金調達率  : +0.0100 %
未決済建玉  : 58234.117

# demo trading 口座に仮想資金を投入する（api-demo.bybit.com 専用）
from pybit.unified_trading import HTTP

session = HTTP(demo=True, api_key="DEMO_KEY", api_secret="DEMO_SECRET")

res = session._submit_request(
    method="POST",
    path=f"{session.endpoint}/v5/account/demo-apply-money",
    query={"adjustType": 0, "utaDemoApplyMoney": [{"coin": "USDT", "amountStr": "100000"}]},
    auth=True,
)
print(res["retCode"], res["retMsg"])

# 既定の接続先: testnet / demo / mainnet
BYBIT_ENV=testnet

# testnet.bybit.com で発行したキー
BYBIT_TESTNET_API_KEY=xxxxxxxxxxxxxxxxxxxx
BYBIT_TESTNET_API_SECRET=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

# デモトレード画面で発行したキー
BYBIT_DEMO_API_KEY=
BYBIT_DEMO_API_SECRET=

# 本番キー（出金権限なし・IP制限あり）
BYBIT_MAINNET_API_KEY=
BYBIT_MAINNET_API_SECRET=

"""config.py — 環境ごとのAPIキーを型安全に読み込む
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import os
from dataclasses import dataclass
from enum import Enum
from pathlib import Path

from dotenv import load_dotenv

load_dotenv(Path(__file__).with_name(".env"))


class Env(str, Enum):
    TESTNET = "testnet"
    DEMO = "demo"
    MAINNET = "mainnet"


_PREFIX = {
    Env.TESTNET: "BYBIT_TESTNET",
    Env.DEMO: "BYBIT_DEMO",
    Env.MAINNET: "BYBIT_MAINNET",
}

# pybit に渡すフラグと、自作クライアント用のベースURL
REST_BASE = {
    Env.TESTNET: "https://api-testnet.bybit.com",
    Env.DEMO: "https://api-demo.bybit.com",
    Env.MAINNET: "https://api.bybit.com",
}


@dataclass(frozen=True)
class Credentials:
    env: Env
    api_key: str
    api_secret: str

    @property
    def testnet(self) -> bool:
        return self.env is Env.TESTNET

    @property
    def demo(self) -> bool:
        return self.env is Env.DEMO

    @property
    def base_url(self) -> str:
        return REST_BASE[self.env]

    def masked(self) -> str:
        """ログ出力用。キーの先頭4文字だけ見せる。"""
        return f"{self.env.value}:{self.api_key[:4]}****"


def load_credentials(env: str | None = None) -> Credentials:
    resolved = Env(env or os.getenv("BYBIT_ENV", "testnet"))
    prefix = _PREFIX[resolved]

    key = os.getenv(f"{prefix}_API_KEY", "").strip()
    secret = os.getenv(f"{prefix}_API_SECRET", "").strip()
    if not key or not secret:
        raise RuntimeError(
            f"{prefix}_API_KEY / {prefix}_API_SECRET が .env に設定されていません "
            f"(BYBIT_ENV={resolved.value})"
        )
    return Credentials(env=resolved, api_key=key, api_secret=secret)

"""client.py — pybitセッションのファクトリと本番誤爆ガード
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import os
from dataclasses import dataclass

from pybit.unified_trading import HTTP

from config import Credentials, Env, load_credentials


@dataclass(frozen=True)
class BybitClient:
    """環境情報を保持したまま持ち回せるクライアント。"""

    env: Env
    http: HTTP
    cred: Credentials

    @property
    def is_live(self) -> bool:
        return self.env is Env.MAINNET


def make_client(env: str | None = None, *, allow_mainnet: bool = False) -> BybitClient:
    cred = load_credentials(env)

    # --- 本番誤爆ガード -------------------------------------------------
    # 引数で明示許可し、かつ環境変数 BYBIT_ALLOW_LIVE=1 が立っていないと
    # mainnetには接続させない。二重の鍵をかけておく。
    if cred.env is Env.MAINNET:
        if not allow_mainnet:
            raise PermissionError(
                "mainnet接続には make_client(..., allow_mainnet=True) が必要です"
            )
        if os.getenv("BYBIT_ALLOW_LIVE") != "1":
            raise PermissionError(
                "mainnet接続には環境変数 BYBIT_ALLOW_LIVE=1 が必要です"
            )
    # -------------------------------------------------------------------

    http = HTTP(
        testnet=cred.testnet,
        demo=cred.demo,
        api_key=cred.api_key,
        api_secret=cred.api_secret,
        recv_window=5000,   # 署名の有効時間(ms)。時計がズレると10002が出る
        timeout=10,         # 秒
        max_retries=3,      # pybit内蔵のリトライ回数
        retry_delay=3,      # 秒
        log_requests=False, # デバッグ時のみTrue（キーはログに出ない）
    )
    return BybitClient(env=cred.env, http=http, cred=cred)


if __name__ == "__main__":
    c = make_client()
    print(f"接続先: {c.cred.masked()}  endpoint={c.http.endpoint}")

"""check_connection.py — 疎通確認・時刻ズレ検出・権限確認
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time

from pybit.unified_trading import HTTP

from client import make_client

RECV_WINDOW_MS = 5000


def check_clock_skew(http: HTTP, samples: int = 5) -> float:
    """サーバー時刻とローカル時刻の差(ms)を返す。正なら自分の時計が遅れている。"""
    skews, rtts = [], []
    for _ in range(samples):
        t0 = time.time()
        res = http.get_server_time()
        t1 = time.time()

        # timeNano(ns) のほうが精度が高い
        server_ms = int(res["result"]["timeNano"]) / 1_000_000
        local_ms = (t0 + t1) / 2 * 1000     # 往復の中点をローカル時刻とみなす
        skews.append(server_ms - local_ms)
        rtts.append((t1 - t0) * 1000)
        time.sleep(0.2)

    skew = sorted(skews)[len(skews) // 2]   # 中央値でRTTのばらつきを吸収
    rtt = sorted(rtts)[len(rtts) // 2]

    print(f"  RTT(中央値)   : {rtt:8.1f} ms")
    print(f"  時刻ズレ      : {skew:+8.1f} ms")

    if abs(skew) > RECV_WINDOW_MS * 0.5:
        print("  [警告] ズレが大きすぎます。retCode=10002 の原因になります。")
        print("         Windows : w32tm /resync")
        print("         Linux   : sudo chronyc makestep")
    return skew


def check_permissions(http: HTTP) -> None:
    """キーの権限・有効期限・IP制限を確認する。"""
    res = http.get_api_key_information()
    info = res["result"]
    print(f"  キー種別      : {'RSA' if info.get('type') == 2 else 'HMAC'}")
    print(f"  有効期限      : {info.get('expiredAt') or '(無期限)'}")
    print(f"  IP制限        : {info.get('ips') or '(なし)'}")
    print(f"  読み取り専用  : {'はい' if info.get('readOnly') == 1 else 'いいえ'}")
    perms = {k: v for k, v in (info.get("permissions") or {}).items() if v}
    print(f"  権限          : {perms}")


def main() -> None:
    c = make_client()
    print(f"[1] 接続先 {c.cred.masked()}  {c.http.endpoint}")

    print("[2] 時刻同期チェック")
    check_clock_skew(c.http)

    print("[3] APIキー権限チェック")
    check_permissions(c.http)

    print("[4] 残高取得（認証テスト）")
    res = c.http.get_wallet_balance(accountType="UNIFIED")
    print(f"  retCode={res['retCode']} retMsg={res['retMsg']}")
    print("  OK: 認証付きリクエストが通りました")


if __name__ == "__main__":
    main()

[1] 接続先 testnet:AbCd****  https://api-testnet.bybit.com
[2] 時刻同期チェック
  RTT(中央値)   :    182.4 ms
  時刻ズレ      :    -12.7 ms
[3] APIキー権限チェック
  キー種別      : HMAC
  有効期限      : 1786579200000
  IP制限        : ['203.0.113.10']
  読み取り専用  : いいえ
  権限          : {'Wallet': ['AccountTransfer'], 'ContractTrade': ['Order', 'Position']}
[4] 残高取得（認証テスト）
  retCode=0 retMsg=OK
  OK: 認証付きリクエストが通りました

"""screener.py — 全銘柄ティッカーをDataFrame化して流動性でスクリーニング
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd
from pybit.unified_trading import HTTP

from client import make_client

# V5は数値も文字列で返るため、明示的に数値化する対象を列挙しておく
NUMERIC_COLS = [
    "lastPrice", "indexPrice", "markPrice", "prevPrice24h", "price24hPcnt",
    "highPrice24h", "lowPrice24h", "turnover24h", "volume24h",
    "openInterest", "openInterestValue", "fundingRate",
    "bid1Price", "bid1Size", "ask1Price", "ask1Size",
]


def fetch_tickers(http: HTTP, category: str = "linear") -> pd.DataFrame:
    res = http.get_tickers(category=category)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_tickers失敗: {res['retCode']} {res['retMsg']}")

    df = pd.DataFrame(res["result"]["list"])
    for col in NUMERIC_COLS:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")

    # 実効スプレッド（bps = 万分率）。執行コストの一次近似になる
    df["spread_bps"] = (df["ask1Price"] - df["bid1Price"]) / df["lastPrice"] * 10_000

    # 資金調達率の年率換算。※8時間ごと収受(1日3回)を仮定した概算。
    #   実際の収受間隔は銘柄ごとに異なる(instruments-infoのfundingInterval)。
    #   正確な計算は記事10で扱う。
    df["funding_apr_pct"] = df["fundingRate"] * 3 * 365 * 100

    df["chg24h_pct"] = df["price24hPcnt"] * 100
    return df


def screen_usdt_perp(df: pd.DataFrame, min_turnover_usd: float = 50_000_000) -> pd.DataFrame:
    """USDT無期限のうち、24h売買代金が閾値以上の銘柄を流動性順に返す。"""
    out = df[df["symbol"].str.endswith("USDT")].copy()
    out = out[out["turnover24h"] >= min_turnover_usd]
    out = out.sort_values("turnover24h", ascending=False)
    return out[[
        "symbol", "lastPrice", "chg24h_pct", "turnover24h",
        "spread_bps", "funding_apr_pct", "openInterestValue",
    ]].reset_index(drop=True)


if __name__ == "__main__":
    c = make_client()
    tickers = fetch_tickers(c.http, category="linear")
    print(f"取得銘柄数: {len(tickers)}")

    top = screen_usdt_perp(tickers)
    pd.set_option("display.width", 160)
    pd.set_option("display.float_format", lambda v: f"{v:,.4f}")
    print(top.head(15).to_string())

    # 資金調達率が高い＝ロングが過熱している銘柄（記事35のアービトラージ候補）
    print("\n--- 資金調達率(年率)上位 ---")
    print(top.nlargest(10, "funding_apr_pct")[
        ["symbol", "funding_apr_pct", "turnover24h", "spread_bps"]
    ].to_string())

"""wallet.py — 統合口座の残高とリスク指標を読む
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from pybit.unified_trading import HTTP

from client import make_client


def f(value: str | None) -> float:
    """V5は「該当なし」を空文字で返すことがあるので安全に数値化する。"""
    if value is None or value == "":
        return 0.0
    return float(value)


def show_wallet(http: HTTP, account_type: str = "UNIFIED") -> dict:
    res = http.get_wallet_balance(accountType=account_type)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_wallet_balance失敗: {res['retCode']} {res['retMsg']}")

    acct = res["result"]["list"][0]

    print("=== 口座サマリ (USD換算) ===")
    print(f"  総資産            : {f(acct['totalEquity']):>14,.2f}")
    print(f"  総証拠金残高      : {f(acct['totalMarginBalance']):>14,.2f}")
    print(f"  利用可能証拠金    : {f(acct['totalAvailableBalance']):>14,.2f}")
    print(f"  使用中(初期証拠金): {f(acct['totalInitialMargin']):>14,.2f}")
    print(f"  維持証拠金        : {f(acct['totalMaintenanceMargin']):>14,.2f}")
    print(f"  無期限の含み損益  : {f(acct['totalPerpUPL']):>+14,.2f}")

    # 口座維持率。1.0(100%)に近づくと強制決済が発生する
    mmr = f(acct.get("accountMMRate"))
    imr = f(acct.get("accountIMRate"))
    print(f"  口座維持率(MMR)   : {mmr * 100:>13.3f} %")
    print(f"  口座初期率(IMR)   : {imr * 100:>13.3f} %")
    if mmr > 0.5:
        print("  [警告] 維持率が50%を超えています。清算距離が近い状態です。")

    print("\n=== 通貨別明細 ===")
    print(f"  {'coin':<8}{'残高':>16}{'含み損益':>14}{'USD換算':>14}")
    for coin in acct["coin"]:
        bal = f(coin["walletBalance"])
        if bal == 0 and f(coin["unrealisedPnl"]) == 0:
            continue
        print(
            f"  {coin['coin']:<8}{bal:>16,.6f}"
            f"{f(coin['unrealisedPnl']):>+14,.2f}{f(coin['usdValue']):>14,.2f}"
        )
    return acct


if __name__ == "__main__":
    c = make_client()
    show_wallet(c.http)

"""symbol_rules.py — 銘柄仕様の取得とDecimalによる丸め
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import ROUND_DOWN, ROUND_HALF_UP, Decimal

from pybit.unified_trading import HTTP

from client import make_client


@dataclass(frozen=True)
class SymbolRules:
    symbol: str
    category: str
    tick_size: Decimal      # 価格の刻み幅
    qty_step: Decimal       # 数量の刻み幅
    min_qty: Decimal        # 最小注文数量
    max_qty: Decimal        # 最大注文数量（成行/指値で異なる場合あり）
    min_notional: Decimal   # 最小注文金額
    max_leverage: Decimal   # 最大レバレッジ


def fetch_rules(http: HTTP, symbol: str, category: str = "linear") -> SymbolRules:
    res = http.get_instruments_info(category=category, symbol=symbol)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_instruments_info失敗: {res['retCode']} {res['retMsg']}")

    items = res["result"]["list"]
    if not items:
        raise LookupError(f"{symbol} は category={category} に存在しません")
    info = items[0]

    pf = info["priceFilter"]
    lf = info["lotSizeFilter"]
    lev = info.get("leverageFilter", {})

    return SymbolRules(
        symbol=info["symbol"],
        category=category,
        tick_size=Decimal(pf["tickSize"]),
        qty_step=Decimal(lf.get("qtyStep") or lf.get("basePrecision") or "0"),
        min_qty=Decimal(lf.get("minOrderQty") or "0"),
        max_qty=Decimal(lf.get("maxOrderQty") or "0"),
        min_notional=Decimal(lf.get("minNotionalValue") or lf.get("minOrderAmt") or "0"),
        max_leverage=Decimal(lev.get("maxLeverage") or "1"),
    )


def floor_to_step(value: Decimal, step: Decimal) -> Decimal:
    """刻み幅に切り捨てで揃える。数量はこちら（過大発注を防ぐ）。"""
    if step == 0:
        return value
    return (value / step).to_integral_value(rounding=ROUND_DOWN) * step


def round_to_step(value: Decimal, step: Decimal) -> Decimal:
    """刻み幅に四捨五入で揃える。価格はこちら。"""
    if step == 0:
        return value
    return (value / step).to_integral_value(rounding=ROUND_HALF_UP) * step


def prepare_order(rules: SymbolRules, price: Decimal, notional_usd: Decimal
                  ) -> tuple[Decimal, Decimal]:
    """想定金額(USD)から、発注可能な (価格, 数量) を組み立てる。"""
    px = round_to_step(price, rules.tick_size)
    qty = floor_to_step(notional_usd / px, rules.qty_step)

    if qty < rules.min_qty:
        raise ValueError(
            f"数量 {qty} が最小注文数量 {rules.min_qty} を下回ります"
            f"（{notional_usd} USD では足りません）"
        )
    if rules.min_notional > 0 and px * qty < rules.min_notional:
        raise ValueError(f"注文金額 {px * qty} が最小 {rules.min_notional} 未満です")
    return px, qty


if __name__ == "__main__":
    c = make_client()
    r = fetch_rules(c.http, "BTCUSDT", "linear")
    print(r)

    # 現在価格の0.5%下に、想定100USDの指値を置く場合の値を作る
    last = Decimal(
        c.http.get_tickers(category="linear", symbol="BTCUSDT")["result"]["list"][0]["lastPrice"]
    )
    px, qty = prepare_order(r, last * Decimal("0.995"), Decimal("100"))
    print(f"発注価格={px}  数量={qty}  想定金額={px * qty:.2f} USD")

"""bybit_raw.py — pybit非依存のV5最小クライアント（HMAC-SHA256署名）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import hashlib
import hmac
import json
import time
from typing import Any
from urllib.parse import urlencode

import requests

BASE_URLS = {
    "mainnet": "https://api.bybit.com",
    "testnet": "https://api-testnet.bybit.com",
    "demo": "https://api-demo.bybit.com",
}


class BybitAPIError(RuntimeError):
    def __init__(self, ret_code: int, ret_msg: str, ext: Any = None):
        super().__init__(f"retCode={ret_code} retMsg={ret_msg}")
        self.ret_code = ret_code
        self.ret_msg = ret_msg
        self.ext = ext


class BybitRaw:
    def __init__(
        self,
        api_key: str = "",
        api_secret: str = "",
        env: str = "testnet",
        recv_window: int = 5000,
        timeout: int = 10,
    ):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base = BASE_URLS[env]
        self.recv_window = recv_window
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Referer": "codetribe"})  # 任意

    # ---- 署名 --------------------------------------------------------
    def _sign(self, timestamp: str, payload: str) -> str:
        """V5署名: timestamp + apiKey + recvWindow + (queryString | jsonBody)"""
        message = f"{timestamp}{self.api_key}{self.recv_window}{payload}"
        return hmac.new(
            self.api_secret.encode("utf-8"),
            message.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()

    def _auth_headers(self, payload: str) -> dict[str, str]:
        ts = str(int(time.time() * 1000))
        return {
            "X-BAPI-API-KEY": self.api_key,
            "X-BAPI-TIMESTAMP": ts,
            "X-BAPI-RECV-WINDOW": str(self.recv_window),
            "X-BAPI-SIGN": self._sign(ts, payload),
            "X-BAPI-SIGN-TYPE": "2",          # 2 = HMAC-SHA256
            "Content-Type": "application/json",
        }

    # ---- リクエスト --------------------------------------------------
    def get(self, path: str, params: dict | None = None, *, auth: bool = True) -> dict:
        # 署名対象のqueryStringと、実際に送るqueryStringを同一にするため
        # ここで一度だけ文字列化し、それをURLに直接連結する
        query = urlencode(params or {})
        headers = self._auth_headers(query) if auth else {}
        url = f"{self.base}{path}" + (f"?{query}" if query else "")
        r = self.session.get(url, headers=headers, timeout=self.timeout)
        return self._unwrap(r)

    def post(self, path: str, body: dict | None = None) -> dict:
        # 署名した文字列と送信する文字列を完全に一致させる。
        # requestsのjson=は独自に直列化するので使わない（署名不一致の原因）。
        payload = json.dumps(body or {}, separators=(",", ":"))
        headers = self._auth_headers(payload)
        r = self.session.post(
            f"{self.base}{path}", data=payload, headers=headers, timeout=self.timeout
        )
        return self._unwrap(r)

    # ---- レスポンス処理 ----------------------------------------------
    @staticmethod
    def _unwrap(r: requests.Response) -> dict:
        # レート制限の残量はヘッダで返る（記事05で活用）
        limit_status = r.headers.get("X-Bapi-Limit-Status")
        if limit_status is not None and int(limit_status) < 3:
            print(f"[warn] レート残量わずか: {limit_status}")

        if r.status_code == 403:
            raise BybitAPIError(-403, "IP単位のレート制限に到達（約10分の制限）")
        r.raise_for_status()

        data = r.json()
        if data.get("retCode") != 0:
            raise BybitAPIError(data.get("retCode"), data.get("retMsg", ""),
                                data.get("retExtInfo"))
        return data.get("result") or {}


if __name__ == "__main__":
    from config import load_credentials

    cred = load_credentials()
    api = BybitRaw(cred.api_key, cred.api_secret, env=cred.env.value)

    # 1) 公開エンドポイント（認証なし）
    srv = api.get("/v5/market/time", auth=False)
    print("server time:", srv["timeSecond"])

    # 2) 認証付きGET
    wallet = api.get("/v5/account/wallet-balance", {"accountType": "UNIFIED"})
    print("総資産:", wallet["list"][0]["totalEquity"])

    # 3) 認証付きPOST（テストネットでのみ実行してください）
    # order = api.post("/v5/order/create", {
    #     "category": "linear", "symbol": "BTCUSDT", "side": "Buy",
    #     "orderType": "Limit", "qty": "0.001", "price": "50000",
    #     "timeInForce": "PostOnly",
    # })
    # print("orderId:", order["orderId"])

"""debug_sign.py — 署名エラー(10004)の切り分けヘルパ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import hashlib
import hmac
import json
import time
from urllib.parse import urlencode


def dump_signature(api_key: str, api_secret: str, *,
                   method: str = "GET",
                   params: dict | None = None,
                   recv_window: int = 5000) -> None:
    ts = str(int(time.time() * 1000))

    if method.upper() == "GET":
        payload = urlencode(params or {})
    else:
        payload = json.dumps(params or {}, separators=(",", ":"))

    message = f"{ts}{api_key}{recv_window}{payload}"
    sign = hmac.new(api_secret.encode(), message.encode(), hashlib.sha256).hexdigest()

    print("--- 署名の内訳 -------------------------------------------")
    print(f"timestamp   : {ts}")
    print(f"apiKey      : {api_key[:4]}****  (長さ {len(api_key)})")
    print(f"recvWindow  : {recv_window}")
    print(f"payload     : {payload!r}")
    print(f"署名対象     : {message[:20]}...{message[-40:]!r}")
    print(f"X-BAPI-SIGN : {sign}")
    print("----------------------------------------------------------")
    print("チェック項目:")
    print("  1. payload は実際に送信する文字列と1文字も違わないか")
    print("     （POSTで requests の json= を使っていないか）")
    print("  2. apiKey / apiSecret の前後に改行・空白が混じっていないか")
    print("  3. 接続先ドメインとキーの環境が一致しているか")
    print("     （testnetのキーを api.bybit.com に投げていないか）")
    print("  4. timestamp のズレが recvWindow 以内か（10002ならこれ）")


if __name__ == "__main__":
    from config import load_credentials

    cred = load_credentials()
    dump_signature(cred.api_key, cred.api_secret,
                   method="GET", params={"accountType": "UNIFIED"})

{
  "retCode": 0,                  // 0 が成功。それ以外はすべてエラー
  "retMsg": "OK",                // エラーメッセージ
  "result": { ... },             // 実データ。多くは {"list": [...], "nextPageCursor": "..."}
  "retExtInfo": {},              // バッチ処理時の個別結果などが入る
  "time": 1785000000000          // サーバー時刻(ms)
}

"""unwrap.py — retCodeを1箇所で処理するラッパ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from typing import Any


class BybitAPIError(RuntimeError):
    def __init__(self, ret_code: int, ret_msg: str, ext: Any = None):
        super().__init__(f"retCode={ret_code} retMsg={ret_msg}")
        self.ret_code = ret_code
        self.ret_msg = ret_msg
        self.ext = ext

    @property
    def retryable(self) -> bool:
        # レート制限・サーバー内部エラーは時間を置けば通る可能性がある
        return self.ret_code in {10006, 10016}

    @property
    def needs_reconcile(self) -> bool:
        # 「注文が見つからない」系は、送れているのに結果を取り違えている
        # 可能性があるため、取引所側の状態照会が必要（記事17）
        return self.ret_code in {110001, 110004}


# エラーだが実質成功として扱ってよいコード
BENIGN_CODES = {
    110043,   # レバレッジ変更なし（既に同じ値が設定されている）
}


def unwrap(res: dict) -> dict:
    """V5レスポンスからresultを取り出す。retCode≠0なら例外にする。"""
    code = res.get("retCode")
    if code == 0:
        return res.get("result") or {}
    if code in BENIGN_CODES:
        return res.get("result") or {}
    raise BybitAPIError(code, res.get("retMsg", ""), res.get("retExtInfo"))


# 使用例
if __name__ == "__main__":
    from client import make_client

    c = make_client()
    try:
        result = unwrap(c.http.get_wallet_balance(accountType="UNIFIED"))
        print(result["list"][0]["totalEquity"])
    except BybitAPIError as e:
        print(f"APIエラー: {e}  リトライ可={e.retryable}")

from collections.abc import Callable, Iterator


def paginate(fetch: Callable[..., dict], *, limit: int = 200,
             max_pages: int = 100, **kwargs) -> Iterator[dict]:
    """cursorベースのV5エンドポイントを全件たどる。

    使用例:
        for row in paginate(c.http.get_executions,
                            category="linear", symbol="BTCUSDT"):
            print(row["execTime"], row["execPrice"])
    """
    cursor = None
    for _ in range(max_pages):
        params = dict(kwargs, limit=limit)
        if cursor:
            params["cursor"] = cursor
        result = unwrap(fetch(**params))

        rows = result.get("list") or []
        yield from rows

        cursor = result.get("nextPageCursor")
        if not cursor or not rows:
            return