Bybit APIで板情報（オーダーブック）を取得し
流動性を測るPython実装
OFI・Kyleのλ

{
  "s": "BTCUSDT",
  "b": [["118342.4", "1.234"], ["118342.3", "0.500"], ...],   // 買い（降順）
  "a": [["118342.5", "0.812"], ["118342.6", "2.100"], ...],   // 売り（昇順）
  "ts": 1785000000123,    // システム時刻(ms)
  "u": 12345678,          // update id（WebSocketのdeltaと突き合わせる）
  "seq": 98765432,        // クロスシーケンス
  "cts": 1785000000100    // マッチングエンジンの時刻（publicTradeと対応）
}

各要素は [価格, 数量] の文字列ペア。
b は価格の降順（先頭が最良買い）、a は昇順（先頭が最良売り）。
注: RPI注文はレスポンスに含まれない。

"""orderbook.py — 板情報のモデル化と基本指標
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal

import pandas as pd


@dataclass(frozen=True)
class Level:
    price: Decimal
    size: Decimal

    @property
    def notional(self) -> Decimal:
        return self.price * self.size


@dataclass
class OrderBook:
    symbol: str
    category: str
    bids: list[Level]          # 降順（先頭が最良買い）
    asks: list[Level]          # 昇順（先頭が最良売り）
    ts_ms: int
    update_id: int = 0
    seq: int = 0
    engine_ts_ms: int = 0

    # ---- 基本 -----------------------------------------------------------
    @property
    def best_bid(self) -> Decimal:
        return self.bids[0].price if self.bids else Decimal(0)

    @property
    def best_ask(self) -> Decimal:
        return self.asks[0].price if self.asks else Decimal(0)

    @property
    def mid(self) -> Decimal:
        if not self.bids or not self.asks:
            return Decimal(0)
        return (self.best_bid + self.best_ask) / 2

    @property
    def spread(self) -> Decimal:
        return self.best_ask - self.best_bid

    @property
    def spread_bps(self) -> Decimal:
        m = self.mid
        return (self.spread / m * 10_000) if m else Decimal(0)

    @property
    def is_healthy(self) -> bool:
        """板が壊れていないか（クロスしていないか）。"""
        if not self.bids or not self.asks:
            return False
        if self.best_bid >= self.best_ask:
            return False        # クロス = 異常
        # 価格の単調性を確認
        for i in range(len(self.bids) - 1):
            if self.bids[i].price <= self.bids[i + 1].price:
                return False
        for i in range(len(self.asks) - 1):
            if self.asks[i].price >= self.asks[i + 1].price:
                return False
        return True

    # ---- 板厚 -----------------------------------------------------------
    def depth_notional(self, side: str, pct: Decimal = Decimal("0.1")
                       ) -> Decimal:
        """ミッドから pct% 以内に存在する板の合計金額。"""
        m = self.mid
        if not m:
            return Decimal(0)
        if side == "bid":
            floor_ = m * (1 - pct / 100)
            return sum((l.notional for l in self.bids if l.price >= floor_),
                       Decimal(0))
        cap = m * (1 + pct / 100)
        return sum((l.notional for l in self.asks if l.price <= cap),
                   Decimal(0))

    def depth_imbalance(self, pct: Decimal = Decimal("0.1")) -> Decimal:
        """板の偏り。+1に近ければ買いが厚い、-1なら売りが厚い。"""
        b = self.depth_notional("bid", pct)
        a = self.depth_notional("ask", pct)
        total = b + a
        return ((b - a) / total) if total else Decimal(0)

    def cumulative(self, side: str) -> pd.DataFrame:
        """累積板厚（デプスカーブ）。可視化やスリッページ計算に使う。"""
        levels = self.bids if side == "bid" else self.asks
        rows = []
        cum_size = Decimal(0)
        cum_notional = Decimal(0)
        for l in levels:
            cum_size += l.size
            cum_notional += l.notional
            dist = abs(l.price - self.mid) / self.mid * 10_000 if self.mid else 0
            rows.append({"price": float(l.price), "size": float(l.size),
                         "cum_size": float(cum_size),
                         "cum_notional": float(cum_notional),
                         "dist_bps": float(dist)})
        return pd.DataFrame(rows)

    def summary(self) -> str:
        return "\n".join([
            f"{self.symbol} ({self.category})  {len(self.bids)}x{len(self.asks)}段",
            f"  bid/ask    : {self.best_bid} / {self.best_ask}",
            f"  ミッド      : {self.mid}",
            f"  スプレッド  : {self.spread} ({self.spread_bps:.2f} bps)",
            f"  板厚(±0.1%) : bid {self.depth_notional('bid'):>12,.0f} / "
            f"ask {self.depth_notional('ask'):>12,.0f}",
            f"  板厚(±0.5%) : bid "
            f"{self.depth_notional('bid', Decimal('0.5')):>12,.0f} / "
            f"ask {self.depth_notional('ask', Decimal('0.5')):>12,.0f}",
            f"  偏り        : {self.depth_imbalance():+.3f}",
            f"  健全性      : {'OK' if self.is_healthy else '異常'}",
        ])


def fetch_orderbook(http, symbol: str, category: str = "linear",
                    limit: int = 200) -> OrderBook:
    """板を取得する。limitは必ず明示する（既定値がカテゴリで違う）。"""
    res = http.get_orderbook(category=category, symbol=symbol, limit=limit)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_orderbook失敗: {res['retCode']} {res['retMsg']}")
    r = res["result"]

    return OrderBook(
        symbol=r["s"], category=category,
        bids=[Level(Decimal(p), Decimal(q)) for p, q in r["b"]],
        asks=[Level(Decimal(p), Decimal(q)) for p, q in r["a"]],
        ts_ms=int(r.get("ts") or 0),
        update_id=int(r.get("u") or 0),
        seq=int(r.get("seq") or 0),
        engine_ts_ms=int(r.get("cts") or 0),
    )


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

    c = make_client()
    for category, symbol in (("linear", "BTCUSDT"), ("spot", "BTCUSDT"),
                             ("linear", "SOLUSDT")):
        ob = fetch_orderbook(c.http, symbol, category, limit=200)
        print(ob.summary())
        print()

    # limitを省略した場合の違いを確認する
    print("--- limitを省略した場合 ---")
    for category in ("linear", "spot"):
        res = c.http.get_orderbook(category=category, symbol="BTCUSDT")
        n = len(res["result"]["b"])
        print(f"  {category:<8} 既定で {n}段しか返らない")

"""microprice.py — マイクロプライスと加重ミッドの実装
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

