Bybit WebSocket Trade（WS発注）で
発注レイテンシを最小化する

"""ws_trade_client.py — WS Trade への接続・認証
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install websockets

  エンドポイント:
    本番    wss://stream.bybit.com/v5/trade
    テスト  wss://stream-testnet.bybit.com/v5/trade
    デモ    wss://stream-demo.bybit.com/v5/trade

  認証（private WS と同じ形式）:
    expires   = 現在時刻(ms) + 有効期間(ms)
    signature = HMAC_SHA256(api_secret, "GET/realtime" + expires)
    → {"op": "auth", "args": [api_key, expires, signature]}

  認証が完了する前に発注を送ると拒否される。
  必ず auth の成功応答を待ってから使うこと。
"""
from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import time
from dataclasses import dataclass, field

import websockets

from config import load_config

ENDPOINTS = {
    "mainnet": "wss://stream.bybit.com/v5/trade",
    "testnet": "wss://stream-testnet.bybit.com/v5/trade",
    "demo": "wss://stream-demo.bybit.com/v5/trade",
}


def sign_auth(api_secret: str, expires_ms: int) -> str:
    """WS認証用の署名を作る。

    private WS（記事27）と同じ形式。
    署名対象は "GET/realtime" + expires という固定文字列。
    """
    payload = f"GET/realtime{expires_ms}"
    return hmac.new(api_secret.encode(), payload.encode(),
                    hashlib.sha256).hexdigest()


@dataclass(frozen=True)
class WsTradeConfig:
    recv_window_ms: int = 5000
    auth_ttl_ms: int = 10_000       # 認証の有効期間
    ping_interval_sec: float = 20.0
    connect_timeout_sec: float = 10.0
    auth_timeout_sec: float = 10.0
    referer: str = ""               # ブローカー用（通常は不要）


@dataclass
class WsTradeConnection:
    """WS Trade への接続を管理する（送受信の生層）。"""
    config: WsTradeConfig = field(default_factory=WsTradeConfig)

    _ws: object = None
    _connected: bool = False
    _authenticated: bool = False
    _url: str = ""
    connected_at: float = 0.0
    auth_expires_ms: int = 0

    on_message: object = None       # callable(dict)

    @property
    def is_ready(self) -> bool:
        return self._connected and self._authenticated

    def _endpoint(self, cfg) -> str:
        if getattr(cfg, "demo", False):
            return ENDPOINTS["demo"]
        return ENDPOINTS["testnet"] if cfg.testnet else ENDPOINTS["mainnet"]

    async def connect(self) -> None:
        """接続して認証する。"""
        cfg = load_config()
        self._url = self._endpoint(cfg)

        self._ws = await asyncio.wait_for(
            websockets.connect(
                self._url,
                ping_interval=self.config.ping_interval_sec,
                ping_timeout=self.config.ping_interval_sec / 2,
                max_queue=1024),
            timeout=self.config.connect_timeout_sec)
        self._connected = True
        self.connected_at = time.time()
        print(f"  接続: {self._url}")

        await self._authenticate(cfg)

    async def _authenticate(self, cfg) -> None:
        expires = int(time.time() * 1000) + self.config.auth_ttl_ms
        signature = sign_auth(cfg.api_secret, expires)

        await self._ws.send(json.dumps({
            "op": "auth",
            "args": [cfg.api_key, expires, signature],
        }))

        deadline = time.time() + self.config.auth_timeout_sec
        while time.time() < deadline:
            raw = await asyncio.wait_for(
                self._ws.recv(), timeout=self.config.auth_timeout_sec)
            msg = json.loads(raw)

            if msg.get("op") != "auth":
                # 認証前に他のメッセージが来ることは基本ないが、念のため
                continue

            if msg.get("success") is True or msg.get("retCode") == 0:
                self._authenticated = True
                self.auth_expires_ms = expires
                print(f"  認証成功（有効期限 "
                      f"{self.config.auth_ttl_ms / 1000:.0f}秒）")
                return

            raise RuntimeError(
                f"認証失敗: {msg.get('retMsg') or msg.get('ret_msg')}\n"
                "  APIキー・シークレット・時計のズレを確認してください")

        raise TimeoutError("認証応答がタイムアウトしました")

    async def send(self, payload: dict) -> None:
        if not self.is_ready:
            raise RuntimeError("未接続または未認証です")
        await self._ws.send(json.dumps(payload))

    async def recv_loop(self) -> None:
        """受信ループ。別タスクで動かす。"""
        try:
            async for raw in self._ws:
                try:
                    msg = json.loads(raw)
                except json.JSONDecodeError:
                    continue
                if self.on_message is not None:
                    self.on_message(msg)
        except websockets.ConnectionClosed as e:
            print(f"  接続が閉じられました: {e.code} {e.reason}")
        finally:
            self._connected = False
            self._authenticated = False

    async def close(self) -> None:
        self._connected = False
        self._authenticated = False
        if self._ws is not None:
            await self._ws.close()
            self._ws = None


async def main() -> None:
    conn = WsTradeConnection()
    conn.on_message = lambda m: print(f"  受信: {json.dumps(m, ensure_ascii=False)[:200]}")

    await conn.connect()
    print(f"  準備完了: {conn.is_ready}")

    task = asyncio.create_task(conn.recv_loop())
    await asyncio.sleep(3)

    await conn.close()
    task.cancel()

    print("\n※ 認証の有効期限（expires）は接続の寿命ではない。")
    print("  一度認証すれば接続が続く限り有効。")
    print("  ただし再接続時は必ず認証をやり直すこと。")


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

"""ws_trade_rpc.py — reqIdでリクエストと応答を対応付ける
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  リクエスト:
    {
      "reqId": "自分で付ける一意なID",
      "header": {"X-BAPI-TIMESTAMP": "...", "X-BAPI-RECV-WINDOW": "5000"},
      "op": "order.create",
      "args": [{ ...place_order と同じパラメータ... }]
    }

  応答:
    {
      "reqId": "同じID",
      "retCode": 0,
      "retMsg": "OK",
      "op": "order.create",
      "data": {"orderId": "...", "orderLinkId": "..."},
      "header": {"X-Bapi-Limit": "...", ...}
    }

  設計:
    reqId → asyncio.Future の辞書を持ち、
    応答が来たら対応する Future に結果を set する。
    タイムアウトしたら Future をキャンセルして辞書から消す。

  重要: 応答が来ない可能性を常に考慮する。
        タイムアウト時に「注文が通っていないとは限らない」。
        orderLinkId による冪等化（記事17）が必須。
"""
from __future__ import annotations

import asyncio
import time
import uuid
from dataclasses import dataclass, field

from ws_trade_client import WsTradeConfig, WsTradeConnection


@dataclass
class RpcResponse:
    req_id: str
    op: str
    ret_code: int
    ret_msg: str
    data: dict = field(default_factory=dict)
    ret_ext_info: dict = field(default_factory=dict)
    header: dict = field(default_factory=dict)
    latency_ms: float = 0.0

    @property
    def ok(self) -> bool:
        return self.ret_code == 0

    @property
    def order_id(self) -> str:
        return self.data.get("orderId", "")

    @property
    def order_link_id(self) -> str:
        return self.data.get("orderLinkId", "")

    @property
    def rate_limit_remaining(self) -> int:
        """X-Bapi-Limit-Status（残り回数）。記事05の流量管理に使う。"""
        try:
            return int(self.header.get("X-Bapi-Limit-Status", -1))
        except (TypeError, ValueError):
            return -1

    def line(self) -> str:
        if self.ok:
            return (f"OK {self.op} orderId={self.order_id[:12]}... "
                    f"({self.latency_ms:.1f}ms)")
        return (f"NG {self.op} retCode={self.ret_code} {self.ret_msg} "
                f"({self.latency_ms:.1f}ms)")


