ローカル板（オーダーブック）をWebSocketで
正確に復元する差分更新実装

"""local_book.py — ローカル板のデータ構造
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install sortedcontainers

  設計:
    - 価格をキーにした SortedDict で保持する（挿入・削除・最良取得が高速）
    - bid は降順、ask は昇順で参照したいので、
      bid 側はキーを負数にして「昇順 = 高い順」にする
    - 価格は Decimal ではなく float で持つ
      （更新が高頻度なので Decimal の精度より速度を優先。
        発注時は記事03の丸めルールを通すので問題ない）

  なぜ dict ではなく SortedDict か:
    ベストプライスの取得が dict だと O(n)。
    1秒に数百回呼ぶので、ここが効いてくる。
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field

from sortedcontainers import SortedDict


@dataclass
class BookSide:
    """板の片側。bid は is_bid=True。"""
    is_bid: bool
    levels: SortedDict = field(default_factory=SortedDict)

    def _key(self, price: float) -> float:
        # bid は高い順に並べたいのでキーを反転する
        return -price if self.is_bid else price

    def apply(self, price: float, size: float) -> None:
        """1階層を更新する。size=0 は削除。"""
        key = self._key(price)
        if size == 0:
            self.levels.pop(key, None)      # 削除（存在しなくてもエラーにしない）
        else:
            self.levels[key] = size

    def clear(self) -> None:
        self.levels.clear()

    def best(self) -> tuple[float, float] | None:
        """最良気配 (価格, 数量)。"""
        if not self.levels:
            return None
        key = self.levels.keys()[0]
        return (-key if self.is_bid else key), self.levels[key]

    def top(self, n: int) -> list[tuple[float, float]]:
        """上位n階層。"""
        out = []
        for key in self.levels.keys()[:n]:
            price = -key if self.is_bid else key
            out.append((price, self.levels[key]))
        return out

    def depth_notional(self, n: int) -> float:
        """上位n階層の建玉額（価格×数量の合計）。"""
        return sum(p * q for p, q in self.top(n))

    def total_size(self, n: int) -> float:
        return sum(q for _, q in self.top(n))

    def __len__(self) -> int:
        return len(self.levels)


@dataclass
class LocalBook:
    """1銘柄分のローカル板。"""
    symbol: str
    depth: int = 50
    bids: BookSide = field(default_factory=lambda: BookSide(is_bid=True))
    asks: BookSide = field(default_factory=lambda: BookSide(is_bid=False))

    last_update_id: int = 0
    last_seq: int = 0
    last_ts: int = 0          # サーバのタイムスタンプ (ms)
    last_local_ts: float = 0.0  # 受信したローカル時刻 (s)
    update_count: int = 0
    snapshot_count: int = 0
    ready: bool = False

    # --- 更新 ------------------------------------------------------

    def apply_snapshot(self, data: dict, ts: int = 0) -> None:
        """snapshot で板を丸ごと置き換える。"""
        self.bids.clear()
        self.asks.clear()

        for price, size in data.get("b", []):
            self.bids.apply(float(price), float(size))
        for price, size in data.get("a", []):
            self.asks.apply(float(price), float(size))

        self.last_update_id = int(data.get("u", 0))
        self.last_seq = int(data.get("seq", 0))
        self.last_ts = ts
        self.last_local_ts = time.time()
        self.snapshot_count += 1
        self.update_count += 1
        self.ready = True

    def apply_delta(self, data: dict, ts: int = 0) -> None:
        """delta を適用する。

        呼び出す前に必ずシーケンス検証を通すこと（サンプル2）。
        検証せずに呼ぶと板が静かに壊れる。
        """
        for price, size in data.get("b", []):
            self.bids.apply(float(price), float(size))
        for price, size in data.get("a", []):
            self.asks.apply(float(price), float(size))

        self.last_update_id = int(data.get("u", 0))
        self.last_seq = int(data.get("seq", 0))
        self.last_ts = ts
        self.last_local_ts = time.time()
        self.update_count += 1

    def reset(self) -> None:
        """破損時に呼ぶ。再購読で snapshot を待つ状態に戻す。"""
        self.bids.clear()
        self.asks.clear()
        self.last_update_id = 0
        self.last_seq = 0
        self.ready = False

    # --- 参照 ------------------------------------------------------

    @property
    def best_bid(self) -> float:
        b = self.bids.best()
        return b[0] if b else 0.0

    @property
    def best_ask(self) -> float:
        a = self.asks.best()
        return a[0] if a else 0.0

    @property
    def best_bid_size(self) -> float:
        b = self.bids.best()
        return b[1] if b else 0.0

    @property
    def best_ask_size(self) -> float:
        a = self.asks.best()
        return a[1] if a else 0.0

    @property
    def mid(self) -> float:
        bid, ask = self.best_bid, self.best_ask
        return (bid + ask) / 2 if bid > 0 and ask > 0 else 0.0

    @property
    def spread(self) -> float:
        bid, ask = self.best_bid, self.best_ask
        return ask - bid if bid > 0 and ask > 0 else 0.0

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

    @property
    def age_ms(self) -> float:
        """最終更新からの経過時間（ミリ秒）。"""
        return (time.time() - self.last_local_ts) * 1000 \
            if self.last_local_ts else float("inf")

    def snapshot_dict(self, n: int = 5) -> dict:
        """現在の板を辞書で返す（ログ・保存用）。"""
        return {
            "symbol": self.symbol,
            "ts": self.last_ts,
            "u": self.last_update_id,
            "seq": self.last_seq,
            "bids": self.bids.top(n),
            "asks": self.asks.top(n),
        }

    def render(self, n: int = 5) -> str:
        """板を人間が読める形で表示する。"""
        lines = [f"=== {self.symbol} (u={self.last_update_id} "
                 f"seq={self.last_seq} age={self.age_ms:.0f}ms) ==="]

        asks = self.asks.top(n)
        for price, size in reversed(asks):
            lines.append(f"       {price:>14,.2f}  {size:>14,.4f}")

        lines.append(f"  ---- spread {self.spread:,.2f} "
                     f"({self.spread_bps:.2f}bps) ----")

        for price, size in self.bids.top(n):
            lines.append(f"       {price:>14,.2f}  {size:>14,.4f}")

        lines.append(f"  階層数: bid {len(self.bids)} / ask {len(self.asks)}")
        return "\n".join(lines)


if __name__ == "__main__":
    book = LocalBook("BTCUSDT", depth=50)

    # snapshot の適用
    book.apply_snapshot({
        "s": "BTCUSDT", "u": 1000, "seq": 500,
        "b": [["120000.0", "1.5"], ["119999.5", "2.0"],
              ["119999.0", "0.8"]],
        "a": [["120000.5", "1.2"], ["120001.0", "3.0"],
              ["120001.5", "0.5"]],
    }, ts=int(time.time() * 1000))

    print(book.render())

    # delta の適用: 119999.5 を削除、120000.5 を更新、新しい階層を追加
    book.apply_delta({
        "s": "BTCUSDT", "u": 1001, "seq": 501,
        "b": [["119999.5", "0"], ["119998.5", "5.0"]],
        "a": [["120000.5", "0.3"]],
    }, ts=int(time.time() * 1000))

    print("\ndelta 適用後:")
    print(book.render())

    print(f"\n  ベストビッド: {book.best_bid:,.2f} x {book.best_bid_size}")
    print(f"  ベストアスク: {book.best_ask:,.2f} x {book.best_ask_size}")
    print(f"  ミッド      : {book.mid:,.2f}")
    print(f"  上位3の厚み : bid {book.bids.depth_notional(3):,.0f} / "
          f"ask {book.asks.depth_notional(3):,.0f}")

    print("\n※ 数量 '0' は『0で上書き』ではなく『階層の削除』。")
    print("  0のまま残すと、ベスト価格が実在しない価格になる。")

"""sequence_guard.py — update id の連続性を検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  Bybit V5 の orderbook トピックの規則:

    type="snapshot" → 板を置き換える。u は任意の値
    type="delta"    → u は「前回の u + 1」でなければならない
    u=1             → snapshot の再送を意味する（type に関わらず）

  判定結果は4種類:
    APPLY        通常の delta。そのまま適用する
    SNAPSHOT     板を置き換える
    DUPLICATE    既に適用済み（u <= last_u）。無視する
    GAP          欠番。ローカル板は信用できない → 再購読

  GAP を握りつぶすと、板は「それらしく見えるが間違っている」
  状態になる。例外が出ないので気づけない。これが最も危険。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum


