Bybit WebSocket privateストリームで
注文・ポジション・残高を
リアルタイム同期する

{
  "req_id": "auth-1785000000000",
  "op": "auth",
  "args": [
    "YOUR_API_KEY",
    1785000005000,                    // expires（ミリ秒）= 現在時刻 + 数秒
    "3f2a...署名..."                   // HMAC-SHA256(secret, "GET/realtime1785000005000")
  ]
}

成功時のレスポンス:
{
  "success": true,
  "ret_msg": "",
  "op": "auth",
  "conn_id": "cxxxxxxx-xxxx-xxxx"
}

"""private_ws_min.py — private WebSocketの最小構成（pybit）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time

from pybit.unified_trading import WebSocket

from config import load_credentials

cred = load_credentials()

# channel_type="private" を指定し、キーを渡すと自動で認証される。
# demo環境では demo=True（接続先が stream-demo になる）
ws = WebSocket(
    channel_type="private",
    testnet=(cred.env.value == "testnet"),
    demo=(cred.env.value == "demo"),
    api_key=cred.api_key,
    api_secret=cred.api_secret,
    ping_interval=20,          # 公式推奨
    ping_timeout=10,
    retries=10,
    restart_on_error=True,
)


def on_order(msg: dict) -> None:
    for o in msg.get("data", []):
        print(f"[order]  {o['symbol']:<11} {o['side']:<5} "
              f"{o['orderStatus']:<18} qty={o['qty']} "
              f"約定={o.get('cumExecQty')} link={o.get('orderLinkId')}")


def on_execution(msg: dict) -> None:
    for e in msg.get("data", []):
        # execTypeにはTrade以外（Funding / Settle / AdlTrade等）も来る
        if e.get("execType") != "Trade":
            print(f"[exec*]  {e['symbol']:<11} execType={e['execType']}")
            continue
        maker = "Maker" if e.get("isMaker") else "Taker"
        print(f"[exec]   {e['symbol']:<11} {e['side']:<5} "
              f"{e['execQty']}@{e['execPrice']} {maker} "
              f"fee={e.get('execFee')} pnl={e.get('execPnl', '-')}")


def on_position(msg: dict) -> None:
    for p in msg.get("data", []):
        # 建値は entryPrice。RESTの avgPrice とは名前が違う
        print(f"[pos]    {p['symbol']:<11} {p.get('side', '-'):<5} "
              f"size={p['size']} 建値={p.get('entryPrice')} "
              f"含み損益={p.get('unrealisedPnl')}")


def on_wallet(msg: dict) -> None:
    for w in msg.get("data", []):
        print(f"[wallet] 総資産={w.get('totalEquity')} "
              f"利用可能={w.get('totalAvailableBalance')} "
              f"維持率={w.get('accountMMRate')}")


ws.order_stream(callback=on_order)
ws.execution_stream(callback=on_execution)
ws.position_stream(callback=on_position)
ws.wallet_stream(callback=on_wallet)

print("購読を開始しました（発注すると通知が届きます）")
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    ws.exit()

"""ws_private.py — asyncioによるprivate WebSocketクライアント
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install websockets orjson
"""
from __future__ import annotations

import asyncio
import contextlib
import hashlib
import hmac
import random
import time
from collections.abc import Awaitable, Callable

import orjson
import websockets

PING_INTERVAL = 20.0
# max_active_time は 30s〜600s。長くすると再接続の頻度が下がる
PRIVATE_URLS = {
    "mainnet": "wss://stream.bybit.com/v5/private?max_active_time=600s",
    "testnet": "wss://stream-testnet.bybit.com/v5/private?max_active_time=600s",
    "demo": "wss://stream-demo.bybit.com/v5/private?max_active_time=600s",
}

Handler = Callable[[dict], Awaitable[None] | None]


def ws_signature(api_secret: str, expires_ms: int) -> str:
    """WebSocket認証の署名。対象は 'GET/realtime' + expires のみ。

    RESTの署名（timestamp+apiKey+recvWindow+body）とは別物。
    """
    message = f"GET/realtime{expires_ms}"
    return hmac.new(api_secret.strip().encode("utf-8"),
                    message.encode("utf-8"), hashlib.sha256).hexdigest()


class BybitPrivateWS:
    def __init__(self, api_key: str, api_secret: str, env: str = "testnet",
                 auth_valid_sec: float = 5.0):
        self.api_key = api_key.strip()
        self.api_secret = api_secret.strip()
        self.url = PRIVATE_URLS[env]
        self.auth_valid_sec = auth_valid_sec
        self._handlers: dict[str, list[Handler]] = {}
        self._ws = None
        self.last_msg_at: float = 0.0
        self.reconnects = 0
        self.authenticated = False
        self.on_reconnect: Callable[[], Awaitable[None] | None] | None = None

    def subscribe(self, topic: str, handler: Handler) -> None:
        self._handlers.setdefault(topic, []).append(handler)

    # ---- 認証 -----------------------------------------------------------
    async def _authenticate(self) -> None:
        expires = int((time.time() + self.auth_valid_sec) * 1000)
        await self._ws.send(orjson.dumps({
            "req_id": f"auth-{expires}",
            "op": "auth",
            "args": [self.api_key, expires,
                     ws_signature(self.api_secret, expires)],
        }))

        raw = await asyncio.wait_for(self._ws.recv(), timeout=10)
        reply = orjson.loads(raw)
        if not reply.get("success", False):
            raise PermissionError(f"WS認証に失敗: {reply}")
        self.authenticated = True
        print(f"[ws-private] 認証成功 conn_id={reply.get('conn_id')}")

    # ---- ping -----------------------------------------------------------
    async def _ping_loop(self) -> None:
        while True:
            await asyncio.sleep(PING_INTERVAL)
            if self._ws is None:
                return
            with contextlib.suppress(Exception):
                await self._ws.send(orjson.dumps({"op": "ping"}))

    # ---- 受信 -----------------------------------------------------------
    async def _recv_loop(self) -> None:
        async for raw in self._ws:
            self.last_msg_at = time.time()
            msg = orjson.loads(raw)

            if "op" in msg and "topic" not in msg:
                if msg["op"] == "subscribe" and not msg.get("success", True):
                    print(f"[ws-private] 購読失敗: {msg}")
                continue

            topic = msg.get("topic", "")
            # 'execution.linear' も 'execution' のハンドラで受けられるようにする
            for key, handlers in self._handlers.items():
                if topic == key or topic.startswith(key + "."):
                    for h in handlers:
                        result = h(msg)
                        if asyncio.iscoroutine(result):
                            await result

    # ---- 本体 -----------------------------------------------------------
    async def run(self) -> None:
        backoff = 1.0
        first = True
        while True:
            try:
                async with websockets.connect(
                    self.url, ping_interval=None, open_timeout=10,
                    close_timeout=5, max_queue=1024,
                ) as ws:
                    self._ws = ws
                    self.last_msg_at = time.time()
                    await self._authenticate()

                    topics = sorted(self._handlers)
                    await ws.send(orjson.dumps({"op": "subscribe",
                                                "args": topics}))
                    print(f"[ws-private] 購読: {topics}")

                    # 再接続時は取りこぼしがあるのでRESTで補完させる
                    if not first and self.on_reconnect:
                        result = self.on_reconnect()
                        if asyncio.iscoroutine(result):
                            await result
                    first = False

                    ping = asyncio.create_task(self._ping_loop())
                    try:
                        await self._recv_loop()
                    finally:
                        ping.cancel()
                        with contextlib.suppress(asyncio.CancelledError):
                            await ping

                backoff = 1.0

            except asyncio.CancelledError:
                raise
            except PermissionError:
                raise            # 認証エラーは再試行しても直らない
            except Exception as e:
                self.reconnects += 1
                self.authenticated = False
                wait = min(backoff, 30.0) * (0.5 + random.random())
                print(f"[ws-private] 切断 ({type(e).__name__}: {e}) -> "
                      f"{wait:.1f}秒後に再接続 (通算{self.reconnects}回)")
                await asyncio.sleep(wait)
                backoff = min(backoff * 2, 30.0)
            finally:
                self._ws = None


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

    cred = load_credentials()
    ws = BybitPrivateWS(cred.api_key, cred.api_secret, cred.env.value)

    async def show(msg: dict) -> None:
        print(f"[{msg['topic']}] {len(msg.get('data', []))}件")

    for topic in ("order", "execution", "position", "wallet"):
        ws.subscribe(topic, show)

    await ws.run()


if __name__ == "__main__":
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(main())

"""account_mirror.py — 口座状態のローカルミラー（WS即時更新 + REST照合）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

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

