自動売買botのサーキットブレーカー実装
日次DD上限・連敗停止・キルスイッチ

"""risk_state.py — リスク状態機械
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  NORMAL → REDUCE_ONLY → FLATTEN → HALTED

  設計:
    - 危険側への遷移は即座（1つの検知で飛べる）
    - 回復は「連続 N 回正常」を要求（フラッピング防止）
    - HALTED からは自動復帰しない（人間の承認が必要）
    - 遷移の理由を必ず記録する

  スレッドセーフにしてあるのは、
  監視スレッドと発注スレッドが同時に触るため。
"""
from __future__ import annotations

import json
import threading
import time
from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path

STATE_PATH = Path("./risk_state.json")


class RiskState(IntEnum):
    NORMAL = 0
    REDUCE_ONLY = 1
    FLATTEN = 2
    HALTED = 3

    @property
    def label(self) -> str:
        return {
            RiskState.NORMAL: "通常",
            RiskState.REDUCE_ONLY: "縮退（新規停止）",
            RiskState.FLATTEN: "全決済中",
            RiskState.HALTED: "完全停止",
        }[self]

    @property
    def can_open(self) -> bool:
        return self is RiskState.NORMAL

    @property
    def can_close(self) -> bool:
        return self is not RiskState.HALTED


@dataclass(frozen=True)
class Transition:
    ts: float
    from_state: int
    to_state: int
    reason: str
    detector: str

    def line(self) -> str:
        return (f"  {time.strftime('%H:%M:%S', time.localtime(self.ts))}  "
                f"{RiskState(self.from_state).name} → "
                f"{RiskState(self.to_state).name}"
                f"  [{self.detector}] {self.reason}")


@dataclass
class RiskStateMachine:
    state: RiskState = RiskState.NORMAL
    recovery_required: int = 12       # 回復に必要な連続正常回数
    _consecutive_ok: int = 0
    _lock: threading.RLock = field(default_factory=threading.RLock)
    history: list[Transition] = field(default_factory=list)
    entered_at: float = field(default_factory=time.time)

    # --- 状態の問い合わせ ------------------------------------------

    @property
    def can_open(self) -> bool:
        with self._lock:
            return self.state.can_open

    @property
    def can_close(self) -> bool:
        with self._lock:
            return self.state.can_close

    def seconds_in_state(self) -> float:
        return time.time() - self.entered_at

    # --- 遷移 ------------------------------------------------------

    def _transit(self, target: RiskState, reason: str,
                 detector: str) -> bool:
        if target is self.state:
            return False
        t = Transition(time.time(), int(self.state), int(target),
                       reason, detector)
        self.history.append(t)
        self.state = target
        self.entered_at = time.time()
        self._consecutive_ok = 0
        print(t.line())
        self.save()
        return True

    def escalate(self, target: RiskState, reason: str,
                 detector: str = "") -> bool:
        """危険側へ遷移する。現在より安全な状態への呼び出しは無視する。"""
        with self._lock:
            if self.state is RiskState.HALTED:
                return False          # HALTED は終端
            if target <= self.state:
                self._consecutive_ok = 0
                return False
            return self._transit(target, reason, detector)

    def report_ok(self, detector: str = "") -> bool:
        """異常が無いことを報告する。連続すれば回復する。"""
        with self._lock:
            if self.state in (RiskState.NORMAL, RiskState.HALTED):
                return False
            if self.state is RiskState.FLATTEN:
                return False          # 決済中は回復しない

            self._consecutive_ok += 1
            if self._consecutive_ok < self.recovery_required:
                return False

            return self._transit(
                RiskState.NORMAL,
                f"{self._consecutive_ok}回連続で正常",
                detector or "recovery")

    def halt(self, reason: str, detector: str = "") -> bool:
        """即座に完全停止する。"""
        with self._lock:
            if self.state is RiskState.HALTED:
                return False
            return self._transit(RiskState.HALTED, reason, detector)

    # --- 永続化 ----------------------------------------------------

    def save(self, path: Path = STATE_PATH) -> None:
        data = {
            "state": int(self.state),
            "state_name": self.state.name,
            "entered_at": self.entered_at,
            "consecutive_ok": self._consecutive_ok,
            "history": [t.__dict__ for t in self.history[-50:]],
        }
        path.write_text(json.dumps(data, ensure_ascii=False, indent=2),
                        encoding="utf-8")

    @classmethod
    def load(cls, path: Path = STATE_PATH) -> "RiskStateMachine":
        """再起動時に前回の状態を引き継ぐ。

        重要: HALTED だった場合は HALTED のまま起動する。
              再起動で勝手に復帰させてはいけない。
        """
        m = cls()
        if not path.exists():
            return m
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            m.state = RiskState(int(data.get("state", 0)))
            m.entered_at = float(data.get("entered_at", time.time()))
            m.history = [Transition(**t) for t in data.get("history", [])]
            if m.state is not RiskState.NORMAL:
                print(f"  [起動] 前回の状態を引き継ぎ: {m.state.label}")
        except Exception as e:
            print(f"  [warn] 状態の読み込みに失敗: {e}")
        return m

    def summary(self) -> str:
        lines = [
            f"現在の状態: {self.state.name} ({self.state.label})",
            f"  滞在時間: {self.seconds_in_state():.0f}秒",
            f"  新規建て: {'可' if self.can_open else '不可'}",
            f"  決済    : {'可' if self.can_close else '不可'}",
        ]
        if self.history:
            lines.append("  --- 遷移履歴（直近5件） ---")
            lines.extend(t.line() for t in self.history[-5:])
        return "\n".join(lines)


if __name__ == "__main__":
    m = RiskStateMachine(recovery_required=3)

    print("=== 状態遷移のデモ ===\n")
    m.escalate(RiskState.REDUCE_ONLY, "日次DDが3%を超過", "drawdown")
    print(f"  新規建て可能? {m.can_open}\n")

    # 1回正常でも戻らない
    m.report_ok("drawdown")
    print(f"  1回正常後: {m.state.name}")
    m.report_ok("drawdown")
    print(f"  2回正常後: {m.state.name}")
    m.report_ok("drawdown")
    print(f"  3回正常後: {m.state.name}\n")

    # より危険な検知は即座に飛ぶ
    m.escalate(RiskState.FLATTEN, "日次DDが8%を超過", "drawdown")
    m.escalate(RiskState.REDUCE_ONLY, "軽微な異常", "spread")
    print(f"  安全側への escalate は無視される: {m.state.name}\n")

    m.halt("全決済完了", "killswitch")
    print(f"  HALTED後の回復試行: {m.report_ok()}  → {m.state.name}\n")

    print(m.summary())
    print("\n※ HALTED からは自動復帰しない。")
    print("  サンプル9のロック解除（人間の操作）が必要。")

"""drawdown_guard.py — 日次DD（実現＋含み）を計算して段階的に発動する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  DDの定義は複数あるので、何を測るかを明示する:

    1. 日次DD     : 当日の開始時点の資産からの下落率
    2. ピークDD   : 直近の資産ピークからの下落率
    3. 実現のみDD : 決済済み損益だけ（含み損を無視する = 危険）

  本実装は 1 と 2 の両方を見る。
  「今日は勝っているが、月初からは大きく負けている」という
  状態を見逃さないため。

  基準となる資産は totalMarginBalance（未実現損益込み）を使う（記事22）。
"""
from __future__ import annotations

import json
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path

from margin_snapshot import fetch_account_margin
from risk_state import RiskState, RiskStateMachine

EQUITY_PATH = Path("./equity_track.json")


@dataclass(frozen=True)
class DrawdownThresholds:
    daily_reduce: Decimal = Decimal("0.03")    # 日次3%で新規停止
    daily_flatten: Decimal = Decimal("0.06")   # 日次6%で全決済
    peak_reduce: Decimal = Decimal("0.10")     # ピークから10%で新規停止
    peak_flatten: Decimal = Decimal("0.20")    # ピークから20%で全決済


