Bybit APIエラーコード実践ハンドブック
retCode別の原因・リトライ可否・
自動リカバリのPython実装

"""bybit_errors.py — retCodeを例外階層にマッピングする
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from enum import Enum
from typing import Any


class Disposition(str, Enum):
    """このエラーをどう処理すべきか。"""

    FATAL = "fatal"        # A: リトライ不可。コードか設定の誤り
    RETRY = "retry"        # B: リトライ可。時間を置けば通る
    RECONCILE = "verify"   # C: 通ったか不明。状態照会が必要
    BENIGN = "benign"      # エラーだが実質成功。無視してよい


class BybitError(Exception):
    """全Bybit APIエラーの基底。"""

    disposition = Disposition.FATAL

    def __init__(self, ret_code: int, ret_msg: str = "",
                 endpoint: str = "", sent: Any = None,
                 ret_ext_info: Any = None):
        super().__init__(f"[{ret_code}] {ret_msg} ({endpoint})")
        self.ret_code = ret_code
        self.ret_msg = ret_msg
        self.endpoint = endpoint
        self.sent = sent                  # 送信パラメータ（調査で必須）
        self.ret_ext_info = ret_ext_info

    def __repr__(self) -> str:
        return (f"{type(self).__name__}(retCode={self.ret_code}, "
                f"disposition={self.disposition.value})")


# ---- A: リトライ不可 --------------------------------------------------
class BybitFatalError(BybitError):
    disposition = Disposition.FATAL


class BybitAuthError(BybitFatalError):
    """署名・キー・権限・IP。人間が設定を直すまで直らない。"""


class BybitParamError(BybitFatalError):
    """パラメータの誤り。コードのバグ。"""


class BybitInsufficientBalance(BybitFatalError):
    """残高・証拠金不足。資金かポジションを整理するまで直らない。"""


class BybitLimitExceeded(BybitFatalError):
    """注文数・TP/SL数などの上限超過。整理が必要。"""


# ---- B: リトライ可 ----------------------------------------------------
class BybitTransientError(BybitError):
    disposition = Disposition.RETRY


class BybitRateLimitError(BybitTransientError):
    """レート制限。バックオフして再試行。"""

    def __init__(self, *args, retry_after: float = 1.0, **kwargs):
        super().__init__(*args, **kwargs)
        self.retry_after = retry_after


class BybitIpBanError(BybitTransientError):
    """IPレート制限。約10分待つ必要がある。"""


# ---- C: 状態照会が必要 ------------------------------------------------
class BybitUnknownStateError(BybitError):
    """通ったかどうか分からない。照会してから判断する。"""

    disposition = Disposition.RECONCILE


# ---- 実質成功 ---------------------------------------------------------
class BybitBenignError(BybitError):
    """エラーだが処理を止める必要がない。"""

    disposition = Disposition.BENIGN


# ---- マッピング -------------------------------------------------------
# 「エラーだが無視してよい」コード
BENIGN_CODES: dict[int, str] = {
    110001: "注文が存在しない（既に約定/キャンセル済み）",
    110008: "注文は完了またはキャンセル済み",
    110010: "注文はキャンセル済み",
    110025: "ポジションモードは変更されていない",
    110026: "マージンモードは変更されていない",
    110027: "マージンは変更されていない",
    110030: "orderIdの重複（同一注文の再送）",
    110043: "レバレッジは変更されていない",
    110072: "orderLinkIdの重複（二重発注を防止）",
    110075: "リスクIDは変更されていない",
    170141: "clientOrderIdの重複（現物）",
    170142: "注文はキャンセル済み（現物）",
    170213: "注文が存在しない（現物）",
    34040: "TP/SLの値が既に設定済み",
}

AUTH_CODES = {10003, 10004, 10005, 10007, 10010}
PARAM_CODES = {10001, 10017, 10029, 110017, 110019, 110032, 110092, 110093,
               110094, 110119, 170105, 170115, 170116, 170117, 170121,
               170130, 170136, 170355}
BALANCE_CODES = {110004, 110006, 110007, 110012, 110014, 110044, 110045,
                 170033, 170131}
LIMIT_CODES = {110009, 110020, 110021, 110061, 170005, 170810}
RATE_CODES = {10006, 170222}
IP_BAN_CODES = {10018}
# 書き込み操作では「通ったか不明」、読み取りでは単なる再試行
AMBIGUOUS_CODES = {10000, 10016, 170007, 170146, 170147, 170234, 170310}


def classify(ret_code: int, *, is_write: bool = False) -> type[BybitError]:
    """retCodeと操作の種類から、対応する例外クラスを返す。

    is_write=True は「副作用のある操作」（発注・修正・キャンセル・振替）。
    同じretCodeでも読み取りと書き込みで扱いが変わる点が要点。
    """
    if ret_code in BENIGN_CODES:
        return BybitBenignError
    if ret_code in AUTH_CODES:
        return BybitAuthError
    if ret_code in PARAM_CODES:
        return BybitParamError
    if ret_code in BALANCE_CODES:
        return BybitInsufficientBalance
    if ret_code in LIMIT_CODES:
        return BybitLimitExceeded
    if ret_code in RATE_CODES:
        return BybitRateLimitError
    if ret_code in IP_BAN_CODES:
        return BybitIpBanError
    if ret_code in AMBIGUOUS_CODES:
        return BybitUnknownStateError if is_write else BybitTransientError
    if ret_code == 10002:
        # 時計ズレ。同期すれば直るのでリトライ扱いにするが、
        # 連続する場合はNTPの問題なのでFATALとして通知したい
        return BybitTransientError
    return BybitFatalError            # 未知のコードは安全側（止める）


def raise_for_ret_code(res: dict, endpoint: str = "", sent: Any = None,
                       *, is_write: bool = False) -> dict:
    """V5レスポンスを検査し、retCode!=0なら適切な例外を投げる。

    BENIGNの場合は例外を投げず、resultを返す。
    """
    code = res.get("retCode")
    if code == 0:
        return res.get("result") or {}

    exc_cls = classify(code, is_write=is_write)
    if exc_cls is BybitBenignError:
        return res.get("result") or {}

    raise exc_cls(code, res.get("retMsg", ""), endpoint=endpoint, sent=sent,
                  ret_ext_info=res.get("retExtInfo"))


def describe(ret_code: int) -> str:
    """retCodeの人間向け説明（ログ・通知用）。"""
    if ret_code in BENIGN_CODES:
        return f"{ret_code}: {BENIGN_CODES[ret_code]}（無視可）"
    cls = classify(ret_code)
    return f"{ret_code}: {cls.__name__} / 対処={cls.disposition.value}"


if __name__ == "__main__":
    for code in (0, 10001, 10004, 10006, 10016, 110007, 110043, 110072, 999999):
        if code == 0:
            print("0: 成功")
            continue
        read = classify(code, is_write=False).__name__
        write = classify(code, is_write=True).__name__
        mark = "  <- 読み書きで分類が変わる" if read != write else ""
        print(f"{code:>7}  読み取り={read:<24} 書き込み={write:<24}{mark}")

"""retry.py — 例外の分類に応じてリトライを制御するデコレータ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import functools
import logging
import random
import time
from collections.abc import Callable
from typing import ParamSpec, TypeVar