class SeqVerdict(str, Enum):
    APPLY = "apply"
    SNAPSHOT = "snapshot"
    DUPLICATE = "duplicate"
    GAP = "gap"


@dataclass
class SeqCheck:
    verdict: SeqVerdict
    expected: int = 0
    received: int = 0
    gap_size: int = 0
    note: str = ""

    @property
    def is_healthy(self) -> bool:
        return self.verdict in (SeqVerdict.APPLY, SeqVerdict.SNAPSHOT,
                                SeqVerdict.DUPLICATE)


@dataclass
class SequenceGuard:
    """銘柄ごとの update id を追跡する。"""
    symbol: str
    last_update_id: int = 0
    last_seq: int = 0
    initialized: bool = False

    gap_count: int = 0
    duplicate_count: int = 0
    snapshot_count: int = 0
    applied_count: int = 0
    gap_history: list = field(default_factory=list)

    def check(self, message_type: str, data: dict) -> SeqCheck:
        """メッセージを検証して、どう扱うべきかを返す。"""
        u = int(data.get("u", 0))
        seq = int(data.get("seq", 0))

        # 1) 明示的な snapshot
        if message_type == "snapshot":
            self.last_update_id = u
            self.last_seq = seq
            self.initialized = True
            self.snapshot_count += 1
            return SeqCheck(SeqVerdict.SNAPSHOT, received=u,
                            note="snapshot を受信")

        # 2) u=1 は snapshot の再送（type が delta でも）
        if u == 1:
            self.last_update_id = u
            self.last_seq = seq
            self.initialized = True
            self.snapshot_count += 1
            return SeqCheck(SeqVerdict.SNAPSHOT, received=u,
                            note="u=1（snapshot 再送）")

        # 3) まだ snapshot を受け取っていない
        if not self.initialized:
            return SeqCheck(SeqVerdict.GAP, received=u,
                            note="snapshot 未受信のまま delta が到着")

        # 4) 既に適用済み
        if u <= self.last_update_id:
            self.duplicate_count += 1
            return SeqCheck(SeqVerdict.DUPLICATE,
                            expected=self.last_update_id + 1, received=u,
                            note="適用済みの update id")

        # 5) 欠番
        expected = self.last_update_id + 1
        if u != expected:
            gap = u - expected
            self.gap_count += 1
            self.gap_history.append((self.last_update_id, u))
            if len(self.gap_history) > 50:
                self.gap_history.pop(0)
            return SeqCheck(SeqVerdict.GAP, expected=expected, received=u,
                            gap_size=gap,
                            note=f"{gap}件の更新を取りこぼしました")

        # 6) 正常
        self.last_update_id = u
        self.last_seq = seq
        self.applied_count += 1
        return SeqCheck(SeqVerdict.APPLY, expected=expected, received=u)

    def reset(self) -> None:
        """再購読するときに呼ぶ。"""
        self.last_update_id = 0
        self.last_seq = 0
        self.initialized = False

    def stats(self) -> dict:
        total = (self.applied_count + self.snapshot_count
                 + self.duplicate_count + self.gap_count)
        return {
            "symbol": self.symbol,
            "total": total,
            "applied": self.applied_count,
            "snapshots": self.snapshot_count,
            "duplicates": self.duplicate_count,
            "gaps": self.gap_count,
            "gap_rate": self.gap_count / total if total else 0.0,
        }

    def report(self) -> str:
        s = self.stats()
        lines = [
            f"{s['symbol']}: 受信 {s['total']:,} / "
            f"適用 {s['applied']:,} / snapshot {s['snapshots']} / "
            f"重複 {s['duplicates']} / 欠番 {s['gaps']}",
            f"  欠番率: {s['gap_rate']:.4%}",
        ]
        if self.gap_history:
            lines.append("  直近の欠番:")
            for last_u, got_u in self.gap_history[-3:]:
                lines.append(f"    u={last_u} の次に u={got_u} "
                             f"（{got_u - last_u - 1}件欠落）")
        return "\n".join(lines)


if __name__ == "__main__":
    guard = SequenceGuard("BTCUSDT")

    print("=== シーケンス検証のデモ ===\n")

    cases = [
        ("snapshot", {"u": 100, "seq": 50}, "初回 snapshot"),
        ("delta", {"u": 101, "seq": 51}, "正常な delta"),
        ("delta", {"u": 102, "seq": 52}, "正常な delta"),
        ("delta", {"u": 102, "seq": 52}, "重複（再送）"),
        ("delta", {"u": 106, "seq": 56}, "欠番（103〜105が欠落）"),
        ("delta", {"u": 1, "seq": 60}, "u=1（snapshot 再送）"),
        ("delta", {"u": 2, "seq": 61}, "再送後の delta"),
    ]

    for msg_type, data, description in cases:
        result = guard.check(msg_type, data)
        mark = {
            SeqVerdict.APPLY: "  ",
            SeqVerdict.SNAPSHOT: "S ",
            SeqVerdict.DUPLICATE: "D ",
            SeqVerdict.GAP: "!!",
        }[result.verdict]
        note = f"  {result.note}" if result.note else ""
        print(f"{mark} type={msg_type:<9} u={data['u']:<5} "
              f"→ {result.verdict.value:<10}{note}")
        print(f"     ({description})")

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

    print("\n※ GAP を検出したら、ローカル板は破棄して再購読する。")
    print("  『少しくらいの欠番なら大丈夫』は成り立たない。")
    print("  削除イベントを落とすと、実在しない価格が板に残り続ける。")

