Bybit V5 APIの認証・署名をPythonで自作実装
HMAC-SHA256とRSA・
X-BAPIヘッダ完全解説

timestamp   : 1785000000000
apiKey      : AbCdEfGhIjKlMnOpQr
recvWindow  : 5000
queryString : accountType=UNIFIED

署名対象     : "1785000000000AbCdEfGhIjKlMnOpQr5000accountType=UNIFIED"
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                区切り文字は一切入らない。単純な連結。

X-BAPI-SIGN = HMAC-SHA256(apiSecret, 署名対象).hexdigest()   ← 小文字16進

"""signer.py — V5署名の生成（HMAC / RSA 両対応）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import base64
import hashlib
import hmac
from abc import ABC, abstractmethod


class Signer(ABC):
    """署名器の共通インターフェース。HMACとRSAを差し替え可能にする。"""

    sign_type: str = "2"

    @abstractmethod
    def sign(self, payload: str) -> str:
        """署名対象文字列を受け取り、X-BAPI-SIGN に入れる値を返す。"""


class HmacSigner(Signer):
    """システム生成キー用。出力は小文字16進。"""

    sign_type = "2"

    def __init__(self, api_secret: str):
        if not api_secret:
            raise ValueError("api_secretが空です")
        # 前後の空白・改行を除去する。コピペ時の混入で10004になるため
        self._secret = api_secret.strip().encode("utf-8")

    def sign(self, payload: str) -> str:
        return hmac.new(self._secret, payload.encode("utf-8"),
                        hashlib.sha256).hexdigest()


class RsaSigner(Signer):
    """自己生成キー（RSA）用。出力はbase64。

      pip install cryptography
    """

    sign_type = "2"

    def __init__(self, private_key_pem: str | bytes):
        from cryptography.hazmat.primitives import serialization

        if isinstance(private_key_pem, str):
            private_key_pem = private_key_pem.strip().encode("utf-8")
        self._key = serialization.load_pem_private_key(private_key_pem,
                                                       password=None)

    def sign(self, payload: str) -> str:
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import padding

        signature = self._key.sign(
            payload.encode("utf-8"),
            padding.PKCS1v15(),
            hashes.SHA256(),
        )
        # HMACは16進だが、RSAはbase64で渡す点が違う
        return base64.b64encode(signature).decode("ascii")


def build_payload(timestamp: str, api_key: str, recv_window: int,
                  body_or_query: str) -> str:
    """V5の署名対象文字列を組み立てる。区切り文字は入らない。"""
    return f"{timestamp}{api_key}{recv_window}{body_or_query}"


if __name__ == "__main__":
    # 既知の値で決定論的に動くことを確認する
    signer = HmacSigner("mysecret")
    payload = build_payload("1785000000000", "mykey", 5000,
                            "accountType=UNIFIED")
    print("署名対象:", payload)
    print("署名    :", signer.sign(payload))
    # 同じ入力なら常に同じ署名になる（これが確認できれば実装は正しい）
    assert signer.sign(payload) == signer.sign(payload)

"""v5_client.py — 署名を自作したV5 RESTクライアント（同期版）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

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

import requests

from signer import HmacSigner, Signer, build_payload


class V5Error(RuntimeError):
    def __init__(self, ret_code: int, ret_msg: str, endpoint: str = ""):
        super().__init__(f"[{ret_code}] {ret_msg} ({endpoint})")
        self.ret_code = ret_code
        self.ret_msg = ret_msg