マイクロプライス = ask * (bidSize / (bidSize + askSize))
                 + bid * (askSize / (bidSize + askSize))

買いが厚いほどaskに寄る（次の約定はaskで起きやすい）という直感を表す。
マーケットメイク（記事37）の基準価格として、ミッドより優れることが多い。
"""
from __future__ import annotations

from decimal import Decimal

from orderbook import OrderBook


def microprice(ob: OrderBook) -> Decimal:
    """サイズ加重のマイクロプライス（ベスト気配のみ使用）。"""
    if not ob.bids or not ob.asks:
        return ob.mid
    bs, as_ = ob.bids[0].size, ob.asks[0].size
    total = bs + as_
    if total == 0:
        return ob.mid
    # 買いが厚い（bs大）ほど ask 寄りになる
    return (ob.best_ask * bs + ob.best_bid * as_) / total


def weighted_mid(ob: OrderBook, levels: int = 5) -> Decimal:
    """複数段を使った加重ミッド。ノイズに強い。"""
    bids = ob.bids[:levels]
    asks = ob.asks[:levels]
    if not bids or not asks:
        return ob.mid

    bid_notional = sum((l.notional for l in bids), Decimal(0))
    ask_notional = sum((l.notional for l in asks), Decimal(0))
    bid_size = sum((l.size for l in bids), Decimal(0))
    ask_size = sum((l.size for l in asks), Decimal(0))

    if bid_size == 0 or ask_size == 0:
        return ob.mid

    vwap_bid = bid_notional / bid_size
    vwap_ask = ask_notional / ask_size
    total = bid_size + ask_size
    return (vwap_ask * bid_size + vwap_bid * ask_size) / total


def imbalance_ratio(ob: OrderBook, levels: int = 1) -> Decimal:
    """ベストN段のサイズ偏り。-1（売り厚い）〜 +1（買い厚い）。"""
    bs = sum((l.size for l in ob.bids[:levels]), Decimal(0))
    as_ = sum((l.size for l in ob.asks[:levels]), Decimal(0))
    total = bs + as_
    return ((bs - as_) / total) if total else Decimal(0)


def compare_prices(ob: OrderBook) -> str:
    mid = ob.mid
    mp = microprice(ob)
    wm = weighted_mid(ob)

    def bps(v: Decimal) -> Decimal:
        return (v - mid) / mid * 10_000 if mid else Decimal(0)

    return "\n".join([
        f"{ob.symbol}  スプレッド {ob.spread_bps:.2f} bps",
        f"  bid            : {ob.best_bid}  (size {ob.bids[0].size})",
        f"  ask            : {ob.best_ask}  (size {ob.asks[0].size})",
        f"  ミッド          : {mid}",
        f"  マイクロプライス : {mp}   ({bps(mp):+.3f} bps vs ミッド)",
        f"  加重ミッド(5段)  : {wm}   ({bps(wm):+.3f} bps vs ミッド)",
        f"  偏り(1段)       : {imbalance_ratio(ob, 1):+.3f}",
        f"  偏り(5段)       : {imbalance_ratio(ob, 5):+.3f}",
    ])


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

    c = make_client()
    for symbol in ("BTCUSDT", "ETHUSDT", "DOGEUSDT"):
        ob = fetch_orderbook(c.http, symbol, "linear", limit=50)
        print(compare_prices(ob))
        print()

    print("※ スプレッドが広い銘柄ほど、ミッドとマイクロプライスの差が大きい。")
    print("  マーケットメイクの基準価格には、マイクロプライスのほうが適する。")

"""slippage.py — 板を食った場合の平均約定価格とスリッページを計算する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

用途:
  - 発注前に「この数量なら何bps滑るか」を知る
  - 許容できるスリッページから最大発注量を逆算する
  - バックテストの約定モデルに実測値を入れる（記事31）
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