import requests

from bybit_errors import (
    BybitError, BybitIpBanError, BybitRateLimitError, BybitTransientError,
    BybitUnknownStateError, Disposition,
)

log = logging.getLogger(__name__)

P = ParamSpec("P")
R = TypeVar("R")

# ネットワーク層の例外。これらも「通ったか不明」に該当する
NETWORK_ERRORS = (
    requests.exceptions.Timeout,
    requests.exceptions.ConnectionError,
    requests.exceptions.ChunkedEncodingError,
)

IP_BAN_WAIT_SEC = 600.0      # IPレート制限は約10分


def with_retry(
    max_attempts: int = 4,
    base_delay: float = 0.5,
    max_delay: float = 30.0,
    is_write: bool = False,
    on_unknown_state: Callable[[Exception], R | None] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """分類に応じてリトライする。

    is_write=True の場合、「通ったか不明」な失敗では
    リトライせずに on_unknown_state（照会処理）へ委ねる。
    これが二重発注を防ぐ要点。
    """

    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            delay = base_delay
            last_exc: Exception | None = None

            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)

                except NETWORK_ERRORS as e:
                    last_exc = e
                    if is_write:
                        log.warning(
                            "%s: 書き込み中に通信エラー。リトライせず照会します (%s)",
                            fn.__name__, type(e).__name__,
                        )
                        if on_unknown_state:
                            return on_unknown_state(e)
                        raise BybitUnknownStateError(
                            -1, f"通信エラー: {e}", endpoint=fn.__name__
                        ) from e
                    log.info("%s: 通信エラー、再試行 %d/%d", fn.__name__,
                             attempt, max_attempts)

                except BybitIpBanError as e:
                    last_exc = e
                    log.error("IPレート制限に到達。%.0f秒待機します", IP_BAN_WAIT_SEC)
                    time.sleep(IP_BAN_WAIT_SEC)
                    continue

                except BybitRateLimitError as e:
                    last_exc = e
                    wait = max(delay, getattr(e, "retry_after", 0.0))
                    log.info("レート制限。%.2f秒待機 (%d/%d)", wait, attempt,
                             max_attempts)
                    time.sleep(wait * (0.5 + random.random()))
                    delay = min(delay * 2, max_delay)
                    continue

                except BybitError as e:
                    last_exc = e
                    if e.disposition is Disposition.RECONCILE:
                        log.warning("%s: 状態不明。照会します [%d] %s",
                                    fn.__name__, e.ret_code, e.ret_msg)
                        if on_unknown_state:
                            return on_unknown_state(e)
                        raise
                    if e.disposition is not Disposition.RETRY:
                        # FATAL: リトライしても無駄なので即座に上げる
                        log.error("%s: 回復不能 [%d] %s / 送信=%r",
                                  fn.__name__, e.ret_code, e.ret_msg, e.sent)
                        raise
                    log.info("%s: 一時エラー [%d]、再試行 %d/%d",
                             fn.__name__, e.ret_code, attempt, max_attempts)

                # 共通のバックオフ（ジッタ付き）
                if attempt < max_attempts:
                    sleep_for = min(delay, max_delay) * (0.5 + random.random())
                    time.sleep(sleep_for)
                    delay = min(delay * 2, max_delay)

            assert last_exc is not None
            log.error("%s: %d回試行して失敗", fn.__name__, max_attempts)
            raise last_exc

        return wrapper

    return decorator


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format="%(levelname)-7s %(message)s")

    calls = {"n": 0}

    @with_retry(max_attempts=4, base_delay=0.1)
    def flaky_read() -> str:
        calls["n"] += 1
        if calls["n"] < 3:
            raise BybitTransientError(10016, "Server error", endpoint="test")
        return f"成功（{calls['n']}回目）"

    print(flaky_read())

    @with_retry(is_write=True, on_unknown_state=lambda e: "照会にフォールバック")
    def flaky_write() -> str:
        raise BybitUnknownStateError(10016, "Server error", endpoint="place_order")

    print(flaky_write())