from position import Position, parse_position


def _d(v, default: str = "0") -> Decimal:
    if v in (None, "", "null"):
        return Decimal(default)
    return Decimal(str(v))


@dataclass
class WalletState:
    total_equity: Decimal = Decimal(0)
    total_margin_balance: Decimal = Decimal(0)
    total_available: Decimal = Decimal(0)
    total_im: Decimal = Decimal(0)
    total_mm: Decimal = Decimal(0)
    total_perp_upl: Decimal = Decimal(0)
    account_mm_rate: Decimal = Decimal(0)
    account_im_rate: Decimal = Decimal(0)
    coins: dict[str, Decimal] = field(default_factory=dict)
    updated_at: float = 0.0
    source: str = ""          # "rest" or "ws"

    @property
    def age_sec(self) -> float:
        return time.time() - self.updated_at if self.updated_at else float("inf")


@dataclass
class OrderState:
    order_id: str
    link_id: str
    symbol: str
    side: str
    status: str
    qty: Decimal
    price: Decimal
    filled_qty: Decimal
    avg_price: Decimal
    updated_at: float = field(default_factory=time.time)

    @property
    def is_open(self) -> bool:
        return self.status in ("New", "PartiallyFilled", "Untriggered",
                               "Triggered")


@dataclass
class FillRecord:
    exec_id: str
    symbol: str
    side: str
    qty: Decimal
    price: Decimal
    fee: Decimal
    is_maker: bool
    exec_pnl: Decimal
    link_id: str
    exec_time: int