from orderbook import OrderBook


@dataclass
class FillEstimate:
    side: str
    requested_notional: Decimal
    filled_notional: Decimal
    filled_qty: Decimal
    avg_price: Decimal
    reference_price: Decimal      # 比較基準（ミッド）
    levels_consumed: int
    fully_filled: bool

    @property
    def slippage_bps(self) -> Decimal:
        """基準価格からの乖離（bps）。プラスが不利側。"""
        if not self.reference_price or not self.avg_price:
            return Decimal(0)
        diff = self.avg_price - self.reference_price
        sign = 1 if self.side == "Buy" else -1
        return diff / self.reference_price * 10_000 * sign

    @property
    def fill_ratio(self) -> Decimal:
        if not self.requested_notional:
            return Decimal(0)
        return self.filled_notional / self.requested_notional

    def line(self) -> str:
        status = "" if self.fully_filled else "  ← 食い切れず"
        return (f"{self.requested_notional:>12,.0f} USDT -> "
                f"{self.slippage_bps:>+7.2f} bps "
                f"({self.levels_consumed:>3}段消費, "
                f"約定率 {self.fill_ratio:>5.1%}){status}")


def estimate_market_fill(ob: OrderBook, side: str,
                         notional: Decimal) -> FillEstimate:
    """指定金額を成行で食った場合の平均約定価格を板から計算する。

    side="Buy" なら ask 側を上から食う。
    """
    levels = ob.asks if side == "Buy" else ob.bids
    ref = ob.mid

    remaining = notional
    cost = Decimal(0)
    qty = Decimal(0)
    consumed = 0

    for l in levels:
        if remaining <= 0:
            break
        available = l.notional
        take_notional = min(remaining, available)
        take_qty = take_notional / l.price
        cost += take_qty * l.price
        qty += take_qty
        remaining -= take_notional
        consumed += 1

    avg = (cost / qty) if qty > 0 else Decimal(0)
    return FillEstimate(
        side=side, requested_notional=notional,
        filled_notional=cost, filled_qty=qty, avg_price=avg,
        reference_price=ref, levels_consumed=consumed,
        fully_filled=(remaining <= 0),
    )


def max_notional_for_slippage(ob: OrderBook, side: str,
                              max_slippage_bps: Decimal,
                              precision: Decimal = Decimal("100")
                              ) -> Decimal:
    """許容スリッページを満たす最大の発注金額を二分探索で求める。"""
    levels = ob.asks if side == "Buy" else ob.bids
    total = sum((l.notional for l in levels), Decimal(0))
    if total <= 0:
        return Decimal(0)

    lo, hi = Decimal(0), total
    while hi - lo > precision:
        mid_n = (lo + hi) / 2
        est = estimate_market_fill(ob, side, mid_n)
        if est.fully_filled and est.slippage_bps <= max_slippage_bps:
            lo = mid_n
        else:
            hi = mid_n
    return lo


def slippage_curve(ob: OrderBook, side: str = "Buy") -> str:
    sizes = [Decimal(x) for x in
             (1_000, 5_000, 10_000, 50_000, 100_000, 500_000, 1_000_000)]
    lines = [f"{ob.symbol} {side} のスリッページ曲線 "
             f"（スプレッド {ob.spread_bps:.2f} bps）"]
    for n in sizes:
        lines.append("  " + estimate_market_fill(ob, side, n).line())
    return "\n".join(lines)


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

    c = make_client()

    for symbol in ("BTCUSDT", "SOLUSDT", "DOGEUSDT"):
        ob = fetch_orderbook(c.http, symbol, "linear", limit=500)
        print(slippage_curve(ob, "Buy"))

        for tol in (Decimal("2"), Decimal("5"), Decimal("10")):
            cap = max_notional_for_slippage(ob, "Buy", tol)
            print(f"  スリッページ {tol} bps以内で発注できる上限: "
                  f"{cap:>12,.0f} USDT")
        print()

    print("※ この数値を記事31のバックテストの slippage_bps に入れると、")
    print("  自分のロットに見合った現実的な検証になる。")