"""recover.py — 状態不明になった注文の結末を確定させる
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

前提: 発注時に必ず orderLinkId を自分で付けていること。
      これが無いと「自分が出した注文」を特定できず、照会もできない。
"""
from __future__ import annotations

import logging
import time
from dataclasses import dataclass
from enum import Enum

log = logging.getLogger(__name__)


class OrderOutcome(str, Enum):
    FILLED = "filled"              # 約定済み
    PARTIALLY = "partially"        # 部分約定
    OPEN = "open"                  # 板にある（未約定）
    CANCELLED = "cancelled"        # キャンセル済み
    REJECTED = "rejected"          # 拒否された
    NOT_FOUND = "not_found"        # 存在しない = 届いていなかった


@dataclass
class OrderState:
    outcome: OrderOutcome
    order_id: str | None = None
    filled_qty: float = 0.0
    avg_price: float = 0.0
    status: str = ""

    @property
    def exists(self) -> bool:
        return self.outcome is not OrderOutcome.NOT_FOUND

    @property
    def safe_to_resend(self) -> bool:
        """再送してよいのは「届いていなかった」と確認できた場合のみ。"""
        return self.outcome is OrderOutcome.NOT_FOUND


# V5のOrderStatus -> 結末のマッピング
_STATUS_MAP = {
    "New": OrderOutcome.OPEN,
    "Untriggered": OrderOutcome.OPEN,
    "Triggered": OrderOutcome.OPEN,
    "PartiallyFilled": OrderOutcome.PARTIALLY,
    "Filled": OrderOutcome.FILLED,
    "Cancelled": OrderOutcome.CANCELLED,
    "PartiallyFilledCanceled": OrderOutcome.PARTIALLY,
    "Deactivated": OrderOutcome.CANCELLED,
    "Rejected": OrderOutcome.REJECTED,
}