@dataclass
class EquityTracker:
    """資産の推移を追跡する。プロセス再起動をまたいで永続化する。"""
    day_start_equity: Decimal = Decimal(0)
    day_start_date: str = ""
    peak_equity: Decimal = Decimal(0)
    current_equity: Decimal = Decimal(0)
    updated_at: float = 0.0
    path: Path = field(default=EQUITY_PATH)

    @staticmethod
    def _today() -> str:
        return datetime.now(timezone.utc).strftime("%Y-%m-%d")

    def update(self, equity: Decimal) -> None:
        today = self._today()

        # 日付が変わったら日次基準をリセット
        if today != self.day_start_date:
            self.day_start_date = today
            self.day_start_equity = equity
            print(f"  [日次リセット] {today} 開始資産 {equity:.2f} USD")

        self.current_equity = equity
        if equity > self.peak_equity:
            self.peak_equity = equity
        self.updated_at = time.time()
        self.save()

    @property
    def daily_drawdown(self) -> Decimal:
        """当日の下落率（正の値が下落）。"""
        if self.day_start_equity <= 0:
            return Decimal(0)
        return max(Decimal(0),
                   (self.day_start_equity - self.current_equity)
                   / self.day_start_equity)

    @property
    def peak_drawdown(self) -> Decimal:
        """ピークからの下落率。"""
        if self.peak_equity <= 0:
            return Decimal(0)
        return max(Decimal(0),
                   (self.peak_equity - self.current_equity) / self.peak_equity)

    @property
    def daily_pnl(self) -> Decimal:
        return self.current_equity - self.day_start_equity

    def save(self) -> None:
        self.path.write_text(json.dumps({
            "day_start_equity": str(self.day_start_equity),
            "day_start_date": self.day_start_date,
            "peak_equity": str(self.peak_equity),
            "current_equity": str(self.current_equity),
            "updated_at": self.updated_at,
        }, ensure_ascii=False, indent=2), encoding="utf-8")

    @classmethod
    def load(cls, path: Path = EQUITY_PATH) -> "EquityTracker":
        t = cls(path=path)
        if not path.exists():
            return t
        try:
            d = json.loads(path.read_text(encoding="utf-8"))
            t.day_start_equity = Decimal(d.get("day_start_equity", "0"))
            t.day_start_date = d.get("day_start_date", "")
            t.peak_equity = Decimal(d.get("peak_equity", "0"))
            t.current_equity = Decimal(d.get("current_equity", "0"))
            t.updated_at = float(d.get("updated_at", 0))
        except Exception as e:
            print(f"  [warn] 資産履歴の読み込みに失敗: {e}")
        return t

    def report(self) -> str:
        return "\n".join([
            f"  当日開始 : {self.day_start_equity:>12.2f} USD "
            f"({self.day_start_date})",
            f"  現在     : {self.current_equity:>12.2f} USD",
            f"  ピーク   : {self.peak_equity:>12.2f} USD",
            f"  日次損益 : {self.daily_pnl:>+12.2f} USD",
            f"  日次DD   : {self.daily_drawdown:>12.2%}",
            f"  ピークDD : {self.peak_drawdown:>12.2%}",
        ])


class DrawdownDetector:
    """DDを監視して状態機械に報告する。"""

    name = "drawdown"

    def __init__(self, http, machine: RiskStateMachine,
                 thresholds: DrawdownThresholds = DrawdownThresholds()):
        self.http = http
        self.machine = machine
        self.th = thresholds
        self.tracker = EquityTracker.load()

    def check(self) -> dict:
        margin = fetch_account_margin(self.http)
        self.tracker.update(margin.total_margin_balance)

        daily = self.tracker.daily_drawdown
        peak = self.tracker.peak_drawdown

        result = {
            "daily_dd": float(daily), "peak_dd": float(peak),
            "equity": float(self.tracker.current_equity),
            "triggered": None,
        }

        # 厳しい条件から順に判定する
        if daily >= self.th.daily_flatten:
            self.machine.escalate(
                RiskState.FLATTEN,
                f"日次DD {daily:.2%} が上限 {self.th.daily_flatten:.2%} を超過",
                self.name)
            result["triggered"] = "daily_flatten"
        elif peak >= self.th.peak_flatten:
            self.machine.escalate(
                RiskState.FLATTEN,
                f"ピークDD {peak:.2%} が上限 {self.th.peak_flatten:.2%} を超過",
                self.name)
            result["triggered"] = "peak_flatten"
        elif daily >= self.th.daily_reduce:
            self.machine.escalate(
                RiskState.REDUCE_ONLY,
                f"日次DD {daily:.2%} が閾値 {self.th.daily_reduce:.2%} を超過",
                self.name)
            result["triggered"] = "daily_reduce"
        elif peak >= self.th.peak_reduce:
            self.machine.escalate(
                RiskState.REDUCE_ONLY,
                f"ピークDD {peak:.2%} が閾値 {self.th.peak_reduce:.2%} を超過",
                self.name)
            result["triggered"] = "peak_reduce"
        else:
            self.machine.report_ok(self.name)

        return result


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

    c = make_client()
    machine = RiskStateMachine.load()
    detector = DrawdownDetector(c.http, machine)

    result = detector.check()
    print("=== ドローダウン監視 ===")
    print(detector.tracker.report())
    print(f"\n  発動: {result['triggered'] or 'なし'}")
    print(f"\n{machine.summary()}")

    print("\n※ 含み損を含めない『実現損益だけのDD』は危険。")
    print("  大きな含み損を抱えたまま『今日はまだ損失0』と")
    print("  判定してしまう。totalMarginBalance を基準にすること。")

"""anomaly_detector.py — 連敗・約定頻度の異常を検知する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  検知する2つの異常:

  1. 連敗
     戦略が相場に合わなくなったサイン。
     ただし「勝率40%の戦略で5連敗」は確率的に普通に起きる
     （0.6^5 ≒ 7.8%）。閾値は勝率から逆算すること。

  2. 約定頻度の異常増加
     バグで発注ループに入った場合の最重要シグナル。
     「通常の10倍の頻度で約定している」は、
     相場ではなくコードの異常である可能性が高い。

  約定履歴は7日制約があるため、直近だけを見る（記事18）。
"""
from __future__ import annotations

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

import numpy as np

from margin_snapshot import d
from risk_state import RiskState, RiskStateMachine


@dataclass(frozen=True)
class AnomalyThresholds:
    max_consecutive_losses: int = 6
    halt_consecutive_losses: int = 10
    max_fills_per_minute: float = 10.0     # 通常運用の上限
    fill_rate_multiplier: float = 5.0      # 平常の何倍で異常とするか
    baseline_window_sec: float = 3600.0    # 平常値の学習窓
    min_baseline_samples: int = 20


@dataclass
class TradeOutcome:
    ts: float
    symbol: str
    pnl: Decimal
    exec_id: str

    @property
    def is_win(self) -> bool:
        return self.pnl > 0