"""liquidity_gate.py — 板の状態で発注を制限するゲート
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

思想: 「戦略が出したい数量」ではなく
      「板が受け入れられる数量」で発注する。
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

from orderbook import OrderBook, fetch_orderbook
from slippage import estimate_market_fill, max_notional_for_slippage


class LiquidityRejected(RuntimeError):
    """流動性が足りないため発注を拒否した。"""


@dataclass
class GatePolicy:
    max_slippage_bps: Decimal = Decimal("5.0")
    max_spread_bps: Decimal = Decimal("10.0")
    min_depth_notional: Decimal = Decimal("50000")   # ±0.1%の板厚の下限
    max_participation: Decimal = Decimal("0.2")      # 板厚に対する参加率
    require_healthy_book: bool = True


@dataclass
class GateResult:
    approved_notional: Decimal
    requested_notional: Decimal
    reasons: list[str]
    estimated_slippage_bps: Decimal

    @property
    def was_reduced(self) -> bool:
        return self.approved_notional < self.requested_notional

    @property
    def rejected(self) -> bool:
        return self.approved_notional <= 0

    def line(self) -> str:
        if self.rejected:
            return f"拒否: {' / '.join(self.reasons)}"
        prefix = "縮小" if self.was_reduced else "承認"
        detail = f" ({' / '.join(self.reasons)})" if self.reasons else ""
        return (f"{prefix}: {self.requested_notional:,.0f} -> "
                f"{self.approved_notional:,.0f} USDT "
                f"（想定スリッページ {self.estimated_slippage_bps:+.2f} bps）"
                f"{detail}")


def check_liquidity(ob: OrderBook, side: str, notional: Decimal,
                    policy: GatePolicy | None = None) -> GateResult:
    """板の状態を検査し、承認する発注金額を返す。"""
    policy = policy or GatePolicy()
    reasons: list[str] = []
    approved = notional

    # 1) 板が壊れていないか
    if policy.require_healthy_book and not ob.is_healthy:
        return GateResult(Decimal(0), notional,
                          ["板が異常（クロスまたは欠損）"], Decimal(0))

    # 2) スプレッドが広すぎないか
    if ob.spread_bps > policy.max_spread_bps:
        return GateResult(
            Decimal(0), notional,
            [f"スプレッドが広い（{ob.spread_bps:.2f} > "
             f"{policy.max_spread_bps} bps）"], Decimal(0))

    # 3) 板厚が最低水準を満たすか
    book_side = "ask" if side == "Buy" else "bid"
    depth = ob.depth_notional(book_side, Decimal("0.1"))
    if depth < policy.min_depth_notional:
        return GateResult(
            Decimal(0), notional,
            [f"板が薄い（±0.1%で {depth:,.0f} < "
             f"{policy.min_depth_notional:,.0f}）"], Decimal(0))

    # 4) 板厚に対する参加率で上限を設ける
    participation_cap = depth * policy.max_participation
    if approved > participation_cap:
        approved = participation_cap
        reasons.append(f"参加率上限で縮小（板厚の"
                       f"{policy.max_participation:.0%}）")

    # 5) スリッページ上限で上限を設ける
    slip_cap = max_notional_for_slippage(ob, side, policy.max_slippage_bps)
    if approved > slip_cap:
        approved = slip_cap
        reasons.append(f"スリッページ上限で縮小"
                       f"（{policy.max_slippage_bps} bps以内）")

    if approved <= 0:
        return GateResult(Decimal(0), notional,
                          reasons or ["承認可能な数量がゼロ"], Decimal(0))

    est = estimate_market_fill(ob, side, approved)
    return GateResult(approved, notional, reasons, est.slippage_bps)


def gated_market_order(client, symbol: str, side: str,
                       notional: Decimal, registry,
                       category: str = "linear",
                       policy: GatePolicy | None = None,
                       dry_run: bool = True) -> dict | None:
    """板を検査してから成行を出す。"""
    from order_validate import size_from_notional, validate_order

    ob = fetch_orderbook(client.http, symbol, category, limit=500)
    gate = check_liquidity(ob, side, notional, policy)
    print(f"[gate] {symbol} {side} {gate.line()}")

    if gate.rejected:
        raise LiquidityRejected(f"{symbol}: {' / '.join(gate.reasons)}")

    rules = registry.get(symbol, category)
    qty = size_from_notional(rules, gate.approved_notional, ob.mid)
    v = validate_order(rules, side, qty, order_type="Market",
                       ref_price=ob.mid)

    if dry_run:
        print(f"  [dry-run] qty={v.qty_str} を発注する予定")
        return None
    return client.http.place_order(**v.to_params("Market"))


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

    c = make_client()
    reg = SymbolRegistry(c.http)
    policy = GatePolicy(max_slippage_bps=Decimal("5"),
                        max_spread_bps=Decimal("10"),
                        min_depth_notional=Decimal("50000"))

    for symbol in ("BTCUSDT", "SOLUSDT", "DOGEUSDT"):
        ob = fetch_orderbook(c.http, symbol, "linear", limit=500)
        for notional in (Decimal("10000"), Decimal("200000")):
            print(f"{symbol:<11} {notional:>9,.0f} USDT: "
                  f"{check_liquidity(ob, 'Buy', notional, policy).line()}")
        print()

    # 実際の発注（dry_run=Trueなので送信しない）
    try:
        gated_market_order(c, "BTCUSDT", "Buy", Decimal("5000"), reg,
                           policy=policy, dry_run=True)
    except LiquidityRejected as e:
        print(f"発注を拒否しました: {e}")

"""ofi.py — Order Flow Imbalance の計算
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Cont, Kukanov & Stoikov (2014)
      "The Price Impact of Order Book Events"

OFIは連続する板スナップショットの差分から、
「買い圧力 − 売り圧力」を数量ベースで測る指標。

  e_n = I(P_b >= P_b') * q_b - I(P_b <= P_b') * q_b'
      - I(P_a <= P_a') * q_a + I(P_a >= P_a') * q_a'