@dataclass
class PendingRequest:
    req_id: str
    op: str
    future: asyncio.Future
    sent_at: float


@dataclass
class WsTradeRpc:
    """WS Trade を RPC 的に使えるようにする層。"""
    config: WsTradeConfig = field(default_factory=WsTradeConfig)
    default_timeout_sec: float = 5.0

    connection: WsTradeConnection = field(init=False)
    _pending: dict = field(default_factory=dict)
    _recv_task: asyncio.Task | None = None

    sent_count: int = 0
    received_count: int = 0
    timeout_count: int = 0
    orphan_count: int = 0        # 対応するリクエストが無い応答

    def __post_init__(self) -> None:
        self.connection = WsTradeConnection(self.config)
        self.connection.on_message = self._handle

    # --- 受信 ------------------------------------------------------

    def _handle(self, msg: dict) -> None:
        """受信メッセージを対応する Future へ届ける。"""
        req_id = msg.get("reqId", "")
        if not req_id:
            # pong など。ここでは無視する
            return

        pending = self._pending.pop(req_id, None)
        if pending is None:
            self.orphan_count += 1
            return

        self.received_count += 1
        latency = (time.perf_counter() - pending.sent_at) * 1000

        response = RpcResponse(
            req_id=req_id,
            op=msg.get("op", pending.op),
            ret_code=int(msg.get("retCode", -1)),
            ret_msg=msg.get("retMsg", ""),
            data=msg.get("data") or {},
            ret_ext_info=msg.get("retExtInfo") or {},
            header=msg.get("header") or {},
            latency_ms=latency)

        if not pending.future.done():
            pending.future.set_result(response)

    # --- 送信 ------------------------------------------------------

    def _build_header(self) -> dict:
        return {
            "X-BAPI-TIMESTAMP": str(int(time.time() * 1000)),
            "X-BAPI-RECV-WINDOW": str(self.config.recv_window_ms),
        }

    async def call(self, op: str, args: list,
                   timeout_sec: float = 0.0) -> RpcResponse:
        """1回のRPC呼び出し。応答が返るまで待つ。"""
        req_id = str(uuid.uuid4())
        timeout = timeout_sec or self.default_timeout_sec

        loop = asyncio.get_running_loop()
        future: asyncio.Future = loop.create_future()

        payload = {
            "reqId": req_id,
            "header": self._build_header(),
            "op": op,
            "args": args,
        }

        self._pending[req_id] = PendingRequest(
            req_id, op, future, time.perf_counter())
        self.sent_count += 1

        try:
            await self.connection.send(payload)
            return await asyncio.wait_for(future, timeout=timeout)
        except asyncio.TimeoutError:
            self._pending.pop(req_id, None)
            self.timeout_count += 1
            raise TimeoutError(
                f"{op} の応答が {timeout}秒以内に返りませんでした。\n"
                "  注文が通っていない保証はありません。"
                "orderLinkId で確認してください（記事17）")
        except Exception:
            self._pending.pop(req_id, None)
            raise

    # --- ライフサイクル --------------------------------------------

    async def start(self) -> None:
        await self.connection.connect()
        self._recv_task = asyncio.create_task(self.connection.recv_loop())

    async def stop(self) -> None:
        # 未応答のリクエストを片付ける
        for pending in list(self._pending.values()):
            if not pending.future.done():
                pending.future.cancel()
        self._pending.clear()

        await self.connection.close()
        if self._recv_task:
            self._recv_task.cancel()

    def stats(self) -> dict:
        return {
            "sent": self.sent_count,
            "received": self.received_count,
            "timeout": self.timeout_count,
            "orphan": self.orphan_count,
            "pending": len(self._pending),
        }

    def report(self) -> str:
        s = self.stats()
        return (f"  送信 {s['sent']} / 受信 {s['received']} / "
                f"タイムアウト {s['timeout']} / "
                f"孤児応答 {s['orphan']} / 待機中 {s['pending']}")


async def main() -> None:
    rpc = WsTradeRpc()
    await rpc.start()

    print("\n=== 接続状態の確認 ===")
    print(f"  準備完了: {rpc.connection.is_ready}")
    print(rpc.report())

    await rpc.stop()

    print("\n※ タイムアウトは『注文が通らなかった』を意味しない。")
    print("  送信は成功していて応答だけが失われた可能性がある。")
    print("  必ず orderLinkId を付けて、後から照会できるようにすること。")


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

"""ws_trade_ops.py — WS Trade の各オペレーションのラッパ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  対応オペレーション（これ以外は WS では送れない）:
    order.create        単発の発注
    order.amend         単発の修正
    order.cancel        単発のキャンセル
    order.create-batch  一括発注
    order.amend-batch   一括修正
    order.cancel-batch  一括キャンセル

  注意:
    - cancel_all_orders は WS では送れない（RESTのみ）
      → 緊急停止（記事24）は必ずREST経路を残すこと
    - args のパラメータは REST の place_order と同一
    - batch の args は「注文の配列」を1要素として渡す形
"""
from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass
from decimal import Decimal

from ws_trade_rpc import RpcResponse, WsTradeRpc

BATCH_LIMIT = {"linear": 20, "inverse": 20, "option": 20, "spot": 10}


@dataclass(frozen=True)
class OrderSpec:
    """発注1件分。REST の place_order と同じパラメータ体系。"""
    symbol: str
    side: str
    order_type: str
    qty: str
    price: str = ""
    time_in_force: str = "GTC"
    order_link_id: str = ""
    reduce_only: bool = False
    trigger_price: str = ""
    trigger_direction: int = 0
    position_idx: int = 0

    def to_args(self, category: str) -> dict:
        p: dict = {
            "category": category,
            "symbol": self.symbol,
            "side": self.side,
            "orderType": self.order_type,
            "qty": self.qty,
        }
        if self.price:
            p["price"] = self.price
        if self.order_type == "Limit":
            p["timeInForce"] = self.time_in_force
        if self.order_link_id:
            p["orderLinkId"] = self.order_link_id
        if self.reduce_only:
            p["reduceOnly"] = True
        if self.trigger_price:
            p["triggerPrice"] = self.trigger_price
            p["triggerDirection"] = self.trigger_direction
        if self.position_idx:
            p["positionIdx"] = self.position_idx
        return p


@dataclass(frozen=True)
class AmendSpec:
    symbol: str
    order_id: str = ""
    order_link_id: str = ""
    price: str = ""
    qty: str = ""
    trigger_price: str = ""

    def to_args(self, category: str) -> dict:
        if not (self.order_id or self.order_link_id):
            raise ValueError("orderId か orderLinkId が必要です")
        p: dict = {"category": category, "symbol": self.symbol}
        if self.order_id:
            p["orderId"] = self.order_id
        else:
            p["orderLinkId"] = self.order_link_id
        # 変更しない項目は渡さない（空文字はエラーになる。記事16）
        if self.price:
            p["price"] = self.price
        if self.qty:
            p["qty"] = self.qty
        if self.trigger_price:
            p["triggerPrice"] = self.trigger_price
        return p


@dataclass(frozen=True)
class CancelSpec:
    symbol: str
    order_id: str = ""
    order_link_id: str = ""

    def to_args(self, category: str) -> dict:
        if not (self.order_id or self.order_link_id):
            raise ValueError("orderId か orderLinkId が必要です")
        p: dict = {"category": category, "symbol": self.symbol}
        if self.order_id:
            p["orderId"] = self.order_id
        else:
            p["orderLinkId"] = self.order_link_id
        return p