@dataclass
class AnomalyDetector:
    """連敗と約定頻度を監視する。"""
    http: object
    machine: RiskStateMachine
    thresholds: AnomalyThresholds = field(default_factory=AnomalyThresholds)
    category: str = "linear"
    settle_coin: str = "USDT"
    name: str = "anomaly"

    _fill_times: deque = field(default_factory=lambda: deque(maxlen=2000))
    _seen_exec_ids: set = field(default_factory=set)
    _last_poll_ms: int = 0
    consecutive_losses: int = 0
    recent_outcomes: deque = field(default_factory=lambda: deque(maxlen=100))

    # --- 約定の取り込み --------------------------------------------

    def poll_closed_pnl(self) -> list[TradeOutcome]:
        """決済済み損益を取得する。

        get_closed_pnl は7日以内の範囲のみ。
        startTime を省略すると直近が返る。
        spot は非対応（linear/inverse のみ）。
        """
        params = {"category": self.category, "limit": 50}
        if self.settle_coin:
            params["settleCoin"] = self.settle_coin
        if self._last_poll_ms:
            params["startTime"] = self._last_poll_ms

        r = self.http.get_closed_pnl(**params)
        if r["retCode"] != 0:
            raise RuntimeError(
                f"get_closed_pnl失敗: {r['retCode']} {r['retMsg']}")

        new: list[TradeOutcome] = []
        for item in r["result"]["list"]:
            exec_id = item.get("orderId", "") + item.get("updatedTime", "")
            if exec_id in self._seen_exec_ids:
                continue
            self._seen_exec_ids.add(exec_id)
            new.append(TradeOutcome(
                ts=int(item.get("updatedTime", 0)) / 1000,
                symbol=item.get("symbol", ""),
                pnl=d(item.get("closedPnl")),
                exec_id=exec_id))

        self._last_poll_ms = int(time.time() * 1000) - 60_000
        # 集合が無限に膨らむのを防ぐ
        if len(self._seen_exec_ids) > 5000:
            self._seen_exec_ids = set(list(self._seen_exec_ids)[-2000:])

        return sorted(new, key=lambda x: x.ts)

    def record_fill(self, ts: float | None = None) -> None:
        """約定を1件記録する（WSから呼ぶと精度が上がる）。"""
        self._fill_times.append(ts or time.time())

    # --- 検知 ------------------------------------------------------

    def _update_streak(self, outcomes: list[TradeOutcome]) -> None:
        for o in outcomes:
            self.recent_outcomes.append(o)
            if o.is_win:
                self.consecutive_losses = 0
            else:
                self.consecutive_losses += 1

    def fills_per_minute(self, window_sec: float = 60.0) -> float:
        now = time.time()
        recent = [t for t in self._fill_times if now - t <= window_sec]
        return len(recent) / (window_sec / 60.0)

    def baseline_fills_per_minute(self) -> float:
        """平常時の約定頻度を推定する。"""
        now = time.time()
        window = self.thresholds.baseline_window_sec
        times = [t for t in self._fill_times if now - t <= window]
        if len(times) < self.thresholds.min_baseline_samples:
            return -1.0
        # 直近1分は除いて平常値を推定する
        older = [t for t in times if now - t > 60]
        if not older:
            return -1.0
        span_min = max((now - min(older)) / 60.0, 1.0)
        return len(older) / span_min

    def expected_streak_probability(self, win_rate: float) -> float:
        """現在の連敗が偶然起きる確率。閾値の妥当性を判断する材料。"""
        if not 0 < win_rate < 1 or self.consecutive_losses <= 0:
            return 1.0
        return float((1 - win_rate) ** self.consecutive_losses)

    def check(self) -> dict:
        result = {"consecutive_losses": self.consecutive_losses,
                  "triggered": None}

        try:
            outcomes = self.poll_closed_pnl()
            self._update_streak(outcomes)
            for o in outcomes:
                self.record_fill(o.ts)
            result["new_trades"] = len(outcomes)
        except Exception as e:
            result["error"] = f"{type(e).__name__}: {e}"

        result["consecutive_losses"] = self.consecutive_losses

        # 1) 連敗
        if self.consecutive_losses >= self.thresholds.halt_consecutive_losses:
            self.machine.escalate(
                RiskState.FLATTEN,
                f"{self.consecutive_losses}連敗（停止水準）", self.name)
            result["triggered"] = "halt_losses"
            return result

        if self.consecutive_losses >= self.thresholds.max_consecutive_losses:
            self.machine.escalate(
                RiskState.REDUCE_ONLY,
                f"{self.consecutive_losses}連敗", self.name)
            result["triggered"] = "consecutive_losses"

        # 2) 約定頻度
        current = self.fills_per_minute()
        baseline = self.baseline_fills_per_minute()
        result["fills_per_min"] = current
        result["baseline_fills_per_min"] = baseline

        if current > self.thresholds.max_fills_per_minute:
            self.machine.escalate(
                RiskState.FLATTEN,
                f"約定頻度 {current:.1f}件/分 が絶対上限を超過。"
                "発注ループの可能性", self.name)
            result["triggered"] = "fill_rate_absolute"
            return result

        if (baseline > 0
                and current > baseline * self.thresholds.fill_rate_multiplier):
            self.machine.escalate(
                RiskState.REDUCE_ONLY,
                f"約定頻度 {current:.1f}件/分 が平常 {baseline:.1f} の"
                f"{current / baseline:.1f}倍", self.name)
            result["triggered"] = "fill_rate_relative"

        if result["triggered"] is None:
            self.machine.report_ok(self.name)

        return result

    def report(self) -> str:
        wins = sum(1 for o in self.recent_outcomes if o.is_win)
        total = len(self.recent_outcomes)
        win_rate = wins / total if total else 0.0
        prob = self.expected_streak_probability(win_rate)

        return "\n".join([
            f"  直近 {total} 取引: 勝率 {win_rate:.1%}",
            f"  現在の連敗   : {self.consecutive_losses}",
            f"  この連敗の確率: {prob:.2%}"
            + ("  ← 偶然の範囲" if prob > 0.05 else "  ← 異常の可能性"),
            f"  約定頻度     : {self.fills_per_minute():.2f}件/分",
            f"  平常時       : {self.baseline_fills_per_minute():.2f}件/分",
        ])


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

    c = make_client()
    machine = RiskStateMachine.load()
    detector = AnomalyDetector(c.http, machine)

    result = detector.check()
    print("=== 異常検知 ===")
    print(detector.report())
    print(f"\n  発動: {result['triggered'] or 'なし'}")

    print("\n=== 連敗閾値の妥当性 ===")
    print(f"{'勝率':>8}{'5連敗':>10}{'8連敗':>10}{'10連敗':>10}")
    print("-" * 40)
    for wr in (0.35, 0.45, 0.55, 0.65):
        row = f"{wr:>7.0%}"
        for k in (5, 8, 10):
            row += f"{(1 - wr) ** k:>10.2%}"
        print(row)

    print("\n  ※ 勝率45%なら5連敗は5%の確率で起きる（珍しくない）。")
    print("    閾値を低くしすぎると、正常な範囲で止まり続ける。")
    print("    自分の戦略の勝率から逆算して決めること。")

"""market_health.py — スプレッド急拡大・板消失を検知してエントリーを止める
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  相場が異常なときにエントリーすると、
  スリッページで想定の何倍もの損失を負う。

  検知するもの:
    - スプレッドが平常値の N 倍に拡大
    - 板の厚みが平常の 1/N に減少
    - マーク価格と最終価格の乖離（システム側の異常）
    - 直近の価格変動が異常（フラッシュクラッシュ）

  これらは「止める」というより「新規を控える」ための判定。
  既存ポジションの決済はむしろ急ぐべき局面もある。
"""
from __future__ import annotations

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

from margin_snapshot import d
from risk_state import RiskState, RiskStateMachine


@dataclass(frozen=True)
class HealthThresholds:
    spread_multiplier: float = 4.0        # 平常の何倍で異常
    spread_abs_bps: float = 30.0          # 絶対値の上限
    depth_ratio: float = 0.25             # 平常の何割を下回ると異常
    mark_last_divergence_bps: float = 50.0
    flash_move_bps: float = 300.0         # 直近1分の変動
    baseline_samples: int = 30


@dataclass
class MarketSnapshot:
    ts: float
    symbol: str
    bid: Decimal
    ask: Decimal
    mid: Decimal
    spread_bps: float
    depth_usd: float
    mark_price: Decimal
    last_price: Decimal

    @property
    def mark_last_bps(self) -> float:
        if self.last_price <= 0:
            return 0.0
        return float(abs(self.mark_price - self.last_price)
                     / self.last_price * 10_000)