"""book_manager.py — WSと繋いで板を維持し、壊れたら再購読する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  責務:
    - orderbook トピックの購読
    - シーケンス検証（サンプル2）を通してから板に適用（サンプル1）
    - GAP を検出したら板をリセットして再購読
    - 再購読が連続する場合は指数バックオフで間隔を空ける

  再購読の実装上の注意:
    pybit の WebSocket は購読の解除APIを直接は提供しないため、
    「接続ごと作り直す」のが最も確実。
    複数銘柄を1接続で購読している場合、1銘柄の再購読で
    全銘柄が一時的に途切れる点に注意する。
    重要銘柄は接続を分けるのが安全。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field

from pybit.unified_trading import WebSocket

from config import load_config
from local_book import LocalBook
from sequence_guard import SeqVerdict, SequenceGuard


@dataclass(frozen=True)
class BookConfig:
    category: str = "linear"
    depth: int = 50
    resubscribe_backoff_base: float = 1.0
    resubscribe_backoff_max: float = 30.0
    max_resubscribes_per_min: int = 10


@dataclass
class BookManager:
    """複数銘柄のローカル板を維持する。"""
    symbols: list
    config: BookConfig = field(default_factory=BookConfig)

    books: dict = field(default_factory=dict)
    guards: dict = field(default_factory=dict)
    _lock: threading.RLock = field(default_factory=threading.RLock)
    _ws: WebSocket | None = None
    _resubscribe_times: list = field(default_factory=list)
    _resubscribe_count: int = 0
    on_update: object = None       # callable(book) — 更新のたびに呼ぶ

    def __post_init__(self) -> None:
        for s in self.symbols:
            self.books[s] = LocalBook(s, depth=self.config.depth)
            self.guards[s] = SequenceGuard(s)

    # --- 受信処理 --------------------------------------------------

    def handle(self, message: dict) -> None:
        """WSコールバック。ここは可能な限り軽くする。"""
        data = message.get("data") or {}
        symbol = data.get("s", "")
        if not symbol or symbol not in self.books:
            return

        msg_type = message.get("type", "delta")
        ts = int(message.get("ts", 0))

        with self._lock:
            guard = self.guards[symbol]
            book = self.books[symbol]

            check = guard.check(msg_type, data)

            if check.verdict is SeqVerdict.SNAPSHOT:
                book.apply_snapshot(data, ts)
            elif check.verdict is SeqVerdict.APPLY:
                book.apply_delta(data, ts)
            elif check.verdict is SeqVerdict.DUPLICATE:
                return
            else:   # GAP
                print(f"  [GAP] {symbol}: {check.note} "
                      f"(expected u={check.expected}, got u={check.received})")
                book.reset()
                guard.reset()
                self._schedule_resubscribe(symbol)
                return

        if self.on_update is not None:
            try:
                self.on_update(book)
            except Exception as e:
                print(f"  [warn] on_update でエラー: {type(e).__name__}: {e}")

    # --- 再購読 ----------------------------------------------------

    def _backoff_delay(self) -> float:
        n = min(self._resubscribe_count, 6)
        delay = self.config.resubscribe_backoff_base * (2 ** n)
        return min(delay, self.config.resubscribe_backoff_max)

    def _schedule_resubscribe(self, symbol: str) -> None:
        """再購読を別スレッドで行う。コールバックを塞がないため。"""
        now = time.time()
        self._resubscribe_times = [t for t in self._resubscribe_times
                                   if now - t < 60]
        if len(self._resubscribe_times) >= self.config.max_resubscribes_per_min:
            print("  [警告] 再購読が頻発しています。")
            print("         回線品質かサーバ側の問題を疑ってください。")
            print("         記事24のサーキットブレーカーで停止を検討。")
            return

        self._resubscribe_times.append(now)
        self._resubscribe_count += 1
        delay = self._backoff_delay()

        def worker() -> None:
            print(f"  {delay:.1f}秒後に再購読します（{symbol}）")
            time.sleep(delay)
            self.start()

        threading.Thread(target=worker, daemon=True,
                         name=f"resub-{symbol}").start()

    # --- 接続 ------------------------------------------------------

    def start(self) -> None:
        """接続して購読する。既存の接続があれば作り直す。"""
        cfg = load_config()

        if self._ws is not None:
            try:
                self._ws.exit()
            except Exception:
                pass
            self._ws = None

        self._ws = WebSocket(testnet=cfg.testnet,
                             channel_type=self.config.category)

        for symbol in self.symbols:
            self._ws.orderbook_stream(
                depth=self.config.depth, symbol=symbol,
                callback=self.handle)

        print(f"  購読開始: {', '.join(self.symbols)} "
              f"(depth={self.config.depth})")

    def stop(self) -> None:
        if self._ws is not None:
            try:
                self._ws.exit()
            except Exception:
                pass
            self._ws = None

    # --- 参照 ------------------------------------------------------

    def book(self, symbol: str) -> LocalBook | None:
        with self._lock:
            return self.books.get(symbol)

    def ready_symbols(self) -> list:
        with self._lock:
            return [s for s, b in self.books.items() if b.ready]

    def report(self) -> str:
        with self._lock:
            lines = [f"再購読回数: {self._resubscribe_count}"]
            for symbol in self.symbols:
                book = self.books[symbol]
                guard = self.guards[symbol]
                status = "準備完了" if book.ready else "未同期"
                lines.append(
                    f"  {symbol:<12}[{status}] "
                    f"mid={book.mid:>12,.2f} "
                    f"spread={book.spread_bps:>6.2f}bps "
                    f"age={book.age_ms:>6.0f}ms")
                lines.append("    " + guard.report().replace("\n", "\n    "))
            return "\n".join(lines)


if __name__ == "__main__":
    manager = BookManager(["BTCUSDT", "ETHUSDT"],
                          BookConfig(category="linear", depth=50))
    manager.start()

    print("\n=== 板を維持中（Ctrl+C で終了） ===\n")
    try:
        while True:
            time.sleep(10)
            print(manager.report())
            print()

            btc = manager.book("BTCUSDT")
            if btc and btc.ready:
                print(btc.render(3))
                print()
    except KeyboardInterrupt:
        print("\n終了します。")
        manager.stop()

    # 注意: 1接続で複数銘柄を購読していると、
    #   1銘柄のGAPで全銘柄が再購読される。
    #   レイテンシが重要な銘柄は接続を分けること。

"""book_health.py — ローカル板の健全性を検査する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  検査項目:
    1. クロス      : best_bid >= best_ask （論理的にありえない）
    2. 単調性      : bid が降順、ask が昇順になっているか
    3. 深さ        : 階層数が極端に少なくないか
    4. 鮮度        : 最終更新から時間が経っていないか
    5. 異常値      : 数量が負、価格がゼロ以下
    6. スプレッド  : 平常値から大きく外れていないか

  1〜5 は「板が壊れている」ことを示す決定的な証拠。
  検出したら即座に再購読する。
  6 は相場の異常なので、板は正しいが取引を控えるべき状態（記事24）。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum

from local_book import LocalBook


class HealthLevel(str, Enum):
    OK = "正常"
    SUSPECT = "疑わしい"
    CORRUPT = "破損"


@dataclass
class HealthReport:
    symbol: str
    level: HealthLevel
    issues: list = field(default_factory=list)
    metrics: dict = field(default_factory=dict)

    @property
    def needs_resubscribe(self) -> bool:
        return self.level is HealthLevel.CORRUPT

    def report(self) -> str:
        lines = [f"[{self.level.value}] {self.symbol}"]
        for issue in self.issues:
            lines.append(f"    - {issue}")
        if self.metrics:
            lines.append("    metrics: " + "  ".join(
                f"{k}={v}" for k, v in self.metrics.items()))
        return "\n".join(lines)


@dataclass(frozen=True)
class HealthConfig:
    min_levels: int = 5
    max_age_ms: float = 5000.0
    max_spread_bps: float = 100.0
    check_monotonic_depth: int = 10


def check_book(book: LocalBook,
               cfg: HealthConfig = HealthConfig()) -> HealthReport:
    """ローカル板を検査する。"""
    issues: list = []
    level = HealthLevel.OK

    if not book.ready:
        return HealthReport(book.symbol, HealthLevel.SUSPECT,
                            ["snapshot をまだ受信していません"])

    bid, ask = book.best_bid, book.best_ask

    # 1) 空
    if bid <= 0 or ask <= 0:
        return HealthReport(book.symbol, HealthLevel.CORRUPT,
                            [f"最良気配が不正: bid={bid} ask={ask}"])

    # 2) クロス（最も決定的な破損の証拠）
    if bid >= ask:
        issues.append(
            f"板がクロスしています: bid={bid:,.2f} >= ask={ask:,.2f}。"
            "削除イベントを取りこぼした可能性が高い")
        level = HealthLevel.CORRUPT

    # 3) 単調性
    bids = book.bids.top(cfg.check_monotonic_depth)
    asks = book.asks.top(cfg.check_monotonic_depth)

    for i in range(1, len(bids)):
        if bids[i][0] >= bids[i - 1][0]:
            issues.append(
                f"bid が降順になっていません: "
                f"{bids[i - 1][0]} → {bids[i][0]}")
            level = HealthLevel.CORRUPT
            break

    for i in range(1, len(asks)):
        if asks[i][0] <= asks[i - 1][0]:
            issues.append(
                f"ask が昇順になっていません: "
                f"{asks[i - 1][0]} → {asks[i][0]}")
            level = HealthLevel.CORRUPT
            break

    # 4) 数量の異常
    for side_name, levels in (("bid", bids), ("ask", asks)):
        for price, size in levels:
            if size <= 0:
                issues.append(
                    f"{side_name} に数量 {size} の階層が残っています "
                    f"(price={price})。削除処理の実装を確認してください")
                level = HealthLevel.CORRUPT
                break

    # 5) 深さ不足
    if len(book.bids) < cfg.min_levels or len(book.asks) < cfg.min_levels:
        issues.append(
            f"階層数が不足: bid {len(book.bids)} / ask {len(book.asks)} "
            f"（最低 {cfg.min_levels}）")
        if level is HealthLevel.OK:
            level = HealthLevel.SUSPECT

    # 6) 鮮度
    if book.age_ms > cfg.max_age_ms:
        issues.append(
            f"最終更新から {book.age_ms:.0f}ms 経過しています。"
            "WSが切れている可能性があります")
        if level is HealthLevel.OK:
            level = HealthLevel.SUSPECT

    # 7) スプレッド（相場の異常。板自体は正しいかもしれない）
    if book.spread_bps > cfg.max_spread_bps:
        issues.append(
            f"スプレッドが {book.spread_bps:.1f}bps と異常に広いです")
        if level is HealthLevel.OK:
            level = HealthLevel.SUSPECT

    return HealthReport(
        book.symbol, level, issues,
        metrics={
            "bid": f"{bid:,.2f}",
            "ask": f"{ask:,.2f}",
            "spread_bps": f"{book.spread_bps:.2f}",
            "levels": f"{len(book.bids)}/{len(book.asks)}",
            "age_ms": f"{book.age_ms:.0f}",
            "updates": book.update_count,
        })