（b=bid, a=ask, ' は前スナップショット、I は指示関数）
"""
from __future__ import annotations

from collections import deque
from dataclasses import dataclass, field
from decimal import Decimal

import numpy as np
import pandas as pd

from orderbook import OrderBook


@dataclass
class OfiSample:
    ts_ms: int
    ofi: Decimal
    mid: Decimal
    spread_bps: Decimal


class OfiCalculator:
    """連続する板スナップショットからOFIを計算し、履歴を保持する。"""

    def __init__(self, maxlen: int = 5000):
        self._prev: tuple[Decimal, Decimal, Decimal, Decimal] | None = None
        self.samples: deque[OfiSample] = deque(maxlen=maxlen)

    def update(self, ob: OrderBook) -> OfiSample | None:
        if not ob.bids or not ob.asks:
            return None

        pb, qb = ob.best_bid, ob.bids[0].size
        pa, qa = ob.best_ask, ob.asks[0].size

        if self._prev is None:
            self._prev = (pb, qb, pa, qa)
            return None

        pb0, qb0, pa0, qa0 = self._prev

        # 買い側の寄与: 価格が上がるか同じなら現在の量が増加圧力
        e_bid = Decimal(0)
        if pb >= pb0:
            e_bid += qb
        if pb <= pb0:
            e_bid -= qb0

        # 売り側の寄与: 符号が逆
        e_ask = Decimal(0)
        if pa <= pa0:
            e_ask += qa
        if pa >= pa0:
            e_ask -= qa0

        ofi = e_bid - e_ask
        self._prev = (pb, qb, pa, qa)

        sample = OfiSample(ts_ms=ob.ts_ms, ofi=ofi, mid=ob.mid,
                           spread_bps=ob.spread_bps)
        self.samples.append(sample)
        return sample

    # ---- 集計 -----------------------------------------------------------
    def to_frame(self) -> pd.DataFrame:
        if not self.samples:
            return pd.DataFrame()
        df = pd.DataFrame([{
            "ts": pd.Timestamp(s.ts_ms, unit="ms", tz="UTC"),
            "ofi": float(s.ofi), "mid": float(s.mid),
            "spread_bps": float(s.spread_bps),
        } for s in self.samples]).set_index("ts")
        df["mid_change_bps"] = df["mid"].diff() / df["mid"].shift(1) * 10_000
        return df

    def rolling_ofi(self, window: int = 20) -> Decimal:
        """直近window件のOFI合計。短期の方向性の指標。"""
        recent = list(self.samples)[-window:]
        return sum((s.ofi for s in recent), Decimal(0))

    def regress_impact(self) -> dict:
        """OFI -> ミッド変化の回帰。係数がKyleのλに相当する。"""
        df = self.to_frame().dropna()
        if len(df) < 30:
            return {"error": f"サンプル不足 ({len(df)})"}

        x = df["ofi"].to_numpy()
        y = df["mid_change_bps"].to_numpy()
        if np.std(x) == 0:
            return {"error": "OFIに変動がありません"}

        slope, intercept = np.polyfit(x, y, 1)
        corr = float(np.corrcoef(x, y)[0, 1])

        return {
            "n": len(df),
            "lambda_bps_per_unit": float(slope),
            "intercept_bps": float(intercept),
            "correlation": corr,
            "r_squared": corr ** 2,
        }


def collect_ofi(client, symbol: str, seconds: float = 60.0,
                interval: float = 0.5, category: str = "linear",
                limit: int = 25) -> OfiCalculator:
    """一定時間、板をポーリングしてOFIを収集する。

    本番ではWebSocketのorderbookトピック（記事28）を使うほうが
    レート消費が少なく、時間解像度も高い。
    """
    import time

    from orderbook import fetch_orderbook

    calc = OfiCalculator()
    deadline = time.time() + seconds
    while time.time() < deadline:
        try:
            calc.update(fetch_orderbook(client.http, symbol, category, limit))
        except Exception as e:
            print(f"[ofi] 取得失敗: {type(e).__name__}: {e}")
        time.sleep(interval)
    return calc


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

    c = make_client()
    SYMBOL = "BTCUSDT"

    print(f"{SYMBOL} の板を60秒間収集します...")
    calc = collect_ofi(c, SYMBOL, seconds=60.0, interval=0.5)

    df = calc.to_frame()
    print(f"\n収集: {len(df)}サンプル")
    print(f"  OFIの合計   : {df['ofi'].sum():+.3f}")
    print(f"  OFIの標準偏差: {df['ofi'].std():.3f}")
    print(f"  直近20件のOFI: {calc.rolling_ofi(20):+.3f}")

    print("\n--- OFI -> ミッド変化の回帰（Kyleのλに相当） ---")
    for k, v in calc.regress_impact().items():
        print(f"  {k:<22}: {v}")

    print("\n※ r_squaredが高いほど、短期の価格変化がOFIで説明できる。")
    print("  マーケットメイク（記事37）では、OFIを気配の非対称化に使う。")

"""kyle_lambda.py — Kyleのλ（価格インパクト係数）を約定データから推定
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Kyle (1985) "Continuous Auctions and Insider Trading"

モデル: ΔP = λ * Q + ε
        Q = 符号付き出来高（買い出来高 − 売り出来高）

λが大きいほど「浅い」市場。同じ発注量でも価格が大きく動く。
銘柄間の流動性比較と、最適執行のスライス設計（記事19）に使える。
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd


