Bybit APIのレート制限を攻略するPython実装
X-Bapi-Limitヘッダ・
トークンバケット・指数バックオフ

"""rate_headers.py — レート制限ヘッダを読んで自律的に減速する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field


@dataclass
class RateStatus:
    """1エンドポイントの最新レート状況。"""

    limit: int = 0
    remaining: int = 0
    reset_at_ms: int = 0
    observed_at: float = field(default_factory=time.time)

    @property
    def usage_ratio(self) -> float:
        """使用率。1.0に近いほど危険。"""
        if self.limit <= 0:
            return 0.0
        return 1.0 - (self.remaining / self.limit)

    @property
    def seconds_to_reset(self) -> float:
        if self.reset_at_ms <= 0:
            return 0.0
        return max(0.0, self.reset_at_ms / 1000 - time.time())


class RateHeaderMonitor:
    """全エンドポイントのレート状況を保持し、必要なら待機時間を返す。"""

    # 残量がこの割合を下回ったら自主的に減速する
    SLOW_DOWN_RATIO = 0.2

    def __init__(self, verbose: bool = True):
        self.status: dict[str, RateStatus] = {}
        self.verbose = verbose
        self.slowdowns = 0

    def observe(self, endpoint: str, headers) -> RateStatus | None:
        """レスポンスヘッダを取り込む。headersはdictでもCaseInsensitiveDictでも可。"""

        def get(name: str) -> str | None:
            return headers.get(name) or headers.get(name.lower())

        raw_limit = get("X-Bapi-Limit")
        raw_remaining = get("X-Bapi-Limit-Status")
        if raw_limit is None or raw_remaining is None:
            return None       # 公開エンドポイントなど、ヘッダが無い場合

        try:
            st = RateStatus(
                limit=int(raw_limit),
                remaining=int(raw_remaining),
                reset_at_ms=int(get("X-Bapi-Limit-Reset-Timestamp") or 0),
            )
        except ValueError:
            return None

        self.status[endpoint] = st

        if st.limit > 0 and st.remaining / st.limit < self.SLOW_DOWN_RATIO:
            self.slowdowns += 1
            if self.verbose:
                print(f"[rate] {endpoint}: 残 {st.remaining}/{st.limit} "
                      f"(使用率 {st.usage_ratio:.0%}) -> 減速します")
        return st

    def suggested_wait(self, endpoint: str) -> float:
        """このエンドポイントを次に呼ぶ前に待つべき秒数。"""
        st = self.status.get(endpoint)
        if st is None or st.limit <= 0:
            return 0.0

        # 残量がゼロならリセットまで待つ
        if st.remaining <= 0:
            return min(st.seconds_to_reset + 0.05, 5.0)

        # 残量が少ないほど長く待つ（線形に減速）
        if st.remaining / st.limit < self.SLOW_DOWN_RATIO:
            return (1.0 - st.remaining / st.limit) * 0.5
        return 0.0

    def report(self) -> None:
        if not self.status:
            print("レート情報がまだありません")
            return
        print(f"{'endpoint':<40}{'残/上限':>12}{'使用率':>8}")
        print("-" * 62)
        for ep, st in sorted(self.status.items(),
                             key=lambda kv: -kv[1].usage_ratio):
            print(f"{ep:<40}{st.remaining:>6}/{st.limit:<5}"
                  f"{st.usage_ratio:>7.0%}")
        print(f"\n減速の発生回数: {self.slowdowns}")


if __name__ == "__main__":
    mon = RateHeaderMonitor()
    # 実際のレスポンスヘッダを模した例
    mon.observe("/v5/order/create", {
        "X-Bapi-Limit": "20",
        "X-Bapi-Limit-Status": "3",
        "X-Bapi-Limit-Reset-Timestamp": str(int((time.time() + 1) * 1000)),
    })
    mon.report()
    print(f"\n次の呼び出しまでの推奨待機: "
          f"{mon.suggested_wait('/v5/order/create'):.3f}秒")

"""batch_planner.py — バッチ注文のレート消費を見積もり、実行計画を立てる
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import math
from dataclasses import dataclass

# 1リクエストで送れる注文本数の上限（categoryごと。公式Docsで確認すること）
MAX_PER_BATCH = {"linear": 20, "inverse": 20, "spot": 10, "option": 20}

# 発注系の秒間上限
CREATE_RPS = {"linear": 20, "spot": 20, "inverse": 10, "option": 20}