if __name__ == "__main__":
    import time

    # 正常な板
    good = LocalBook("BTCUSDT")
    good.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [[str(120000 - i * 0.5), "1.0"] for i in range(10)],
        "a": [[str(120000.5 + i * 0.5), "1.0"] for i in range(10)],
    }, ts=int(time.time() * 1000))

    print("=== 正常な板 ===")
    print(check_book(good).report())

    # クロスした板（削除イベントを取りこぼした状態を再現）
    broken = LocalBook("BTCUSDT")
    broken.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [["120001.0", "1.0"], ["120000.0", "1.0"]],
        "a": [["120000.5", "1.0"], ["120001.5", "1.0"]],
    }, ts=int(time.time() * 1000))

    print("\n=== クロスした板 ===")
    result = check_book(broken)
    print(result.report())
    print(f"\n  再購読が必要? {result.needs_resubscribe}")

    # 数量0が残った板
    zero_left = LocalBook("BTCUSDT")
    zero_left.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [["120000.0", "1.0"], ["119999.0", "1.0"]],
        "a": [["120001.0", "1.0"], ["120002.0", "1.0"]],
    }, ts=int(time.time() * 1000))
    # 削除ではなく0で上書きしてしまった状態を人工的に作る
    zero_left.bids.levels[-119999.0] = 0.0

    print("\n=== 数量0が残った板 ===")
    print(check_book(zero_left).report())

    print("\n※ クロスは『削除イベントを取りこぼした』ことの決定的な証拠。")
    print("  シーケンス検証を通っていてもクロスすることがある")
    print("  （実装のバグ）ので、独立した検査が必要。")

"""book_reconcile.py — ローカル板とREST板を定期照合する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  シーケンス検証と健全性チェックを両方通っていても、
  実装のバグで板が「それらしく見えるが間違っている」ことがある。
  最終的な答え合わせは REST の get_orderbook() と比較するしかない。

  照合の注意:
    - WSとRESTは取得時点が違うので、完全一致はしない
    - 上位数階層の価格が一致するかを見る（数量は動くので参考程度）
    - 照合は数分に1回で十分（記事05のレート制限を圧迫しない）

  照合で不一致が続くなら、それは実装のバグ。
  再購読しても直らないので、コードを見直すこと。
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field

from client import make_client
from local_book import LocalBook


@dataclass
class ReconcileResult:
    symbol: str
    matched_levels: int
    compared_levels: int
    bid_diff_bps: float
    ask_diff_bps: float
    ws_age_ms: float
    issues: list = field(default_factory=list)

    @property
    def match_rate(self) -> float:
        return (self.matched_levels / self.compared_levels
                if self.compared_levels else 0.0)

    @property
    def ok(self) -> bool:
        return not self.issues

    def report(self) -> str:
        status = "一致" if self.ok else "不一致"
        lines = [
            f"[{status}] {self.symbol}",
            f"    価格の一致: {self.matched_levels}/{self.compared_levels} "
            f"({self.match_rate:.0%})",
            f"    最良気配の差: bid {self.bid_diff_bps:+.3f}bps / "
            f"ask {self.ask_diff_bps:+.3f}bps",
            f"    WS板の鮮度: {self.ws_age_ms:.0f}ms",
        ]
        lines.extend(f"    - {i}" for i in self.issues)
        return "\n".join(lines)


def fetch_rest_book(http, symbol: str, limit: int = 50,
                    category: str = "linear") -> dict:
    """RESTで板を取得する。

    limit の既定値は category ごとに違う（spot=1, linear=25, option=1）。
    比較のため明示的に指定すること。
    """
    r = http.get_orderbook(category=category, symbol=symbol, limit=limit)
    if r["retCode"] != 0:
        raise RuntimeError(f"get_orderbook失敗: {r['retCode']} {r['retMsg']}")
    return r["result"]


def reconcile(http, book: LocalBook, *, category: str = "linear",
              compare_levels: int = 5,
              tolerance_bps: float = 1.0,
              min_match_rate: float = 0.8) -> ReconcileResult:
    """ローカル板とREST板を比較する。"""
    issues: list = []

    if not book.ready:
        return ReconcileResult(book.symbol, 0, 0, 0, 0, book.age_ms,
                               ["ローカル板が未同期です"])

    rest = fetch_rest_book(http, book.symbol,
                           limit=max(compare_levels, 25),
                           category=category)

    rest_bids = [(float(p), float(q)) for p, q in rest.get("b", [])]
    rest_asks = [(float(p), float(q)) for p, q in rest.get("a", [])]
    ws_bids = book.bids.top(compare_levels)
    ws_asks = book.asks.top(compare_levels)

    if not rest_bids or not rest_asks:
        return ReconcileResult(book.symbol, 0, 0, 0, 0, book.age_ms,
                               ["REST板が空です"])

    # 最良気配の乖離
    bid_diff = ((book.best_bid - rest_bids[0][0]) / rest_bids[0][0]
                * 10_000 if rest_bids[0][0] > 0 else 0.0)
    ask_diff = ((book.best_ask - rest_asks[0][0]) / rest_asks[0][0]
                * 10_000 if rest_asks[0][0] > 0 else 0.0)

    if abs(bid_diff) > tolerance_bps:
        issues.append(
            f"ベストビッドが {bid_diff:+.3f}bps 乖離 "
            f"(WS {book.best_bid:,.2f} vs REST {rest_bids[0][0]:,.2f})")
    if abs(ask_diff) > tolerance_bps:
        issues.append(
            f"ベストアスクが {ask_diff:+.3f}bps 乖離 "
            f"(WS {book.best_ask:,.2f} vs REST {rest_asks[0][0]:,.2f})")

    # 価格レベルの一致数
    matched = 0
    compared = 0
    for ws_side, rest_side in ((ws_bids, rest_bids), (ws_asks, rest_asks)):
        for i in range(min(len(ws_side), len(rest_side), compare_levels)):
            compared += 1
            if abs(ws_side[i][0] - rest_side[i][0]) < 1e-9:
                matched += 1

    rate = matched / compared if compared else 0.0
    if rate < min_match_rate:
        issues.append(
            f"価格レベルの一致率が {rate:.0%} と低いです。"
            "ローカル板が壊れている可能性があります")

    if book.age_ms > 3000:
        issues.append(
            f"WS板が {book.age_ms:.0f}ms 更新されていません")

    return ReconcileResult(book.symbol, matched, compared,
                           bid_diff, ask_diff, book.age_ms, issues)


def reconcile_loop(manager, interval_sec: float = 180.0,
                   category: str = "linear",
                   on_mismatch=None) -> None:
    """定期照合を回す。別スレッドで動かす想定。"""
    c = make_client()
    consecutive_failures: dict = {}

    while True:
        for symbol in manager.symbols:
            book = manager.book(symbol)
            if book is None:
                continue
            try:
                result = reconcile(c.http, book, category=category)
                if result.ok:
                    consecutive_failures[symbol] = 0
                else:
                    consecutive_failures[symbol] = \
                        consecutive_failures.get(symbol, 0) + 1
                    print(result.report())

                    if consecutive_failures[symbol] >= 3:
                        print(f"  [警告] {symbol}: 照合の失敗が3回続いています。")
                        print("         再購読では直らない実装バグの可能性。")
                        if on_mismatch is not None:
                            on_mismatch(symbol, result)
            except Exception as e:
                print(f"  [error] 照合失敗 {symbol}: "
                      f"{type(e).__name__}: {e}")
            time.sleep(0.5)

        time.sleep(interval_sec)


if __name__ == "__main__":
    import threading

    from book_manager import BookConfig, BookManager

    manager = BookManager(["BTCUSDT"], BookConfig(depth=50))
    manager.start()

    print("  snapshot を待機中...")
    for _ in range(30):
        book = manager.book("BTCUSDT")
        if book and book.ready:
            break
        time.sleep(1)

    c = make_client()
    book = manager.book("BTCUSDT")

    print("\n=== 照合を5回実行 ===\n")
    for i in range(5):
        try:
            result = reconcile(c.http, book)
            print(f"[{i + 1}/5] {result.report()}")
        except Exception as e:
            print(f"[{i + 1}/5] エラー: {e}")
        print()
        time.sleep(5)

    manager.stop()

    print("※ 最良気配が数bps乖離するのは正常（取得時点が違うため）。")
    print("  問題は『価格レベルの一致率が低い』ケース。")
    print("  これは板の構造そのものがずれている証拠。")

"""book_metrics.py — ローカル板からリアルタイムに指標を算出する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Stoikov, S. (2018) "The Micro-Price: A High Frequency Estimator
        of Future Prices", Quantitative Finance
      Cont, R., Kukanov, A. & Stoikov, S. (2014)
        "The Price Impact of Order Book Events",
        Journal of Financial Econometrics 12(1)

  指標:
    mid          = (bid + ask) / 2
                   最も単純だが、数量の偏りを無視する

    micro price  = (bid*ask_size + ask*bid_size) / (bid_size + ask_size)
                   買い板が厚ければ中値より上に寄る。
                   短期の将来価格の推定量として mid より優れる

    imbalance    = (bid_size − ask_size) / (bid_size + ask_size)
                   −1（売り優勢）〜 +1（買い優勢）

    OFI          = 板イベントの符号付き流量。
                   価格変化と線形関係を持つことが知られている

  注意: これらは「短期の方向性」を示すが、
        手数料とスプレッドを超える優位性があるとは限らない。