class AccountMirror:
    """WSイベントで更新される口座状態。読み取りはここからのみ行う。"""

    def __init__(self, client, max_fills: int = 2000):
        self.c = client
        self.wallet = WalletState()
        self.positions: dict[tuple, Position] = {}
        self.orders: dict[str, OrderState] = {}
        self.fills: list[FillRecord] = []
        self._seen_exec: set[str] = set()
        self._max_fills = max_fills
        self._lock = threading.RLock()
        self.counters = {"order": 0, "execution": 0, "position": 0,
                         "wallet": 0, "dup_exec": 0}
        self.last_event_at = 0.0
        self.on_fill = None       # 約定時に呼ぶコールバック

    # ---- 各トピックのハンドラ -------------------------------------------
    def on_order(self, msg: dict) -> None:
        with self._lock:
            for o in msg.get("data", []):
                st = OrderState(
                    order_id=o.get("orderId", ""),
                    link_id=o.get("orderLinkId", ""),
                    symbol=o.get("symbol", ""),
                    side=o.get("side", ""),
                    status=o.get("orderStatus", ""),
                    qty=_d(o.get("qty")),
                    price=_d(o.get("price")),
                    filled_qty=_d(o.get("cumExecQty")),
                    avg_price=_d(o.get("avgPrice")),
                )
                if st.is_open:
                    self.orders[st.order_id] = st
                else:
                    self.orders.pop(st.order_id, None)   # 終了したら外す
            self.counters["order"] += 1
            self.last_event_at = time.time()

    def on_execution(self, msg: dict) -> None:
        new_fills: list[FillRecord] = []
        with self._lock:
            for e in msg.get("data", []):
                exec_id = e.get("execId", "")
                # 再接続時の重複配信に備えて約定IDで排除する
                if exec_id and exec_id in self._seen_exec:
                    self.counters["dup_exec"] += 1
                    continue
                if exec_id:
                    self._seen_exec.add(exec_id)

                # Funding / Settle / AdlTrade も届くので区別する
                if e.get("execType") != "Trade":
                    continue

                rec = FillRecord(
                    exec_id=exec_id, symbol=e.get("symbol", ""),
                    side=e.get("side", ""), qty=_d(e.get("execQty")),
                    price=_d(e.get("execPrice")), fee=_d(e.get("execFee")),
                    is_maker=bool(e.get("isMaker")),
                    exec_pnl=_d(e.get("execPnl")),
                    link_id=e.get("orderLinkId", ""),
                    exec_time=int(e.get("execTime") or 0),
                )
                self.fills.append(rec)
                new_fills.append(rec)

            if len(self.fills) > self._max_fills:
                del self.fills[: len(self.fills) - self._max_fills]
            if len(self._seen_exec) > 20_000:
                self._seen_exec.clear()

            self.counters["execution"] += 1
            self.last_event_at = time.time()

        # コールバックはロックの外で呼ぶ（デッドロック回避）
        if self.on_fill:
            for rec in new_fills:
                try:
                    self.on_fill(rec)
                except Exception as e:
                    print(f"[mirror] on_fillでエラー: {type(e).__name__}: {e}")

    def on_position(self, msg: dict) -> None:
        with self._lock:
            for raw in msg.get("data", []):
                try:
                    p = parse_position(raw, raw.get("category", "linear"))
                except Exception:
                    continue
                old = self.positions.get(p.key)
                if old and p.seq and old.seq and p.seq < old.seq:
                    continue          # 古いイベントは無視
                if p.size > 0:
                    self.positions[p.key] = p
                else:
                    self.positions.pop(p.key, None)
            self.counters["position"] += 1
            self.last_event_at = time.time()

    def on_wallet(self, msg: dict) -> None:
        with self._lock:
            for w in msg.get("data", []):
                self.wallet = WalletState(
                    total_equity=_d(w.get("totalEquity")),
                    total_margin_balance=_d(w.get("totalMarginBalance")),
                    total_available=_d(w.get("totalAvailableBalance")),
                    total_im=_d(w.get("totalInitialMargin")),
                    total_mm=_d(w.get("totalMaintenanceMargin")),
                    total_perp_upl=_d(w.get("totalPerpUPL")),
                    account_mm_rate=_d(w.get("accountMMRate")),
                    account_im_rate=_d(w.get("accountIMRate")),
                    coins={c["coin"]: _d(c.get("walletBalance"))
                           for c in (w.get("coin") or [])},
                    updated_at=time.time(), source="ws",
                )
            self.counters["wallet"] += 1
            self.last_event_at = time.time()

    # ---- REST初期化・照合 ------------------------------------------------
    def refresh_from_rest(self) -> None:
        """RESTで全状態を取り直す。起動時と再接続時に呼ぶ。"""
        from position import fetch_positions

        # 1) 残高（walletトピックはスナップショットを送らないので必須）
        res = self.c.http.get_wallet_balance(accountType="UNIFIED")
        if res.get("retCode") == 0:
            a = res["result"]["list"][0]
            with self._lock:
                self.wallet = WalletState(
                    total_equity=_d(a.get("totalEquity")),
                    total_margin_balance=_d(a.get("totalMarginBalance")),
                    total_available=_d(a.get("totalAvailableBalance")),
                    total_im=_d(a.get("totalInitialMargin")),
                    total_mm=_d(a.get("totalMaintenanceMargin")),
                    total_perp_upl=_d(a.get("totalPerpUPL")),
                    account_mm_rate=_d(a.get("accountMMRate")),
                    account_im_rate=_d(a.get("accountIMRate")),
                    coins={c["coin"]: _d(c.get("walletBalance"))
                           for c in (a.get("coin") or [])},
                    updated_at=time.time(), source="rest",
                )

        # 2) ポジション
        with self._lock:
            self.positions = {p.key: p for p in fetch_positions(self.c.http)}

        # 3) 未約定注文
        fresh: dict[str, OrderState] = {}
        for category, extra in (("linear", {"settleCoin": "USDT"}),
                                ("linear", {"settleCoin": "USDC"}),
                                ("spot", {})):
            r = self.c.http.get_open_orders(category=category, limit=50,
                                            **extra)
            if r.get("retCode") != 0:
                continue
            for o in r["result"]["list"]:
                fresh[o.get("orderId", "")] = OrderState(
                    order_id=o.get("orderId", ""),
                    link_id=o.get("orderLinkId", ""),
                    symbol=o.get("symbol", ""), side=o.get("side", ""),
                    status=o.get("orderStatus", ""),
                    qty=_d(o.get("qty")), price=_d(o.get("price")),
                    filled_qty=_d(o.get("cumExecQty")),
                    avg_price=_d(o.get("avgPrice")),
                )
        with self._lock:
            self.orders = fresh

    # ---- 読み取り --------------------------------------------------------
    def open_positions(self) -> list[Position]:
        with self._lock:
            return [p for p in self.positions.values() if p.size > 0]

    def open_orders(self, prefix: str | None = None) -> list[OrderState]:
        with self._lock:
            rows = [o for o in self.orders.values() if o.is_open]
        if prefix:
            rows = [o for o in rows if o.link_id.startswith(prefix)]
        return rows

    def recent_fills(self, n: int = 20) -> list[FillRecord]:
        with self._lock:
            return self.fills[-n:]

    def fee_summary(self) -> dict:
        with self._lock:
            fills = list(self.fills)
        maker = [f for f in fills if f.is_maker]
        taker = [f for f in fills if not f.is_maker]
        return {
            "約定数": len(fills),
            "Maker": len(maker),
            "Taker": len(taker),
            "Maker率": f"{len(maker) / len(fills):.1%}" if fills else "-",
            "手数料合計": sum((f.fee for f in fills), Decimal(0)),
            "実現損益": sum((f.exec_pnl for f in fills), Decimal(0)),
        }

    def status(self) -> str:
        w = self.wallet
        age = time.time() - self.last_event_at if self.last_event_at else -1
        return (f"総資産 {w.total_equity:,.2f} / 利用可能 "
                f"{w.total_available:,.2f} / 維持率 "
                f"{w.account_mm_rate * 100:.3f}% ({w.source}, "
                f"{w.age_sec:.0f}秒前)\n"
                f"  建玉 {len(self.open_positions())} / 未約定 "
                f"{len(self.open_orders())} / 約定 {len(self.fills)}\n"
                f"  イベント {self.counters} / 最終 {age:.0f}秒前")