def resolve_order(http, order_link_id: str, symbol: str,
                  category: str = "linear",
                  attempts: int = 4, delay: float = 1.0) -> OrderState:
    """orderLinkIdで注文の結末を確定させる。

    取引所側の反映に遅延があるため、NOT_FOUNDは数回確認してから確定させる。
    ここを1回で判定すると「実は通っていた」を見逃して二重発注になる。
    """
    for attempt in range(1, attempts + 1):
        # 1) まず未約定注文を探す
        res = http.get_open_orders(category=category, symbol=symbol,
                                   orderLinkId=order_link_id)
        rows = res.get("result", {}).get("list") or []

        # 2) 見つからなければ履歴を探す（約定済み・キャンセル済み）
        if not rows:
            res = http.get_order_history(category=category, symbol=symbol,
                                         orderLinkId=order_link_id)
            rows = res.get("result", {}).get("list") or []

        if rows:
            o = rows[0]
            status = o.get("orderStatus", "")
            state = OrderState(
                outcome=_STATUS_MAP.get(status, OrderOutcome.OPEN),
                order_id=o.get("orderId"),
                filled_qty=float(o.get("cumExecQty") or 0),
                avg_price=float(o.get("avgPrice") or 0),
                status=status,
            )
            log.info("照会: %s -> %s (status=%s, 約定=%s)",
                     order_link_id, state.outcome.value, status, state.filled_qty)
            return state

        if attempt < attempts:
            log.info("照会: %s は未検出。%.1f秒後に再確認 (%d/%d)",
                     order_link_id, delay, attempt, attempts)
            time.sleep(delay)
            delay *= 2

    log.warning("照会: %s は存在しません（届いていなかったと判断）", order_link_id)
    return OrderState(outcome=OrderOutcome.NOT_FOUND)