"""
from __future__ import annotations

from collections import deque
from dataclasses import dataclass, field

from local_book import LocalBook


@dataclass
class BookMetrics:
    symbol: str
    mid: float
    micro_price: float
    imbalance: float
    spread_bps: float
    depth_imbalance: float       # 上位N階層での偏り
    bid_depth: float
    ask_depth: float
    ts: int = 0

    @property
    def micro_minus_mid_bps(self) -> float:
        """マイクロプライスと中値の差。正なら上向きの圧力。"""
        return ((self.micro_price - self.mid) / self.mid * 10_000
                if self.mid > 0 else 0.0)

    def line(self) -> str:
        return (f"  {self.symbol:<10}"
                f"mid={self.mid:>12,.2f}  "
                f"micro={self.micro_price:>12,.2f}  "
                f"({self.micro_minus_mid_bps:>+6.2f}bps)  "
                f"imb={self.imbalance:>+6.3f}  "
                f"spread={self.spread_bps:>5.2f}bps")


def compute_metrics(book: LocalBook, depth: int = 5) -> BookMetrics | None:
    """板から指標を計算する。"""
    if not book.ready:
        return None

    bid, ask = book.best_bid, book.best_ask
    bid_size, ask_size = book.best_bid_size, book.best_ask_size

    if bid <= 0 or ask <= 0 or (bid_size + ask_size) <= 0:
        return None

    mid = (bid + ask) / 2

    # マイクロプライス: 反対側の数量で重み付けする
    # 買い板が厚い(bid_size大) → ask 寄りになる
    micro = (bid * ask_size + ask * bid_size) / (bid_size + ask_size)

    imbalance = (bid_size - ask_size) / (bid_size + ask_size)

    bid_depth = book.bids.total_size(depth)
    ask_depth = book.asks.total_size(depth)
    total_depth = bid_depth + ask_depth
    depth_imbalance = ((bid_depth - ask_depth) / total_depth
                       if total_depth > 0 else 0.0)

    return BookMetrics(
        symbol=book.symbol, mid=mid, micro_price=micro,
        imbalance=imbalance, spread_bps=book.spread_bps,
        depth_imbalance=depth_imbalance,
        bid_depth=bid_depth, ask_depth=ask_depth,
        ts=book.last_ts)


@dataclass
class OfiTracker:
    """Order Flow Imbalance を追跡する。

    Cont-Kukanov-Stoikov の定義:

      e_n = I(P^b_n >= P^b_{n-1}) * q^b_n − I(P^b_n <= P^b_{n-1}) * q^b_{n-1}
          − I(P^a_n <= P^a_{n-1}) * q^a_n + I(P^a_n >= P^a_{n-1}) * q^a_{n-1}

    直感的には:
      買い板が上がった/厚くなった → 正（買い圧力）
      売り板が下がった/厚くなった → 負（売り圧力）
    """
    symbol: str
    window: int = 100
    events: deque = field(default_factory=lambda: deque(maxlen=1000))

    _prev_bid: float = 0.0
    _prev_ask: float = 0.0
    _prev_bid_size: float = 0.0
    _prev_ask_size: float = 0.0
    _initialized: bool = False

    def update(self, book: LocalBook) -> float:
        """板の更新ごとに呼ぶ。今回のOFI寄与分を返す。"""
        bid, ask = book.best_bid, book.best_ask
        bid_size, ask_size = book.best_bid_size, book.best_ask_size

        if bid <= 0 or ask <= 0:
            return 0.0

        if not self._initialized:
            self._prev_bid, self._prev_ask = bid, ask
            self._prev_bid_size, self._prev_ask_size = bid_size, ask_size
            self._initialized = True
            return 0.0

        # bid 側の寄与
        if bid > self._prev_bid:
            bid_contrib = bid_size                  # 買い板が上がった
        elif bid < self._prev_bid:
            bid_contrib = -self._prev_bid_size      # 買い板が下がった
        else:
            bid_contrib = bid_size - self._prev_bid_size

        # ask 側の寄与（符号が逆）
        if ask < self._prev_ask:
            ask_contrib = -ask_size                 # 売り板が下がった
        elif ask > self._prev_ask:
            ask_contrib = self._prev_ask_size       # 売り板が上がった
        else:
            ask_contrib = -(ask_size - self._prev_ask_size)

        ofi = bid_contrib + ask_contrib

        self.events.append((book.last_ts or 0, ofi))
        self._prev_bid, self._prev_ask = bid, ask
        self._prev_bid_size, self._prev_ask_size = bid_size, ask_size

        return ofi

    def rolling_ofi(self, n: int = 0) -> float:
        """直近n件のOFI合計。"""
        n = n or self.window
        recent = list(self.events)[-n:]
        return sum(v for _, v in recent)

    def normalized_ofi(self, n: int = 0) -> float:
        """標準化したOFI。銘柄間で比較できる。"""
        n = n or self.window
        recent = [v for _, v in list(self.events)[-n:]]
        if len(recent) < 10:
            return 0.0
        total_abs = sum(abs(v) for v in recent)
        return sum(recent) / total_abs if total_abs > 0 else 0.0

    def reset(self) -> None:
        self.events.clear()
        self._initialized = False


class MetricsCollector:
    """BookManager の on_update に差し込んで使う。"""

    def __init__(self, symbols: list, depth: int = 5):
        self.depth = depth
        self.latest: dict = {}
        self.ofi: dict = {s: OfiTracker(s) for s in symbols}

    def __call__(self, book: LocalBook) -> None:
        metrics = compute_metrics(book, self.depth)
        if metrics is None:
            return
        self.latest[book.symbol] = metrics

        tracker = self.ofi.get(book.symbol)
        if tracker is not None:
            tracker.update(book)

    def report(self) -> str:
        lines = []
        for symbol, m in self.latest.items():
            tracker = self.ofi.get(symbol)
            ofi_str = ""
            if tracker:
                ofi_str = (f"  OFI={tracker.rolling_ofi():>+10.2f} "
                           f"(正規化 {tracker.normalized_ofi():>+6.3f})")
            lines.append(m.line() + ofi_str)
        return "\n".join(lines) if lines else "  データがありません"


if __name__ == "__main__":
    import time

    from book_manager import BookConfig, BookManager

    symbols = ["BTCUSDT", "ETHUSDT"]
    collector = MetricsCollector(symbols, depth=5)

    manager = BookManager(symbols, BookConfig(depth=50))
    manager.on_update = collector
    manager.start()

    print("=== リアルタイム指標（Ctrl+C で終了） ===\n")
    try:
        while True:
            time.sleep(3)
            print(collector.report())
            print()
    except KeyboardInterrupt:
        print("\n終了します。")
        manager.stop()

    print("\n【解釈】")
    print("  micro − mid が正 → 買い板が厚い → 短期的に上向きの圧力")
    print("  OFI が正        → 買い圧力が優勢")
    print("\n※ これらは『短期の方向性』を示すが、")
    print("  手数料とスプレッドを超える優位性があるとは限らない。")
    print("  記事31のバックテストでコスト込みで検証すること。")

"""book_profile.py — 板処理のレイテンシを測って最適化する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  測る3つの遅延:
    1. ネットワーク遅延 : サーバの ts と受信時のローカル時刻の差
       （時計が同期している前提。記事38のNTP設定が必須）
    2. 処理時間        : コールバック内の処理にかかる時間
    3. 滞留            : 更新間隔。処理が追いつかないと伸びる

  コールバックが重いと、WSの受信バッファに未処理メッセージが溜まり、
  遅延が雪だるま式に増える。
  「処理時間 < 更新間隔」を必ず満たすこと。

  最適化の勘所:
    - コールバック内では板の更新だけを行い、重い計算は別スレッドへ
    - json のデコードを orjson に置き換える（pybit の内部なので効果は限定的）
    - float 変換を減らす（毎回 float() を呼ぶのは意外に重い）