if __name__ == "__main__":
    from client import make_client
    from pybit.unified_trading import WebSocket

    c = make_client()
    mirror = AccountMirror(c)

    # 1) RESTで初期化（walletにスナップショットが無いため必須）
    mirror.refresh_from_rest()
    print("初期状態:")
    print(mirror.status())

    # 2) WSで即時更新
    ws = WebSocket(channel_type="private",
                   testnet=(c.env.value == "testnet"),
                   demo=(c.env.value == "demo"),
                   api_key=c.cred.api_key, api_secret=c.cred.api_secret)
    ws.order_stream(callback=mirror.on_order)
    ws.execution_stream(callback=mirror.on_execution)
    ws.position_stream(callback=mirror.on_position)
    ws.wallet_stream(callback=mirror.on_wallet)

    try:
        while True:
            time.sleep(15)
            print(f"\n[{time.strftime('%H:%M:%S')}]")
            print(mirror.status())
            if mirror.fills:
                print(f"  手数料内訳: {mirror.fee_summary()}")
    except KeyboardInterrupt:
        ws.exit()

"""equity_estimator.py — walletイベントの隙間を埋めて純資産を推定する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

背景: walletトピックは「含み損益の変化では発火しない」。
      価格が動いて含み損が膨らんでも通知が来ないため、
      positionトピックのマーク価格から自前で推定する必要がある。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass
from decimal import Decimal

from account_mirror import AccountMirror


@dataclass
class EquityEstimate:
    wallet_balance: Decimal      # 確定済み残高（walletイベント由来）
    unrealised: Decimal          # positionから計算した含み損益
    estimated_equity: Decimal
    maintenance_margin: Decimal
    estimated_mm_rate: Decimal
    rest_mm_rate: Decimal        # RESTで取得した実測値
    drift: Decimal               # 推定と実測の差
    positions: int

    @property
    def is_dangerous(self) -> bool:
        return self.estimated_mm_rate > Decimal("0.5")

    def line(self) -> str:
        flag = "  ← 危険水準" if self.is_dangerous else ""
        return (f"推定純資産 {self.estimated_equity:>12,.2f} "
                f"(残高 {self.wallet_balance:,.2f} + 含み "
                f"{self.unrealised:+,.2f})\n"
                f"  推定維持率 {self.estimated_mm_rate * 100:>7.3f}%  "
                f"実測 {self.rest_mm_rate * 100:>7.3f}%  "
                f"乖離 {self.drift * 100:+.3f}pt{flag}")


class EquityEstimator:
    """WSのpositionイベントから純資産と維持率を推定し、RESTで補正する。"""

    def __init__(self, mirror: AccountMirror, rest_interval_sec: float = 30.0):
        self.mirror = mirror
        self.rest_interval = rest_interval_sec
        self._stop = threading.Event()
        self._last_rest = 0.0

    def estimate(self) -> EquityEstimate:
        w = self.mirror.wallet
        positions = self.mirror.open_positions()

        # 確定残高 = 総証拠金 - 無期限の含み損益（含みを除いた部分）
        wallet_balance = w.total_margin_balance - w.total_perp_upl

        # positionトピックの最新マーク価格から含み損益を再計算する。
        # これがwalletイベントの隙間を埋める部分。
        unrealised = sum((p.unrealised_pnl for p in positions), Decimal(0))
        maintenance = sum((p.position_mm for p in positions), Decimal(0))

        equity = wallet_balance + unrealised
        est_rate = (maintenance / equity) if equity > 0 else Decimal(0)

        return EquityEstimate(
            wallet_balance=wallet_balance, unrealised=unrealised,
            estimated_equity=equity, maintenance_margin=maintenance,
            estimated_mm_rate=est_rate, rest_mm_rate=w.account_mm_rate,
            drift=est_rate - w.account_mm_rate, positions=len(positions),
        )

    def start(self, on_danger=None) -> None:
        """定期的にRESTで補正し、危険水準なら通知する。"""

        def loop() -> None:
            while not self._stop.wait(self.rest_interval):
                try:
                    # walletは含みで発火しないので、RESTで定期的に取り直す
                    self.mirror.refresh_from_rest()
                    self._last_rest = time.time()

                    est = self.estimate()
                    if est.is_dangerous and on_danger:
                        on_danger(est)
                    if abs(est.drift) > Decimal("0.05"):
                        print(f"[equity] 推定と実測の乖離が大きい: "
                              f"{est.drift * 100:+.2f}pt")
                except Exception as e:
                    print(f"[equity] 補正に失敗: {type(e).__name__}: {e}")

        threading.Thread(target=loop, daemon=True, name="equity").start()

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


if __name__ == "__main__":
    from client import make_client
    from pybit.unified_trading import WebSocket

    c = make_client()
    mirror = AccountMirror(c)
    mirror.refresh_from_rest()

    est = EquityEstimator(mirror, rest_interval_sec=30.0)

    ws = WebSocket(channel_type="private",
                   testnet=(c.env.value == "testnet"),
                   demo=(c.env.value == "demo"),
                   api_key=c.cred.api_key, api_secret=c.cred.api_secret)
    ws.position_stream(callback=mirror.on_position)
    ws.wallet_stream(callback=mirror.on_wallet)

    def alert(e: EquityEstimate) -> None:
        print(f"[!!] 維持率が危険水準です\n{e.line()}")

    est.start(on_danger=alert)

    try:
        while True:
            time.sleep(10)
            print(f"\n[{time.strftime('%H:%M:%S')}] {est.estimate().line()}")
    except KeyboardInterrupt:
        est.stop()
        ws.exit()

"""on_fill_actions.py — 約定を起点にTP/SLを設定し、記録を残す
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