@dataclass
class BatchPlan:
    category: str
    total_orders: int
    chunks: list[int]
    estimated_seconds: float
    rate_cost: int

    def describe(self) -> str:
        return (f"{self.category}: {self.total_orders}本を"
                f"{len(self.chunks)}リクエストに分割\n"
                f"  レート消費 : {self.rate_cost}件分"
                f"（上限 {CREATE_RPS[self.category]}/秒）\n"
                f"  所要時間   : 約{self.estimated_seconds:.1f}秒\n"
                f"  分割       : {self.chunks}")


def plan_batch(total_orders: int, category: str = "linear",
               safety: float = 0.7) -> BatchPlan:
    """発注本数から、分割方法と所要時間を見積もる。

    safety は上限に対する使用率。0.7なら上限の70%で流す。
    """
    if category not in MAX_PER_BATCH:
        raise ValueError(f"未知のcategory: {category}")

    per_batch = MAX_PER_BATCH[category]
    chunks = [per_batch] * (total_orders // per_batch)
    if total_orders % per_batch:
        chunks.append(total_orders % per_batch)

    effective_rps = CREATE_RPS[category] * safety
    # 消費は「本数」で決まるので、本数 / 実効rps が所要時間
    seconds = total_orders / effective_rps if effective_rps > 0 else float("inf")

    return BatchPlan(category=category, total_orders=total_orders,
                     chunks=chunks, estimated_seconds=seconds,
                     rate_cost=total_orders)


def chunk_orders(orders: list[dict], category: str = "linear") -> list[list[dict]]:
    """注文リストを、1リクエストの上限本数で分割する。"""
    size = MAX_PER_BATCH[category]
    return [orders[i:i + size] for i in range(0, len(orders), size)]


if __name__ == "__main__":
    for n, cat in ((10, "linear"), (50, "linear"), (50, "spot"),
                   (200, "linear")):
        print(plan_batch(n, cat).describe())
        print()

    print("--- グリッドbotの現実的な見積り ---")
    p = plan_batch(60, "linear")
    print(f"60本のグリッドを張り直すのに約{p.estimated_seconds:.1f}秒。")
    print("この間に相場が動くため、価格を都度再計算する設計が必要"
          "（記事36ではamend_batchで価格だけ更新する方式を使う）")

"""token_bucket.py — スレッドセーフなトークンバケットと階層的レートゲート
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field


class TokenBucket:
    """一定速度で補充されるトークンバケット（スレッドセーフ）。

    rate     : 1秒あたりの補充トークン数
    capacity : 溜められる最大トークン数（バースト許容量）
    """

    def __init__(self, rate: float, capacity: float | None = None,
                 name: str = ""):
        if rate <= 0:
            raise ValueError("rateは正の値でなければなりません")
        self.rate = rate
        self.capacity = capacity if capacity is not None else rate
        self.name = name
        self._tokens = float(self.capacity)
        self._last = time.monotonic()
        self._lock = threading.Lock()
        self.total_waited = 0.0
        self.total_acquired = 0

    def _refill(self, now: float) -> None:
        elapsed = now - self._last
        if elapsed > 0:
            self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
            self._last = now

    def acquire(self, cost: float = 1.0, timeout: float | None = None) -> float:
        """トークンを取得する。足りなければ待つ。待った秒数を返す。

        cost にバッチの本数を渡せば、消費量を正しく反映できる。
        """
        if cost > self.capacity:
            raise ValueError(
                f"{self.name}: cost({cost})がcapacity({self.capacity})を超えています。"
                f"分割してください"
            )

        deadline = None if timeout is None else time.monotonic() + timeout
        waited = 0.0

        while True:
            with self._lock:
                now = time.monotonic()
                self._refill(now)

                if self._tokens >= cost:
                    self._tokens -= cost
                    self.total_acquired += 1
                    self.total_waited += waited
                    return waited

                # 足りない分が補充されるまでの時間
                need = cost - self._tokens
                sleep_for = need / self.rate

            if deadline is not None and time.monotonic() + sleep_for > deadline:
                raise TimeoutError(f"{self.name}: トークン取得がタイムアウト")

            time.sleep(min(sleep_for, 0.25))    # 細かく刻んで応答性を保つ
            waited += min(sleep_for, 0.25)

    @property
    def available(self) -> float:
        with self._lock:
            self._refill(time.monotonic())
            return self._tokens


@dataclass
class RateGate:
    """IP全体（グローバル）とエンドポイント別の二層でトークンを取る。

    Bybitの制限構造をそのまま写した設計。
    """

    # IP制限は5秒600件。安全側に倒して秒間40件・バースト80件にする
    global_rate: float = 40.0
    global_capacity: float = 80.0
    default_endpoint_rate: float = 10.0

    # エンドポイントごとの上限（公式の値から安全マージンを引いたもの）
    endpoint_rates: dict[str, float] = field(default_factory=lambda: {
        "/v5/order/create": 15.0,
        "/v5/order/amend": 8.0,
        "/v5/order/cancel": 15.0,
        "/v5/order/cancel-all": 8.0,
        "/v5/order/create-batch": 15.0,
        "/v5/order/amend-batch": 15.0,
        "/v5/order/cancel-batch": 15.0,
        "/v5/position/list": 40.0,
        "/v5/position/set-leverage": 8.0,
        "/v5/position/trading-stop": 8.0,
        "/v5/account/wallet-balance": 40.0,
        "/v5/account/transaction-log": 20.0,
        "/v5/account/fee-rate": 4.0,
        "/v5/market/instruments-info": 8.0,
    })

    def __post_init__(self) -> None:
        self._global = TokenBucket(self.global_rate, self.global_capacity,
                                   name="GLOBAL")
        self._buckets: dict[str, TokenBucket] = {}
        self._lock = threading.Lock()

    def bucket(self, endpoint: str) -> TokenBucket:
        with self._lock:
            if endpoint not in self._buckets:
                rate = self.endpoint_rates.get(endpoint,
                                               self.default_endpoint_rate)
                self._buckets[endpoint] = TokenBucket(rate, rate,
                                                      name=endpoint)
            return self._buckets[endpoint]

    def acquire(self, endpoint: str, cost: float = 1.0) -> float:
        """エンドポイント別 -> グローバル の順に取得する。"""
        waited = self.bucket(endpoint).acquire(cost)
        waited += self._global.acquire(cost)
        return waited

    def report(self) -> None:
        print(f"{'bucket':<38}{'残':>8}{'取得数':>8}{'待機合計':>10}")
        print("-" * 66)
        b = self._global
        print(f"{'GLOBAL (IP)':<38}{b.available:>8.1f}{b.total_acquired:>8}"
              f"{b.total_waited:>9.2f}s")
        for ep, bk in sorted(self._buckets.items(),
                             key=lambda kv: -kv[1].total_waited):
            print(f"{ep:<38}{bk.available:>8.1f}{bk.total_acquired:>8}"
                  f"{bk.total_waited:>9.2f}s")


if __name__ == "__main__":
    gate = RateGate()

    started = time.monotonic()
    for i in range(30):
        w = gate.acquire("/v5/order/create")
        if w > 0:
            print(f"  {i:>2}回目: {w:.3f}秒待機")
    print(f"\n30リクエストに{time.monotonic() - started:.2f}秒 "
          f"(上限15/秒なので約2秒が妥当)\n")

    # バッチは本数分のコストを渡す
    print("バッチ20本を発注:")
    w = gate.acquire("/v5/order/create-batch", cost=20)
    print(f"  {w:.3f}秒待機（20件分のトークンを消費）\n")

    gate.report()

"""gated_http.py — pybitのHTTPをラップして全リクエストを流量制御下に置く
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import functools
import time
from collections.abc import Callable

from pybit.unified_trading import HTTP

from rate_headers import RateHeaderMonitor
from token_bucket import RateGate

# pybitのメソッド名 -> エンドポイントパス
# バケットをエンドポイント単位で分けるために必要
METHOD_ENDPOINT: dict[str, str] = {
    "place_order": "/v5/order/create",
    "amend_order": "/v5/order/amend",
    "cancel_order": "/v5/order/cancel",
    "cancel_all_orders": "/v5/order/cancel-all",
    "place_batch_order": "/v5/order/create-batch",
    "amend_batch_order": "/v5/order/amend-batch",
    "cancel_batch_order": "/v5/order/cancel-batch",
    "get_open_orders": "/v5/order/realtime",
    "get_order_history": "/v5/order/history",
    "get_executions": "/v5/execution/list",
    "get_positions": "/v5/position/list",
    "set_leverage": "/v5/position/set-leverage",
    "set_trading_stop": "/v5/position/trading-stop",
    "get_closed_pnl": "/v5/position/closed-pnl",
    "get_wallet_balance": "/v5/account/wallet-balance",
    "get_fee_rates": "/v5/account/fee-rate",
    "get_transaction_log": "/v5/account/transaction-log",
    "get_instruments_info": "/v5/market/instruments-info",
    "get_kline": "/v5/market/kline",
    "get_tickers": "/v5/market/tickers",
    "get_orderbook": "/v5/market/orderbook",
    "get_funding_rate_history": "/v5/market/history-fund-rate",
}

# バッチ系メソッドの「本数」を取り出すためのキー
BATCH_KEYS = ("request",)


def batch_cost(kwargs: dict) -> int:
    """バッチリクエストの注文本数を数える。無ければ1。"""
    for key in BATCH_KEYS:
        items = kwargs.get(key)
        if isinstance(items, list):
            return max(1, len(items))
    return 1


class GatedHTTP:
    """pybitのHTTPと同じインターフェースで、流量制御を挟む。

    使い方:
        c = make_client()
        http = GatedHTTP(c.http)
        http.get_tickers(category="linear")     # 通常どおり呼べる
    """

    def __init__(self, http: HTTP, gate: RateGate | None = None,
                 monitor: RateHeaderMonitor | None = None,
                 obey_headers: bool = True):
        self._http = http
        self.gate = gate or RateGate()
        self.monitor = monitor or RateHeaderMonitor()
        self.obey_headers = obey_headers
        self._wrapped: dict[str, Callable] = {}

    def __getattr__(self, name: str):
        # HTTP以外の属性はそのまま通す
        attr = getattr(self._http, name)
        if not callable(attr):
            return attr
        if name in self._wrapped:
            return self._wrapped[name]

        endpoint = METHOD_ENDPOINT.get(name, f"/unknown/{name}")

        @functools.wraps(attr)
        def wrapper(*args, **kwargs):
            cost = batch_cost(kwargs)

            # 1) トークンを取る（足りなければここで待つ）
            self.gate.acquire(endpoint, cost=cost)

            # 2) 直近のヘッダ情報が「危険」なら追加で待つ
            if self.obey_headers:
                extra = self.monitor.suggested_wait(endpoint)
                if extra > 0:
                    time.sleep(extra)

            result = attr(*args, **kwargs)

            # 3) レスポンスヘッダを取り込む
            #    pybitは return_response_headers=True のとき (result, headers)
            #    のタプルを返す。単体ではdictのみ。
            if isinstance(result, tuple) and len(result) == 2:
                body, headers = result
                self.monitor.observe(endpoint, headers)
                return body
            return result

        self._wrapped[name] = wrapper
        return wrapper

    def report(self) -> None:
        print("=== トークンバケット ===")
        self.gate.report()
        print("\n=== レート制限ヘッダ ===")
        self.monitor.report()


def make_gated_client(env: str | None = None) -> GatedHTTP:
    """ヘッダを取得できるようにHTTPを構成して包む。"""
    from config import load_credentials

    cred = load_credentials(env)
    http = HTTP(
        testnet=cred.testnet, demo=cred.demo,
        api_key=cred.api_key, api_secret=cred.api_secret,
        recv_window=5000, timeout=10,
        return_response_headers=True,   # ヘッダ監視のために必要
    )
    return GatedHTTP(http)


if __name__ == "__main__":
    http = make_gated_client()

    started = time.monotonic()
    for _ in range(25):
        http.get_tickers(category="linear", symbol="BTCUSDT")
    print(f"25リクエスト: {time.monotonic() - started:.2f}秒\n")

    http.report()

"""backoff.py — レート制限に特化した指数バックオフ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import functools
import random
import time
from collections.abc import Callable

import requests

RATE_LIMIT_CODES = {10006, 170222}     # UID単位のレート制限
IP_LIMIT_CODES = {10018}               # IP単位のレート制限
IP_BAN_WAIT = 620.0                    # 約10分 + 余裕


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


def decorrelated_jitter(base: float, cap: float, previous: float) -> float:
    """AWSが推奨する decorrelated jitter。
    単純な full jitter より収束が良く、群れになりにくい。
    """
    return min(cap, random.uniform(base, previous * 3))


def with_rate_backoff(max_attempts: int = 6, base: float = 0.2,
                      cap: float = 20.0, verbose: bool = True) -> Callable:
    """レート制限に対してのみバックオフする（他のエラーは即座に上げる）。"""

    def decorator(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            sleep_for = base

            for attempt in range(1, max_attempts + 1):
                try:
                    result = fn(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    resp = getattr(e, "response", None)
                    if resp is not None and resp.status_code == 403:
                        if verbose:
                            print(f"[rate] HTTP 403（IP制限）。"
                                  f"{IP_BAN_WAIT:.0f}秒待機します")
                        time.sleep(IP_BAN_WAIT)
                        continue
                    raise
                else:
                    # pybitはretCodeを辞書で返すので自分で確認する
                    code = result.get("retCode") if isinstance(result, dict) else 0

                    if code in IP_LIMIT_CODES:
                        if verbose:
                            print(f"[rate] retCode={code}（IP制限）。"
                                  f"{IP_BAN_WAIT:.0f}秒待機します")
                        time.sleep(IP_BAN_WAIT)
                        continue

                    if code in RATE_LIMIT_CODES:
                        if attempt >= max_attempts:
                            raise RateLimited(code,
                                              result.get("retMsg", ""))
                        sleep_for = decorrelated_jitter(base, cap, sleep_for)
                        if verbose:
                            print(f"[rate] retCode={code}、{sleep_for:.2f}秒後に"
                                  f"再試行 ({attempt}/{max_attempts})")
                        time.sleep(sleep_for)
                        continue

                    return result

            raise RateLimited(-1, f"{max_attempts}回試行して失敗")

        return wrapper

    return decorator


if __name__ == "__main__":
    # バックオフの間隔がどう伸びるかを確認する
    prev = 0.2
    print("decorrelated jitter の待機時間（cap=20秒）:")
    for i in range(1, 9):
        prev = decorrelated_jitter(0.2, 20.0, prev)
        print(f"  {i}回目: {prev:6.2f}秒")

    calls = {"n": 0}

    @with_rate_backoff(max_attempts=5, base=0.05)
    def flaky():
        calls["n"] += 1
        if calls["n"] < 3:
            return {"retCode": 10006, "retMsg": "Too many visits"}
        return {"retCode": 0, "result": {"ok": True}}

    print("\n実行:", flaky(), f"（{calls['n']}回目で成功）")

"""async_gate.py — asyncio向けのトークンバケットと流量ゲート
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import asyncio
import time


class AsyncTokenBucket:
    """asyncio用トークンバケット。待機中もイベントループを止めない。"""

    def __init__(self, rate: float, capacity: float | None = None,
                 name: str = ""):
        self.rate = rate
        self.capacity = capacity if capacity is not None else rate
        self.name = name
        self._tokens = float(self.capacity)
        self._last = time.monotonic()
        self._lock = asyncio.Lock()
        self.total_waited = 0.0

    async def acquire(self, cost: float = 1.0) -> float:
        if cost > self.capacity:
            raise ValueError(f"{self.name}: cost({cost}) > capacity")

        waited = 0.0
        while True:
            async with self._lock:
                now = time.monotonic()
                elapsed = now - self._last
                if elapsed > 0:
                    self._tokens = min(self.capacity,
                                       self._tokens + elapsed * self.rate)
                    self._last = now

                if self._tokens >= cost:
                    self._tokens -= cost
                    self.total_waited += waited
                    return waited

                sleep_for = (cost - self._tokens) / self.rate

            await asyncio.sleep(min(sleep_for, 0.25))
            waited += min(sleep_for, 0.25)


class AsyncRateGate:
    """IP全体とエンドポイント別の二層ゲート（非同期版）。"""

    def __init__(self, global_rate: float = 40.0,
                 endpoint_rates: dict[str, float] | None = None,
                 default_rate: float = 10.0,
                 max_concurrency: int = 8):
        self._global = AsyncTokenBucket(global_rate, global_rate * 2, "GLOBAL")
        self._rates = endpoint_rates or {}
        self._default = default_rate
        self._buckets: dict[str, AsyncTokenBucket] = {}
        # 同時実行数の上限。ソケットとメモリを守るために別途必要
        self._sem = asyncio.Semaphore(max_concurrency)

    def _bucket(self, endpoint: str) -> AsyncTokenBucket:
        if endpoint not in self._buckets:
            rate = self._rates.get(endpoint, self._default)
            self._buckets[endpoint] = AsyncTokenBucket(rate, rate, endpoint)
        return self._buckets[endpoint]

    async def acquire(self, endpoint: str, cost: float = 1.0) -> float:
        waited = await self._bucket(endpoint).acquire(cost)
        waited += await self._global.acquire(cost)
        return waited

    def limited(self, endpoint: str, cost: float = 1.0):
        """async with で使うコンテキストマネージャ。"""
        gate = self

        class _Ctx:
            async def __aenter__(self):
                await gate._sem.acquire()
                try:
                    await gate.acquire(endpoint, cost)
                except BaseException:
                    gate._sem.release()
                    raise
                return gate

            async def __aexit__(self, *exc):
                gate._sem.release()
                return False

        return _Ctx()


async def demo() -> None:
    gate = AsyncRateGate(global_rate=20.0,
                         endpoint_rates={"/v5/market/tickers": 10.0})

    async def task(i: int) -> None:
        async with gate.limited("/v5/market/tickers"):
            await asyncio.sleep(0.05)      # 実際のHTTPリクエストの代わり
            print(f"  リクエスト {i:>2} 完了 ({time.monotonic() - t0:.2f}秒)")

    t0 = time.monotonic()
    await asyncio.gather(*(task(i) for i in range(20)))
    print(f"\n20リクエストに {time.monotonic() - t0:.2f}秒 "
          f"(上限10/秒なので約2秒が妥当)")


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

"""rate_profiler.py — エンドポイント別のリクエスト消費を計測・可視化する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import threading
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field


@dataclass
class EndpointStat:
    calls: int = 0
    cost: int = 0
    waited: float = 0.0
    errors: int = 0
    recent: deque = field(default_factory=lambda: deque(maxlen=2000))

    def rps(self, window: float = 5.0) -> float:
        """直近window秒の実効リクエスト毎秒。"""
        now = time.monotonic()
        n = sum(1 for t in self.recent if now - t <= window)
        return n / window


class RateProfiler:
    def __init__(self):
        self.stats: dict[str, EndpointStat] = defaultdict(EndpointStat)
        self._lock = threading.Lock()
        self.started = time.monotonic()

    def record(self, endpoint: str, cost: int = 1, waited: float = 0.0,
               error: bool = False) -> None:
        with self._lock:
            s = self.stats[endpoint]
            s.calls += 1
            s.cost += cost
            s.waited += waited
            if error:
                s.errors += 1
            s.recent.append(time.monotonic())

    def total_rps(self, window: float = 5.0) -> float:
        return sum(s.rps(window) for s in self.stats.values())

    def report(self, top: int = 15) -> None:
        elapsed = time.monotonic() - self.started
        with self._lock:
            rows = sorted(self.stats.items(), key=lambda kv: -kv[1].cost)

        total_cost = sum(s.cost for _, s in rows)
        print(f"稼働 {elapsed:.0f}秒 / 総消費 {total_cost}件 / "
              f"平均 {total_cost / max(elapsed, 1):.1f} 件/秒")
        print(f"直近5秒の実効レート: {self.total_rps():.1f} 件/秒 "
              f"(IP上限の目安 120件/秒)\n")

        print(f"{'endpoint':<36}{'呼出':>7}{'消費':>7}{'割合':>7}"
              f"{'直近rps':>9}{'待機':>9}{'err':>5}")
        print("-" * 82)
        for ep, s in rows[:top]:
            share = s.cost / total_cost if total_cost else 0
            print(f"{ep:<36}{s.calls:>7}{s.cost:>7}{share:>6.0%}"
                  f"{s.rps():>9.1f}{s.waited:>8.1f}s{s.errors:>5}")

        # 改善の指針を出す
        print("\n--- 改善の指針 ---")
        for ep, s in rows[:3]:
            share = s.cost / total_cost if total_cost else 0
            if share < 0.15:
                continue
            hint = {
                "/v5/market/tickers": "WebSocketのtickersトピックに移行（記事26）",
                "/v5/position/list": "WebSocketのpositionトピックに移行（記事27）",
                "/v5/order/realtime": "WebSocketのorderトピックに移行（記事27）",
                "/v5/market/orderbook": "WebSocketのorderbookに移行（記事28）",
                "/v5/market/instruments-info": "起動時に取得してキャッシュ（記事03）",
                "/v5/account/fee-rate": "起動時に1回だけ取得してキャッシュ",
                "/v5/account/wallet-balance": "WebSocketのwalletトピックに移行",
            }.get(ep)
            print(f"  {ep} が全体の{share:.0%}を消費")
            if hint:
                print(f"      -> {hint}")


if __name__ == "__main__":
    prof = RateProfiler()

    # 典型的な「悪い」ポーリングbotの消費パターンを模擬する
    for _ in range(300):
        prof.record("/v5/market/tickers")
    for _ in range(180):
        prof.record("/v5/position/list")
    for _ in range(120):
        prof.record("/v5/order/realtime")
    for _ in range(40):
        prof.record("/v5/market/instruments-info")
    for _ in range(12):
        prof.record("/v5/order/create", cost=1, waited=0.3)
    prof.record("/v5/order/create-batch", cost=20, waited=1.2)

    prof.report()

"""ttl_cache.py — 変化しないデータをキャッシュしてリクエストを削る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import functools
import json
import threading
import time
from pathlib import Path
from typing import Any


class TtlCache:
    """メモリ + ディスクの二段キャッシュ。プロセス再起動でも効く。"""

    def __init__(self, path: str | Path = "data/api_cache.json",
                 default_ttl: float = 3600.0):
        self.path = Path(path)
        self.default_ttl = default_ttl
        self._mem: dict[str, tuple[float, Any]] = {}
        self._lock = threading.Lock()
        self.hits = 0
        self.misses = 0
        self._load()

    def _load(self) -> None:
        if not self.path.exists():
            return
        try:
            raw = json.loads(self.path.read_text(encoding="utf-8"))
            self._mem = {k: (v["exp"], v["val"]) for k, v in raw.items()}
        except Exception:
            self._mem = {}

    def _save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        data = {k: {"exp": exp, "val": val}
                for k, (exp, val) in self._mem.items()}
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
        tmp.replace(self.path)

    def get(self, key: str) -> Any | None:
        with self._lock:
            entry = self._mem.get(key)
            if entry is None or entry[0] < time.time():
                self.misses += 1
                return None
            self.hits += 1
            return entry[1]

    def set(self, key: str, value: Any, ttl: float | None = None) -> None:
        with self._lock:
            self._mem[key] = (time.time() + (ttl or self.default_ttl), value)
            self._save()

    def stats(self) -> str:
        total = self.hits + self.misses
        rate = self.hits / total if total else 0
        return (f"キャッシュ: hit={self.hits} miss={self.misses} "
                f"ヒット率={rate:.0%} → 節約したリクエスト {self.hits}件")


_cache = TtlCache()


def cached(ttl: float = 3600.0, key_prefix: str = ""):
    """引数をキーにして結果をキャッシュするデコレータ。"""

    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # 第1引数がself/httpの場合は除外してキーを作る
            key_args = [a for a in args if isinstance(a, (str, int, float, bool))]
            key = (f"{key_prefix or fn.__name__}:"
                   f"{','.join(map(str, key_args))}:"
                   f"{sorted(kwargs.items())}")
            hit = _cache.get(key)
            if hit is not None:
                return hit
            result = fn(*args, **kwargs)
            _cache.set(key, result, ttl)
            return result

        return wrapper

    return decorator


# ---- 使用例 -----------------------------------------------------------
@cached(ttl=86400.0)          # 銘柄仕様は1日キャッシュ
def get_symbol_rules(http, symbol: str, category: str = "linear") -> dict:
    res = http.get_instruments_info(category=category, symbol=symbol)
    info = res["result"]["list"][0]
    return {
        "tickSize": info["priceFilter"]["tickSize"],
        "qtyStep": info["lotSizeFilter"].get("qtyStep"),
        "minOrderQty": info["lotSizeFilter"].get("minOrderQty"),
        "maxLeverage": info.get("leverageFilter", {}).get("maxLeverage"),
    }


@cached(ttl=3600.0)           # 手数料率は1時間キャッシュ（上限5/秒と低いので重要）
def get_fee_rate(http, symbol: str, category: str = "linear") -> dict:
    res = http.get_fee_rates(category=category, symbol=symbol)
    row = res["result"]["list"][0]
    return {"maker": row["makerFeeRate"], "taker": row["takerFeeRate"]}


if __name__ == "__main__":
    from client import make_client

    c = make_client()

    # 100回呼んでも、実際のAPIリクエストは1回だけ
    for _ in range(100):
        rules = get_symbol_rules(c.http, "BTCUSDT")
    print("銘柄仕様:", rules)

    for _ in range(50):
        fee = get_fee_rate(c.http, "BTCUSDT")
    print("手数料  :", fee)

    print(_cache.stats())