"""
from __future__ import annotations

import statistics
import time
from collections import deque
from dataclasses import dataclass, field

from local_book import LocalBook


@dataclass
class LatencyStats:
    name: str
    samples: deque = field(default_factory=lambda: deque(maxlen=5000))

    def add(self, value_ms: float) -> None:
        self.samples.append(value_ms)

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

    def line(self) -> str:
        p = self.percentiles()
        if not p:
            return f"  {self.name:<20}サンプル不足 ({len(self.samples)}件)"
        return (f"  {self.name:<20}"
                f"n={p['n']:>6}  "
                f"p50={p['p50']:>7.2f}  "
                f"p90={p['p90']:>7.2f}  "
                f"p99={p['p99']:>7.2f}  "
                f"max={p['max']:>8.2f}  (ms)")


@dataclass
class BookProfiler:
    """板処理のプロファイラ。BookManager.on_update に差し込む。"""
    network: LatencyStats = field(
        default_factory=lambda: LatencyStats("ネットワーク遅延"))
    processing: LatencyStats = field(
        default_factory=lambda: LatencyStats("処理時間"))
    interval: LatencyStats = field(
        default_factory=lambda: LatencyStats("更新間隔"))

    _last_update: dict = field(default_factory=dict)
    _last_process_start: float = 0.0
    inner: object = None       # 実際の処理（連鎖させる）

    def __call__(self, book: LocalBook) -> None:
        now = time.time()

        # 1) ネットワーク遅延（サーバtsとの差）
        if book.last_ts > 0:
            delay_ms = now * 1000 - book.last_ts
            # 時計がずれていると負になる。その場合は記録しない
            if 0 <= delay_ms < 60_000:
                self.network.add(delay_ms)

        # 2) 更新間隔
        prev = self._last_update.get(book.symbol)
        if prev is not None:
            self.interval.add((now - prev) * 1000)
        self._last_update[book.symbol] = now

        # 3) 処理時間
        start = time.perf_counter()
        if self.inner is not None:
            try:
                self.inner(book)
            except Exception as e:
                print(f"  [warn] inner でエラー: {type(e).__name__}: {e}")
        self.processing.add((time.perf_counter() - start) * 1000)

    def report(self) -> str:
        lines = ["=== レイテンシ統計 ==="]
        for stats in (self.network, self.interval, self.processing):
            lines.append(stats.line())

        proc = self.processing.percentiles()
        intv = self.interval.percentiles()
        if proc and intv:
            ratio = proc["p99"] / intv["p50"] if intv["p50"] > 0 else 0
            lines.append("")
            lines.append(f"  処理時間p99 / 更新間隔p50 = {ratio:.2%}")
            if ratio > 0.5:
                lines.append("  → 処理が重すぎます。コールバックを軽くしてください")
            elif ratio > 0.2:
                lines.append("  → 余裕は少なめ。銘柄数を増やすときは注意")
            else:
                lines.append("  → 十分な余裕があります")

        net = self.network.percentiles()
        if net:
            lines.append("")
            if net["p50"] > 200:
                lines.append(f"  ネットワーク遅延p50={net['p50']:.0f}ms は大きいです。")
                lines.append("  VPSのリージョンを見直してください（記事38）")
            elif net["p50"] < 0:
                lines.append("  遅延が負です。時計がずれています（NTP同期を確認）")

        return "\n".join(lines)


def benchmark_book_update(n_updates: int = 100_000) -> None:
    """板更新そのものの速度を測る（ネットワーク無し）。"""
    import random

    book = LocalBook("BENCH")
    book.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [[str(120000 - i * 0.5), "1.0"] for i in range(50)],
        "a": [[str(120000.5 + i * 0.5), "1.0"] for i in range(50)],
    })

    rng = random.Random(42)
    deltas = []
    for i in range(n_updates):
        side = "b" if rng.random() < 0.5 else "a"
        offset = rng.randint(0, 49) * 0.5
        price = (120000 - offset) if side == "b" else (120000.5 + offset)
        size = "0" if rng.random() < 0.2 else f"{rng.random() * 5:.4f}"
        deltas.append({"u": i + 2, "seq": i + 2, side: [[str(price), size]]})

    start = time.perf_counter()
    for d in deltas:
        book.apply_delta(d)
    elapsed = time.perf_counter() - start

    print(f"=== 板更新のベンチマーク ===")
    print(f"  {n_updates:,}件の delta を適用: {elapsed:.3f}秒")
    print(f"  1件あたり: {elapsed / n_updates * 1e6:.2f} マイクロ秒")
    print(f"  スループット: {n_updates / elapsed:,.0f} 件/秒")
    print(f"  最終階層数: bid {len(book.bids)} / ask {len(book.asks)}")

    print("\n  参考: orderbook.50 は 20ms 間隔 = 50件/秒。")
    print("  1銘柄なら余裕があるが、100銘柄なら 5,000件/秒 になる。")


if __name__ == "__main__":
    import sys

    if "--bench" in sys.argv:
        benchmark_book_update()
        sys.exit(0)

    from book_manager import BookConfig, BookManager
    from book_metrics import MetricsCollector

    symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
    profiler = BookProfiler(inner=MetricsCollector(symbols))

    manager = BookManager(symbols, BookConfig(depth=50))
    manager.on_update = profiler
    manager.start()

    print("=== プロファイリング中（60秒） ===\n")
    try:
        for i in range(6):
            time.sleep(10)
            print(f"[{(i + 1) * 10}秒]")
            print(profiler.report())
            print()
    except KeyboardInterrupt:
        pass
    finally:
        manager.stop()

    print("※ 処理時間p99が更新間隔の50%を超えたら危険信号。")
    print("  コールバックでは板の更新だけを行い、")
    print("  シグナル計算や発注判断は別スレッド・別タスクへ逃がすこと（記事30）。")

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

  保存する2種類:
    1. スナップショット : 一定間隔での板の状態（上位N階層）
       → 分析・バックテスト用。データ量が予測しやすい
    2. 生イベント       : delta をそのまま
       → 完全な再現が可能だが、データ量が膨大

  実務的には 1 で十分なことが多い。
  100ms 間隔・上位10階層なら 1銘柄1日で数百MB程度。

  重要:
    書き込みは必ずバッファリングして別スレッドで行う。
    コールバック内で I/O を行うと、そこで詰まって板が壊れる。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

from local_book import LocalBook


@dataclass(frozen=True)
class RecorderConfig:
    out_dir: Path = Path("./book_data")
    levels: int = 10               # 保存する階層数
    interval_ms: float = 100.0     # スナップショットの間隔
    flush_rows: int = 5000         # この行数ごとにファイルへ書く
    flush_sec: float = 60.0        # または、この秒数ごとに
    compression: str = "zstd"


@dataclass
class BookRecorder:
    """板を定期的にスナップショットして保存する。"""
    config: RecorderConfig = field(default_factory=RecorderConfig)

    _buffer: list = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock)
    _last_snapshot: dict = field(default_factory=dict)
    _last_flush: float = field(default_factory=time.time)
    _writer_thread: threading.Thread | None = None
    _stop_event: threading.Event = field(default_factory=threading.Event)

    rows_written: int = 0
    files_written: int = 0
    inner: object = None

    def __post_init__(self) -> None:
        self.config.out_dir.mkdir(parents=True, exist_ok=True)

    # --- 収集（コールバックから呼ばれる。軽く保つ） ----------------

    def __call__(self, book: LocalBook) -> None:
        now = time.time()
        last = self._last_snapshot.get(book.symbol, 0.0)

        if (now - last) * 1000 >= self.config.interval_ms:
            self._last_snapshot[book.symbol] = now
            self._capture(book, now)

        if self.inner is not None:
            self.inner(book)

    def _capture(self, book: LocalBook, now: float) -> None:
        """板を1行のフラットな辞書にする。"""
        n = self.config.levels
        bids = book.bids.top(n)
        asks = book.asks.top(n)

        row = {
            "local_ts": now,
            "server_ts": book.last_ts,
            "symbol": book.symbol,
            "u": book.last_update_id,
            "seq": book.last_seq,
            "mid": book.mid,
            "spread": book.spread,
        }

        # 階層を bid_px_0, bid_sz_0, ... という列に展開する
        for i in range(n):
            if i < len(bids):
                row[f"bid_px_{i}"] = bids[i][0]
                row[f"bid_sz_{i}"] = bids[i][1]
            else:
                row[f"bid_px_{i}"] = 0.0
                row[f"bid_sz_{i}"] = 0.0

            if i < len(asks):
                row[f"ask_px_{i}"] = asks[i][0]
                row[f"ask_sz_{i}"] = asks[i][1]
            else:
                row[f"ask_px_{i}"] = 0.0
                row[f"ask_sz_{i}"] = 0.0

        with self._lock:
            self._buffer.append(row)

    # --- 書き出し（別スレッド） ------------------------------------

    def _should_flush(self) -> bool:
        return (len(self._buffer) >= self.config.flush_rows
                or time.time() - self._last_flush >= self.config.flush_sec)

    def flush(self) -> int:
        """バッファをParquetに書き出す。"""
        with self._lock:
            if not self._buffer:
                return 0
            rows = self._buffer
            self._buffer = []

        self._last_flush = time.time()

        df = pd.DataFrame(rows)
        df["datetime"] = pd.to_datetime(df["local_ts"], unit="s", utc=True)

        # 銘柄・日付ごとにファイルを分ける
        for (symbol, day), group in df.groupby(
                ["symbol", df["datetime"].dt.strftime("%Y%m%d")]):
            path = self.config.out_dir / f"{symbol}_{day}.parquet"

            if path.exists():
                # 追記できないので読み込んで結合する
                # （行数が多くなったら日付でさらに細かく分けること）
                existing = pd.read_parquet(path)
                group = pd.concat([existing, group], ignore_index=True)
            else:
                self.files_written += 1

            group.to_parquet(path, index=False,
                             compression=self.config.compression)

        self.rows_written += len(rows)
        return len(rows)

    def start(self) -> None:
        """書き出しスレッドを開始する。"""
        if self._writer_thread and self._writer_thread.is_alive():
            return

        def loop() -> None:
            while not self._stop_event.is_set():
                try:
                    if self._should_flush():
                        n = self.flush()
                        if n:
                            print(f"  [recorder] {n:,}行を書き出し "
                                  f"(累計 {self.rows_written:,})")
                except Exception as e:
                    print(f"  [recorder error] {type(e).__name__}: {e}")
                self._stop_event.wait(5.0)

        self._stop_event.clear()
        self._writer_thread = threading.Thread(
            target=loop, daemon=True, name="book-recorder")
        self._writer_thread.start()
        print(f"  レコーダ起動: {self.config.out_dir} "
              f"({self.config.interval_ms:.0f}ms間隔 / "
              f"{self.config.levels}階層)")

    def stop(self) -> None:
        self._stop_event.set()
        if self._writer_thread:
            self._writer_thread.join(timeout=10.0)
        remaining = self.flush()
        if remaining:
            print(f"  [recorder] 終了時に {remaining:,}行を書き出し")

    def report(self) -> str:
        with self._lock:
            buffered = len(self._buffer)
        return (f"  記録: {self.rows_written:,}行 / "
                f"バッファ {buffered:,}行 / "
                f"ファイル {self.files_written}個")


def estimate_size(symbols: int, levels: int, interval_ms: float,
                  hours: float = 24.0) -> None:
    """データ量を見積もる。走らせる前に確認すること。"""
    rows_per_symbol = hours * 3600 * 1000 / interval_ms
    total_rows = rows_per_symbol * symbols
    # 1行の列数: 固定7列 + 階層ごと4列
    cols = 7 + levels * 4
    # float64 = 8バイト。zstd圧縮で概ね 1/4 程度
    raw_mb = total_rows * cols * 8 / 1024 / 1024
    compressed_mb = raw_mb / 4

    print(f"  {symbols}銘柄 × {levels}階層 × {interval_ms:.0f}ms間隔 "
          f"× {hours:.0f}時間")
    print(f"    行数    : {total_rows:,.0f}")
    print(f"    非圧縮  : {raw_mb:,.0f} MB")
    print(f"    圧縮後  : {compressed_mb:,.0f} MB（概算）")


if __name__ == "__main__":
    import sys

    if "--estimate" in sys.argv:
        print("=== データ量の見積もり ===\n")
        for symbols, levels, interval in ((1, 10, 100), (1, 10, 1000),
                                          (5, 10, 100), (5, 25, 100),
                                          (20, 5, 1000)):
            estimate_size(symbols, levels, interval)
            print()
        sys.exit(0)

    from book_manager import BookConfig, BookManager

    recorder = BookRecorder(RecorderConfig(
        levels=10, interval_ms=100.0, flush_rows=2000, flush_sec=30.0))

    manager = BookManager(["BTCUSDT"], BookConfig(depth=50))
    manager.on_update = recorder
    manager.start()
    recorder.start()

    print("\n=== 記録中（Ctrl+C で終了） ===\n")
    try:
        while True:
            time.sleep(15)
            print(recorder.report())
    except KeyboardInterrupt:
        print("\n終了します。")
    finally:
        recorder.stop()
        manager.stop()

    print("\n※ コールバック内で I/O を行うと、そこで詰まって板が壊れる。")
    print("  必ずバッファリングして別スレッドで書き出すこと。")
    print("  python book_recorder.py --estimate でデータ量を試算できる。")

"""test_local_book.py — 板ロジックをネットワーク無しで検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pytest test_local_book.py -v

  板の実装は「実際のWSに繋がないと確認できない」と思われがちだが、
  ロジックの大半はオフラインでテストできる。
  むしろネットワークに依存させると、
  バグの再現ができず修正の確認もできない。