注意: コールバック内で重い処理・REST呼び出しをすると受信が詰まる。
      キューに載せて別スレッドで処理する（記事26と同じ原則）。
"""
from __future__ import annotations

import queue
import threading
import time
from dataclasses import dataclass
from decimal import Decimal

from account_mirror import AccountMirror, FillRecord


@dataclass
class FillPolicy:
    """約定時に何をするか。"""

    tp_pct: Decimal = Decimal("2.0")
    sl_pct: Decimal = Decimal("1.0")
    set_tpsl: bool = True
    only_prefix: str | None = None      # この戦略の約定にのみ反応する
    min_qty_for_tpsl: Decimal = Decimal("0")


class FillReactor:
    """約定イベントを受けて後続処理を実行する（別スレッドで処理）。"""

    def __init__(self, client, mirror: AccountMirror,
                 policy: FillPolicy | None = None, registry=None):
        self.c = client
        self.mirror = mirror
        self.policy = policy or FillPolicy()
        self.registry = registry
        self._q: queue.Queue[FillRecord] = queue.Queue(maxsize=1000)
        self._stop = threading.Event()
        self.processed = 0
        self.errors = 0

    # ---- コールバック（絶対にブロックしない） ---------------------------
    def enqueue(self, fill: FillRecord) -> None:
        if self.policy.only_prefix and \
           not fill.link_id.startswith(self.policy.only_prefix):
            return
        try:
            self._q.put_nowait(fill)
        except queue.Full:
            print("[reactor] キューが満杯。約定処理が追いついていません")

    # ---- ワーカー --------------------------------------------------------
    def _worker(self) -> None:
        while not self._stop.is_set():
            try:
                fill = self._q.get(timeout=0.5)
            except queue.Empty:
                continue
            try:
                self._handle(fill)
                self.processed += 1
            except Exception as e:
                self.errors += 1
                print(f"[reactor] 処理失敗 {fill.exec_id}: "
                      f"{type(e).__name__}: {e}")
            finally:
                self._q.task_done()

    def _handle(self, fill: FillRecord) -> None:
        maker = "Maker" if fill.is_maker else "Taker"
        print(f"[reactor] 約定 {fill.symbol} {fill.side} "
              f"{fill.qty}@{fill.price} {maker} "
              f"fee={fill.fee} pnl={fill.exec_pnl:+}")

        if not self.policy.set_tpsl:
            return

        # 約定でポジションが増えた場合のみTP/SLを設定する。
        # 決済約定（execPnl != 0）では設定しない。
        if fill.exec_pnl != 0:
            return

        # ポジションが反映されるまで少し待つ（約定通知が先に来ることがある）
        for _ in range(5):
            positions = [p for p in self.mirror.open_positions()
                         if p.symbol == fill.symbol]
            if positions:
                break
            time.sleep(0.3)
        else:
            print(f"[reactor] {fill.symbol}: ポジションが反映されません")
            return

        pos = positions[0]
        if pos.size < self.policy.min_qty_for_tpsl:
            return
        if pos.has_stop:
            return          # 既に設定済み

        from tpsl import set_tpsl_from_entry

        result = set_tpsl_from_entry(
            self.c.http, fill.symbol,
            tp_pct=self.policy.tp_pct, sl_pct=self.policy.sl_pct,
            category=pos.category,
        )
        print(f"[reactor] TP/SL設定: TP={result.take_profit} "
              f"SL={result.stop_loss} (RR {result.rr:.2f})")

    # ---- 起動・停止 ------------------------------------------------------
    def start(self, workers: int = 2) -> None:
        self.mirror.on_fill = self.enqueue
        for i in range(workers):
            threading.Thread(target=self._worker, daemon=True,
                             name=f"reactor-{i}").start()
        print(f"[reactor] 約定リアクタを起動しました（worker {workers}）")

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


if __name__ == "__main__":
    from client import make_client
    from pybit.unified_trading import WebSocket
    from symbol_rules import SymbolRegistry

    c = make_client()
    mirror = AccountMirror(c)
    mirror.refresh_from_rest()

    reactor = FillReactor(
        c, mirror,
        FillPolicy(tp_pct=Decimal("2"), sl_pct=Decimal("1"),
                   only_prefix=None),
        registry=SymbolRegistry(c.http),
    )
    reactor.start()

    ws = WebSocket(channel_type="private",
                   testnet=(c.env.value == "testnet"),
                   demo=(c.env.value == "demo"),
                   api_key=c.cred.api_key, api_secret=c.cred.api_secret)
    ws.execution_stream(callback=mirror.on_execution)
    ws.position_stream(callback=mirror.on_position)
    ws.order_stream(callback=mirror.on_order)

    print("約定を待機しています（testnetで発注してみてください）")
    try:
        while True:
            time.sleep(20)
            print(f"[reactor] 処理済 {reactor.processed} / "
                  f"エラー {reactor.errors}")
    except KeyboardInterrupt:
        reactor.stop()
        ws.exit()

"""gap_recovery.py — 切断中に起きた約定をRESTで拾い直す
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