def place_order_safely(http, params: dict, category: str = "linear",
                       max_resend: int = 1) -> dict:
    """タイムアウトしても二重発注しない発注。

    params には必ず orderLinkId を含めること。
    """
    link_id = params.get("orderLinkId")
    if not link_id:
        raise ValueError("orderLinkIdは必須です（冪等性が確保できません）")

    symbol = params["symbol"]

    for attempt in range(max_resend + 1):
        try:
            res = http.place_order(**params)
            code = res["retCode"]

            if code == 0:
                return res["result"]

            # 重複ID = 既に受理されている。照会して結果を返す
            if code in (110072, 170141, 110030):
                log.info("orderLinkId重複を検出。既存注文を照会します: %s", link_id)
                state = resolve_order(http, link_id, symbol, category)
                return {"orderLinkId": link_id, "orderId": state.order_id,
                        "_recovered": True, "_outcome": state.outcome.value}

            raise RuntimeError(f"発注失敗 [{code}] {res['retMsg']}")

        except Exception as e:
            # 通信エラー・サーバーエラー -> 状態を確定させる
            log.warning("発注が不確定になりました (%s)。照会します",
                        type(e).__name__)
            state = resolve_order(http, link_id, symbol, category)

            if state.exists:
                log.info("実は通っていました: %s", state.outcome.value)
                return {"orderLinkId": link_id, "orderId": state.order_id,
                        "_recovered": True, "_outcome": state.outcome.value}

            if attempt < max_resend:
                log.info("届いていなかったので再送します")
                continue
            raise

    raise RuntimeError("発注を確定できませんでした")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format="%(levelname)-7s %(message)s")
    from client import make_client

    c = make_client()
    result = place_order_safely(c.http, {
        "category": "linear", "symbol": "BTCUSDT", "side": "Buy",
        "orderType": "Limit", "qty": "0.001", "price": "50000",
        "timeInForce": "PostOnly",
        "orderLinkId": f"safe-{int(time.time())}",
    })
    print(result)

"""http_errors.py — HTTPステータスとネットワーク例外の扱い
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import logging

import requests

from bybit_errors import (
    BybitAuthError, BybitError, BybitIpBanError, BybitRateLimitError,
    BybitTransientError, BybitUnknownStateError, raise_for_ret_code,
)

log = logging.getLogger(__name__)


def handle_response(resp: requests.Response, endpoint: str = "",
                    sent=None, *, is_write: bool = False) -> dict:
    """HTTPステータス -> retCode の順に検査する。"""

    # ---- 1) レート制限ヘッダの監視（記事05で活用） -------------------
    remaining = resp.headers.get("X-Bapi-Limit-Status")
    limit = resp.headers.get("X-Bapi-Limit")
    if remaining is not None and limit is not None:
        try:
            if int(remaining) <= 2:
                log.warning("%s: レート残量わずか %s/%s", endpoint, remaining, limit)
        except ValueError:
            pass

    # ---- 2) HTTPステータス --------------------------------------------
    status = resp.status_code

    if status == 403:
        # IP単位のレート制限。約10分の制限がかかる
        raise BybitIpBanError(
            -403, "IP単位のレート制限に到達（約10分の制限）", endpoint=endpoint
        )
    if status == 401:
        raise BybitAuthError(-401, "認証エラー（キー・署名を確認）",
                             endpoint=endpoint)
    if status == 429:
        raise BybitRateLimitError(-429, "リクエスト過多", endpoint=endpoint,
                                  retry_after=float(resp.headers.get("Retry-After", 1)))
    if 500 <= status < 600:
        # サーバー側の障害。書き込みなら通っている可能性がある
        cls = BybitUnknownStateError if is_write else BybitTransientError
        raise cls(-status, f"サーバーエラー HTTP {status}", endpoint=endpoint)
    if status != 200:
        raise BybitError(-status, f"予期しないHTTPステータス {status}: "
                                  f"{resp.text[:200]}", endpoint=endpoint)

    # ---- 3) JSONのパース ---------------------------------------------
    try:
        data = resp.json()
    except ValueError as e:
        # HTMLのエラーページが返るケース（メンテナンス・CDNエラー）
        cls = BybitUnknownStateError if is_write else BybitTransientError
        raise cls(-1, f"JSONとして解釈できない応答: {resp.text[:200]}",
                  endpoint=endpoint) from e

    # ---- 4) retCode ---------------------------------------------------
    return raise_for_ret_code(data, endpoint=endpoint, sent=sent,
                              is_write=is_write)


HTTP_TROUBLE = """
症状ごとの一次切り分け:

  HTTP 403        -> IP単位のレート制限。並列数を落とし10分待つ（記事05）
  HTTP 401        -> キー・署名。接続先とキーの環境が一致しているか（記事02）
  HTTP 5xx        -> 取引所側の障害。書き込みは必ず照会（サンプル3）
  タイムアウト      -> 読み取りは再試行、書き込みは照会
  ConnectionError -> ネットワークまたはDNS。VPSの疎通を確認
  JSONパース失敗    -> メンテナンス中のHTMLページが返っている可能性