@dataclass
class MarketHealthDetector:
    http: object
    machine: RiskStateMachine
    symbols: list[str]
    thresholds: HealthThresholds = field(default_factory=HealthThresholds)
    category: str = "linear"
    name: str = "market"

    _spread_history: dict = field(default_factory=dict)
    _depth_history: dict = field(default_factory=dict)
    _price_history: dict = field(default_factory=dict)

    def snapshot(self, symbol: str) -> MarketSnapshot:
        r = self.http.get_orderbook(category=self.category,
                                    symbol=symbol, limit=25)
        if r["retCode"] != 0:
            raise RuntimeError(f"板取得失敗: {r['retCode']} {r['retMsg']}")
        ob = r["result"]

        if not ob.get("b") or not ob.get("a"):
            raise RuntimeError(f"{symbol}: 板が空です")

        bid, ask = Decimal(ob["b"][0][0]), Decimal(ob["a"][0][0])
        mid = (bid + ask) / 2
        spread_bps = float((ask - bid) / mid * 10_000) if mid > 0 else 0.0

        depth = sum(float(p) * float(q) for p, q in ob["b"][:10])
        depth += sum(float(p) * float(q) for p, q in ob["a"][:10])

        t = self.http.get_tickers(category=self.category, symbol=symbol)
        tick = t["result"]["list"][0] if t["retCode"] == 0 else {}

        return MarketSnapshot(
            ts=time.time(), symbol=symbol, bid=bid, ask=ask, mid=mid,
            spread_bps=spread_bps, depth_usd=depth,
            mark_price=d(tick.get("markPrice"), str(mid)),
            last_price=d(tick.get("lastPrice"), str(mid)))

    def _baseline(self, history: deque) -> float:
        if len(history) < self.thresholds.baseline_samples:
            return -1.0
        return float(statistics.median(history))

    def check_symbol(self, symbol: str) -> dict:
        try:
            snap = self.snapshot(symbol)
        except Exception as e:
            return {"symbol": symbol, "error": str(e), "issues": ["取得失敗"]}

        sh = self._spread_history.setdefault(symbol, deque(maxlen=200))
        dh = self._depth_history.setdefault(symbol, deque(maxlen=200))
        ph = self._price_history.setdefault(symbol, deque(maxlen=200))

        issues: list[str] = []

        # 1) スプレッド
        base_spread = self._baseline(sh)
        if snap.spread_bps > self.thresholds.spread_abs_bps:
            issues.append(
                f"スプレッド {snap.spread_bps:.1f}bps が絶対上限を超過")
        elif (base_spread > 0
              and snap.spread_bps > base_spread * self.thresholds.spread_multiplier):
            issues.append(
                f"スプレッド {snap.spread_bps:.1f}bps が"
                f"平常 {base_spread:.1f}bps の"
                f"{snap.spread_bps / base_spread:.1f}倍")

        # 2) 板の厚み
        base_depth = self._baseline(dh)
        if base_depth > 0 and snap.depth_usd < base_depth * self.thresholds.depth_ratio:
            issues.append(
                f"板の厚み {snap.depth_usd:,.0f} が"
                f"平常 {base_depth:,.0f} の"
                f"{snap.depth_usd / base_depth:.0%}に減少")

        # 3) マーク価格との乖離
        if snap.mark_last_bps > self.thresholds.mark_last_divergence_bps:
            issues.append(
                f"マーク価格との乖離 {snap.mark_last_bps:.1f}bps")

        # 4) 急変
        if ph:
            recent = [p for t, p in ph if snap.ts - t <= 60]
            if recent:
                move_bps = abs(float(snap.mid) - recent[0]) / recent[0] * 10_000
                if move_bps > self.thresholds.flash_move_bps:
                    issues.append(f"直近1分で {move_bps:.0f}bps 変動")

        # 履歴に追加（異常値も含めて記録する。平常値は中央値なので頑健）
        sh.append(snap.spread_bps)
        dh.append(snap.depth_usd)
        ph.append((snap.ts, float(snap.mid)))

        return {"symbol": symbol, "spread_bps": snap.spread_bps,
                "depth_usd": snap.depth_usd, "issues": issues}

    def check(self) -> dict:
        results = [self.check_symbol(s) for s in self.symbols]
        problem_symbols = [r for r in results if r.get("issues")]

        if problem_symbols:
            details = "; ".join(
                f"{r['symbol']}: {r['issues'][0]}" for r in problem_symbols[:3])
            self.machine.escalate(RiskState.REDUCE_ONLY,
                                  f"板の異常 ({details})", self.name)
        else:
            self.machine.report_ok(self.name)

        return {"results": results, "problems": len(problem_symbols)}


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

    c = make_client()
    machine = RiskStateMachine.load()
    detector = MarketHealthDetector(
        c.http, machine, ["BTCUSDT", "ETHUSDT", "SOLUSDT"])

    print("=== 板の健全性チェック（平常値を学習中） ===")
    print("  中央値ベースの平常値を作るため、数回計測します。\n")

    for i in range(5):
        out = detector.check()
        print(f"[{i + 1}/5]")
        for r in out["results"]:
            if "error" in r:
                print(f"  {r['symbol']:<10} エラー: {r['error']}")
                continue
            status = ("正常" if not r["issues"]
                      else " / ".join(r["issues"]))
            print(f"  {r['symbol']:<10}"
                  f"spread {r['spread_bps']:>6.2f}bps  "
                  f"depth {r['depth_usd']:>12,.0f}  {status}")
        time.sleep(2)

    print(f"\n{machine.summary()}")
    print("\n※ 平常値は中央値で推定しているため、")
    print("  一時的な異常値に引っ張られにくい。")
    print("  ただし起動直後は平常値が無いので、")
    print("  絶対閾値（spread_abs_bps）だけが機能する。")