get_executions の時間範囲は最大7日。
切断が短時間であれば、直近の約定を取り直せば十分。
"""
from __future__ import annotations

import time
from decimal import Decimal

from account_mirror import AccountMirror, FillRecord


def _d(v, default: str = "0") -> Decimal:
    if v in (None, "", "null"):
        return Decimal(default)
    return Decimal(str(v))


class GapRecovery:
    """切断区間の約定をRESTで取得し、ミラーに反映する。"""

    def __init__(self, client, mirror: AccountMirror):
        self.c = client
        self.mirror = mirror
        self.disconnected_at: float | None = None
        self.recovered_total = 0

    def mark_disconnected(self) -> None:
        if self.disconnected_at is None:
            self.disconnected_at = time.time()
            print(f"[gap] 切断を記録しました")

    def recover(self, categories=("linear", "spot"),
                lookback_sec: float = 600.0) -> int:
        """切断区間（不明なら直近lookback_sec）の約定を拾い直す。"""
        since = self.disconnected_at or (time.time() - lookback_sec)
        # 少し余裕を持たせて取りこぼしを防ぐ
        start_ms = int((since - 30) * 1000)
        end_ms = int(time.time() * 1000)

        recovered = 0
        for category in categories:
            cursor = None
            for _ in range(50):          # 安全上限
                kw: dict = {"category": category, "limit": 100,
                            "startTime": start_ms, "endTime": end_ms}
                if cursor:
                    kw["cursor"] = cursor
                res = self.c.http.get_executions(**kw)
                if res.get("retCode") != 0:
                    print(f"[gap] {category}: 取得失敗 {res.get('retMsg')}")
                    break

                rows = res["result"]["list"]
                for e in rows:
                    if e.get("execType") != "Trade":
                        continue
                    exec_id = e.get("execId", "")
                    # ミラーが既に持っている約定は追加しない
                    with self.mirror._lock:
                        if exec_id in self.mirror._seen_exec:
                            continue
                        self.mirror._seen_exec.add(exec_id)
                        self.mirror.fills.append(FillRecord(
                            exec_id=exec_id, symbol=e.get("symbol", ""),
                            side=e.get("side", ""),
                            qty=_d(e.get("execQty")),
                            price=_d(e.get("execPrice")),
                            fee=_d(e.get("execFee")),
                            is_maker=bool(e.get("isMaker")),
                            exec_pnl=_d(e.get("execPnl")),
                            link_id=e.get("orderLinkId", ""),
                            exec_time=int(e.get("execTime") or 0),
                        ))
                    recovered += 1

                cursor = res["result"].get("nextPageCursor")
                if not cursor or not rows:
                    break

        # 状態も取り直す（ポジション・注文・残高）
        self.mirror.refresh_from_rest()

        self.disconnected_at = None
        self.recovered_total += recovered
        print(f"[gap] {recovered}件の約定を補完し、状態を再同期しました")
        return recovered

    def on_reconnect(self) -> None:
        """WSクライアントの on_reconnect に渡す。"""
        self.recover()


if __name__ == "__main__":
    import asyncio

    from client import make_client
    from ws_private import BybitPrivateWS

    c = make_client()
    mirror = AccountMirror(c)
    mirror.refresh_from_rest()
    gap = GapRecovery(c, mirror)

    ws = BybitPrivateWS(c.cred.api_key, c.cred.api_secret, c.env.value)
    ws.subscribe("order", mirror.on_order)
    ws.subscribe("execution", mirror.on_execution)
    ws.subscribe("position", mirror.on_position)
    ws.subscribe("wallet", mirror.on_wallet)

    # 再接続時に自動で補完させる
    ws.on_reconnect = gap.on_reconnect

    async def monitor() -> None:
        while True:
            await asyncio.sleep(20)
            print(f"\n{mirror.status()}")
            print(f"  補完した約定: 通算 {gap.recovered_total}件")

    async def main() -> None:
        await asyncio.gather(ws.run(), monitor())

    asyncio.run(main())

"""fill_latency.py — 発注から約定通知が届くまでのレイテンシを測る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