@dataclass
class KyleEstimate:
    symbol: str
    n_buckets: int
    lambda_bps_per_musd: float    # 100万USD当たりの価格変化(bps)
    r_squared: float
    p_value_proxy: float          # 相関の強さの目安
    avg_bucket_volume_usd: float

    @property
    def impact_for(self, ) -> float:
        return self.lambda_bps_per_musd

    def impact_bps(self, notional_usd: float) -> float:
        """指定金額を一方向に流したときの想定価格インパクト(bps)。"""
        return self.lambda_bps_per_musd * (notional_usd / 1_000_000)

    def line(self) -> str:
        return (f"{self.symbol:<12}{self.lambda_bps_per_musd:>10.2f}"
                f"{self.r_squared:>9.3f}{self.n_buckets:>8}"
                f"{self.avg_bucket_volume_usd:>14,.0f}")


def estimate_kyle_lambda(trades: pd.DataFrame, symbol: str,
                         bucket: str = "1min") -> KyleEstimate:
    """約定データからλを推定する。

    trades は index=UTC時刻、列に price / qty / side（テイカー方向）が必要。
    記事08の publicTrade データ、または記事26のティック収集を使う。
    """
    df = trades.copy()
    df["notional"] = df["price"] * df["qty"]
    # テイカーがBuyならプラス、Sellならマイナス
    df["signed_notional"] = df["notional"] * df["side"].map(
        {"Buy": 1, "Sell": -1}).fillna(0)

    agg = df.resample(bucket).agg(
        first_price=("price", "first"),
        last_price=("price", "last"),
        signed=("signed_notional", "sum"),
        volume=("notional", "sum"),
    ).dropna()

    agg = agg[agg["volume"] > 0]
    if len(agg) < 30:
        raise ValueError(f"{symbol}: バケット数が不足 ({len(agg)})")

    # 価格変化(bps)と符号付き出来高(百万USD)
    agg["ret_bps"] = ((agg["last_price"] - agg["first_price"])
                      / agg["first_price"] * 10_000)
    agg["signed_musd"] = agg["signed"] / 1_000_000

    x = agg["signed_musd"].to_numpy()
    y = agg["ret_bps"].to_numpy()
    if np.std(x) == 0:
        raise ValueError(f"{symbol}: 符号付き出来高に変動がありません")

    slope, _ = np.polyfit(x, y, 1)
    corr = float(np.corrcoef(x, y)[0, 1])

    return KyleEstimate(
        symbol=symbol, n_buckets=len(agg),
        lambda_bps_per_musd=float(slope),
        r_squared=corr ** 2,
        p_value_proxy=abs(corr) * np.sqrt(len(agg)),
        avg_bucket_volume_usd=float(agg["volume"].mean()),
    )


def collect_trades(client, symbol: str, category: str = "linear",
                   limit: int = 1000) -> pd.DataFrame:
    """直近の約定をRESTで取得する（長期はpublic.bybit.comを使う・記事08）。"""
    res = client.http.get_public_trade_history(
        category=category, symbol=symbol, limit=min(limit, 1000))
    if res["retCode"] != 0:
        raise RuntimeError(f"get_public_trade_history失敗: {res['retMsg']}")

    rows = res["result"]["list"]
    df = pd.DataFrame([{
        "ts": pd.Timestamp(int(r["time"]), unit="ms", tz="UTC"),
        "price": float(r["price"]),
        "qty": float(r["size"]),
        "side": r["side"],           # テイカーの方向
    } for r in rows])
    return df.set_index("ts").sort_index()


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

    c = make_client()

    print(f"{'symbol':<12}{'λ(bps/M$)':>10}{'R^2':>9}"
          f"{'バケット':>8}{'平均出来高':>14}")
    print("-" * 54)

    estimates = []
    for symbol in ("BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT"):
        try:
            trades = collect_trades(c, symbol)
            est = estimate_kyle_lambda(trades, symbol, bucket="1min")
            estimates.append(est)
            print(est.line())
        except Exception as e:
            print(f"{symbol:<12} 推定失敗: {e}")

    if estimates:
        print("\n--- 発注金額別の想定価格インパクト(bps) ---")
        print(f"{'symbol':<12}{'1万$':>9}{'10万$':>9}{'100万$':>9}")
        print("-" * 39)
        for e in estimates:
            print(f"{e.symbol:<12}{e.impact_bps(10_000):>9.2f}"
                  f"{e.impact_bps(100_000):>9.2f}"
                  f"{e.impact_bps(1_000_000):>9.2f}")

    print("\n※ λが大きい銘柄では、同じ金額でも価格が大きく動く。")
    print("  分割執行（記事19のTWAP/VWAP）の必要性はλで判断できる。")
    print("  注: 直近1000件の約定からの推定なので、時間帯の偏りがある。")
    print("     長期の安定した推定にはpublic.bybit.comのデータを使う（記事08）。")

"""book_recorder.py — 板スナップショットをParquetに蓄積する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

用途:
  - 流動性の時間帯パターンを調べる（いつ薄いか）
  - スリッページの実測値を時系列で持つ
  - マーケットメイクのパラメータ推定（記事37）
"""
from __future__ import annotations