class V5Client:
    def __init__(self, base_url: str, api_key: str = "",
                 signer: Signer | None = None,
                 recv_window: int = 5000, timeout: float = 10.0):
        self.base = base_url.rstrip("/")
        self.api_key = api_key.strip()
        self.signer = signer
        self.recv_window = recv_window
        self.timeout = timeout
        self.session = requests.Session()
        self.time_offset_ms: int = 0        # サーバーとのズレ補正（後述）

    # ---- タイムスタンプ ------------------------------------------------
    def _timestamp(self) -> str:
        return str(int(time.time() * 1000) + self.time_offset_ms)

    # ---- 認証ヘッダ ----------------------------------------------------
    def _auth_headers(self, body_or_query: str) -> dict[str, str]:
        if not self.api_key or self.signer is None:
            raise RuntimeError("認証が必要なリクエストにキーが設定されていません")

        ts = self._timestamp()
        payload = build_payload(ts, self.api_key, self.recv_window,
                                body_or_query)
        return {
            "X-BAPI-API-KEY": self.api_key,
            "X-BAPI-TIMESTAMP": ts,
            "X-BAPI-RECV-WINDOW": str(self.recv_window),
            "X-BAPI-SIGN": self.signer.sign(payload),
            "X-BAPI-SIGN-TYPE": self.signer.sign_type,
        }

    # ---- GET -----------------------------------------------------------
    def get(self, path: str, params: dict | None = None, *,
            auth: bool = True) -> dict:
        # 署名対象と送信URLを完全一致させるため、ここで1回だけ文字列化し、
        # requestsのparams=は使わず自分でURLに連結する
        query = urlencode(params or {}, doseq=True)
        headers = self._auth_headers(query) if auth else {}
        url = f"{self.base}{path}" + (f"?{query}" if query else "")

        resp = self.session.get(url, headers=headers, timeout=self.timeout)
        return self._unwrap(resp, path)

    # ---- POST ----------------------------------------------------------
    def post(self, path: str, body: dict | None = None) -> dict:
        # 署名した文字列と送信バイト列を完全一致させる。
        # requestsのjson=は独自に直列化するので絶対に使わない
        payload = json.dumps(body or {}, separators=(",", ":"),
                             ensure_ascii=False)
        headers = self._auth_headers(payload)
        headers["Content-Type"] = "application/json"

        resp = self.session.post(
            f"{self.base}{path}",
            data=payload.encode("utf-8"),   # ヘッダ生成に使った文字列そのまま
            headers=headers,
            timeout=self.timeout,
        )
        return self._unwrap(resp, path, sent=body)

    # ---- レスポンス処理 --------------------------------------------------
    def _unwrap(self, resp: requests.Response, endpoint: str,
                sent: Any = None) -> dict:
        remaining = resp.headers.get("X-Bapi-Limit-Status")
        if remaining is not None and remaining.isdigit() and int(remaining) <= 2:
            print(f"[warn] {endpoint}: レート残量 {remaining}")

        if resp.status_code == 403:
            raise V5Error(-403, "IP単位のレート制限（約10分）", endpoint)
        resp.raise_for_status()

        data = resp.json()
        if data.get("retCode") != 0:
            raise V5Error(data.get("retCode", -1), data.get("retMsg", ""),
                          endpoint)
        return data.get("result") or {}

    # ---- 便利メソッド --------------------------------------------------
    def server_time_ms(self) -> int:
        r = self.get("/v5/market/time", auth=False)
        return int(int(r["timeNano"]) / 1_000_000)

    def wallet_balance(self, account_type: str = "UNIFIED") -> dict:
        return self.get("/v5/account/wallet-balance",
                        {"accountType": account_type})

    def tickers(self, category: str = "linear", symbol: str | None = None) -> dict:
        params: dict = {"category": category}
        if symbol:
            params["symbol"] = symbol
        return self.get("/v5/market/tickers", params, auth=False)


def make_v5_client() -> V5Client:
    from config import load_credentials

    cred = load_credentials()
    return V5Client(base_url=cred.base_url, api_key=cred.api_key,
                    signer=HmacSigner(cred.api_secret))


if __name__ == "__main__":
    api = make_v5_client()
    print("server time:", api.server_time_ms())
    print("BTC:", api.tickers(symbol="BTCUSDT")["list"][0]["lastPrice"])
    print("総資産:", api.wallet_balance()["list"][0]["totalEquity"])