REST発注 -> WS約定通知 の一巡の遅延を測る。
戦略の反応速度の上限がここで決まる。
"""
from __future__ import annotations

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


@dataclass
class LatencySample:
    link_id: str
    sent_at: float
    acked_at: float | None = None      # orderイベント受信
    filled_at: float | None = None     # executionイベント受信
    exec_time_ms: int = 0              # 取引所側の約定時刻

    @property
    def ack_ms(self) -> float | None:
        if self.acked_at is None:
            return None
        return (self.acked_at - self.sent_at) * 1000

    @property
    def fill_ms(self) -> float | None:
        if self.filled_at is None:
            return None
        return (self.filled_at - self.sent_at) * 1000

    @property
    def notify_ms(self) -> float | None:
        """取引所の約定時刻から、通知が届くまでの片道遅延。"""
        if self.filled_at is None or not self.exec_time_ms:
            return None
        return self.filled_at * 1000 - self.exec_time_ms


class FillLatencyMonitor:
    def __init__(self):
        self.samples: dict[str, LatencySample] = {}
        self._lock = threading.Lock()

    def mark_sent(self, link_id: str) -> None:
        with self._lock:
            self.samples[link_id] = LatencySample(link_id=link_id,
                                                 sent_at=time.time())

    def on_order(self, msg: dict) -> None:
        now = time.time()
        with self._lock:
            for o in msg.get("data", []):
                s = self.samples.get(o.get("orderLinkId", ""))
                if s and s.acked_at is None:
                    s.acked_at = now

    def on_execution(self, msg: dict) -> None:
        now = time.time()
        with self._lock:
            for e in msg.get("data", []):
                if e.get("execType") != "Trade":
                    continue
                s = self.samples.get(e.get("orderLinkId", ""))
                if s and s.filled_at is None:
                    s.filled_at = now
                    s.exec_time_ms = int(e.get("execTime") or 0)

    def report(self) -> str:
        with self._lock:
            rows = list(self.samples.values())

        def stats(values: list[float], label: str) -> str:
            if len(values) < 2:
                return f"  {label:<26} サンプル不足 ({len(values)})"
            s = sorted(values)
            p90 = s[min(int(len(s) * 0.9), len(s) - 1)]
            return (f"  {label:<26} n={len(s):<4} "
                    f"中央値 {statistics.median(s):>8.1f}ms  "
                    f"p90 {p90:>8.1f}ms  最大 {s[-1]:>8.1f}ms")

        return "\n".join([
            "発注からの経過時間:",
            stats([r.ack_ms for r in rows if r.ack_ms is not None],
                  "受理通知(order)まで"),
            stats([r.fill_ms for r in rows if r.fill_ms is not None],
                  "約定通知(execution)まで"),
            stats([r.notify_ms for r in rows
                   if r.notify_ms is not None], "約定時刻からの通知遅延"),
        ])


if __name__ == "__main__":
    from decimal import Decimal

    from client import make_client
    from pybit.unified_trading import WebSocket

    c = make_client()
    mon = FillLatencyMonitor()

    ws = WebSocket(channel_type="private",
                   testnet=(c.env.value == "testnet"),
                   demo=(c.env.value == "demo"),
                   api_key=c.cred.api_key, api_secret=c.cred.api_secret)
    ws.order_stream(callback=mon.on_order)
    ws.execution_stream(callback=mon.on_execution)
    time.sleep(2)      # 購読の確立を待つ

    # 最小ロットの成行を数回出して測る（testnetまたはdemoで実行）
    for i in range(5):
        link_id = f"lat-{int(time.time() * 1000)}"
        mon.mark_sent(link_id)
        res = c.http.place_order(
            category="linear", symbol="BTCUSDT",
            side="Buy" if i % 2 == 0 else "Sell",
            orderType="Market", qty="0.001",
            reduceOnly=(i % 2 == 1),      # 交互に建てて決済する
            orderLinkId=link_id,
        )
        if res["retCode"] != 0:
            print(f"発注失敗: {res['retCode']} {res['retMsg']}")
        time.sleep(3)

    time.sleep(3)
    print(mon.report())
    print("\n※ この値が戦略の反応速度の下限になる。")
    print("  さらに詰めたい場合はWS発注（記事29）を検討する。")
    ws.exit()

"""consistency_check.py — WSミラーとRESTの整合性を検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