"""
from __future__ import annotations

import pytest

from book_health import HealthLevel, check_book
from local_book import BookSide, LocalBook
from sequence_guard import SeqVerdict, SequenceGuard


# --- BookSide -------------------------------------------------------

def test_bid_side_orders_high_to_low():
    """bid は高い順に並ぶこと。"""
    side = BookSide(is_bid=True)
    for price in (100.0, 102.0, 101.0):
        side.apply(price, 1.0)
    assert [p for p, _ in side.top(3)] == [102.0, 101.0, 100.0]


def test_ask_side_orders_low_to_high():
    """ask は安い順に並ぶこと。"""
    side = BookSide(is_bid=False)
    for price in (102.0, 100.0, 101.0):
        side.apply(price, 1.0)
    assert [p for p, _ in side.top(3)] == [100.0, 101.0, 102.0]


def test_zero_size_deletes_level():
    """数量0は『0で上書き』ではなく削除であること。"""
    side = BookSide(is_bid=True)
    side.apply(100.0, 1.0)
    side.apply(101.0, 2.0)
    assert len(side) == 2

    side.apply(101.0, 0.0)
    assert len(side) == 1
    assert side.best() == (100.0, 1.0)


def test_delete_nonexistent_level_is_safe():
    """存在しない階層の削除でエラーにならないこと。"""
    side = BookSide(is_bid=True)
    side.apply(100.0, 0.0)      # 何もしないだけ
    assert len(side) == 0


def test_empty_side_best_is_none():
    assert BookSide(is_bid=True).best() is None


# --- LocalBook ------------------------------------------------------

def make_book() -> LocalBook:
    book = LocalBook("TEST")
    book.apply_snapshot({
        "u": 100, "seq": 50,
        "b": [["100.0", "1.0"], ["99.0", "2.0"], ["98.0", "3.0"]],
        "a": [["101.0", "1.5"], ["102.0", "2.5"], ["103.0", "3.5"]],
    })
    return book


def test_snapshot_replaces_entire_book():
    """snapshot は板を丸ごと置き換えること（追記ではない）。"""
    book = make_book()
    book.apply_snapshot({
        "u": 200, "seq": 100,
        "b": [["50.0", "1.0"]], "a": [["51.0", "1.0"]],
    })
    assert len(book.bids) == 1
    assert len(book.asks) == 1
    assert book.best_bid == 50.0


def test_delta_updates_and_deletes():
    book = make_book()
    book.apply_delta({
        "u": 101, "seq": 51,
        "b": [["100.0", "0"], ["97.0", "5.0"]],
        "a": [["101.0", "0.5"]],
    })
    assert book.best_bid == 99.0        # 100.0 が消えた
    assert book.best_ask_size == 0.5    # 101.0 の数量が更新された
    assert len(book.bids) == 3          # 97.0 が追加された


def test_mid_and_spread():
    book = make_book()
    assert book.mid == pytest.approx(100.5)
    assert book.spread == pytest.approx(1.0)
    assert book.spread_bps == pytest.approx(1.0 / 100.5 * 10_000)


def test_reset_clears_everything():
    book = make_book()
    book.reset()
    assert not book.ready
    assert book.last_update_id == 0
    assert len(book.bids) == 0


# --- SequenceGuard --------------------------------------------------

def test_sequential_deltas_are_applied():
    guard = SequenceGuard("TEST")
    assert guard.check("snapshot", {"u": 10}).verdict is SeqVerdict.SNAPSHOT
    assert guard.check("delta", {"u": 11}).verdict is SeqVerdict.APPLY
    assert guard.check("delta", {"u": 12}).verdict is SeqVerdict.APPLY


def test_gap_is_detected():
    """欠番を確実に検出すること。これが最重要のテスト。"""
    guard = SequenceGuard("TEST")
    guard.check("snapshot", {"u": 10})
    guard.check("delta", {"u": 11})

    result = guard.check("delta", {"u": 15})
    assert result.verdict is SeqVerdict.GAP
    assert result.expected == 12
    assert result.received == 15
    assert result.gap_size == 3


def test_duplicate_is_ignored():
    guard = SequenceGuard("TEST")
    guard.check("snapshot", {"u": 10})
    guard.check("delta", {"u": 11})
    assert guard.check("delta", {"u": 11}).verdict is SeqVerdict.DUPLICATE
    assert guard.last_update_id == 11    # 巻き戻らないこと


def test_u_equals_one_is_snapshot():
    """type=delta でも u=1 なら snapshot 扱いであること。"""
    guard = SequenceGuard("TEST")
    guard.check("snapshot", {"u": 500})
    guard.check("delta", {"u": 501})

    result = guard.check("delta", {"u": 1})
    assert result.verdict is SeqVerdict.SNAPSHOT
    assert guard.last_update_id == 1


def test_delta_before_snapshot_is_gap():
    """snapshot 未受信の delta は適用してはいけない。"""
    guard = SequenceGuard("TEST")
    assert guard.check("delta", {"u": 100}).verdict is SeqVerdict.GAP


def test_gap_does_not_advance_state():
    """GAP のとき last_update_id を進めないこと。

    進めてしまうと、再購読せずに壊れた板を使い続けることになる。
    """
    guard = SequenceGuard("TEST")
    guard.check("snapshot", {"u": 10})
    guard.check("delta", {"u": 15})     # GAP
    assert guard.last_update_id == 10


# --- 健全性チェック --------------------------------------------------

def test_healthy_book_passes():
    book = LocalBook("TEST")
    book.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [[str(100 - i), "1.0"] for i in range(10)],
        "a": [[str(101 + i), "1.0"] for i in range(10)],
    })
    assert check_book(book).level is HealthLevel.OK


def test_crossed_book_is_corrupt():
    """クロスした板を破損と判定すること。"""
    book = LocalBook("TEST")
    book.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [["102.0", "1.0"], ["101.0", "1.0"]],
        "a": [["101.5", "1.0"], ["102.5", "1.0"]],
    })
    result = check_book(book)
    assert result.level is HealthLevel.CORRUPT
    assert result.needs_resubscribe


def test_unready_book_is_suspect():
    book = LocalBook("TEST")
    assert check_book(book).level is HealthLevel.SUSPECT


# --- 統合: delta の連続適用で板が壊れないこと ------------------------

def test_many_deltas_keep_book_consistent():
    """ランダムな delta を大量に適用しても、板が不整合にならないこと。"""
    import random

    book = LocalBook("TEST")
    book.apply_snapshot({
        "u": 1, "seq": 1,
        "b": [[str(1000 - i), "1.0"] for i in range(50)],
        "a": [[str(1001 + i), "1.0"] for i in range(50)],
    })

    rng = random.Random(123)
    for i in range(10_000):
        # bid は 1000 以下、ask は 1001 以上に保つ（クロスさせない）
        if rng.random() < 0.5:
            price = 1000 - rng.randint(0, 49)
            key = "b"
        else:
            price = 1001 + rng.randint(0, 49)
            key = "a"
        size = "0" if rng.random() < 0.3 else f"{rng.random() * 3:.4f}"
        book.apply_delta({"u": i + 2, "seq": i + 2,
                          key: [[str(price), size]]})

    # 数量0の階層が残っていないこと
    assert all(q > 0 for _, q in book.bids.top(50))
    assert all(q > 0 for _, q in book.asks.top(50))

    # クロスしていないこと
    if book.best_bid > 0 and book.best_ask > 0:
        assert book.best_bid < book.best_ask