import time
from dataclasses import asdict, dataclass
from decimal import Decimal
from pathlib import Path

import pandas as pd

from orderbook import OrderBook, fetch_orderbook
from slippage import estimate_market_fill

OUT_DIR = Path("data/orderbook")


@dataclass
class BookSnapshot:
    """1時点の板を、指標に圧縮して保存する（生の板は容量が大きすぎる）。"""

    ts_ms: int
    symbol: str
    best_bid: float
    best_ask: float
    bid_size: float
    ask_size: float
    mid: float
    microprice: float
    spread_bps: float
    depth_bid_01: float          # ±0.1%の板厚（金額）
    depth_ask_01: float
    depth_bid_05: float
    depth_ask_05: float
    imbalance_01: float
    slip_buy_10k_bps: float
    slip_sell_10k_bps: float
    slip_buy_100k_bps: float
    update_id: int


def snapshot(ob: OrderBook) -> BookSnapshot:
    from microprice import microprice as mp

    b10 = estimate_market_fill(ob, "Buy", Decimal("10000"))
    s10 = estimate_market_fill(ob, "Sell", Decimal("10000"))
    b100 = estimate_market_fill(ob, "Buy", Decimal("100000"))

    return BookSnapshot(
        ts_ms=ob.ts_ms, symbol=ob.symbol,
        best_bid=float(ob.best_bid), best_ask=float(ob.best_ask),
        bid_size=float(ob.bids[0].size) if ob.bids else 0.0,
        ask_size=float(ob.asks[0].size) if ob.asks else 0.0,
        mid=float(ob.mid), microprice=float(mp(ob)),
        spread_bps=float(ob.spread_bps),
        depth_bid_01=float(ob.depth_notional("bid", Decimal("0.1"))),
        depth_ask_01=float(ob.depth_notional("ask", Decimal("0.1"))),
        depth_bid_05=float(ob.depth_notional("bid", Decimal("0.5"))),
        depth_ask_05=float(ob.depth_notional("ask", Decimal("0.5"))),
        imbalance_01=float(ob.depth_imbalance(Decimal("0.1"))),
        slip_buy_10k_bps=float(b10.slippage_bps),
        slip_sell_10k_bps=float(s10.slippage_bps),
        slip_buy_100k_bps=float(b100.slippage_bps),
        update_id=ob.update_id,
    )


class BookRecorder:
    def __init__(self, symbol: str, category: str = "linear",
                 out_dir: Path = OUT_DIR, flush_every: int = 200):
        self.symbol = symbol
        self.category = category
        self.out_dir = out_dir
        self.out_dir.mkdir(parents=True, exist_ok=True)
        self.flush_every = flush_every
        self._buf: list[BookSnapshot] = []
        self.total = 0

    @property
    def path(self) -> Path:
        return self.out_dir / f"{self.category}_{self.symbol}_book.parquet"

    def add(self, ob: OrderBook) -> None:
        self._buf.append(snapshot(ob))
        if len(self._buf) >= self.flush_every:
            self.flush()

    def flush(self) -> None:
        if not self._buf:
            return
        df = pd.DataFrame([asdict(s) for s in self._buf])
        df["ts"] = pd.to_datetime(df["ts_ms"], unit="ms", utc=True)
        df = df.drop(columns=["ts_ms"]).set_index("ts")

        if self.path.exists():
            df = pd.concat([pd.read_parquet(self.path), df])
            df = df[~df.index.duplicated(keep="last")].sort_index()

        tmp = self.path.with_suffix(".tmp")
        df.to_parquet(tmp, compression="zstd")
        tmp.replace(self.path)

        self.total += len(self._buf)
        self._buf.clear()

    def run(self, client, seconds: float = 300.0,
            interval: float = 1.0) -> None:
        deadline = time.time() + seconds
        while time.time() < deadline:
            try:
                self.add(fetch_orderbook(client.http, self.symbol,
                                         self.category, limit=200))
            except Exception as e:
                print(f"[rec] 失敗: {type(e).__name__}: {e}")
            time.sleep(interval)
        self.flush()
        print(f"[rec] {self.symbol}: 累計 {self.total}件を保存しました")


def analyse_liquidity_pattern(path: Path) -> None:
    """時間帯ごとの流動性パターンを集計する。"""
    if not path.exists():
        print(f"{path} がありません")
        return
    df = pd.read_parquet(path)
    df["hour_utc"] = df.index.hour

    agg = df.groupby("hour_utc").agg(
        spread=("spread_bps", "median"),
        depth=("depth_ask_01", "median"),
        slip10k=("slip_buy_10k_bps", "median"),
        n=("mid", "count"),
    )
    print(f"{'時(UTC)':>7}{'JST':>6}{'スプレッド':>11}{'板厚(±0.1%)':>14}"
          f"{'1万$滑り':>10}{'件数':>7}")
    print("-" * 56)
    for h, r in agg.iterrows():
        jst = (h + 9) % 24
        print(f"{h:>7}{jst:>6}{r['spread']:>10.2f}b{r['depth']:>14,.0f}"
              f"{r['slip10k']:>9.2f}b{int(r['n']):>7}")
    print("\n※ 流動性が薄い時間帯を避けるだけで執行コストは下がる。")


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

    c = make_client()
    rec = BookRecorder("BTCUSDT", "linear")
    rec.run(c, seconds=60.0, interval=1.0)
    analyse_liquidity_pattern(rec.path)