定期実行して「WSだけで運用しても大丈夫か」を数値で確認する。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal

from account_mirror import AccountMirror


@dataclass
class ConsistencyReport:
    position_diffs: list[str] = field(default_factory=list)
    order_diffs: list[str] = field(default_factory=list)
    wallet_diff: Decimal = Decimal(0)
    checks: int = 0

    @property
    def is_consistent(self) -> bool:
        return not self.position_diffs and not self.order_diffs

    def report(self) -> str:
        if self.is_consistent:
            return (f"整合しています（残高の差 "
                    f"{self.wallet_diff:+.4f}）")
        lines = ["不整合を検出:"]
        for d in self.position_diffs:
            lines.append(f"  [position] {d}")
        for d in self.order_diffs:
            lines.append(f"  [order] {d}")
        lines.append(f"  [wallet] 差 {self.wallet_diff:+.4f}")
        return "\n".join(lines)


def check_consistency(mirror: AccountMirror) -> ConsistencyReport:
    """WSミラーの内容と、RESTの実測を比較する。"""
    from position import fetch_positions

    rep = ConsistencyReport()

    # WS由来の状態を退避
    ws_positions = {p.key: p.size for p in mirror.open_positions()}
    ws_orders = {o.order_id for o in mirror.open_orders()}
    ws_equity = mirror.wallet.total_equity

    # RESTで実測
    rest_positions = {p.key: p.size for p in fetch_positions(mirror.c.http)}

    rest_order_ids: set[str] = set()
    for category, extra in (("linear", {"settleCoin": "USDT"}),
                            ("spot", {})):
        r = mirror.c.http.get_open_orders(category=category, limit=50, **extra)
        if r.get("retCode") == 0:
            rest_order_ids |= {o["orderId"] for o in r["result"]["list"]}

    bal = mirror.c.http.get_wallet_balance(accountType="UNIFIED")
    rest_equity = Decimal(
        bal["result"]["list"][0]["totalEquity"] or "0"
    ) if bal.get("retCode") == 0 else Decimal(0)

    # 比較
    for key in set(ws_positions) | set(rest_positions):
        a = ws_positions.get(key, Decimal(0))
        b = rest_positions.get(key, Decimal(0))
        if a != b:
            rep.position_diffs.append(f"{key[1]}: WS={a} REST={b}")

    for oid in ws_orders - rest_order_ids:
        rep.order_diffs.append(f"{oid}: WSにあるがRESTに無い")
    for oid in rest_order_ids - ws_orders:
        rep.order_diffs.append(f"{oid}: RESTにあるがWSに無い")

    rep.wallet_diff = ws_equity - rest_equity
    rep.checks = 1
    return rep


if __name__ == "__main__":
    import time

    from client import make_client
    from pybit.unified_trading import WebSocket

    c = make_client()
    mirror = AccountMirror(c)
    mirror.refresh_from_rest()

    ws = WebSocket(channel_type="private",
                   testnet=(c.env.value == "testnet"),
                   demo=(c.env.value == "demo"),
                   api_key=c.cred.api_key, api_secret=c.cred.api_secret)
    ws.order_stream(callback=mirror.on_order)
    ws.position_stream(callback=mirror.on_position)
    ws.wallet_stream(callback=mirror.on_wallet)

    try:
        while True:
            time.sleep(60)
            print(f"\n[{time.strftime('%H:%M:%S')}] "
                  f"{check_consistency(mirror).report()}")
    except KeyboardInterrupt:
        ws.exit()