"""watchdog.py — WS断・データ遅延・時計ズレを検知する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  最も危険な故障は「静かに止まる」こと。
  例外も出さず、ただデータが来なくなる。
  botは最後に受け取った価格で判断し続け、
  実際の相場と乖離したまま発注する。

  検知するもの:
    1. トピックごとの最終受信時刻（ハートビート）
    2. bot本体のループが回っているか（自己申告）
    3. サーバ時刻とローカル時刻のズレ（署名エラーの前兆）

  ウォッチドッグは必ず別スレッドで動かす。
  bot本体のループに組み込むと、そのループが固まったとき
  ウォッチドッグも一緒に止まる。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field

from risk_state import RiskState, RiskStateMachine


@dataclass(frozen=True)
class WatchdogThresholds:
    topic_stale_sec: float = 30.0       # トピックの許容無音時間
    topic_halt_sec: float = 120.0       # これを超えたら全決済
    loop_stale_sec: float = 60.0        # bot本体の許容無応答時間
    clock_skew_warn_ms: float = 2000.0
    clock_skew_halt_ms: float = 10000.0  # 署名が通らなくなる水準
    clock_check_interval_sec: float = 300.0


@dataclass
class Heartbeat:
    name: str
    last_seen: float = field(default_factory=time.time)
    count: int = 0
    expected_interval_sec: float = 5.0

    def touch(self) -> None:
        self.last_seen = time.time()
        self.count += 1

    @property
    def age(self) -> float:
        return time.time() - self.last_seen


@dataclass
class Watchdog:
    http: object
    machine: RiskStateMachine
    thresholds: WatchdogThresholds = field(default_factory=WatchdogThresholds)
    name: str = "watchdog"

    beats: dict = field(default_factory=dict)
    _lock: threading.RLock = field(default_factory=threading.RLock)
    _thread: threading.Thread | None = None
    _stop_event: threading.Event = field(default_factory=threading.Event)
    _last_clock_check: float = 0.0
    clock_skew_ms: float = 0.0

    # --- ハートビート ----------------------------------------------

    def register(self, name: str, expected_interval_sec: float = 5.0) -> None:
        with self._lock:
            self.beats[name] = Heartbeat(
                name, expected_interval_sec=expected_interval_sec)

    def touch(self, name: str) -> None:
        """データを受信したら呼ぶ。WSのコールバック先頭に置く。"""
        with self._lock:
            beat = self.beats.get(name)
            if beat is None:
                beat = Heartbeat(name)
                self.beats[name] = beat
            beat.touch()

    def stale_topics(self, threshold_sec: float) -> list[tuple[str, float]]:
        with self._lock:
            return [(b.name, b.age) for b in self.beats.values()
                    if b.age > threshold_sec]

    # --- 時計ズレ --------------------------------------------------

    def check_clock_skew(self) -> float:
        """サーバ時刻とローカル時刻のズレをミリ秒で返す。

        Bybitの署名は recv_window（既定5000ms）内でないと弾かれる。
        ズレが大きくなると 10002 (invalid timestamp) が頻発する。
        """
        t0 = time.time()
        r = self.http.get_server_time()
        t1 = time.time()

        if r["retCode"] != 0:
            return self.clock_skew_ms

        server_ms = int(r["result"]["timeNano"]) / 1_000_000
        # 往復時間の半分を通信遅延として差し引く
        local_ms = (t0 + t1) / 2 * 1000
        self.clock_skew_ms = local_ms - server_ms
        self._last_clock_check = time.time()
        return self.clock_skew_ms

    # --- 判定 ------------------------------------------------------

    def check(self) -> dict:
        result = {"triggered": None, "stale": [], "clock_skew_ms": 0.0}

        # 1) トピックの無音
        halt_stale = self.stale_topics(self.thresholds.topic_halt_sec)
        if halt_stale:
            names = ", ".join(f"{n}({a:.0f}秒)" for n, a in halt_stale)
            self.machine.escalate(
                RiskState.FLATTEN,
                f"データ断が継続: {names}", self.name)
            result["triggered"] = "topic_halt"
            result["stale"] = halt_stale
            return result

        warn_stale = self.stale_topics(self.thresholds.topic_stale_sec)
        if warn_stale:
            names = ", ".join(f"{n}({a:.0f}秒)" for n, a in warn_stale)
            self.machine.escalate(
                RiskState.REDUCE_ONLY,
                f"データ遅延: {names}", self.name)
            result["triggered"] = "topic_stale"
            result["stale"] = warn_stale

        # 2) 時計ズレ（頻繁に叩く必要は無い）
        if (time.time() - self._last_clock_check
                > self.thresholds.clock_check_interval_sec):
            try:
                skew = self.check_clock_skew()
                result["clock_skew_ms"] = skew

                if abs(skew) > self.thresholds.clock_skew_halt_ms:
                    self.machine.escalate(
                        RiskState.FLATTEN,
                        f"時計ズレ {skew:+.0f}ms。署名が通らなくなります",
                        self.name)
                    result["triggered"] = "clock_halt"
                    return result
                if abs(skew) > self.thresholds.clock_skew_warn_ms:
                    print(f"  [警告] 時計ズレ {skew:+.0f}ms。"
                          "NTP同期を確認してください（記事38）")
            except Exception as e:
                print(f"  [warn] 時刻確認に失敗: {e}")

        if result["triggered"] is None:
            self.machine.report_ok(self.name)

        return result

    # --- 独立スレッドでの常時監視 ----------------------------------

    def start(self, interval_sec: float = 5.0) -> None:
        """別スレッドで監視を開始する。

        bot本体のループとは独立して動く。
        本体が固まってもウォッチドッグは動き続ける。
        """
        if self._thread and self._thread.is_alive():
            return

        def loop() -> None:
            while not self._stop_event.is_set():
                try:
                    self.check()
                except Exception as e:
                    print(f"  [watchdog error] {type(e).__name__}: {e}")
                self._stop_event.wait(interval_sec)

        self._stop_event.clear()
        self._thread = threading.Thread(target=loop, daemon=True,
                                        name="watchdog")
        self._thread.start()
        print(f"  ウォッチドッグ起動（{interval_sec}秒間隔）")

    def stop(self) -> None:
        self._stop_event.set()
        if self._thread:
            self._thread.join(timeout=5.0)

    def report(self) -> str:
        with self._lock:
            lines = [f"  時計ズレ: {self.clock_skew_ms:+.0f}ms"]
            if not self.beats:
                lines.append("  監視対象が登録されていません")
            for b in sorted(self.beats.values(), key=lambda x: -x.age):
                flag = ""
                if b.age > self.thresholds.topic_halt_sec:
                    flag = "  ← 停止水準"
                elif b.age > self.thresholds.topic_stale_sec:
                    flag = "  ← 遅延"
                lines.append(
                    f"  {b.name:<24}{b.age:>7.1f}秒前"
                    f"  受信 {b.count:>6}件{flag}")
            return "\n".join(lines)


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

    c = make_client()
    machine = RiskStateMachine.load()

    wd = Watchdog(c.http, machine,
                  WatchdogThresholds(topic_stale_sec=5.0,
                                     topic_halt_sec=10.0,
                                     clock_check_interval_sec=0.0))

    wd.register("orderbook.BTCUSDT", expected_interval_sec=0.1)
    wd.register("tickers.BTCUSDT", expected_interval_sec=1.0)
    wd.register("private.order", expected_interval_sec=60.0)

    print("=== ウォッチドッグのデモ ===\n")
    print("正常時:")
    for name in wd.beats:
        wd.touch(name)
    wd.check()
    print(wd.report())

    print("\n6秒待機（orderbookだけ更新を止める）...")
    for _ in range(6):
        wd.touch("tickers.BTCUSDT")
        wd.touch("private.order")
        time.sleep(1)

    wd.check()
    print(wd.report())
    print(f"\n{machine.summary()}")

    print("\n※ WSのコールバック先頭で wd.touch(トピック名) を呼ぶだけで、")
    print("  『データが来なくなった』を確実に検知できる。")
    print("  例外が出ない静かな故障こそ、この仕組みが必要な理由。")

"""kill_switch.py — 全注文キャンセル → 全ポジション決済 → 停止
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  実行順序（この順序が絶対）:
    1. 全注文をキャンセル（条件注文・TP/SLを含む）
       ← 先にやらないと、決済中に指値が約定して新しい玉ができる
    2. 全ポジションを reduceOnly 成行で決済
    3. 再度キャンセル（1と2の間に発火した条件注文の掃除）
    4. 残ポジション・残注文がゼロであることを検証
    5. HALTED に遷移してロックファイルを作成

  検証を省くと「止めたつもりで止まっていない」状態になる。
  必ず 4 を実行すること。