"""

if __name__ == "__main__":
    print(HTTP_TROUBLE)

"""notify.py — エラー通知（Discord Webhook）と重複抑制
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import logging
import os
import time
from collections import defaultdict

import requests

from bybit_errors import BybitError, Disposition

log = logging.getLogger(__name__)

WEBHOOK = os.getenv("DISCORD_WEBHOOK_URL", "")

# 同じ種類のエラーは、この秒数内は再通知しない
COOLDOWN_SEC = {
    Disposition.FATAL: 300.0,
    Disposition.RECONCILE: 60.0,
    Disposition.RETRY: 1800.0,     # 一時エラーは頻繁なので長めに抑制
    Disposition.BENIGN: 0.0,       # 通知しない
}

# 機密情報のマスク対象（キーが含まれる可能性のあるフィールド）
SECRET_KEYS = {"api_key", "apiKey", "api_secret", "sign", "X-BAPI-SIGN",
               "X-BAPI-API-KEY", "secret", "password"}


def mask(obj):
    """送信パラメータからキー・署名を除去する。ログに残す前に必ず通す。"""
    if isinstance(obj, dict):
        return {k: ("****" if k in SECRET_KEYS else mask(v))
                for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [mask(v) for v in obj]
    return obj


class Notifier:
    def __init__(self, webhook: str = WEBHOOK, enabled: bool = True):
        self.webhook = webhook
        self.enabled = enabled and bool(webhook)
        self._last_sent: dict[str, float] = {}
        self._suppressed: dict[str, int] = defaultdict(int)

    def _key(self, err: BybitError) -> str:
        return f"{type(err).__name__}:{err.ret_code}:{err.endpoint}"

    def send(self, text: str, *, force: bool = False) -> bool:
        if not self.enabled and not force:
            log.info("[通知(無効)] %s", text)
            return False
        try:
            r = requests.post(self.webhook, json={"content": text[:1900]},
                              timeout=10)
            return r.status_code < 300
        except Exception as e:
            log.error("通知の送信に失敗: %s", e)
            return False

    def on_error(self, err: BybitError) -> bool:
        """エラーを通知する。分類ごとにクールダウンを適用する。"""
        cooldown = COOLDOWN_SEC.get(err.disposition, 300.0)
        if cooldown <= 0:
            return False

        key = self._key(err)
        now = time.monotonic()
        last = self._last_sent.get(key, 0.0)

        if now - last < cooldown:
            self._suppressed[key] += 1
            return False

        suppressed = self._suppressed.pop(key, 0)
        self._last_sent[key] = now

        icon = {"fatal": "🛑", "verify": "⚠️", "retry": "🔁"}.get(
            err.disposition.value, "ℹ️")
        lines = [
            f"{icon} **Bybit APIエラー** `{err.disposition.value}`",
            f"retCode: `{err.ret_code}`  retMsg: `{err.ret_msg}`",
            f"endpoint: `{err.endpoint}`",
        ]
        if err.sent:
            lines.append(f"送信: `{mask(err.sent)}`")
        if suppressed:
            lines.append(f"_（同種のエラーを{suppressed}件抑制しました）_")

        return self.send("\n".join(lines))


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    from bybit_errors import BybitAuthError

    n = Notifier(enabled=False)
    err = BybitAuthError(10004, "error sign", endpoint="place_order",
                         sent={"symbol": "BTCUSDT", "api_key": "SECRET123"})
    n.on_error(err)
    n.on_error(err)      # クールダウン中なので抑制される
    print("マスク確認:", mask({"api_key": "abc", "symbol": "BTCUSDT"}))

"""error_stats.py — retCodeの発生統計を取り、改善すべき箇所を特定する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import json
import sqlite3
import time
from pathlib import Path

from bybit_errors import BybitError, classify, describe

DB = Path("data/error_log.sqlite")

SCHEMA = """
CREATE TABLE IF NOT EXISTS errors (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    ts         REAL    NOT NULL,
    ret_code   INTEGER NOT NULL,
    ret_msg    TEXT,
    endpoint   TEXT,
    disposition TEXT,
    sent       TEXT
);
CREATE INDEX IF NOT EXISTS idx_errors_ts ON errors(ts);
CREATE INDEX IF NOT EXISTS idx_errors_code ON errors(ret_code);
"""


class ErrorRecorder:
    def __init__(self, path: Path = DB):
        path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(str(path))
        self.conn.executescript(SCHEMA)
        self.conn.commit()

    def record(self, err: BybitError) -> None:
        from notify import mask

        self.conn.execute(
            "INSERT INTO errors (ts, ret_code, ret_msg, endpoint, disposition, sent)"
            " VALUES (?, ?, ?, ?, ?, ?)",
            (time.time(), err.ret_code, err.ret_msg, err.endpoint,
             err.disposition.value,
             json.dumps(mask(err.sent), ensure_ascii=False, default=str)),
        )
        self.conn.commit()

    def summary(self, hours: float = 24.0) -> list[tuple]:
        since = time.time() - hours * 3600
        cur = self.conn.execute(
            "SELECT ret_code, endpoint, disposition, COUNT(*) AS n,"
            "       MAX(ret_msg) AS sample"
            "  FROM errors WHERE ts >= ?"
            " GROUP BY ret_code, endpoint"
            " ORDER BY n DESC",
            (since,),
        )
        return cur.fetchall()

    def report(self, hours: float = 24.0) -> None:
        rows = self.summary(hours)
        if not rows:
            print(f"直近{hours:.0f}時間のエラーはありません")
            return

        total = sum(r[3] for r in rows)
        print(f"直近{hours:.0f}時間: 合計{total}件\n")
        print(f"{'code':>8} {'endpoint':<22}{'分類':<10}{'件数':>7}  メッセージ")
        print("-" * 88)
        for code, endpoint, disp, n, sample in rows[:25]:
            print(f"{code:>8} {(endpoint or '-'):<22}{disp:<10}{n:>7}  "
                  f"{(sample or '')[:34]}")

        print("\n--- 改善の指針 ---")
        by_code: dict[int, int] = {}
        for code, _, _, n, _ in rows:
            by_code[code] = by_code.get(code, 0) + n

        hints = {
            110017: "数量の丸め処理を見直す（記事13の発注前検証）",
            110094: "最小注文金額のチェックを追加する",
            110007: "未約定注文が証拠金を拘束していないか確認",
            110020: "有効注文数が上限に近い。注文の寿命管理を追加",
            10006: "レート制限。トークンバケットで流量制御（記事05）",
            10018: "IPレート制限。並列数を削減する",
            10002: "時計のズレ。NTP同期を設定する（記事38）",
            10004: "署名の生成を確認（記事02）",
            110072: "冪等性が機能している。照会処理があるか確認（記事17）",
        }
        for code, n in sorted(by_code.items(), key=lambda kv: -kv[1])[:8]:
            hint = hints.get(code)
            print(f"  {describe(code)}  ({n}件)")
            if hint:
                print(f"      -> {hint}")


if __name__ == "__main__":
    from bybit_errors import BybitParamError, BybitRateLimitError

    rec = ErrorRecorder()
    rec.record(BybitParamError(110017, "OrderQty will be truncated to zero",
                               endpoint="place_order",
                               sent={"qty": "0.0000001", "api_key": "SECRET"}))
    rec.record(BybitRateLimitError(10006, "Too many visits",
                                   endpoint="get_tickers"))
    rec.report()

"""diagnose.py — 「動かない」ときの一次切り分けを自動化する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time

import requests

from config import REST_BASE, load_credentials


def check(name: str, fn) -> bool:
    try:
        detail = fn()
        print(f"  [OK]   {name}" + (f" — {detail}" if detail else ""))
        return True
    except Exception as e:
        print(f"  [NG]   {name} — {type(e).__name__}: {e}")
        return False


def diagnose() -> None:
    cred = load_credentials()
    base = cred.base_url
    print(f"環境: {cred.env.value}  接続先: {base}\n")

    results: list[bool] = []

    # 1) ネットワーク（公開エンドポイント・認証なし）
    def net():
        r = requests.get(f"{base}/v5/market/time", timeout=10)
        r.raise_for_status()
        return f"HTTP {r.status_code}"

    results.append(check("1. ネットワーク疎通", net))

    # 2) 時計のズレ
    def clock():
        t0 = time.time()
        r = requests.get(f"{base}/v5/market/time", timeout=10)
        t1 = time.time()
        server_ms = int(r.json()["result"]["timeNano"]) / 1e6
        skew = server_ms - (t0 + t1) / 2 * 1000
        if abs(skew) > 2500:
            raise RuntimeError(f"ズレ {skew:+.0f}ms（recv_window超過の危険 -> 10002）")
        return f"ズレ {skew:+.0f}ms"

    results.append(check("2. 時計の同期", clock))

    # 3) 認証（署名が正しいか）
    from client import make_client

    c = make_client()

    def auth():
        res = c.http.get_api_key_information()
        if res["retCode"] != 0:
            raise RuntimeError(f"retCode={res['retCode']} {res['retMsg']}")
        info = res["result"]
        return (f"IP制限={info.get('ips') or 'なし'} "
                f"期限={info.get('expiredAt') or '無期限'}")

    results.append(check("3. 認証（キー・署名）", auth))

    # 4) 権限
    def perms():
        info = c.http.get_api_key_information()["result"]
        if info.get("readOnly") == 1:
            raise RuntimeError("Read-Onlyキーです（発注はできません）")
        active = {k: v for k, v in (info.get("permissions") or {}).items() if v}
        return str(active)

    results.append(check("4. キーの権限", perms))

    # 5) 口座種別
    def account():
        res = c.http.get_wallet_balance(accountType="UNIFIED")
        if res["retCode"] != 0:
            raise RuntimeError(f"retCode={res['retCode']} {res['retMsg']} "
                               f"(accountType=CONTRACTかもしれません)")
        eq = res["result"]["list"][0]["totalEquity"]
        return f"統合口座 総資産={eq}"

    results.append(check("5. 口座（残高取得）", account))

    # 6) 銘柄仕様
    def symbol():
        res = c.http.get_instruments_info(category="linear", symbol="BTCUSDT")
        row = res["result"]["list"][0]
        return (f"tickSize={row['priceFilter']['tickSize']} "
                f"qtyStep={row['lotSizeFilter']['qtyStep']} "
                f"status={row.get('status')}")

    results.append(check("6. 銘柄仕様の取得", symbol))

    print()
    if all(results):
        print("すべて正常です。問題は個別のリクエストのパラメータにあります。")
        print("失敗したリクエストの送信パラメータとretCodeをログから確認してください。")
    else:
        first_ng = results.index(False) + 1
        print(f"最初に失敗したのは項目{first_ng}です。ここを先に解決してください。")
        print("（上流の項目が失敗している状態で下流を調べても意味がありません）")


if __name__ == "__main__":
    diagnose()