body = {"category": "linear", "symbol": "BTCUSDT"}

# 署名は自分でdumpsした文字列に対して作る
payload = json.dumps(body)                    # -> '{"category": "linear", ...}'
sign = hmac_sign(payload)                     # 空白入りの文字列で署名

# ところが送信は requests が独自に直列化する
requests.post(url, json=body, headers={"X-BAPI-SIGN": sign})
#                  ^^^^^^^^^ requestsは separators=(',', ':') 相当で送る
#                            -> '{"category":"linear",...}' 空白なし
# 署名した文字列と送信内容が食い違い、retCode=10004 になる

payload = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
sign = hmac_sign(payload)

requests.post(
    url,
    data=payload.encode("utf-8"),     # json= ではなく data= に同じ文字列
    headers={"X-BAPI-SIGN": sign, "Content-Type": "application/json"},
)

"""time_sync.py — サーバー時刻とのオフセットを測定して署名に反映する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import statistics
import threading
import time

from v5_client import V5Client


def measure_offset(client: V5Client, samples: int = 7) -> tuple[int, float]:
    """(オフセットms, RTT中央値ms) を返す。

    NTPと同じ考え方で、往復時間の中点をローカル時刻とみなして差を取る。
    複数回測って中央値を使うことで、通信のばらつきを吸収する。
    """
    offsets: list[float] = []
    rtts: list[float] = []

    for _ in range(samples):
        t0 = time.time()
        server_ms = client.server_time_ms()
        t1 = time.time()

        local_mid_ms = (t0 + t1) / 2 * 1000
        offsets.append(server_ms - local_mid_ms)
        rtts.append((t1 - t0) * 1000)
        time.sleep(0.1)

    return int(statistics.median(offsets)), statistics.median(rtts)


class TimeSyncedClient(V5Client):
    """定期的にオフセットを測り直すクライアント。"""

    def __init__(self, *args, resync_sec: float = 300.0, **kwargs):
        super().__init__(*args, **kwargs)
        self.resync_sec = resync_sec
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def sync_now(self, verbose: bool = True) -> int:
        offset, rtt = measure_offset(self)
        self.time_offset_ms = offset
        if verbose:
            warn = ""
            if abs(offset) > self.recv_window * 0.5:
                warn = "  ← NTP同期を確認してください"
            print(f"[timesync] オフセット {offset:+d}ms  RTT {rtt:.0f}ms{warn}")
        return offset

    def start_auto_sync(self) -> None:
        """バックグラウンドで定期的に再同期する。"""

        def loop() -> None:
            while not self._stop.wait(self.resync_sec):
                try:
                    self.sync_now(verbose=False)
                except Exception as e:
                    print(f"[timesync] 再同期に失敗: {e}")

        self.sync_now()
        self._thread = threading.Thread(target=loop, daemon=True,
                                        name="timesync")
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()


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

    cred = load_credentials()
    api = TimeSyncedClient(base_url=cred.base_url, api_key=cred.api_key,
                           signer=HmacSigner(cred.api_secret))
    api.start_auto_sync()

    print("総資産:", api.wallet_balance()["list"][0]["totalEquity"])
    api.stop()

"""rsa_setup.py — RSA鍵ペアの生成と、RSA署名クライアントの使用例
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install cryptography

公式のジェネレータもあります:
  https://github.com/bybit-exchange/api-rsa-generator