"""
from __future__ import annotations

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

from margin_snapshot import d
from risk_state import RiskStateMachine

ORDER_FILTERS = ("Order", "StopOrder", "tpslOrder")


@dataclass
class KillResult:
    cancelled_orders: int = 0
    closed_positions: int = 0
    remaining_orders: int = 0
    remaining_positions: int = 0
    elapsed_sec: float = 0.0
    errors: list[str] = field(default_factory=list)

    @property
    def clean(self) -> bool:
        return (self.remaining_orders == 0
                and self.remaining_positions == 0
                and not self.errors)

    def report(self) -> str:
        status = "完全停止" if self.clean else "未完了（要手動確認）"
        lines = [
            f"=== キルスイッチ結果: {status} ===",
            f"  キャンセルした注文 : {self.cancelled_orders}",
            f"  決済したポジション : {self.closed_positions}",
            f"  残注文             : {self.remaining_orders}",
            f"  残ポジション       : {self.remaining_positions}",
            f"  所要時間           : {self.elapsed_sec:.2f}秒",
        ]
        if self.errors:
            lines.append("  --- エラー ---")
            lines.extend(f"    {e}" for e in self.errors)
        return "\n".join(lines)


def cancel_everything(http, *, category: str = "linear",
                      settle_coin: str = "USDT") -> tuple[int, list[str]]:
    """全注文をキャンセルする。3種類すべてを対象にする。

    cancel_all_orders はレート消費1で最速（記事16）。
    緊急時にバッチキャンセルを使ってはいけない。
    """
    total, errors = 0, []
    filters = list(ORDER_FILTERS) if category != "spot" else ["Order", "StopOrder"]

    for f in filters:
        params = {"category": category, "orderFilter": f}
        if category in ("linear", "inverse"):
            params["settleCoin"] = settle_coin
        try:
            r = http.cancel_all_orders(**params)
            if r["retCode"] == 0:
                total += len(r["result"].get("list", []))
            else:
                errors.append(
                    f"cancel_all({f}): {r['retCode']} {r['retMsg']}")
        except Exception as e:
            errors.append(f"cancel_all({f}): {type(e).__name__}: {e}")

    return total, errors


def close_all_positions(http, *, category: str = "linear",
                        settle_coin: str = "USDT") -> tuple[int, list[str]]:
    """全ポジションを reduceOnly 成行で決済する。

    ヘッジモードでは positionIdx を指定する必要がある（記事20）。
    """
    closed, errors = 0, []

    try:
        r = http.get_positions(category=category, settleCoin=settle_coin)
        if r["retCode"] != 0:
            return 0, [f"get_positions: {r['retCode']} {r['retMsg']}"]
    except Exception as e:
        return 0, [f"get_positions: {type(e).__name__}: {e}"]

    for p in r["result"]["list"]:
        size = d(p.get("size"))
        if size <= 0:
            continue

        params = {
            "category": category,
            "symbol": p["symbol"],
            "side": "Sell" if p["side"] == "Buy" else "Buy",
            "orderType": "Market",
            "qty": str(size),
            "reduceOnly": True,
            "orderLinkId": f"kill-{int(time.time() * 1000) % 1_000_000}",
        }
        # ヘッジモードなら positionIdx が必須
        idx = int(p.get("positionIdx", 0) or 0)
        if idx != 0:
            params["positionIdx"] = idx

        try:
            res = http.place_order(**params)
            if res["retCode"] == 0:
                closed += 1
            else:
                errors.append(
                    f"close {p['symbol']}: {res['retCode']} {res['retMsg']}")
        except Exception as e:
            errors.append(f"close {p['symbol']}: {type(e).__name__}: {e}")

    return closed, errors


def verify_flat(http, *, category: str = "linear",
                settle_coin: str = "USDT") -> tuple[int, int]:
    """注文とポジションが本当にゼロかを確認する。"""
    orders = 0
    for f in ORDER_FILTERS:
        if category == "spot" and f == "tpslOrder":
            continue
        try:
            params = {"category": category, "orderFilter": f, "limit": 50}
            if category in ("linear", "inverse"):
                params["settleCoin"] = settle_coin
            r = http.get_open_orders(**params)
            if r["retCode"] == 0:
                orders += len(r["result"]["list"])
        except Exception:
            orders += 999      # 確認できない = 安全側に倒す

    positions = 0
    try:
        r = http.get_positions(category=category, settleCoin=settle_coin)
        if r["retCode"] == 0:
            positions = sum(1 for p in r["result"]["list"]
                            if d(p.get("size")) > 0)
    except Exception:
        positions = 999

    return orders, positions


def kill(http, machine: RiskStateMachine | None = None,
         reason: str = "手動実行", *,
         category: str = "linear", settle_coin: str = "USDT",
         settle_wait_sec: float = 2.0) -> KillResult:
    """キルスイッチ本体。"""
    started = time.perf_counter()
    result = KillResult()

    print(f"=== キルスイッチ発動: {reason} ===")

    # 1) 全注文キャンセル（必ず先）
    print("  [1/5] 全注文をキャンセル...")
    n, errors = cancel_everything(http, category=category,
                                  settle_coin=settle_coin)
    result.cancelled_orders += n
    result.errors.extend(errors)
    print(f"        {n}件をキャンセル")

    # 2) 全ポジション決済
    print("  [2/5] 全ポジションを決済...")
    closed, errors = close_all_positions(http, category=category,
                                         settle_coin=settle_coin)
    result.closed_positions = closed
    result.errors.extend(errors)
    print(f"        {closed}件を決済")

    time.sleep(settle_wait_sec)

    # 3) 再キャンセル（決済中に発火した条件注文の掃除）
    print("  [3/5] 再キャンセル（発火した条件注文の掃除）...")
    n, errors = cancel_everything(http, category=category,
                                  settle_coin=settle_coin)
    result.cancelled_orders += n
    result.errors.extend(errors)
    print(f"        {n}件を追加キャンセル")

    # 4) 検証
    print("  [4/5] 検証...")
    orders, positions = verify_flat(http, category=category,
                                    settle_coin=settle_coin)
    result.remaining_orders = orders
    result.remaining_positions = positions
    print(f"        残注文 {orders} / 残ポジション {positions}")

    # 5) 状態遷移
    print("  [5/5] 状態を HALTED に遷移...")
    if machine is not None:
        machine.halt(f"キルスイッチ: {reason}", "killswitch")

    result.elapsed_sec = time.perf_counter() - started
    return result


if __name__ == "__main__":
    import sys

    from client import make_client

    if "--yes" not in sys.argv:
        print("!! このスクリプトは全注文と全ポジションを消します !!")
        print("   実行するには --yes を付けてください:")
        print("     python kill_switch.py --yes")
        sys.exit(0)

    c = make_client()
    machine = RiskStateMachine.load()

    result = kill(c.http, machine, reason="手動テスト")
    print()
    print(result.report())

    if not result.clean:
        print("\n!! 完全には停止できていません !!")
        print("   Bybitのウェブ画面で手動確認してください。")
        print("   API経由で消せない状態（メンテナンス中など）の可能性があります。")

    print("\n※ 順序が重要: 注文キャンセル → ポジション決済 → 再キャンセル。")
    print("  先にポジションを閉じると、残った指値が新しい玉を作る。")

"""control_channel.py — 外部からbotを止めるコントロールチャネル
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  3つの経路を用意する。多重化が要点。

  1. ファイル: ./control/HALT を作れば停止
     → SSHで touch するだけ。最も確実で、依存が無い。
  2. 環境変数: BOT_MODE=halt
     → コンテナの再起動時に効かせたい場合（記事38）
  3. Discord/Webhook: 外出先から止める
     → 通信できない状況では使えないので、あくまで補助

  優先順位: ファイル > 環境変数 > リモート
  「最も単純な経路が最も確実」という原則で設計する。

  重要: リモート経路には必ず認証を付ける。
        誰でも叩ける停止APIは、それ自体が攻撃対象になる。