class WsTradeOps:
    """WS Trade のオペレーションを型付きで呼べるようにする。"""

    def __init__(self, rpc: WsTradeRpc, category: str = "linear"):
        self.rpc = rpc
        self.category = category

    # --- 単発 ------------------------------------------------------

    async def create(self, spec: OrderSpec,
                     timeout_sec: float = 5.0) -> RpcResponse:
        return await self.rpc.call(
            "order.create", [spec.to_args(self.category)], timeout_sec)

    async def amend(self, spec: AmendSpec,
                    timeout_sec: float = 5.0) -> RpcResponse:
        return await self.rpc.call(
            "order.amend", [spec.to_args(self.category)], timeout_sec)

    async def cancel(self, spec: CancelSpec,
                     timeout_sec: float = 5.0) -> RpcResponse:
        return await self.rpc.call(
            "order.cancel", [spec.to_args(self.category)], timeout_sec)

    # --- バッチ ----------------------------------------------------

    def _check_batch_size(self, n: int) -> None:
        limit = BATCH_LIMIT.get(self.category, 10)
        if n > limit:
            raise ValueError(
                f"{self.category} のバッチ上限は {limit}件です（{n}件指定）")
        if n == 0:
            raise ValueError("空のバッチは送れません")

    async def create_batch(self, specs: list,
                           timeout_sec: float = 10.0) -> RpcResponse:
        """一括発注。args は {"category":..., "request": [...]} の形。"""
        self._check_batch_size(len(specs))
        args = [{
            "category": self.category,
            "request": [
                {k: v for k, v in s.to_args(self.category).items()
                 if k != "category"}
                for s in specs
            ],
        }]
        return await self.rpc.call("order.create-batch", args, timeout_sec)

    async def amend_batch(self, specs: list,
                          timeout_sec: float = 10.0) -> RpcResponse:
        self._check_batch_size(len(specs))
        args = [{
            "category": self.category,
            "request": [
                {k: v for k, v in s.to_args(self.category).items()
                 if k != "category"}
                for s in specs
            ],
        }]
        return await self.rpc.call("order.amend-batch", args, timeout_sec)

    async def cancel_batch(self, specs: list,
                           timeout_sec: float = 10.0) -> RpcResponse:
        self._check_batch_size(len(specs))
        args = [{
            "category": self.category,
            "request": [
                {k: v for k, v in s.to_args(self.category).items()
                 if k != "category"}
                for s in specs
            ],
        }]
        return await self.rpc.call("order.cancel-batch", args, timeout_sec)


def parse_batch_result(response: RpcResponse) -> list:
    """バッチ応答から個別の成否を取り出す。

    REST と同じ構造（記事16）:
      data.list       → 各注文の orderId
      retExtInfo.list → 各注文の code / msg
    トップレベルの retCode が 0 でも個別に失敗していることがある。
    """
    results = response.data.get("list", []) if response.data else []
    ext = response.ret_ext_info.get("list", []) if response.ret_ext_info else []

    out = []
    for i in range(max(len(results), len(ext))):
        res = results[i] if i < len(results) else {}
        info = ext[i] if i < len(ext) else {"code": 0, "msg": ""}
        code = int(info.get("code", 0))
        order_id = res.get("orderId", "")
        out.append({
            "index": i,
            "ok": code == 0 and bool(order_id),
            "order_id": order_id,
            "order_link_id": res.get("orderLinkId", ""),
            "code": code,
            "msg": info.get("msg", ""),
        })
    return out


async def main() -> None:
    from client import make_client

    rpc = WsTradeRpc()
    await rpc.start()
    ops = WsTradeOps(rpc, category="linear")

    c = make_client()
    r = c.http.get_tickers(category="linear", symbol="BTCUSDT")
    last = Decimal(r["result"]["list"][0]["lastPrice"])

    stamp = int(time.time())

    print("=== 単発の発注 ===")
    spec = OrderSpec(
        symbol="BTCUSDT", side="Buy", order_type="Limit",
        qty="0.001",
        price=str((last * Decimal("0.95")).quantize(Decimal("0.1"))),
        time_in_force="PostOnly",
        order_link_id=f"wst-{stamp}")
    res = await ops.create(spec)
    print(f"  {res.line()}")
    print(f"  レート残り: {res.rate_limit_remaining}")

    if res.ok:
        print("\n=== 修正 ===")
        amend = AmendSpec(
            symbol="BTCUSDT", order_id=res.order_id,
            price=str((last * Decimal("0.94")).quantize(Decimal("0.1"))))
        print(f"  {(await ops.amend(amend)).line()}")

        print("\n=== キャンセル ===")
        cancel = CancelSpec(symbol="BTCUSDT", order_id=res.order_id)
        print(f"  {(await ops.cancel(cancel)).line()}")

    print("\n=== バッチ発注（5件） ===")
    specs = [
        OrderSpec(
            symbol="BTCUSDT", side="Buy", order_type="Limit", qty="0.001",
            price=str((last * (Decimal(1) - Decimal("0.01") * k))
                      .quantize(Decimal("0.1"))),
            time_in_force="PostOnly",
            order_link_id=f"wsb-{stamp}-{k}")
        for k in range(1, 6)
    ]
    batch_res = await ops.create_batch(specs)
    print(f"  {batch_res.line()}")
    for item in parse_batch_result(batch_res):
        status = "OK" if item["ok"] else f"NG {item['code']} {item['msg']}"
        print(f"    [{item['index']}] {status}")

    # 後片付けは REST（cancel_all_orders は WS で送れない）
    c.http.cancel_all_orders(category="linear", symbol="BTCUSDT")
    print("\n  全注文をキャンセルしました（REST経由）")

    print(f"\n{rpc.report()}")
    await rpc.stop()

    print("\n※ cancel_all_orders は WS では送れない。")
    print("  緊急停止（記事24）は必ずREST経路を残すこと。")


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

"""latency_bench.py — WS Trade と REST の往復レイテンシを比較する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  測るもの: 「発注を送ってから応答が返るまで」の往復時間

  公平な比較のための工夫:
    - 同じ銘柄・同じサイズ・同じ価格帯（約定しないPostOnly指値）
    - WS と REST を交互に実行する（時間帯の偏りを消す）
    - ウォームアップを行う（初回はTLS確立で遅い）
    - 分布で比較する（平均は外れ値に引っ張られる）

  必ず testnet で実行すること。
  本番では大量の注文が板に載る。
"""
from __future__ import annotations

import asyncio
import statistics
import time
from dataclasses import dataclass, field
from decimal import Decimal

from client import make_client
from ws_trade_ops import CancelSpec, OrderSpec, WsTradeOps
from ws_trade_rpc import WsTradeRpc


@dataclass
class LatencySample:
    method: str
    op: str
    latency_ms: float
    ok: bool
    order_id: str = ""