"""
from __future__ import annotations

import os
import stat
from pathlib import Path

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa


def generate_keypair(out_dir: str | Path = "secrets",
                     bits: int = 2048) -> tuple[Path, Path]:
    """RSA鍵ペアをPEM形式で保存する。公開鍵だけをBybitに登録する。"""
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)

    key = rsa.generate_private_key(public_exponent=65537, key_size=bits)

    priv_path = out / "bybit_private.pem"
    pub_path = out / "bybit_public.pem"

    priv_pem = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    pub_pem = key.public_key().public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo,
    )

    priv_path.write_bytes(priv_pem)
    pub_path.write_bytes(pub_pem)

    # 秘密鍵のパーミッションを本人のみ読み書きに絞る（Linux/macOS）
    if os.name != "nt":
        priv_path.chmod(stat.S_IRUSR | stat.S_IWUSR)   # 0o600

    print(f"秘密鍵: {priv_path}  ← 絶対に外部に出さない / .gitignoreに追加")
    print(f"公開鍵: {pub_path}   ← この中身をBybitの管理画面に登録する")
    return priv_path, pub_path


def make_rsa_client(private_key_path: str = "secrets/bybit_private.pem"):
    from config import load_credentials
    from signer import RsaSigner
    from v5_client import V5Client

    cred = load_credentials()
    pem = Path(private_key_path).read_bytes()
    return V5Client(base_url=cred.base_url, api_key=cred.api_key,
                    signer=RsaSigner(pem))


if __name__ == "__main__":
    import sys

    if "--generate" in sys.argv:
        generate_keypair()
    else:
        api = make_rsa_client()
        print("RSA署名で接続します")
        print("総資産:", api.wallet_balance()["list"][0]["totalEquity"])

"""v5_async.py — aiohttpによる非同期V5クライアント
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install aiohttp
"""
from __future__ import annotations

import asyncio
import json
import time
from typing import Any
from urllib.parse import urlencode

import aiohttp

from signer import Signer, build_payload


class AsyncV5Error(RuntimeError):
    def __init__(self, ret_code: int, ret_msg: str, endpoint: str = ""):
        super().__init__(f"[{ret_code}] {ret_msg} ({endpoint})")
        self.ret_code = ret_code
        self.ret_msg = ret_msg


class AsyncV5Client:
    def __init__(self, base_url: str, api_key: str = "",
                 signer: Signer | None = None, recv_window: int = 5000,
                 timeout: float = 10.0, max_concurrency: int = 8):
        self.base = base_url.rstrip("/")
        self.api_key = api_key.strip()
        self.signer = signer
        self.recv_window = recv_window
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.time_offset_ms = 0
        # 同時リクエスト数を絞る。IP制限（5秒600件）への基本的な配慮
        self._sem = asyncio.Semaphore(max_concurrency)
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "AsyncV5Client":
        self._session = aiohttp.ClientSession(timeout=self.timeout)
        return self

    async def __aexit__(self, *exc) -> None:
        if self._session:
            await self._session.close()
            self._session = None

    @property
    def session(self) -> aiohttp.ClientSession:
        if self._session is None:
            raise RuntimeError("async with で使用してください")
        return self._session

    def _auth_headers(self, body_or_query: str) -> dict[str, str]:
        if not self.api_key or self.signer is None:
            raise RuntimeError("キーが設定されていません")
        ts = str(int(time.time() * 1000) + self.time_offset_ms)
        payload = build_payload(ts, self.api_key, self.recv_window,
                                body_or_query)
        return {
            "X-BAPI-API-KEY": self.api_key,
            "X-BAPI-TIMESTAMP": ts,
            "X-BAPI-RECV-WINDOW": str(self.recv_window),
            "X-BAPI-SIGN": self.signer.sign(payload),
            "X-BAPI-SIGN-TYPE": self.signer.sign_type,
        }

    async def get(self, path: str, params: dict | None = None, *,
                  auth: bool = True) -> dict:
        query = urlencode(params or {}, doseq=True)
        headers = self._auth_headers(query) if auth else {}
        url = f"{self.base}{path}" + (f"?{query}" if query else "")

        async with self._sem:
            async with self.session.get(url, headers=headers) as resp:
                return await self._unwrap(resp, path)

    async def post(self, path: str, body: dict | None = None) -> dict:
        payload = json.dumps(body or {}, separators=(",", ":"),
                             ensure_ascii=False)
        headers = self._auth_headers(payload)
        headers["Content-Type"] = "application/json"

        async with self._sem:
            async with self.session.post(f"{self.base}{path}",
                                         data=payload.encode("utf-8"),
                                         headers=headers) as resp:
                return await self._unwrap(resp, path)

    @staticmethod
    async def _unwrap(resp: aiohttp.ClientResponse, endpoint: str) -> dict:
        if resp.status == 403:
            raise AsyncV5Error(-403, "IP単位のレート制限（約10分）", endpoint)
        resp.raise_for_status()
        data = await resp.json()
        if data.get("retCode") != 0:
            raise AsyncV5Error(data.get("retCode", -1),
                               data.get("retMsg", ""), endpoint)
        return data.get("result") or {}

    async def sync_time(self) -> int:
        t0 = time.time()
        r = await self.get("/v5/market/time", auth=False)
        t1 = time.time()
        server_ms = int(int(r["timeNano"]) / 1_000_000)
        self.time_offset_ms = int(server_ms - (t0 + t1) / 2 * 1000)
        return self.time_offset_ms


async def main() -> None:
    from config import load_credentials
    from signer import HmacSigner

    cred = load_credentials()
    async with AsyncV5Client(cred.base_url, cred.api_key,
                             HmacSigner(cred.api_secret)) as api:
        print("オフセット:", await api.sync_time(), "ms")

        # 複数の情報を並列で取得する（非同期化の主な利点）
        balance, positions, tickers = await asyncio.gather(
            api.get("/v5/account/wallet-balance", {"accountType": "UNIFIED"}),
            api.get("/v5/position/list", {"category": "linear",
                                          "settleCoin": "USDT"}),
            api.get("/v5/market/tickers", {"category": "linear",
                                           "symbol": "BTCUSDT"}, auth=False),
        )
        print("総資産  :", balance["list"][0]["totalEquity"])
        print("建玉数  :", len([p for p in positions["list"]
                              if float(p["size"]) > 0]))
        print("BTC価格 :", tickers["list"][0]["lastPrice"])


if __name__ == "__main__":
    asyncio.run(main())

"""ws_auth.py — private WebSocketの認証（RESTとは署名対象が違う）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install websockets orjson
"""
from __future__ import annotations

import asyncio
import hashlib
import hmac
import time

import orjson
import websockets

# max_active_time は 30s〜600s（"1m" のような分表記も可）
PRIVATE_URL = "wss://stream.bybit.com/v5/private?max_active_time=600s"
TESTNET_URL = "wss://stream-testnet.bybit.com/v5/private"
DEMO_URL = "wss://stream-demo.bybit.com/v5/private"


def ws_signature(api_secret: str, expires_ms: int) -> str:
    """WebSocket認証の署名。対象は 'GET/realtime' + expires のみ。"""
    message = f"GET/realtime{expires_ms}"
    return hmac.new(api_secret.strip().encode("utf-8"),
                    message.encode("utf-8"), hashlib.sha256).hexdigest()


def auth_message(api_key: str, api_secret: str,
                 valid_for_sec: float = 5.0) -> bytes:
    """authオペレーションのJSONを組み立てる。"""
    expires = int((time.time() + valid_for_sec) * 1000)
    return orjson.dumps({
        "req_id": f"auth-{expires}",
        "op": "auth",
        "args": [api_key.strip(), expires, ws_signature(api_secret, expires)],
    })


async def connect_private(api_key: str, api_secret: str,
                          topics: list[str], url: str = PRIVATE_URL) -> None:
    async with websockets.connect(url, ping_interval=None) as ws:
        # 1) 認証
        await ws.send(auth_message(api_key, api_secret))
        reply = orjson.loads(await ws.recv())
        if not reply.get("success", False):
            raise RuntimeError(f"WS認証に失敗: {reply}")
        print(f"[ws] 認証成功 conn_id={reply.get('conn_id')}")

        # 2) 購読
        await ws.send(orjson.dumps({"op": "subscribe", "args": topics}))

        # 3) 20秒ごとのping（Bybit独自のハートビート）
        async def pinger() -> None:
            while True:
                await asyncio.sleep(20)
                await ws.send(orjson.dumps({"op": "ping"}))

        ping_task = asyncio.create_task(pinger())
        try:
            async for raw in ws:
                msg = orjson.loads(raw)
                if "topic" in msg:
                    print(f"[{msg['topic']}] {msg.get('data')}")
                elif msg.get("op") == "subscribe":
                    print(f"[ws] 購読結果: {msg}")
        finally:
            ping_task.cancel()


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

    cred = load_credentials()
    url = {"mainnet": PRIVATE_URL, "testnet": TESTNET_URL,
           "demo": DEMO_URL}[cred.env.value]

    # 署名の形が正しいかを目視確認できるようにしておく
    expires = int((time.time() + 5) * 1000)
    print(f"署名対象 : 'GET/realtime{expires}'")
    print(f"署名     : {ws_signature(cred.api_secret, expires)[:24]}...")

    asyncio.run(connect_private(cred.api_key, cred.api_secret,
                                topics=["order", "position", "wallet"],
                                url=url))

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

import json
import time
from urllib.parse import urlencode

from signer import HmacSigner, build_payload

CHECKLIST = """
--- 10004 のチェックリスト（上から順に確認） ---