"""
from __future__ import annotations

import hashlib
import hmac
import os
import time
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path

from risk_state import RiskState, RiskStateMachine

CONTROL_DIR = Path("./control")


class Command(str, Enum):
    RUN = "run"                 # 通常運用
    REDUCE_ONLY = "reduce"      # 新規停止
    FLATTEN = "flatten"         # 全決済
    HALT = "halt"               # 完全停止

    @property
    def target_state(self) -> RiskState:
        return {
            Command.RUN: RiskState.NORMAL,
            Command.REDUCE_ONLY: RiskState.REDUCE_ONLY,
            Command.FLATTEN: RiskState.FLATTEN,
            Command.HALT: RiskState.HALTED,
        }[self]


@dataclass
class ControlChannel:
    """外部からの停止指示を受け取る。"""
    machine: RiskStateMachine
    control_dir: Path = field(default=CONTROL_DIR)
    env_var: str = "BOT_MODE"
    name: str = "control"
    _last_command: Command | None = None

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

    # --- 経路1: ファイル -------------------------------------------

    def _file_command(self) -> Command | None:
        """control/ ディレクトリのファイルで指示する。

        危険度の高い順に確認する。
        HALT と FLATTEN が同時にあれば HALT が勝つ。
        """
        for cmd in (Command.HALT, Command.FLATTEN, Command.REDUCE_ONLY):
            if (self.control_dir / cmd.name).exists():
                return cmd
        return None

    # --- 経路2: 環境変数 -------------------------------------------

    def _env_command(self) -> Command | None:
        value = os.environ.get(self.env_var, "").strip().lower()
        if not value:
            return None
        for cmd in Command:
            if value == cmd.value:
                return cmd
        return None

    # --- 経路3: リモート（Webhook経由で書かれたファイル） ----------

    def _remote_command(self) -> Command | None:
        """リモート指示は署名付きファイルとして受け取る。

        Webhookサーバ（別プロセス）が検証してこのファイルを書く。
        bot本体がHTTPサーバを持つ設計は避ける。
        bot本体が固まったらそのサーバも固まるため。
        """
        path = self.control_dir / "remote.cmd"
        if not path.exists():
            return None
        try:
            content = path.read_text(encoding="utf-8").strip()
            parts = content.split("|")
            if len(parts) != 3:
                return None
            cmd_str, ts_str, signature = parts

            secret = os.environ.get("BOT_CONTROL_SECRET", "")
            if not secret:
                print("  [warn] BOT_CONTROL_SECRET 未設定。リモート指示を無視")
                return None

            expected = hmac.new(
                secret.encode(), f"{cmd_str}|{ts_str}".encode(),
                hashlib.sha256).hexdigest()
            if not hmac.compare_digest(expected, signature):
                print("  [warn] リモート指示の署名が不正です")
                return None

            # リプレイ攻撃を防ぐため、古い指示は無視する
            if time.time() - float(ts_str) > 300:
                return None

            for cmd in Command:
                if cmd_str == cmd.value:
                    return cmd
        except Exception as e:
            print(f"  [warn] リモート指示の解析に失敗: {e}")
        return None

    # --- 統合 ------------------------------------------------------

    def poll(self) -> Command | None:
        """全経路を確認し、最も危険度の高い指示を返す。"""
        commands = [c for c in (self._file_command(),
                                self._env_command(),
                                self._remote_command()) if c is not None]
        if not commands:
            return None
        return max(commands, key=lambda c: int(c.target_state))

    def apply(self) -> bool:
        """指示を状態機械に反映する。"""
        cmd = self.poll()
        if cmd is None:
            self._last_command = None
            return False

        if cmd is self._last_command:
            return False        # 同じ指示の繰り返しは無視

        self._last_command = cmd

        if cmd is Command.RUN:
            print("  [control] RUN 指示。ただし自動復帰はしません。")
            print("            HALTED からの復帰はロック解除が必要です。")
            return False

        return self.machine.escalate(
            cmd.target_state, f"外部指示: {cmd.value}", self.name)

    # --- 送信側（運用者が使う） ------------------------------------

    def issue(self, cmd: Command) -> None:
        """ローカルで指示を出す（ファイル経路）。"""
        (self.control_dir / cmd.name).touch()
        print(f"  指示を発行: {cmd.name}")

    def clear(self, cmd: Command) -> None:
        path = self.control_dir / cmd.name
        if path.exists():
            path.unlink()
            print(f"  指示を解除: {cmd.name}")

    @staticmethod
    def sign_remote_command(cmd: Command, secret: str) -> str:
        """リモート指示の署名付き文字列を作る（Webhookサーバ側で使う）。"""
        ts = str(time.time())
        payload = f"{cmd.value}|{ts}"
        sig = hmac.new(secret.encode(), payload.encode(),
                       hashlib.sha256).hexdigest()
        return f"{payload}|{sig}"

    def status(self) -> str:
        lines = ["コントロールチャネルの状態:"]
        f = self._file_command()
        lines.append(f"  ファイル  : {f.name if f else '指示なし'}")
        e = self._env_command()
        lines.append(f"  環境変数  : {e.name if e else '指示なし'}"
                     f"  ({self.env_var}="
                     f"{os.environ.get(self.env_var, '未設定')})")
        r = self._remote_command()
        lines.append(f"  リモート  : {r.name if r else '指示なし'}")
        active = self.poll()
        lines.append(f"  有効な指示: {active.name if active else 'なし'}")
        return "\n".join(lines)


if __name__ == "__main__":
    machine = RiskStateMachine.load()
    channel = ControlChannel(machine)

    print("=== コントロールチャネル ===\n")
    print(channel.status())

    print("\n--- ファイル経路のテスト ---")
    channel.issue(Command.REDUCE_ONLY)
    channel.apply()
    print(f"  状態: {machine.state.name}")

    print("\n--- より危険な指示が優先される ---")
    channel.issue(Command.HALT)
    channel.apply()
    print(f"  状態: {machine.state.name}")

    channel.clear(Command.REDUCE_ONLY)
    channel.clear(Command.HALT)

    print("\n--- リモート指示の署名 ---")
    secret = "example-secret-do-not-use-in-production"
    signed = ControlChannel.sign_remote_command(Command.HALT, secret)
    print(f"  {signed}")
    print("  ↑ これを control/remote.cmd に書けば停止する")
    print("    署名が無い・古い指示は無視される")

    print("\n※ 運用中に止めたいときは、SSHで以下を実行するだけ:")
    print("    touch ./control/HALT")
    print("  依存が無く、bot本体が固まっていても確実に効く。")
    print("  Discord経由の停止は便利だが、通信できない状況では使えない。")

"""restart_lock.py — 人間が承認するまで再開させないロック
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  なぜロックが必要か:
    停止したbotをsystemdやDockerが自動再起動すると、
    原因が分からないまま同じ事故を繰り返す。
    「異常検知 → 停止 → 自動再起動 → 即座に再停止」を
    延々と繰り返し、その間ずっと発注し続けることもある。

  ロックの原則:
    - 停止理由と当時の状態をロックファイルに書き込む
    - ロックが存在する限り、起動時に即座に終了する
    - 解除は人間の明示的な操作（ファイル削除）でのみ可能
    - 解除時に「確認した」記録を残す
"""
from __future__ import annotations

import json
import os
import platform
import socket
import sys
import time
from dataclasses import dataclass, asdict, field
from pathlib import Path

LOCK_PATH = Path("./control/HALT.lock")


@dataclass
class LockInfo:
    created_at: float
    reason: str
    detector: str
    hostname: str
    pid: int
    state_snapshot: dict = field(default_factory=dict)
    acknowledged_at: float = 0.0
    acknowledged_by: str = ""

    @property
    def age_hours(self) -> float:
        return (time.time() - self.created_at) / 3600

    def report(self) -> str:
        when = time.strftime("%Y-%m-%d %H:%M:%S",
                             time.localtime(self.created_at))
        lines = [
            "!!! 再開ロックが有効です !!!",
            f"  停止時刻: {when} ({self.age_hours:.1f}時間前)",
            f"  停止理由: {self.reason}",
            f"  検知元  : {self.detector}",
            f"  ホスト  : {self.hostname} (PID {self.pid})",
        ]
        if self.state_snapshot:
            lines.append("  --- 停止時の状態 ---")
            for k, v in self.state_snapshot.items():
                lines.append(f"    {k}: {v}")
        return "\n".join(lines)


def create_lock(reason: str, detector: str = "",
                state_snapshot: dict | None = None,
                path: Path = LOCK_PATH) -> LockInfo:
    """ロックを作成する。既存のロックは上書きしない。"""
    path.parent.mkdir(parents=True, exist_ok=True)

    if path.exists():
        existing = read_lock(path)
        if existing:
            print("  既存のロックを保持します（最初の原因が重要）")
            return existing

    info = LockInfo(
        created_at=time.time(), reason=reason, detector=detector,
        hostname=socket.gethostname(), pid=os.getpid(),
        state_snapshot=state_snapshot or {})

    path.write_text(json.dumps(asdict(info), ensure_ascii=False, indent=2),
                    encoding="utf-8")
    print(f"  ロックを作成: {path}")
    return info


def read_lock(path: Path = LOCK_PATH) -> LockInfo | None:
    if not path.exists():
        return None
    try:
        return LockInfo(**json.loads(path.read_text(encoding="utf-8")))
    except Exception as e:
        # 壊れたロックファイルも「ロックあり」として扱う（安全側）
        print(f"  [warn] ロックファイルの解析に失敗: {e}")
        return LockInfo(created_at=0, reason="ロックファイルが破損",
                        detector="unknown", hostname="unknown", pid=0)


def check_lock_or_exit(path: Path = LOCK_PATH) -> None:
    """起動時に呼ぶ。ロックがあれば即座に終了する。

    main() の一番最初、クライアント生成より前に置くこと。
    """
    info = read_lock(path)
    if info is None:
        return

    print(info.report())
    print("\n  原因を確認し、対処してから以下で解除してください:")
    print(f"    python restart_lock.py --release --by あなたの名前")
    print("\n  起動を中止します。")
    sys.exit(1)


def release_lock(by: str, note: str = "",
                 path: Path = LOCK_PATH,
                 archive_dir: Path = Path("./control/lock_history")) -> bool:
    """ロックを解除する。解除の記録をアーカイブに残す。"""
    info = read_lock(path)
    if info is None:
        print("  ロックはありません。")
        return False

    info.acknowledged_at = time.time()
    info.acknowledged_by = by

    archive_dir.mkdir(parents=True, exist_ok=True)
    stamp = time.strftime("%Y%m%d-%H%M%S",
                          time.localtime(info.created_at))
    archive = archive_dir / f"{stamp}.json"
    record = asdict(info)
    record["release_note"] = note
    archive.write_text(json.dumps(record, ensure_ascii=False, indent=2),
                       encoding="utf-8")

    path.unlink()
    print(f"  ロックを解除しました（{by}）")
    print(f"  記録: {archive}")
    return True


def lock_history(archive_dir: Path = Path("./control/lock_history")) -> None:
    """過去の停止履歴を表示する。同じ原因が繰り返されていないか。"""
    if not archive_dir.exists():
        print("  停止履歴はありません。")
        return

    files = sorted(archive_dir.glob("*.json"), reverse=True)
    print(f"=== 停止履歴 ({len(files)}件) ===\n")

    reasons: dict[str, int] = {}
    for f in files[:20]:
        try:
            d = json.loads(f.read_text(encoding="utf-8"))
        except Exception:
            continue
        when = time.strftime("%Y-%m-%d %H:%M",
                             time.localtime(d.get("created_at", 0)))
        detector = d.get("detector", "?")
        reason = d.get("reason", "?")
        print(f"  {when}  [{detector:<12}] {reason}")
        key = detector
        reasons[key] = reasons.get(key, 0) + 1

    if reasons:
        print("\n  検知元別の頻度:")
        for detector, count in sorted(reasons.items(), key=lambda kv: -kv[1]):
            print(f"    {detector:<14}{count:>4}回")
        print("\n  ※ 同じ検知元で何度も止まっているなら、")
        print("    閾値が厳しすぎるか、戦略に構造的な問題がある。")


if __name__ == "__main__":
    if "--release" in sys.argv:
        idx = sys.argv.index("--by") if "--by" in sys.argv else -1
        by = sys.argv[idx + 1] if 0 <= idx < len(sys.argv) - 1 else ""
        if not by:
            print("解除には --by 名前 が必要です。")
            print("  例: python restart_lock.py --release --by tanaka")
            sys.exit(1)
        release_lock(by, note="手動解除")
        sys.exit(0)

    if "--history" in sys.argv:
        lock_history()
        sys.exit(0)

    # デモ: ロックを作って読む
    print("=== 再開ロックのデモ ===\n")
    info = create_lock(
        reason="日次DD 6.2% が上限を超過",
        detector="drawdown",
        state_snapshot={"equity": "9380.00", "daily_dd": "6.2%",
                        "positions": "0", "python": platform.python_version()})
    print()
    print(info.report())

    print("\n  起動時のチェック（このデモでは終了しないよう別途確認）:")
    existing = read_lock()
    print(f"    ロックあり? {existing is not None}")

    release_lock("demo-user", note="デモのため解除")
    print(f"\n  解除後: ロックあり? {read_lock() is not None}")

    print("\n※ systemd の Restart=always と組み合わせるときは、")
    print("  main() の先頭で check_lock_or_exit() を必ず呼ぶこと。")
    print("  呼ばないと『停止→再起動→即停止』を無限に繰り返す。")

"""supervisor.py — 全ディテクタを統合したサーキットブレーカー
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  サンプル1〜8を1つにまとめる。

  構成:
    - Supervisor が全ディテクタを定期実行
    - Watchdog は独立スレッド（本体が固まっても動く）
    - FLATTEN に入ったらキルスイッチを実行して HALTED へ
    - HALTED でロックを作成し、アラートを送信

  bot本体はこの Supervisor の can_open / can_close だけを見る。
  「なぜ止まったか」を本体が知る必要は無い。
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field