"""test_orderbook.py — 板の指標計算が正しいことを検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pytest test_orderbook.py -v
"""
from __future__ import annotations

from decimal import Decimal

import pytest

from microprice import imbalance_ratio, microprice
from orderbook import Level, OrderBook
from slippage import estimate_market_fill, max_notional_for_slippage


def make_book(bid_sizes=(1, 1, 1), ask_sizes=(1, 1, 1),
              bid_start=Decimal("100"), tick=Decimal("1")) -> OrderBook:
    """テスト用の単純な板を作る。"""
    bids = [Level(bid_start - tick * i, Decimal(str(s)))
            for i, s in enumerate(bid_sizes)]
    asks = [Level(bid_start + tick * (i + 1), Decimal(str(s)))
            for i, s in enumerate(ask_sizes)]
    return OrderBook(symbol="TESTUSDT", category="linear", bids=bids,
                     asks=asks, ts_ms=0)


def test_mid_and_spread():
    ob = make_book()
    assert ob.best_bid == Decimal("100")
    assert ob.best_ask == Decimal("101")
    assert ob.mid == Decimal("100.5")
    assert ob.spread == Decimal("1")


def test_healthy_book():
    assert make_book().is_healthy is True


def test_crossed_book_detected():
    """bid >= ask のクロス板を異常として検出すること。"""
    ob = OrderBook(symbol="X", category="linear",
                   bids=[Level(Decimal("101"), Decimal("1"))],
                   asks=[Level(Decimal("100"), Decimal("1"))], ts_ms=0)
    assert ob.is_healthy is False


def test_non_monotonic_detected():
    """価格順が壊れた板を検出すること。"""
    ob = OrderBook(symbol="X", category="linear",
                   bids=[Level(Decimal("99"), Decimal("1")),
                         Level(Decimal("100"), Decimal("1"))],   # 昇順は異常
                   asks=[Level(Decimal("101"), Decimal("1"))], ts_ms=0)
    assert ob.is_healthy is False


def test_microprice_leans_to_thick_side():
    """買いが厚いとき、マイクロプライスはaskに寄ること。"""
    thick_bid = make_book(bid_sizes=(10,), ask_sizes=(1,))
    mp = microprice(thick_bid)
    assert mp > thick_bid.mid            # ask寄り
    assert mp < thick_bid.best_ask

    thick_ask = make_book(bid_sizes=(1,), ask_sizes=(10,))
    assert microprice(thick_ask) < thick_ask.mid   # bid寄り


def test_microprice_equals_mid_when_balanced():
    ob = make_book(bid_sizes=(5,), ask_sizes=(5,))
    assert microprice(ob) == ob.mid


def test_imbalance_sign():
    assert imbalance_ratio(make_book(bid_sizes=(9,), ask_sizes=(1,))) > 0
    assert imbalance_ratio(make_book(bid_sizes=(1,), ask_sizes=(9,))) < 0


def test_slippage_grows_with_size():
    """発注量が増えるとスリッページが増えること（単調性）。"""
    ob = make_book(ask_sizes=(1, 1, 1, 1, 1))
    small = estimate_market_fill(ob, "Buy", Decimal("100"))
    large = estimate_market_fill(ob, "Buy", Decimal("400"))
    assert large.slippage_bps > small.slippage_bps
    assert large.levels_consumed > small.levels_consumed


def test_slippage_sign_is_unfavourable():
    """買いも売りも、スリッページは不利側（プラス）になること。"""
    ob = make_book(bid_sizes=(1, 1, 1), ask_sizes=(1, 1, 1))
    buy = estimate_market_fill(ob, "Buy", Decimal("250"))
    sell = estimate_market_fill(ob, "Sell", Decimal("250"))
    assert buy.slippage_bps > 0
    assert sell.slippage_bps > 0


def test_not_fully_filled_flagged():
    """板を食い切れない場合にフラグが立つこと。"""
    ob = make_book(ask_sizes=(1,))          # 101 * 1 = 101 USDT分しかない
    est = estimate_market_fill(ob, "Buy", Decimal("10000"))
    assert est.fully_filled is False
    assert est.fill_ratio < 1


def test_max_notional_respects_tolerance():
    ob = make_book(ask_sizes=(1,) * 20)
    cap = max_notional_for_slippage(ob, "Buy", Decimal("50"),
                                    precision=Decimal("1"))
    est = estimate_market_fill(ob, "Buy", cap)
    assert est.slippage_bps <= Decimal("50")


@pytest.mark.parametrize("pct,expected_positive", [
    (Decimal("0.1"), True),
    (Decimal("5.0"), True),
])
def test_depth_notional_positive(pct, expected_positive):
    ob = make_book(bid_sizes=(1, 2, 3), ask_sizes=(1, 2, 3))
    assert (ob.depth_notional("bid", pct) > 0) is expected_positive