1. 接続先とキーの環境が一致しているか
     testnetのキーを api.bybit.com に投げていないか
     （これは 10003 になることも多いが、混乱しやすい）

2. Secretに余分な文字が混じっていないか
     ブラウザからのコピペで改行・全角スペースが入りやすい
     -> 下の「Secretの長さ」を確認する

3. POSTで requests の json= を使っていないか
     署名した文字列と送信バイト列が食い違う最大の原因
     -> data=payload.encode("utf-8") にする

4. GETのqueryStringが署名対象と送信URLで一致しているか
     requests の params= に dict を渡すと、
     自分がurlencodeした文字列と順序が変わる可能性がある

5. RSAキーで16進を送っていないか
     RSAの署名は base64。HMACは小文字16進

6. recvWindowが署名対象とヘッダで同じ値か

7. timestampが署名対象とヘッダ(X-BAPI-TIMESTAMP)で同じ値か
     署名生成とヘッダ生成で別々にtime.time()を呼んでいると
     ミリ秒がずれて不一致になる（ありがちなバグ）
"""


def dump(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":
        body_or_query = urlencode(params or {}, doseq=True)
    else:
        body_or_query = json.dumps(params or {}, separators=(",", ":"),
                                   ensure_ascii=False)

    payload = build_payload(ts, api_key.strip(), recv_window, body_or_query)
    sign = HmacSigner(api_secret).sign(payload)

    print("=" * 66)
    print(f"method          : {method.upper()}")
    print(f"timestamp       : {ts}")
    print(f"apiKey          : {api_key[:6]}****  (長さ {len(api_key)})")
    print(f"apiSecret       : ****        (長さ {len(api_secret)})")
    if api_key != api_key.strip() or api_secret != api_secret.strip():
        print("  [!] キーまたはSecretの前後に空白・改行があります")
    print(f"recvWindow      : {recv_window}")
    print(f"body_or_query   : {body_or_query!r}")
    print(f"署名対象(全文)   : {payload!r}")
    print(f"X-BAPI-SIGN     : {sign}")
    print("=" * 66)
    print(CHECKLIST)


def verify_roundtrip(api_secret: str) -> None:
    """署名関数が決定論的か（同じ入力で同じ出力か）を確認する。"""
    s = HmacSigner(api_secret)
    p = build_payload("1785000000000", "testkey", 5000, "a=1&b=2")
    a, b = s.sign(p), s.sign(p)
    print(f"決定論性: {'OK' if a == b else 'NG（実装に問題があります）'}")
    print(f"署名長  : {len(a)} 文字（HMAC-SHA256は64文字の16進）")


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

    cred = load_credentials()
    verify_roundtrip(cred.api_secret)
    dump(cred.api_key, cred.api_secret, method="GET",
         params={"accountType": "UNIFIED"})
    dump(cred.api_key, cred.api_secret, method="POST",
         params={"category": "linear", "symbol": "BTCUSDT", "side": "Buy",
                 "orderType": "Limit", "qty": "0.001", "price": "50000"})

"""test_signer.py — 署名の単体テスト（pytest）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install pytest
  pytest test_signer.py -v