from anomaly_detector import AnomalyDetector
from control_channel import ControlChannel
from drawdown_guard import DrawdownDetector
from kill_switch import kill
from market_health import MarketHealthDetector
from restart_lock import check_lock_or_exit, create_lock
from risk_state import RiskState, RiskStateMachine
from watchdog import Watchdog


@dataclass
class SupervisorConfig:
    symbols: tuple[str, ...] = ("BTCUSDT",)
    check_interval_sec: float = 10.0
    watchdog_interval_sec: float = 5.0
    category: str = "linear"
    settle_coin: str = "USDT"
    alert_min_state: RiskState = RiskState.REDUCE_ONLY


@dataclass
class Supervisor:
    http: object
    config: SupervisorConfig = field(default_factory=SupervisorConfig)
    machine: RiskStateMachine = field(default_factory=RiskStateMachine.load)
    detectors: list = field(default_factory=list)
    watchdog: Watchdog | None = None
    control: ControlChannel | None = None
    _last_alert_state: RiskState | None = None
    _killed: bool = False

    def setup(self) -> None:
        """全ディテクタを組み立てる。"""
        self.detectors = [
            DrawdownDetector(self.http, self.machine),
            AnomalyDetector(self.http, self.machine,
                            category=self.config.category,
                            settle_coin=self.config.settle_coin),
            MarketHealthDetector(self.http, self.machine,
                                 list(self.config.symbols),
                                 category=self.config.category),
        ]
        self.watchdog = Watchdog(self.http, self.machine)
        self.control = ControlChannel(self.machine)
        self.watchdog.start(self.config.watchdog_interval_sec)

    # --- bot本体が参照する部分 -------------------------------------

    @property
    def can_open(self) -> bool:
        return self.machine.can_open

    @property
    def can_close(self) -> bool:
        return self.machine.can_close

    def touch(self, topic: str) -> None:
        """WSコールバックから呼ぶ。"""
        if self.watchdog:
            self.watchdog.touch(topic)

    # --- 監視ループ ------------------------------------------------

    def alert(self, message: str) -> None:
        """アラートを送信する。記事39の通知基盤に接続する。"""
        print(f"  [ALERT] {message}")
        # send_discord(message) など

    def check_once(self) -> dict:
        results = {}

        # 外部指示を最優先で確認する
        if self.control:
            self.control.apply()

        for detector in self.detectors:
            try:
                results[detector.name] = detector.check()
            except Exception as e:
                results[detector.name] = {"error": f"{type(e).__name__}: {e}"}
                print(f"  [error] {detector.name}: {e}")

        self._handle_state()
        return results

    def _handle_state(self) -> None:
        state = self.machine.state

        # アラート（状態が変わったときだけ）
        if (state >= self.config.alert_min_state
                and state is not self._last_alert_state):
            last = self.machine.history[-1] if self.machine.history else None
            reason = last.reason if last else "不明"
            self.alert(f"リスク状態 {state.name}: {reason}")
            self._last_alert_state = state

        # FLATTEN ならキルスイッチを実行
        if state is RiskState.FLATTEN and not self._killed:
            self._killed = True
            last = self.machine.history[-1] if self.machine.history else None
            reason = last.reason if last else "不明"
            detector = last.detector if last else "unknown"

            result = kill(self.http, self.machine, reason,
                          category=self.config.category,
                          settle_coin=self.config.settle_coin)
            print(result.report())

            create_lock(reason=reason, detector=detector,
                        state_snapshot={
                            "cancelled": str(result.cancelled_orders),
                            "closed": str(result.closed_positions),
                            "clean": str(result.clean),
                        })
            self.alert(
                f"botを停止しました: {reason}\n"
                f"注文キャンセル {result.cancelled_orders}件 / "
                f"決済 {result.closed_positions}件 / "
                f"完全停止: {'はい' if result.clean else 'いいえ（要確認）'}")

    def run(self) -> None:
        print("=== サーキットブレーカー起動 ===")
        print(f"  監視銘柄: {', '.join(self.config.symbols)}")
        print(f"  間隔    : {self.config.check_interval_sec}秒")
        print(f"  停止    : touch ./control/HALT")
        print("  Ctrl+C で監視を終了\n")

        try:
            while True:
                self.check_once()

                if self.machine.state is RiskState.HALTED:
                    print("\n  HALTED に到達しました。監視を終了します。")
                    print("  再開するには restart_lock.py --release が必要です。")
                    break

                time.sleep(self.config.check_interval_sec)
        except KeyboardInterrupt:
            print("\n  監視を終了します。")
        finally:
            if self.watchdog:
                self.watchdog.stop()

    def status(self) -> str:
        lines = [self.machine.summary()]
        if self.watchdog:
            lines.append("\nウォッチドッグ:")
            lines.append(self.watchdog.report())
        if self.control:
            lines.append("")
            lines.append(self.control.status())
        return "\n".join(lines)


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

    # 起動時のロック確認。これより前に何もしない
    check_lock_or_exit()

    c = make_client()
    sup = Supervisor(c.http, SupervisorConfig(
        symbols=("BTCUSDT", "ETHUSDT"),
        check_interval_sec=10.0))
    sup.setup()

    # WSを使う場合、コールバックの先頭で sup.touch(topic) を呼ぶ
    # sup.watchdog.register("orderbook.BTCUSDT")

    sup.run()
    print()
    print(sup.status())


if __name__ == "__main__":
    main()

    # bot本体との統合:
    #   if not supervisor.can_open:
    #       return          # 新規エントリーをスキップ
    #   if not supervisor.can_close:
    #       return          # 決済もできない = 完全停止中
    #
    # 本体は「なぜ止まったか」を知る必要が無い。
    # can_open / can_close の2つだけを見ればよい。