@dataclass
class BenchResult:
    method: str
    samples: list = field(default_factory=list)

    @property
    def latencies(self) -> list:
        return [s.latency_ms for s in self.samples if s.ok]

    def percentiles(self) -> dict:
        data = sorted(self.latencies)
        if len(data) < 5:
            return {}
        n = len(data)
        return {
            "n": n,
            "min": data[0],
            "p50": data[n // 2],
            "p90": data[min(int(n * 0.90), n - 1)],
            "p99": data[min(int(n * 0.99), n - 1)],
            "max": data[-1],
            "mean": statistics.mean(data),
            "stdev": statistics.stdev(data) if n > 1 else 0.0,
        }

    def line(self) -> str:
        p = self.percentiles()
        if not p:
            return f"  {self.method:<10}サンプル不足"
        failures = len(self.samples) - len(self.latencies)
        return (f"  {self.method:<10}"
                f"n={p['n']:>4}  "
                f"p50={p['p50']:>7.1f}  "
                f"p90={p['p90']:>7.1f}  "
                f"p99={p['p99']:>7.1f}  "
                f"max={p['max']:>8.1f}  "
                f"σ={p['stdev']:>6.1f}  "
                f"失敗={failures}")


async def bench_ws(ops: WsTradeOps, symbol: str, price: Decimal,
                   n: int) -> BenchResult:
    result = BenchResult("WS Trade")

    for i in range(n):
        spec = OrderSpec(
            symbol=symbol, side="Buy", order_type="Limit", qty="0.001",
            price=str(price), time_in_force="PostOnly",
            order_link_id=f"bw-{int(time.time() * 1000) % 1_000_000}-{i}")

        start = time.perf_counter()
        try:
            res = await ops.create(spec)
            elapsed = (time.perf_counter() - start) * 1000
            result.samples.append(LatencySample(
                "WS", "create", elapsed, res.ok, res.order_id))

            if res.ok:
                await ops.cancel(CancelSpec(symbol=symbol,
                                            order_id=res.order_id))
        except Exception as e:
            elapsed = (time.perf_counter() - start) * 1000
            result.samples.append(LatencySample("WS", "create", elapsed, False))
            print(f"    [WS {i}] エラー: {type(e).__name__}: {e}")

        await asyncio.sleep(0.15)

    return result


def bench_rest(http, symbol: str, price: Decimal, n: int) -> BenchResult:
    result = BenchResult("REST")

    for i in range(n):
        start = time.perf_counter()
        try:
            r = http.place_order(
                category="linear", symbol=symbol, side="Buy",
                orderType="Limit", qty="0.001", price=str(price),
                timeInForce="PostOnly",
                orderLinkId=f"br-{int(time.time() * 1000) % 1_000_000}-{i}")
            elapsed = (time.perf_counter() - start) * 1000
            ok = r["retCode"] == 0
            order_id = r["result"]["orderId"] if ok else ""
            result.samples.append(LatencySample(
                "REST", "create", elapsed, ok, order_id))

            if ok:
                http.cancel_order(category="linear", symbol=symbol,
                                  orderId=order_id)
        except Exception as e:
            elapsed = (time.perf_counter() - start) * 1000
            result.samples.append(LatencySample("REST", "create",
                                                elapsed, False))
            print(f"    [REST {i}] エラー: {type(e).__name__}: {e}")

        time.sleep(0.15)

    return result


async def run_benchmark(symbol: str = "BTCUSDT", n: int = 30,
                        warmup: int = 3) -> None:
    c = make_client()
    if not getattr(c, "is_testnet", True):
        raise RuntimeError("本番では実行しないでください（実注文が出ます）")

    r = c.http.get_tickers(category="linear", symbol=symbol)
    last = Decimal(r["result"]["list"][0]["lastPrice"])
    # 約定しない価格（現在の -10%）
    price = (last * Decimal("0.90")).quantize(Decimal("0.1"))

    rpc = WsTradeRpc()
    await rpc.start()
    ops = WsTradeOps(rpc, category="linear")

    print(f"=== レイテンシ比較 ({symbol} / 各{n}回) ===")
    print(f"  発注価格: {price}（現在 {last} の -10%。約定しない）\n")

    print(f"  ウォームアップ ({warmup}回)...")
    await bench_ws(ops, symbol, price, warmup)
    bench_rest(c.http, symbol, price, warmup)

    print("  計測中...")
    # 交互に実行して時間帯の偏りを消す
    ws_result = BenchResult("WS Trade")
    rest_result = BenchResult("REST")

    chunk = max(1, n // 5)
    for round_no in range(5):
        ws_part = await bench_ws(ops, symbol, price, chunk)
        ws_result.samples.extend(ws_part.samples)

        rest_part = bench_rest(c.http, symbol, price, chunk)
        rest_result.samples.extend(rest_part.samples)

        print(f"    ラウンド {round_no + 1}/5 完了")

    print(f"\n{'方式':<10}{'件数':>7}{'p50':>10}{'p90':>10}"
          f"{'p99':>10}{'max':>11}{'σ':>9}{'失敗':>8}  (ms)")
    print("-" * 80)
    print(ws_result.line())
    print(rest_result.line())

    ws_p = ws_result.percentiles()
    rest_p = rest_result.percentiles()

    if ws_p and rest_p:
        print("\n=== 差分 ===")
        for key in ("p50", "p90", "p99"):
            diff = rest_p[key] - ws_p[key]
            ratio = (rest_p[key] / ws_p[key]) if ws_p[key] > 0 else 0
            print(f"  {key}: WS が {diff:+.1f}ms 速い "
                  f"({ratio:.2f}倍)")

        gain = rest_p["p50"] - ws_p["p50"]
        print()
        if gain < 3:
            print("  → 差はわずかです。実装の複雑さに見合いません。")
            print("    RESTのままで構いません。")
        elif gain < 15:
            print("  → 一定の効果があります。")
            print("    発注頻度が高い戦略なら導入を検討する価値があります。")
        else:
            print("  → 明確な効果があります。")
            print("    低遅延が重要な戦略では導入すべきです。")

    c.http.cancel_all_orders(category="linear", symbol=symbol)
    print(f"\n{rpc.report()}")
    await rpc.stop()

    print("\n※ この結果は自分の環境（VPSの場所・回線）に固有のもの。")
    print("  他人の測定値を参考にしても意味がない。")
    print("  記事38のリージョン選定と合わせて評価すること。")


if __name__ == "__main__":
    asyncio.run(run_benchmark(n=30))

"""order_gateway.py — WS優先・REST自動フォールバックの発注ゲートウェイ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  bot本体はこのゲートウェイだけを呼ぶ。
  WSかRESTかを意識しない。

  切り替えの規則:
    - WSが準備完了なら WS を使う
    - WSが未接続・タイムアウト・エラーなら REST へ落とす
    - WSの失敗が連続したら、一定時間 REST に固定する（フラッピング防止）
    - 背後で再接続を試み、成功したらWSに戻す

  重要: タイムアウトしたリクエストをRESTで再送するのは危険。
        WS側で通っている可能性があるため、二重発注になる。
        必ず orderLinkId による冪等化（記事17）を前提にする。
"""
from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass, field
from enum import Enum

from ws_trade_ops import (AmendSpec, CancelSpec, OrderSpec, WsTradeOps,
                          parse_batch_result)
from ws_trade_rpc import RpcResponse, WsTradeRpc


class Route(str, Enum):
    WS = "WS"
    REST = "REST"


@dataclass
class GatewayResult:
    ok: bool
    route: Route
    order_id: str = ""
    order_link_id: str = ""
    ret_code: int = 0
    ret_msg: str = ""
    latency_ms: float = 0.0
    fell_back: bool = False

    def line(self) -> str:
        tag = f"{self.route.value}"
        if self.fell_back:
            tag += "(fallback)"
        if self.ok:
            return (f"OK [{tag}] orderId={self.order_id[:12]}... "
                    f"({self.latency_ms:.1f}ms)")
        return (f"NG [{tag}] {self.ret_code} {self.ret_msg} "
                f"({self.latency_ms:.1f}ms)")


@dataclass(frozen=True)
class GatewayConfig:
    category: str = "linear"
    ws_timeout_sec: float = 3.0
    max_consecutive_ws_failures: int = 3
    ws_cooldown_sec: float = 30.0     # 失敗後にRESTへ固定する時間
    reconnect_interval_sec: float = 10.0
    prefer_ws: bool = True


@dataclass
class OrderGateway:
    """WS優先・REST自動フォールバックの発注口。"""
    http: object                       # pybit の HTTP
    config: GatewayConfig = field(default_factory=GatewayConfig)

    rpc: WsTradeRpc = field(default_factory=WsTradeRpc)
    ops: WsTradeOps = field(init=False)

    _ws_failures: int = 0
    _ws_disabled_until: float = 0.0
    _reconnect_task: asyncio.Task | None = None

    ws_calls: int = 0
    rest_calls: int = 0
    fallback_calls: int = 0

    def __post_init__(self) -> None:
        self.ops = WsTradeOps(self.rpc, self.config.category)

    # --- 経路の判断 ------------------------------------------------

    @property
    def ws_available(self) -> bool:
        if not self.config.prefer_ws:
            return False
        if time.time() < self._ws_disabled_until:
            return False
        return self.rpc.connection.is_ready

    def _record_ws_failure(self, reason: str) -> None:
        self._ws_failures += 1
        if self._ws_failures >= self.config.max_consecutive_ws_failures:
            self._ws_disabled_until = time.time() + self.config.ws_cooldown_sec
            print(f"  [gateway] WSを {self.config.ws_cooldown_sec:.0f}秒間"
                  f"無効化します（{reason}）")
            self._schedule_reconnect()

    def _record_ws_success(self) -> None:
        self._ws_failures = 0

    def _schedule_reconnect(self) -> None:
        if self._reconnect_task and not self._reconnect_task.done():
            return

        async def worker() -> None:
            while True:
                await asyncio.sleep(self.config.reconnect_interval_sec)
                try:
                    await self.rpc.stop()
                    await self.rpc.start()
                    self._ws_failures = 0
                    self._ws_disabled_until = 0.0
                    print("  [gateway] WS再接続に成功しました")
                    return
                except Exception as e:
                    print(f"  [gateway] 再接続に失敗: "
                          f"{type(e).__name__}: {e}")

        self._reconnect_task = asyncio.create_task(worker())

    # --- 発注 ------------------------------------------------------

    async def place(self, spec: OrderSpec) -> GatewayResult:
        """発注する。WSが使えなければRESTへ落とす。"""
        if not spec.order_link_id:
            raise ValueError(
                "orderLinkId は必須です。"
                "フォールバック時の二重発注を防げません（記事17）")

        start = time.perf_counter()

        if self.ws_available:
            try:
                res = await self.ops.create(spec, self.config.ws_timeout_sec)
                self.ws_calls += 1
                self._record_ws_success()
                return GatewayResult(
                    ok=res.ok, route=Route.WS,
                    order_id=res.order_id, order_link_id=res.order_link_id,
                    ret_code=res.ret_code, ret_msg=res.ret_msg,
                    latency_ms=res.latency_ms)
            except TimeoutError:
                # タイムアウトは「通っていない」を意味しない。
                # RESTで再送する前に、必ず存在確認を行う。
                self._record_ws_failure("タイムアウト")
                existing = self._find_by_link_id(spec.order_link_id)
                if existing:
                    print("  [gateway] タイムアウトしましたが注文は通っていました")
                    return GatewayResult(
                        ok=True, route=Route.WS,
                        order_id=existing.get("orderId", ""),
                        order_link_id=spec.order_link_id,
                        latency_ms=(time.perf_counter() - start) * 1000)
            except Exception as e:
                self._record_ws_failure(f"{type(e).__name__}: {e}")

        # REST経路
        return self._place_rest(spec, start,
                                fell_back=self.config.prefer_ws)

    def _place_rest(self, spec: OrderSpec, start: float,
                    fell_back: bool = False) -> GatewayResult:
        params = spec.to_args(self.config.category)
        try:
            r = self.http.place_order(**params)
        except Exception as e:
            return GatewayResult(
                ok=False, route=Route.REST, ret_code=-1,
                ret_msg=f"{type(e).__name__}: {e}",
                latency_ms=(time.perf_counter() - start) * 1000,
                fell_back=fell_back)

        self.rest_calls += 1
        if fell_back:
            self.fallback_calls += 1

        ok = r["retCode"] == 0
        return GatewayResult(
            ok=ok, route=Route.REST,
            order_id=r["result"].get("orderId", "") if ok else "",
            order_link_id=spec.order_link_id,
            ret_code=r["retCode"], ret_msg=r["retMsg"],
            latency_ms=(time.perf_counter() - start) * 1000,
            fell_back=fell_back)

    def _find_by_link_id(self, order_link_id: str) -> dict | None:
        """orderLinkId で注文の存在を確認する（冪等性の要）。"""
        for fetch in (self.http.get_open_orders, self.http.get_order_history):
            try:
                r = fetch(category=self.config.category,
                          orderLinkId=order_link_id, limit=1)
                if r["retCode"] == 0 and r["result"]["list"]:
                    return r["result"]["list"][0]
            except Exception:
                continue
        return None

    # --- キャンセル ------------------------------------------------

    async def cancel(self, spec: CancelSpec) -> GatewayResult:
        start = time.perf_counter()

        if self.ws_available:
            try:
                res = await self.ops.cancel(spec, self.config.ws_timeout_sec)
                self.ws_calls += 1
                self._record_ws_success()
                return GatewayResult(
                    ok=res.ok, route=Route.WS, order_id=res.order_id,
                    ret_code=res.ret_code, ret_msg=res.ret_msg,
                    latency_ms=res.latency_ms)
            except Exception as e:
                self._record_ws_failure(f"{type(e).__name__}: {e}")

        params = spec.to_args(self.config.category)
        r = self.http.cancel_order(**params)
        self.rest_calls += 1
        return GatewayResult(
            ok=r["retCode"] in (0, 110001), route=Route.REST,
            ret_code=r["retCode"], ret_msg=r["retMsg"],
            latency_ms=(time.perf_counter() - start) * 1000,
            fell_back=True)

    async def cancel_all(self, symbol: str) -> int:
        """全キャンセル。WSでは送れないので必ずREST。

        緊急停止（記事24）はこの経路を使う。
        """
        r = self.http.cancel_all_orders(
            category=self.config.category, symbol=symbol)
        self.rest_calls += 1
        if r["retCode"] != 0:
            raise RuntimeError(f"全キャンセル失敗: {r['retCode']} {r['retMsg']}")
        return len(r["result"].get("list", []))

    # --- ライフサイクル --------------------------------------------

    async def start(self) -> None:
        if not self.config.prefer_ws:
            print("  [gateway] REST固定モードで起動")
            return
        try:
            await self.rpc.start()
            print("  [gateway] WS Trade を主系として起動")
        except Exception as e:
            print(f"  [gateway] WS起動に失敗、RESTで動作します: {e}")
            self._ws_disabled_until = time.time() + self.config.ws_cooldown_sec
            self._schedule_reconnect()

    async def stop(self) -> None:
        if self._reconnect_task:
            self._reconnect_task.cancel()
        await self.rpc.stop()

    def report(self) -> str:
        total = self.ws_calls + self.rest_calls
        ws_rate = self.ws_calls / total if total else 0.0
        return (f"  経路: WS {self.ws_calls} / REST {self.rest_calls} "
                f"(WS率 {ws_rate:.1%}) / "
                f"フォールバック {self.fallback_calls}回\n"
                f"  WS状態: "
                f"{'利用可' if self.ws_available else '無効'} "
                f"(連続失敗 {self._ws_failures})")


async def main() -> None:
    from decimal import Decimal

    from client import make_client

    c = make_client()
    gateway = OrderGateway(c.http)
    await gateway.start()

    r = c.http.get_tickers(category="linear", symbol="BTCUSDT")
    last = Decimal(r["result"]["list"][0]["lastPrice"])
    price = (last * Decimal("0.90")).quantize(Decimal("0.1"))

    print("\n=== ゲートウェイ経由で発注 ===")
    for i in range(3):
        spec = OrderSpec(
            symbol="BTCUSDT", side="Buy", order_type="Limit",
            qty="0.001", price=str(price), time_in_force="PostOnly",
            order_link_id=f"gw-{int(time.time() * 1000) % 1_000_000}-{i}")
        result = await gateway.place(spec)
        print(f"  {result.line()}")

        if result.ok:
            await gateway.cancel(CancelSpec(symbol="BTCUSDT",
                                            order_id=result.order_id))

    print(f"\n{gateway.report()}")

    n = await gateway.cancel_all("BTCUSDT")
    print(f"\n  残り {n}件をキャンセル（REST経路）")

    await gateway.stop()

    print("\n※ タイムアウト時にRESTで素朴に再送すると二重発注になる。")
    print("  必ず orderLinkId で存在確認してから再送すること。")


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

"""execution_loop.py — 「送信 → 約定確認」までWSで完結させる
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  WS Trade で発注しても、約定の確認をRESTでポーリングしていては
  低遅延化の意味が半減する。
  private WS（記事27）の order / execution トピックを購読し、
  発注から約定確認までを全てWSで完結させる。

  流れ:
    1. WS Trade で発注 → 即座に orderId が返る
    2. private WS の order トピックで状態変化を受信
    3. execution トピックで約定を受信
    4. orderLinkId をキーに、送信側と受信側を突き合わせる

  この構成なら、発注から約定確認までRESTを1回も叩かない。
"""
from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass, field
from decimal import Decimal

from pybit.unified_trading import WebSocket

from config import load_config
from order_gateway import GatewayConfig, OrderGateway
from ws_trade_ops import OrderSpec


@dataclass
class OrderLifecycle:
    """1注文の一生を追跡する。"""
    order_link_id: str
    order_id: str = ""
    submitted_at: float = 0.0
    acked_at: float = 0.0          # WS Trade の応答が返った時刻
    first_event_at: float = 0.0    # private WS で最初に観測した時刻
    filled_at: float = 0.0
    status: str = "pending"
    filled_qty: Decimal = Decimal(0)
    avg_price: Decimal = Decimal(0)
    events: list = field(default_factory=list)

    @property
    def ack_latency_ms(self) -> float:
        return ((self.acked_at - self.submitted_at) * 1000
                if self.acked_at else 0.0)

    @property
    def event_latency_ms(self) -> float:
        """発注応答から private WS の通知までの時間。"""
        return ((self.first_event_at - self.acked_at) * 1000
                if self.first_event_at and self.acked_at else 0.0)

    @property
    def fill_latency_ms(self) -> float:
        return ((self.filled_at - self.submitted_at) * 1000
                if self.filled_at else 0.0)

    def line(self) -> str:
        parts = [f"  {self.order_link_id:<24}{self.status:<16}"]
        parts.append(f"ack={self.ack_latency_ms:>7.1f}ms")
        if self.first_event_at:
            parts.append(f"event=+{self.event_latency_ms:>7.1f}ms")
        if self.filled_at:
            parts.append(f"fill={self.fill_latency_ms:>8.1f}ms")
        return "  ".join(parts)


@dataclass
class ExecutionLoop:
    """WS Trade + private WS を組み合わせた発注ループ。"""
    http: object
    category: str = "linear"

    gateway: OrderGateway = field(init=False)
    private_ws: object = None
    orders: dict = field(default_factory=dict)     # orderLinkId → lifecycle
    _by_order_id: dict = field(default_factory=dict)

    def __post_init__(self) -> None:
        self.gateway = OrderGateway(
            self.http, GatewayConfig(category=self.category))

    # --- private WS の受信 -----------------------------------------

    def _on_order(self, message: dict) -> None:
        now = time.time()
        for o in message.get("data", []):
            link_id = o.get("orderLinkId", "")
            lifecycle = self.orders.get(link_id)
            if lifecycle is None:
                continue

            if not lifecycle.first_event_at:
                lifecycle.first_event_at = now

            status = o.get("orderStatus", "")
            lifecycle.status = status
            lifecycle.order_id = o.get("orderId", lifecycle.order_id)
            lifecycle.filled_qty = Decimal(str(o.get("cumExecQty") or "0"))
            lifecycle.avg_price = Decimal(str(o.get("avgPrice") or "0"))
            lifecycle.events.append((now, status))

            if status == "Filled":
                lifecycle.filled_at = now

    def _on_execution(self, message: dict) -> None:
        now = time.time()
        for e in message.get("data", []):
            link_id = e.get("orderLinkId", "")
            lifecycle = self.orders.get(link_id)
            if lifecycle is None:
                continue
            if not lifecycle.first_event_at:
                lifecycle.first_event_at = now
            lifecycle.events.append(
                (now, f"exec {e.get('execQty')} @{e.get('execPrice')}"))

    # --- ライフサイクル --------------------------------------------

    async def start(self) -> None:
        cfg = load_config()

        await self.gateway.start()

        self.private_ws = WebSocket(
            testnet=cfg.testnet, demo=getattr(cfg, "demo", False),
            channel_type="private",
            api_key=cfg.api_key, api_secret=cfg.api_secret)
        self.private_ws.order_stream(callback=self._on_order)
        self.private_ws.execution_stream(callback=self._on_execution)

        print("  private WS を購読しました（order / execution）")
        print("  ※ order トピックは購読時にスナップショットを送らない。")
        print("    既存の注文は流れてこないので、起動時にRESTで初期状態を作ること。")

        await asyncio.sleep(1.0)   # 購読の確立を待つ

    async def stop(self) -> None:
        await self.gateway.stop()
        if self.private_ws is not None:
            try:
                self.private_ws.exit()
            except Exception:
                pass

    # --- 発注 ------------------------------------------------------

    async def submit(self, spec: OrderSpec) -> OrderLifecycle:
        lifecycle = OrderLifecycle(order_link_id=spec.order_link_id,
                                   submitted_at=time.time())
        self.orders[spec.order_link_id] = lifecycle

        result = await self.gateway.place(spec)
        lifecycle.acked_at = time.time()
        lifecycle.order_id = result.order_id

        if result.ok:
            self._by_order_id[result.order_id] = lifecycle
            lifecycle.status = "submitted"
        else:
            lifecycle.status = f"rejected({result.ret_code})"

        return lifecycle

    async def wait_for_fill(self, lifecycle: OrderLifecycle,
                            timeout_sec: float = 30.0) -> bool:
        """約定を待つ。RESTポーリングは行わない。"""
        deadline = time.time() + timeout_sec
        while time.time() < deadline:
            if lifecycle.status in ("Filled", "Cancelled", "Rejected"):
                return lifecycle.status == "Filled"
            await asyncio.sleep(0.02)
        return False

    def report(self) -> str:
        lines = [f"=== 注文 {len(self.orders)}件 ==="]
        lines.extend(lc.line() for lc in self.orders.values())

        acked = [lc.ack_latency_ms for lc in self.orders.values()
                 if lc.acked_at]
        events = [lc.event_latency_ms for lc in self.orders.values()
                  if lc.first_event_at and lc.acked_at]

        if acked:
            lines.append(f"\n  発注応答の中央値: "
                         f"{sorted(acked)[len(acked) // 2]:.1f}ms")
        if events:
            lines.append(f"  WS通知の中央値  : "
                         f"{sorted(events)[len(events) // 2]:.1f}ms")
        lines.append("")
        lines.append(self.gateway.report())
        return "\n".join(lines)


async def main() -> None:
    from client import make_client

    c = make_client()
    loop = ExecutionLoop(c.http)
    await loop.start()

    r = c.http.get_tickers(category="linear", symbol="BTCUSDT")
    last = Decimal(r["result"]["list"][0]["lastPrice"])

    print("\n=== 成行で発注して約定確認まで（testnet） ===")
    stamp = int(time.time() * 1000) % 1_000_000

    for i in range(2):
        spec = OrderSpec(
            symbol="BTCUSDT", side="Buy", order_type="Market",
            qty="0.001", order_link_id=f"loop-{stamp}-{i}")

        lifecycle = await loop.submit(spec)
        print(f"  発注: {lifecycle.order_link_id} "
              f"({lifecycle.ack_latency_ms:.1f}ms)")

        filled = await loop.wait_for_fill(lifecycle, timeout_sec=10.0)
        print(f"  約定: {filled} "
              f"({lifecycle.filled_qty} @{lifecycle.avg_price})")

        await asyncio.sleep(1)

    print()
    print(loop.report())

    # 後片付け
    pos = c.http.get_positions(category="linear", symbol="BTCUSDT")
    for p in pos["result"]["list"]:
        size = Decimal(p["size"] or "0")
        if size > 0:
            c.http.place_order(
                category="linear", symbol="BTCUSDT",
                side="Sell" if p["side"] == "Buy" else "Buy",
                orderType="Market", qty=str(size), reduceOnly=True)
            print(f"\n  ポジション {size} を解消しました")

    await loop.stop()

    print("\n※ この構成なら発注から約定確認までRESTを1回も叩かない。")
    print("  ただしWSが切れている間は状態が分からなくなるため、")
    print("  記事24のウォッチドッグとRESTによる定期照合は必須。")


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

"""latency_tuning.py — 低遅延化の設定と効果測定
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  順序が大事:
    1. まずリージョンを見直す（効果が桁違いに大きい）
    2. 次に接続の使い回し（WS Trade / Keep-Alive）
    3. 最後に細かいチューニング（orjson・GC・TCP_NODELAY）

  3 から始めても、1 が悪ければ意味がない。
  「1ms を削る前に 100ms を削る」のが正しい順序。
"""
from __future__ import annotations

import asyncio
import gc
import json
import socket
import statistics
import time
from dataclasses import dataclass

try:
    import orjson
    HAS_ORJSON = True
except ImportError:
    HAS_ORJSON = False


# --- JSON ----------------------------------------------------------

def dumps(obj: dict) -> str:
    """orjson があれば使う。無ければ標準 json。"""
    if HAS_ORJSON:
        return orjson.dumps(obj).decode()
    return json.dumps(obj, separators=(",", ":"))


def loads(raw: str | bytes) -> dict:
    if HAS_ORJSON:
        return orjson.loads(raw)
    return json.loads(raw)


def bench_json(n: int = 100_000) -> None:
    """JSON処理の速度を比較する。"""
    payload = {
        "reqId": "abc-123-def-456",
        "header": {"X-BAPI-TIMESTAMP": "1750000000000",
                   "X-BAPI-RECV-WINDOW": "5000"},
        "op": "order.create",
        "args": [{"category": "linear", "symbol": "BTCUSDT",
                  "side": "Buy", "orderType": "Limit",
                  "qty": "0.001", "price": "120000.0",
                  "timeInForce": "PostOnly",
                  "orderLinkId": "bench-000001"}],
    }

    print("=== JSON符号化のベンチマーク ===")

    start = time.perf_counter()
    for _ in range(n):
        json.dumps(payload, separators=(",", ":"))
    std_time = time.perf_counter() - start
    print(f"  標準json : {std_time:.3f}秒 "
          f"({std_time / n * 1e6:.2f}μs/回)")

    if HAS_ORJSON:
        start = time.perf_counter()
        for _ in range(n):
            orjson.dumps(payload)
        orj_time = time.perf_counter() - start
        print(f"  orjson   : {orj_time:.3f}秒 "
              f"({orj_time / n * 1e6:.2f}μs/回)")
        print(f"  高速化   : {std_time / orj_time:.1f}倍 "
              f"（1回あたり {(std_time - orj_time) / n * 1e6:.2f}μs 削減）")
        print("\n  ※ 1回あたり数マイクロ秒。往復数十msの中では誤差。")
        print("    毎秒数千回の処理があるなら効いてくる。")
    else:
        print("  orjson 未インストール（pip install orjson）")


# --- TCP -----------------------------------------------------------

def check_tcp_nodelay() -> None:
    """TCP_NODELAY の効果を説明する。

    Nagleアルゴリズムは「小さいパケットをまとめて送る」最適化。
    帯域は節約できるが、最大40msの遅延が入る。
    発注のような小さくて急ぐデータには有害。
    """
    print("=== TCP_NODELAY ===")

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    default = s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
    print(f"  既定値: {default} "
          f"({'有効（Nagle無効）' if default else '無効（Nagle有効）'})")

    s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    print(f"  設定後: "
          f"{s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)}")
    s.close()

    print("\n  websockets ライブラリは既定で TCP_NODELAY を有効にする。")
    print("  requests / urllib3 は環境によるため、")
    print("  RESTを使う場合は明示的に設定するか、")
    print("  そもそもWS経路を主系にするのが確実。")


# --- GC ------------------------------------------------------------

@dataclass
class GcPause:
    generation: int
    duration_ms: float
    collected: int


def measure_gc_pauses(iterations: int = 200) -> list:
    """GCの停止時間を測る。"""
    pauses = []
    for _ in range(iterations):
        # ゴミを作る
        garbage = [{"a": i, "b": str(i), "c": [i] * 10}
                   for i in range(1000)]
        start = time.perf_counter()
        collected = gc.collect(0)
        elapsed = (time.perf_counter() - start) * 1000
        pauses.append(GcPause(0, elapsed, collected))
        del garbage
    return pauses


def tune_gc() -> None:
    """GCの設定を調整して効果を見る。"""
    print("=== GC の停止時間 ===")

    print(f"  現在の閾値: {gc.get_threshold()}")

    before = measure_gc_pauses(100)
    times = sorted(p.duration_ms for p in before)
    print(f"  既定設定  : p50={times[len(times) // 2]:.3f}ms  "
          f"p99={times[int(len(times) * 0.99)]:.3f}ms  "
          f"max={times[-1]:.3f}ms")

    # 第0世代の閾値を上げると、GCの実行回数が減る（1回は長くなる）
    gc.set_threshold(50_000, 50, 50)
    after = measure_gc_pauses(100)
    times2 = sorted(p.duration_ms for p in after)
    print(f"  閾値を上げた後: p50={times2[len(times2) // 2]:.3f}ms  "
          f"p99={times2[int(len(times2) * 0.99)]:.3f}ms  "
          f"max={times2[-1]:.3f}ms")

    gc.set_threshold(700, 10, 10)     # 既定値に戻す

    print("\n  【方針】")
    print("  - 発注の直前に gc.freeze() を呼ぶと、")
    print("    起動時に作ったオブジェクトがGCの対象から外れる")
    print("  - gc.disable() は危険（メモリが増え続ける）")
    print("  - 最も効くのは『そもそもオブジェクトを作らない』こと")
    print("    ループ内での dict / list の生成を減らす")


# --- イベントループ -------------------------------------------------

async def measure_loop_lag(duration_sec: float = 5.0,
                           interval_sec: float = 0.01) -> None:
    """イベントループの遅延を測る。

    「0.01秒後に起きる」と指定したタスクが、
    実際に何秒後に起きたかの差がループの詰まり具合。
    """
    print("=== イベントループの遅延 ===")

    lags = []
    deadline = time.time() + duration_sec

    while time.time() < deadline:
        start = time.perf_counter()
        await asyncio.sleep(interval_sec)
        actual = time.perf_counter() - start
        lags.append((actual - interval_sec) * 1000)

    lags_sorted = sorted(lags)
    n = len(lags_sorted)
    print(f"  測定回数: {n}")
    print(f"  p50={lags_sorted[n // 2]:.3f}ms  "
          f"p90={lags_sorted[int(n * 0.90)]:.3f}ms  "
          f"p99={lags_sorted[int(n * 0.99)]:.3f}ms  "
          f"max={lags_sorted[-1]:.3f}ms")
    print(f"  平均: {statistics.mean(lags):.3f}ms")

    if lags_sorted[int(n * 0.99)] > 10:
        print("\n  → p99が10msを超えています。")
        print("    重い同期処理がループを塞いでいる可能性があります。")
        print("    run_in_executor で別スレッドに逃がしてください。")
    else:
        print("\n  → ループは健全です。")


async def main() -> None:
    bench_json()
    print()
    check_tcp_nodelay()
    print()
    tune_gc()
    print()
    await measure_loop_lag()

    print("\n=== 優先順位 ===")
    print("  1. VPSのリージョン        効果: 100ms以上")
    print("  2. 接続の使い回し(WS)     効果: 10〜50ms")
    print("  3. イベントループの健全性 効果: 数ms〜数百ms（詰まっていれば）")
    print("  4. GCの調整               効果: 稀に10ms以上")
    print("  5. orjson / TCP_NODELAY   効果: 1ms未満〜40ms")
    print("\n  1 を放置して 5 を最適化しても意味がない。")


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

"""region_latency.py — 接続先とのネットワーク遅延を測る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  VPSを借りる前・借りた後に測る。
  「どのリージョンが速いか」は実測でしか分からない。

  測るもの:
    1. TCP接続の確立時間（3-way handshake の往復）
    2. TLSハンドシェイクを含む接続時間
    3. HTTPリクエストの往復（get_server_time は最も軽い認証不要API）
    4. サーバ時刻との差（時計のズレ。記事24のウォッチドッグと同じ）

  ping（ICMP）は経路が違うことがあるため、
  実際に使うTCPポートで測るのが正確。
"""
from __future__ import annotations

import socket
import ssl
import statistics
import time
from dataclasses import dataclass, field

import requests

ENDPOINTS = {
    "本番 REST": ("api.bybit.com", 443),
    "本番 REST(代替)": ("api.bytick.com", 443),
    "テストネット REST": ("api-testnet.bybit.com", 443),
    "本番 WS": ("stream.bybit.com", 443),
}


@dataclass
class ConnectionLatency:
    name: str
    host: str
    tcp_ms: list = field(default_factory=list)
    tls_ms: list = field(default_factory=list)
    http_ms: list = field(default_factory=list)
    errors: int = 0

    def _stats(self, values: list) -> tuple:
        if not values:
            return (0.0, 0.0, 0.0)
        s = sorted(values)
        return (s[len(s) // 2], s[min(int(len(s) * 0.9), len(s) - 1)], s[-1])

    def line(self) -> str:
        tcp = self._stats(self.tcp_ms)
        tls = self._stats(self.tls_ms)
        http = self._stats(self.http_ms)
        return (f"  {self.name:<20}"
                f"TCP p50={tcp[0]:>7.1f}  "
                f"TLS p50={tls[0]:>7.1f}  "
                f"HTTP p50={http[0]:>7.1f} p99={http[2]:>7.1f}  "
                f"err={self.errors}")


def measure_tcp(host: str, port: int, timeout: float = 5.0) -> float:
    """TCP接続の確立時間（ミリ秒）。"""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    start = time.perf_counter()
    try:
        s.connect((host, port))
        return (time.perf_counter() - start) * 1000
    finally:
        s.close()


def measure_tls(host: str, port: int, timeout: float = 5.0) -> float:
    """TLSハンドシェイクを含む接続時間（ミリ秒）。"""
    context = ssl.create_default_context()
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    start = time.perf_counter()
    try:
        wrapped = context.wrap_socket(s, server_hostname=host)
        wrapped.connect((host, port))
        elapsed = (time.perf_counter() - start) * 1000
        wrapped.close()
        return elapsed
    except Exception:
        s.close()
        raise


def measure_http(host: str, session: requests.Session,
                 timeout: float = 5.0) -> tuple:
    """HTTPリクエストの往復時間と、サーバ時刻との差。"""
    url = f"https://{host}/v5/market/time"
    start = time.perf_counter()
    r = session.get(url, timeout=timeout)
    elapsed = (time.perf_counter() - start) * 1000

    skew_ms = 0.0
    try:
        data = r.json()
        server_ms = int(data["result"]["timeNano"]) / 1_000_000
        local_ms = time.time() * 1000 - elapsed / 2
        skew_ms = local_ms - server_ms
    except Exception:
        pass

    return elapsed, skew_ms


def run(rounds: int = 10) -> None:
    print(f"=== 接続先レイテンシ測定 ({rounds}回) ===\n")

    results = []
    session = requests.Session()      # Keep-Alive を効かせる

    for name, (host, port) in ENDPOINTS.items():
        latency = ConnectionLatency(name, host)
        skews = []

        for i in range(rounds):
            try:
                latency.tcp_ms.append(measure_tcp(host, port))
            except Exception:
                latency.errors += 1

            try:
                latency.tls_ms.append(measure_tls(host, port))
            except Exception:
                latency.errors += 1

            if "REST" in name:
                try:
                    elapsed, skew = measure_http(host, session)
                    latency.http_ms.append(elapsed)
                    skews.append(skew)
                except Exception:
                    latency.errors += 1

            time.sleep(0.2)

        results.append((latency, skews))

    print(f"{'接続先':<20}{'':<20}{'':<20}{'':<30}  (ms)")
    print("-" * 92)
    for latency, _ in results:
        print(latency.line())

    print("\n=== 時計のズレ ===")
    for latency, skews in results:
        if skews:
            print(f"  {latency.name:<20}"
                  f"{statistics.median(skews):>+9.1f}ms")

    print("\n【判断の目安（HTTP p50）】")
    print("   〜 20ms : 取引所と同一リージョン。理想的")
    print("  20〜 60ms: 近隣リージョン。多くの戦略で十分")
    print("  60〜150ms: 遠い。MMや裁定には不利")
    print("  150ms〜  : 日本国内から直接など。低遅延戦略には不向き")

    print("\n【TCP と TLS の差】")
    print("  TLS − TCP がハンドシェイクの追加コスト。")
    print("  この差が大きいほど、接続を使い回す効果（WS Trade）が大きい。")

    for latency, _ in results:
        if latency.tcp_ms and latency.tls_ms:
            tcp_med = statistics.median(latency.tcp_ms)
            tls_med = statistics.median(latency.tls_ms)
            print(f"  {latency.name:<20}追加コスト "
                  f"{tls_med - tcp_med:>7.1f}ms")

    print("\n※ 時計のズレが±2秒を超えるならNTP同期を確認（記事38）。")
    print("  ±5秒を超えると署名が通らなくなる（retCode=10002）。")


if __name__ == "__main__":
    run(rounds=10)