"""
from __future__ import annotations

import json
from urllib.parse import urlencode

import pytest

from signer import HmacSigner, build_payload

# 固定の入力（実在しないダミーのキー）
TS = "1785000000000"
KEY = "testApiKey123456"
SECRET = "testApiSecret7890"
RECV = 5000


def test_payload_has_no_delimiters():
    """署名対象は単純連結。区切り文字が混入していないこと。"""
    p = build_payload(TS, KEY, RECV, "accountType=UNIFIED")
    assert p == f"{TS}{KEY}{RECV}accountType=UNIFIED"
    assert "&" not in p.replace("accountType=UNIFIED", "")
    assert p.startswith(TS) and p.endswith("accountType=UNIFIED")


def test_hmac_is_deterministic():
    """同じ入力なら常に同じ署名になること。"""
    s = HmacSigner(SECRET)
    p = build_payload(TS, KEY, RECV, "a=1")
    assert s.sign(p) == s.sign(p)


def test_hmac_format():
    """HMAC-SHA256は64文字の小文字16進であること。"""
    sign = HmacSigner(SECRET).sign(build_payload(TS, KEY, RECV, ""))
    assert len(sign) == 64
    assert sign == sign.lower()
    assert all(c in "0123456789abcdef" for c in sign)


def test_secret_is_stripped():
    """Secretの前後の空白・改行が無視されること（コピペ事故対策）。"""
    p = build_payload(TS, KEY, RECV, "a=1")
    assert HmacSigner(SECRET).sign(p) == HmacSigner(f"  {SECRET}\n").sign(p)


def test_empty_secret_rejected():
    with pytest.raises(ValueError):
        HmacSigner("")


def test_different_input_different_sign():
    """入力が1文字違えば署名が変わること。"""
    s = HmacSigner(SECRET)
    a = s.sign(build_payload(TS, KEY, RECV, "qty=0.001"))
    b = s.sign(build_payload(TS, KEY, RECV, "qty=0.002"))
    assert a != b


def test_post_body_serialization_is_compact():
    """POSTボディに余分な空白が入らないこと（10004の主要因）。"""
    body = {"category": "linear", "symbol": "BTCUSDT", "qty": "0.001"}
    payload = json.dumps(body, separators=(",", ":"), ensure_ascii=False)
    assert " " not in payload
    assert payload == '{"category":"linear","symbol":"BTCUSDT","qty":"0.001"}'


def test_get_query_matches_sent_url():
    """署名対象のqueryStringと、URLに付けるqueryStringが同一であること。"""
    params = {"category": "linear", "symbol": "BTCUSDT"}
    query = urlencode(params, doseq=True)

    signed_payload = build_payload(TS, KEY, RECV, query)
    sent_url = f"https://api.bybit.com/v5/market/tickers?{query}"

    # URLから切り出したクエリ部分が、署名に使った文字列と一致する
    assert sent_url.split("?", 1)[1] == query
    assert signed_payload.endswith(query)


def test_timestamp_used_once():
    """タイムスタンプは署名とヘッダで同じ値を使う、という契約の確認。"""
    ts = TS                       # 1回だけ生成した値を両方に使う
    payload = build_payload(ts, KEY, RECV, "a=1")
    headers = {
        "X-BAPI-TIMESTAMP": ts,
        "X-BAPI-SIGN": HmacSigner(SECRET).sign(payload),
    }
    assert payload.startswith(headers["X-BAPI-TIMESTAMP"])