Bybitのテストネット・デモトレードAPIで
botを安全に検証する
本番誤爆を防ぐPython設計

"""env_config.py — 3環境のエンドポイントとキーを1箇所で管理する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

記事01の config.py を拡張し、WebSocketのURLと対応状況まで含める。
"""
from __future__ import annotations

import os
from dataclasses import dataclass
from enum import Enum
from pathlib import Path

from dotenv import load_dotenv

load_dotenv(Path(__file__).with_name(".env"))


class Env(str, Enum):
    TESTNET = "testnet"
    DEMO = "demo"
    MAINNET = "mainnet"


@dataclass(frozen=True)
class Endpoints:
    rest: str
    ws_public_base: str        # 末尾に /v5/public/{channel} を付ける
    ws_private: str
    ws_trade: str


# デモのpublicは本番を使う（デモ専用のpublicストリームは存在しない）
ENDPOINTS: dict[Env, Endpoints] = {
    Env.TESTNET: Endpoints(
        rest="https://api-testnet.bybit.com",
        ws_public_base="wss://stream-testnet.bybit.com/v5/public",
        ws_private="wss://stream-testnet.bybit.com/v5/private",
        ws_trade="wss://stream-testnet.bybit.com/v5/trade",
    ),
    Env.DEMO: Endpoints(
        rest="https://api-demo.bybit.com",
        ws_public_base="wss://stream.bybit.com/v5/public",   # 本番を使う
        ws_private="wss://stream-demo.bybit.com/v5/private",
        ws_trade="wss://stream-demo.bybit.com/v5/trade",
    ),
    Env.MAINNET: Endpoints(
        rest="https://api.bybit.com",
        ws_public_base="wss://stream.bybit.com/v5/public",
        ws_private="wss://stream.bybit.com/v5/private",
        ws_trade="wss://stream.bybit.com/v5/trade",
    ),
}

_PREFIX = {Env.TESTNET: "BYBIT_TESTNET", Env.DEMO: "BYBIT_DEMO",
           Env.MAINNET: "BYBIT_MAINNET"}


@dataclass(frozen=True)
class EnvConfig:
    env: Env
    api_key: str
    api_secret: str

    @property
    def endpoints(self) -> Endpoints:
        return ENDPOINTS[self.env]

    # pybitのHTTP / WebSocket に渡すフラグ
    @property
    def testnet(self) -> bool:
        return self.env is Env.TESTNET

    @property
    def demo(self) -> bool:
        return self.env is Env.DEMO

    @property
    def is_live(self) -> bool:
        """実資金が動く環境かどうか。ガードの判定に使う。"""
        return self.env is Env.MAINNET

    def ws_public(self, channel: str = "linear") -> str:
        return f"{self.endpoints.ws_public_base}/{channel}"

    def masked(self) -> str:
        return f"{self.env.value}:{self.api_key[:4]}****"

    def banner(self) -> str:
        mark = "  ***** 実資金 *****" if self.is_live else ""
        return (f"[{self.env.value.upper()}]{mark}\n"
                f"  REST       : {self.endpoints.rest}\n"
                f"  WS public  : {self.ws_public('linear')}\n"
                f"  WS private : {self.endpoints.ws_private}\n"
                f"  key        : {self.masked()}")


def load_env(env: str | None = None) -> EnvConfig:
    resolved = Env(env or os.getenv("BYBIT_ENV", "testnet"))
    prefix = _PREFIX[resolved]
    key = os.getenv(f"{prefix}_API_KEY", "").strip()
    secret = os.getenv(f"{prefix}_API_SECRET", "").strip()
    if not key or not secret:
        raise RuntimeError(
            f"{prefix}_API_KEY / {prefix}_API_SECRET が .env にありません "
            f"(BYBIT_ENV={resolved.value})"
        )
    return EnvConfig(env=resolved, api_key=key, api_secret=secret)


if __name__ == "__main__":
    for name in ("testnet", "demo", "mainnet"):
        try:
            print(load_env(name).banner())
        except RuntimeError as e:
            print(f"[{name.upper()}] 未設定: {e}")
        print()

"""demo_money.py — デモトレード口座に仮想資金を投入する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

エンドポイント: POST /v5/account/demo-apply-money （demo環境専用）
1リクエストの上限: BTC 15 / ETH 200 / USDT 100,000 / USDC 100,000
"""
from __future__ import annotations

from decimal import Decimal

from env_config import Env, load_env
from v5_client import V5Client          # 記事02の自作クライアント
from signer import HmacSigner

# 1リクエストあたりの上限
MAX_PER_REQUEST = {
    "BTC": Decimal("15"),
    "ETH": Decimal("200"),
    "USDT": Decimal("100000"),
    "USDC": Decimal("100000"),
}


def apply_demo_money(client: V5Client, coin: str,
                     amount: Decimal) -> dict:
    """デモ口座に仮想資金を投入する。上限を超える場合は分割する。"""
    coin = coin.upper()
    cap = MAX_PER_REQUEST.get(coin)
    if cap is None:
        raise ValueError(f"{coin} はデモ資金の対象外です "
                         f"（対象: {', '.join(MAX_PER_REQUEST)}）")

    remaining = amount
    results = []
    while remaining > 0:
        chunk = min(remaining, cap)
        res = client.post("/v5/account/demo-apply-money", {
            "adjustType": 0,        # 0 = 追加
            "utaDemoApplyMoney": [
                {"coin": coin, "amountStr": str(chunk)},
            ],
        })
        results.append(res)
        print(f"  {coin} {chunk} を投入しました")
        remaining -= chunk
    return {"requests": len(results), "total": str(amount)}


def show_demo_balance(client: V5Client) -> None:
    r = client.get("/v5/account/wallet-balance", {"accountType": "UNIFIED"})
    acct = r["list"][0]
    print(f"総資産: {acct['totalEquity']} USD")
    for c in acct["coin"]:
        if float(c["walletBalance"] or 0) != 0:
            print(f"  {c['coin']:<6} {c['walletBalance']}")


if __name__ == "__main__":
    cfg = load_env("demo")
    if cfg.env is not Env.DEMO:
        raise SystemExit("このスクリプトは BYBIT_ENV=demo 専用です")

    api = V5Client(base_url=cfg.endpoints.rest, api_key=cfg.api_key,
                   signer=HmacSigner(cfg.api_secret))

    print("投入前:")
    show_demo_balance(api)

    apply_demo_money(api, "USDT", Decimal("100000"))

    print("\n投入後:")
    show_demo_balance(api)

"""compare_liquidity.py — testnetと本番の板の厚さ・価格乖離を比較する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

from pybit.unified_trading import HTTP


@dataclass
class BookStats:
    label: str
    last: Decimal
    bid: Decimal
    ask: Decimal
    spread_bps: Decimal
    depth_1pct_usd: Decimal        # ミッド±1%以内に存在する板の合計金額
    slip_10k_bps: Decimal | None   # 1万USDの成行で想定されるスリッページ

    def line(self) -> str:
        slip = (f"{self.slip_10k_bps:>8.1f}" if self.slip_10k_bps is not None
                else "    食い切れず")
        return (f"{self.label:<10}{self.last:>12}{self.spread_bps:>10.2f}"
                f"{self.depth_1pct_usd:>16,.0f}{slip:>12}")


def analyse_book(http: HTTP, label: str, symbol: str = "BTCUSDT",
                 category: str = "linear",
                 test_notional: Decimal = Decimal("10000")) -> BookStats:
    ob = http.get_orderbook(category=category, symbol=symbol,
                            limit=200)["result"]
    bids = [(Decimal(p), Decimal(q)) for p, q in ob["b"]]
    asks = [(Decimal(p), Decimal(q)) for p, q in ob["a"]]

    ticker = http.get_tickers(category=category, symbol=symbol)["result"]["list"][0]
    last = Decimal(ticker["lastPrice"])

    bid, ask = bids[0][0], asks[0][0]
    mid = (bid + ask) / 2
    spread_bps = (ask - bid) / mid * 10_000

    # ミッド±1%以内の板の厚さ（金額）
    lo, hi = mid * Decimal("0.99"), mid * Decimal("1.01")
    depth = sum(p * q for p, q in bids if p >= lo)
    depth += sum(p * q for p, q in asks if p <= hi)

    # 1万USDを成行で買った場合の平均約定価格 -> スリッページ
    remaining = test_notional
    cost = Decimal(0)
    filled = Decimal(0)
    for p, q in asks:
        take_value = min(remaining, p * q)
        take_qty = take_value / p
        cost += take_qty * p
        filled += take_qty
        remaining -= take_value
        if remaining <= 0:
            break

    slip = None
    if remaining <= 0 and filled > 0:
        avg = cost / filled
        slip = (avg - mid) / mid * 10_000

    return BookStats(label=label, last=last, bid=bid, ask=ask,
                     spread_bps=spread_bps, depth_1pct_usd=depth,
                     slip_10k_bps=slip)


if __name__ == "__main__":
    stats = [
        analyse_book(HTTP(testnet=False), "mainnet"),
        analyse_book(HTTP(testnet=True), "testnet"),
    ]

    print(f"{'環境':<10}{'最終価格':>12}{'スプレッド':>10}"
          f"{'±1%の板厚(USD)':>16}{'1万USDの':>12}")
    print(f"{'':<10}{'':>12}{'(bps)':>10}{'':>16}{'スリッページ':>12}")
    print("-" * 62)
    for s in stats:
        print(s.line())

    main, test = stats
    gap = (test.last - main.last) / main.last * 100
    print(f"\n価格乖離     : {gap:+.2f}%")
    print(f"スプレッド比 : {test.spread_bps / main.spread_bps:.1f}倍")
    if main.depth_1pct_usd > 0:
        print(f"板の厚さ比   : {test.depth_1pct_usd / main.depth_1pct_usd:.3f}倍")
    print("\n※ この差があるため、testnetでの成行約定価格・スリッページ・")
    print("  PostOnlyの約定率は、いずれも本番の参考にならない。")

"""live_guard.py — 本番誤爆を構造的に防ぐ多層ガード
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

思想: 「本番に接続する」という判断を、複数の独立した条件が
      すべて揃わないと成立しないようにする。
"""
from __future__ import annotations

import functools
import os
import sys
from dataclasses import dataclass, field
from decimal import Decimal
from pathlib import Path

from env_config import Env, EnvConfig, load_env

# 本番接続を許可するファイル（存在しなければ本番に繋がらない）
LIVE_TOKEN_FILE = Path("LIVE_ENABLED")


class LiveAccessDenied(PermissionError):
    """本番接続の条件が揃っていない。"""


class RiskLimitExceeded(RuntimeError):
    """本番での発注が上限を超えている。"""


@dataclass
class LiveLimits:
    """本番でのみ適用される上限。ロジックのバグによる大量発注を止める。"""

    max_notional_per_order: Decimal = Decimal("500")
    max_total_notional: Decimal = Decimal("2000")
    max_orders_per_minute: int = 20
    allowed_symbols: set[str] = field(default_factory=lambda: {"BTCUSDT"})


def assert_live_allowed(cfg: EnvConfig, *, explicit: bool = False) -> None:
    """本番接続の4条件をすべて検査する。"""
    if not cfg.is_live:
        return

    reasons: list[str] = []

    # 条件1: 呼び出し側がコードで明示的に許可している
    if not explicit:
        reasons.append("コード側で allow_live=True を指定していない")

    # 条件2: 環境変数が立っている
    if os.getenv("BYBIT_ALLOW_LIVE") != "1":
        reasons.append("環境変数 BYBIT_ALLOW_LIVE=1 が設定されていない")

    # 条件3: 許可ファイルが存在する（VPSに置き忘れを防ぐ）
    if not LIVE_TOKEN_FILE.exists():
        reasons.append(f"許可ファイル {LIVE_TOKEN_FILE} が存在しない")

    # 条件4: pytest実行中は絶対に本番に繋がない
    if "pytest" in sys.modules:
        reasons.append("pytest実行中は本番接続を禁止")

    if reasons:
        raise LiveAccessDenied(
            "本番接続が拒否されました:\n  - " + "\n  - ".join(reasons)
        )


class GuardedTrader:
    """本番では上限を強制し、それ以外の環境では素通しするラッパ。"""

    def __init__(self, http, cfg: EnvConfig,
                 limits: LiveLimits | None = None):
        self.http = http
        self.cfg = cfg
        self.limits = limits or LiveLimits()
        self._order_times: list[float] = []
        self._total_notional = Decimal(0)

    def _check(self, symbol: str, notional: Decimal) -> None:
        if not self.cfg.is_live:
            return          # testnet / demo は無制限

        import time

        lim = self.limits
        if symbol not in lim.allowed_symbols:
            raise RiskLimitExceeded(
                f"{symbol} は本番の許可銘柄ではありません "
                f"（許可: {sorted(lim.allowed_symbols)}）"
            )
        if notional > lim.max_notional_per_order:
            raise RiskLimitExceeded(
                f"1注文の想定元本 {notional} > 上限 {lim.max_notional_per_order}"
            )
        if self._total_notional + notional > lim.max_total_notional:
            raise RiskLimitExceeded(
                f"累計想定元本 {self._total_notional + notional} "
                f"> 上限 {lim.max_total_notional}"
            )

        now = time.time()
        self._order_times = [t for t in self._order_times if now - t < 60]
        if len(self._order_times) >= lim.max_orders_per_minute:
            raise RiskLimitExceeded(
                f"1分あたりの発注数が上限 {lim.max_orders_per_minute} に到達"
                f"（ロジックの暴走を疑ってください）"
            )

    def place_order(self, *, notional: Decimal, **params) -> dict:
        import time

        self._check(params["symbol"], notional)
        res = self.http.place_order(**params)
        if res.get("retCode") == 0 and self.cfg.is_live:
            self._order_times.append(time.time())
            self._total_notional += notional
        return res


def live_only_warning(fn):
    """本番実行時に、実行前に内容を表示するデコレータ（対話環境向け）。"""

    @functools.wraps(fn)
    def wrapper(cfg: EnvConfig, *args, **kwargs):
        if cfg.is_live:
            print("=" * 60)
            print(f"  本番環境で {fn.__name__} を実行します")
            print(f"  接続先: {cfg.endpoints.rest}")
            print("=" * 60)
        return fn(cfg, *args, **kwargs)

    return wrapper


def make_guarded_client(env: str | None = None, *, allow_live: bool = False):
    """ガードを通したクライアントを返す。"""
    from pybit.unified_trading import HTTP

    cfg = load_env(env)
    assert_live_allowed(cfg, explicit=allow_live)

    http = HTTP(testnet=cfg.testnet, demo=cfg.demo,
                api_key=cfg.api_key, api_secret=cfg.api_secret,
                recv_window=5000, timeout=10)
    print(cfg.banner())
    return GuardedTrader(http, cfg), cfg


if __name__ == "__main__":
    # testnetは素通り
    trader, cfg = make_guarded_client("testnet")
    print("\ntestnet: ガードなしで発注可能\n")

    # mainnetは条件が揃わないと拒否される
    try:
        make_guarded_client("mainnet", allow_live=True)
    except LiveAccessDenied as e:
        print(e)
    except RuntimeError as e:
        print(f"（本番キー未設定: {e}）")

"""mock_exchange.py — V5のレスポンス形式を模した最小の取引所モック
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

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


def ok(result: dict | None = None) -> dict:
    return {"retCode": 0, "retMsg": "OK", "result": result or {},
            "retExtInfo": {}, "time": int(time.time() * 1000)}


def err(code: int, msg: str) -> dict:
    return {"retCode": code, "retMsg": msg, "result": {},
            "retExtInfo": {}, "time": int(time.time() * 1000)}


@dataclass
class MockOrder:
    order_id: str
    order_link_id: str
    symbol: str
    side: str
    qty: Decimal
    price: Decimal | None
    order_type: str
    status: str = "New"
    cum_exec_qty: Decimal = Decimal(0)
    avg_price: Decimal = Decimal(0)


@dataclass
class MockExchange:
    """pybitのHTTPと同じメソッド名を持つモック。

    - 銘柄仕様・価格は固定値
    - orderLinkIdの重複は110072で拒否（冪等性のテストに使う）
    - inject_error で任意のエラーを注入できる
    """

    price: Decimal = Decimal("100000")
    tick_size: str = "0.1"
    qty_step: str = "0.001"
    min_qty: str = "0.001"
    min_notional: str = "5"
    orders: dict[str, MockOrder] = field(default_factory=dict)
    positions: dict[str, Decimal] = field(default_factory=dict)
    calls: list[tuple[str, dict]] = field(default_factory=list)
    inject_error: dict[str, dict] = field(default_factory=dict)
    _ids: itertools.count = field(default_factory=lambda: itertools.count(1))

    # ---- 呼び出し記録とエラー注入 ---------------------------------------
    def _record(self, name: str, kwargs: dict) -> dict | None:
        self.calls.append((name, dict(kwargs)))
        if name in self.inject_error:
            return self.inject_error[name]
        return None

    def called(self, name: str) -> list[dict]:
        return [kw for n, kw in self.calls if n == name]

    # ---- Market ---------------------------------------------------------
    def get_server_time(self) -> dict:
        now = int(time.time() * 1000)
        return ok({"timeSecond": str(now // 1000), "timeNano": str(now * 1_000_000)})

    def get_tickers(self, **kw) -> dict:
        if (e := self._record("get_tickers", kw)):
            return e
        return ok({"category": kw.get("category", "linear"), "list": [{
            "symbol": kw.get("symbol", "BTCUSDT"),
            "lastPrice": str(self.price),
            "markPrice": str(self.price),
            "indexPrice": str(self.price),
            "bid1Price": str(self.price - Decimal("1")),
            "ask1Price": str(self.price + Decimal("1")),
            "bid1Size": "10", "ask1Size": "10",
            "price24hPcnt": "0.01", "turnover24h": "1000000000",
            "volume24h": "10000", "openInterest": "50000",
            "fundingRate": "0.0001",
            "nextFundingTime": str(int(time.time() * 1000) + 3600_000),
        }]})

    def get_orderbook(self, **kw) -> dict:
        if (e := self._record("get_orderbook", kw)):
            return e
        p = self.price
        return ok({
            "s": kw.get("symbol", "BTCUSDT"),
            "b": [[str(p - Decimal(i)), "1.0"] for i in range(1, 11)],
            "a": [[str(p + Decimal(i)), "1.0"] for i in range(1, 11)],
            "ts": int(time.time() * 1000), "u": 1, "seq": 1,
        })

    def get_instruments_info(self, **kw) -> dict:
        if (e := self._record("get_instruments_info", kw)):
            return e
        return ok({"category": kw.get("category", "linear"), "list": [{
            "symbol": kw.get("symbol", "BTCUSDT"), "status": "Trading",
            "baseCoin": "BTC", "quoteCoin": "USDT", "settleCoin": "USDT",
            "contractType": "LinearPerpetual", "fundingInterval": 480,
            "priceFilter": {"tickSize": self.tick_size},
            "lotSizeFilter": {"qtyStep": self.qty_step,
                              "minOrderQty": self.min_qty,
                              "maxOrderQty": "100",
                              "minNotionalValue": self.min_notional},
            "leverageFilter": {"minLeverage": "1", "maxLeverage": "100"},
        }], "nextPageCursor": ""})

    def get_price_limit(self, **kw) -> dict:
        self._record("get_price_limit", kw)
        return ok({"symbol": kw.get("symbol"),
                   "buyLmt": str(self.price * Decimal("1.1")),
                   "sellLmt": str(self.price * Decimal("0.9"))})

    def get_kline(self, **kw) -> dict:
        if (e := self._record("get_kline", kw)):
            return e
        now = int(time.time() * 1000) // 3600_000 * 3600_000
        rows = []
        for i in range(int(kw.get("limit", 200))):
            ts = now - i * 3600_000
            p = self.price
            rows.append([str(ts), str(p), str(p * Decimal("1.01")),
                         str(p * Decimal("0.99")), str(p), "100", "10000000"])
        return ok({"symbol": kw.get("symbol"), "category": kw.get("category"),
                   "list": rows})

    # ---- Account --------------------------------------------------------
    def get_wallet_balance(self, **kw) -> dict:
        if (e := self._record("get_wallet_balance", kw)):
            return e
        if kw.get("accountType") not in ("UNIFIED", "CONTRACT", "FUND"):
            return err(10001, "accountType invalid")
        return ok({"list": [{
            "accountType": kw["accountType"], "totalEquity": "10000",
            "totalMarginBalance": "10000", "totalAvailableBalance": "9500",
            "totalInitialMargin": "500", "totalMaintenanceMargin": "50",
            "totalPerpUPL": "0", "accountMMRate": "0.005",
            "accountIMRate": "0.05",
            "coin": [{"coin": "USDT", "walletBalance": "10000",
                      "unrealisedPnl": "0", "usdValue": "10000"}],
        }]})

    def get_fee_rates(self, **kw) -> dict:
        self._record("get_fee_rates", kw)
        return ok({"list": [{"symbol": kw.get("symbol", "BTCUSDT"),
                             "makerFeeRate": "0.0002",
                             "takerFeeRate": "0.00055"}]})

    # ---- Trade ----------------------------------------------------------
    def place_order(self, **kw) -> dict:
        if (e := self._record("place_order", kw)):
            return e

        link_id = kw.get("orderLinkId") or f"auto-{next(self._ids)}"
        if link_id in {o.order_link_id for o in self.orders.values()}:
            return err(110072, "OrderLinkedID is duplicate")

        qty = Decimal(str(kw["qty"]))
        if qty < Decimal(self.min_qty):
            return err(110017, "OrderQty will be truncated to zero")

        price = Decimal(str(kw["price"])) if kw.get("price") else self.price
        if price * qty < Decimal(self.min_notional):
            return err(110094, "Order notional value below the lower limit")

        order_id = f"mock-{next(self._ids):06d}"
        o = MockOrder(order_id=order_id, order_link_id=link_id,
                      symbol=kw["symbol"], side=kw["side"], qty=qty,
                      price=Decimal(str(kw["price"])) if kw.get("price") else None,
                      order_type=kw["orderType"])

        # 成行は即約定させる
        if kw["orderType"] == "Market":
            o.status = "Filled"
            o.cum_exec_qty = qty
            o.avg_price = self.price
            sign = 1 if kw["side"] == "Buy" else -1
            self.positions[kw["symbol"]] = (
                self.positions.get(kw["symbol"], Decimal(0)) + qty * sign
            )

        self.orders[order_id] = o
        return ok({"orderId": order_id, "orderLinkId": link_id})

    def cancel_order(self, **kw) -> dict:
        if (e := self._record("cancel_order", kw)):
            return e
        for o in self.orders.values():
            if kw.get("orderId") == o.order_id or \
               kw.get("orderLinkId") == o.order_link_id:
                if o.status != "New":
                    return err(110008, "The order has been completed or cancelled")
                o.status = "Cancelled"
                return ok({"orderId": o.order_id,
                           "orderLinkId": o.order_link_id})
        return err(110001, "Order does not exist")

    def cancel_all_orders(self, **kw) -> dict:
        self._record("cancel_all_orders", kw)
        n = 0
        for o in self.orders.values():
            if o.status == "New":
                o.status = "Cancelled"
                n += 1
        return ok({"list": [], "cancelled": n})

    def amend_order(self, **kw) -> dict:
        if (e := self._record("amend_order", kw)):
            return e
        for o in self.orders.values():
            if kw.get("orderId") == o.order_id or \
               kw.get("orderLinkId") == o.order_link_id:
                if o.status != "New":
                    return err(110008, "The order has been completed or cancelled")
                if kw.get("price"):
                    o.price = Decimal(str(kw["price"]))
                return ok({"orderId": o.order_id})
        return err(110001, "Order does not exist")

    def get_open_orders(self, **kw) -> dict:
        self._record("get_open_orders", kw)
        rows = [o for o in self.orders.values() if o.status == "New"]
        if kw.get("orderLinkId"):
            rows = [o for o in rows if o.order_link_id == kw["orderLinkId"]]
        return ok({"list": [self._order_row(o) for o in rows],
                   "nextPageCursor": ""})

    def get_order_history(self, **kw) -> dict:
        self._record("get_order_history", kw)
        rows = list(self.orders.values())
        if kw.get("orderLinkId"):
            rows = [o for o in rows if o.order_link_id == kw["orderLinkId"]]
        return ok({"list": [self._order_row(o) for o in rows],
                   "nextPageCursor": ""})

    @staticmethod
    def _order_row(o: MockOrder) -> dict:
        return {
            "orderId": o.order_id, "orderLinkId": o.order_link_id,
            "symbol": o.symbol, "side": o.side, "qty": str(o.qty),
            "price": str(o.price) if o.price else "",
            "orderType": o.order_type, "orderStatus": o.status,
            "cumExecQty": str(o.cum_exec_qty), "avgPrice": str(o.avg_price),
            "createdTime": str(int(time.time() * 1000)),
        }

    # ---- Position -------------------------------------------------------
    def get_positions(self, **kw) -> dict:
        self._record("get_positions", kw)
        rows = []
        for symbol, qty in self.positions.items():
            if qty == 0:
                continue
            rows.append({
                "symbol": symbol, "side": "Buy" if qty > 0 else "Sell",
                "size": str(abs(qty)), "avgPrice": str(self.price),
                "markPrice": str(self.price), "positionIdx": 0,
                "leverage": "10", "unrealisedPnl": "0",
                "liqPrice": "", "positionIM": "100", "positionMM": "10",
                "takeProfit": "", "stopLoss": "", "trailingStop": "0",
                "tpslMode": "Full", "positionStatus": "Normal",
            })
        return ok({"list": rows, "nextPageCursor": ""})

    def set_trading_stop(self, **kw) -> dict:
        if (e := self._record("set_trading_stop", kw)):
            return e
        if not self.positions:
            return err(110033, "Cannot set margin without an open position")
        return ok({})

    def set_leverage(self, **kw) -> dict:
        self._record("set_leverage", kw)
        return err(110043, "Set leverage has not been modified")

"""test_with_mock.py — モック取引所を使った発注ロジックのテスト
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

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

from decimal import Decimal

import pytest

from mock_exchange import MockExchange, err
from order_validate import OrderRejected, size_from_notional, validate_order
from symbol_rules import fetch_rules


@pytest.fixture
def ex() -> MockExchange:
    return MockExchange(price=Decimal("100000"))


def test_valid_order_passes(ex):
    rules = fetch_rules(ex, "BTCUSDT", "linear")
    qty = size_from_notional(rules, Decimal("1000"), Decimal("100000"))
    v = validate_order(rules, "Buy", qty, price=Decimal("99000"))
    res = ex.place_order(**v.to_params("Limit"))
    assert res["retCode"] == 0
    assert res["result"]["orderId"].startswith("mock-")


def test_tiny_qty_rejected_before_api(ex):
    """発注前検証で弾かれ、APIには送られないこと。"""
    rules = fetch_rules(ex, "BTCUSDT", "linear")
    before = len(ex.called("place_order"))
    with pytest.raises(OrderRejected):
        validate_order(rules, "Buy", Decimal("0.0000001"),
                       price=Decimal("100000"))
    assert len(ex.called("place_order")) == before   # 送信されていない


def test_duplicate_link_id_blocked(ex):
    """同じorderLinkIdの再送が110072で拒否されること（冪等性）。"""
    params = {"category": "linear", "symbol": "BTCUSDT", "side": "Buy",
              "orderType": "Limit", "qty": "0.001", "price": "99000",
              "orderLinkId": "test-idem-001"}
    first = ex.place_order(**params)
    second = ex.place_order(**params)
    assert first["retCode"] == 0
    assert second["retCode"] == 110072


def test_market_order_updates_position(ex):
    ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                   orderType="Market", qty="0.010")
    pos = ex.get_positions(category="linear")["result"]["list"]
    assert len(pos) == 1
    assert pos[0]["side"] == "Buy"
    assert Decimal(pos[0]["size"]) == Decimal("0.010")


def test_cancel_twice(ex):
    r = ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                       orderType="Limit", qty="0.001", price="99000",
                       orderLinkId="c1")
    oid = r["result"]["orderId"]
    assert ex.cancel_order(category="linear", symbol="BTCUSDT",
                           orderId=oid)["retCode"] == 0
    # 2回目は110008（正常な競合として扱うべきコード）
    assert ex.cancel_order(category="linear", symbol="BTCUSDT",
                           orderId=oid)["retCode"] == 110008


def test_error_injection(ex):
    """任意のエラーを注入して、リトライ処理を検証できる。"""
    ex.inject_error["place_order"] = err(10006, "Too many visits")
    res = ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                         orderType="Market", qty="0.001")
    assert res["retCode"] == 10006

    ex.inject_error.clear()
    assert ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                          orderType="Market", qty="0.001")["retCode"] == 0


def test_set_leverage_benign_error(ex):
    """110043は『変更なし』なので、実質成功として扱うべきこと。"""
    res = ex.set_leverage(category="linear", symbol="BTCUSDT",
                          buyLeverage="10", sellLeverage="10")
    assert res["retCode"] == 110043      # 呼び出し側で無視する対象


def test_min_notional_enforced(ex):
    res = ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                         orderType="Limit", qty="0.001", price="1000")
    assert res["retCode"] == 110094      # 1000 * 0.001 = 1 < 最小5

"""recorder.py — WebSocketメッセージの記録とリプレイ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

用途: 急落時・スプレッド拡大時の挙動を、何度でも同じ条件で再現する。
"""
from __future__ import annotations

import gzip
import json
import time
from collections.abc import Iterator
from pathlib import Path


class WsRecorder:
    """受信メッセージを1行1JSONでgzip保存する。"""

    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._fh = gzip.open(self.path, "at", encoding="utf-8")
        self.count = 0

    def __call__(self, msg: dict) -> None:
        """WebSocketのコールバックとしてそのまま渡せる。"""
        self._fh.write(json.dumps(
            {"_recv_at": time.time(), "msg": msg}, ensure_ascii=False) + "\n")
        self.count += 1
        if self.count % 1000 == 0:
            self._fh.flush()

    def close(self) -> None:
        self._fh.flush()
        self._fh.close()
        size_mb = self.path.stat().st_size / 1024 / 1024
        print(f"{self.count}件を記録しました ({self.path}, {size_mb:.1f} MB)")


class WsReplayer:
    """記録を再生する。speed=0で待ち時間なし（テスト用）。"""

    def __init__(self, path: str | Path):
        self.path = Path(path)

    def __iter__(self) -> Iterator[tuple[float, dict]]:
        with gzip.open(self.path, "rt", encoding="utf-8") as fh:
            for line in fh:
                rec = json.loads(line)
                yield rec["_recv_at"], rec["msg"]

    def replay(self, handler, speed: float = 0.0,
               topic_filter: str | None = None) -> int:
        """記録を順に handler へ渡す。

        speed=1.0 で実時間どおり、0.0で即時（テストでは0.0）。
        """
        n = 0
        prev_at: float | None = None
        for recv_at, msg in self:
            if topic_filter and msg.get("topic", "") != topic_filter:
                continue
            if speed > 0 and prev_at is not None:
                time.sleep(max(0.0, (recv_at - prev_at) / speed))
            prev_at = recv_at
            handler(msg)
            n += 1
        return n

    def stats(self) -> dict:
        topics: dict[str, int] = {}
        first = last = None
        for recv_at, msg in self:
            t = msg.get("topic", "(control)")
            topics[t] = topics.get(t, 0) + 1
            first = first or recv_at
            last = recv_at
        return {"topics": topics, "duration_sec": (last - first) if first else 0,
                "total": sum(topics.values())}


def record_session(symbol: str = "BTCUSDT", seconds: float = 300.0,
                   out: str = "data/replay/session.jsonl.gz") -> None:
    """指定時間だけ記録する。急変時に手動で走らせておくと後で役に立つ。"""
    from pybit.unified_trading import WebSocket

    rec = WsRecorder(out)
    ws = WebSocket(testnet=False, channel_type="linear")
    ws.ticker_stream(symbol=symbol, callback=rec)
    ws.trade_stream(symbol=symbol, callback=rec)
    ws.kline_stream(interval=1, symbol=symbol, callback=rec)

    print(f"{seconds:.0f}秒間 {symbol} を記録します...")
    try:
        time.sleep(seconds)
    except KeyboardInterrupt:
        pass
    finally:
        ws.exit()
        rec.close()


if __name__ == "__main__":
    import sys

    path = "data/replay/session.jsonl.gz"

    if "--record" in sys.argv:
        record_session(seconds=60.0, out=path)
    elif Path(path).exists():
        rp = WsReplayer(path)
        print("記録の統計:", rp.stats())

        # 戦略を再生データに通す例
        from bt.types import Bar

        closes: list[float] = []

        def handler(msg: dict) -> None:
            if msg.get("topic", "").startswith("kline"):
                for k in msg["data"]:
                    if k.get("confirm"):
                        closes.append(float(k["close"]))

        n = rp.replay(handler, speed=0.0)
        print(f"{n}件を再生 / 確定足 {len(closes)}本")
    else:
        print(f"{path} がありません。--record で記録してください")

"""rollout_check.py — 次の段階に進んでよいかを機械的に判定する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal


@dataclass
class StageCriteria:
    """段階を上げるための最低条件。"""

    name: str
    min_days: int
    min_trades: int
    max_unexpected_errors: int = 0
    max_restarts: int = 2
    require_no_liquidation: bool = True
    max_slippage_bps: Decimal | None = None


STAGES = [
    StageCriteria("1. モック", min_days=0, min_trades=0),
    StageCriteria("2. testnet", min_days=3, min_trades=20),
    StageCriteria("3. デモ", min_days=14, min_trades=100,
                  max_slippage_bps=Decimal("15")),
    StageCriteria("4. 本番(最小)", min_days=14, min_trades=50,
                  max_slippage_bps=Decimal("10")),
    StageCriteria("5. 本番(拡大)", min_days=30, min_trades=200),
]


@dataclass
class Observed:
    days: int
    trades: int
    unexpected_errors: int = 0
    restarts: int = 0
    liquidations: int = 0
    median_slippage_bps: Decimal | None = None
    notes: list[str] = field(default_factory=list)


def can_advance(stage: StageCriteria, obs: Observed) -> tuple[bool, list[str]]:
    blockers: list[str] = []

    if obs.days < stage.min_days:
        blockers.append(f"稼働日数 {obs.days} < 必要 {stage.min_days}日")
    if obs.trades < stage.min_trades:
        blockers.append(f"約定回数 {obs.trades} < 必要 {stage.min_trades}回")
    if obs.unexpected_errors > stage.max_unexpected_errors:
        blockers.append(
            f"未処理エラー {obs.unexpected_errors}件 "
            f"（許容 {stage.max_unexpected_errors}件）"
        )
    if obs.restarts > stage.max_restarts:
        blockers.append(f"異常再起動 {obs.restarts}回 "
                        f"（許容 {stage.max_restarts}回）")
    if stage.require_no_liquidation and obs.liquidations > 0:
        blockers.append(f"清算が{obs.liquidations}回発生している"
                        "（レバレッジ設計の見直しが必要）")
    if (stage.max_slippage_bps is not None
            and obs.median_slippage_bps is not None
            and obs.median_slippage_bps > stage.max_slippage_bps):
        blockers.append(
            f"スリッページ中央値 {obs.median_slippage_bps} bps "
            f"> 許容 {stage.max_slippage_bps} bps"
        )
    return (not blockers), blockers


if __name__ == "__main__":
    # デモ段階の観測値の例
    obs = Observed(days=16, trades=142, unexpected_errors=0, restarts=1,
                   liquidations=0, median_slippage_bps=Decimal("11"))

    stage = STAGES[2]
    ok, blockers = can_advance(stage, obs)

    print(f"現在の段階: {stage.name}")
    print(f"観測: {obs.days}日 / {obs.trades}回 / "
          f"エラー{obs.unexpected_errors}件 / 再起動{obs.restarts}回 / "
          f"スリッページ中央値{obs.median_slippage_bps}bps\n")

    if ok:
        print(f"-> {STAGES[3].name} に進んでよい条件を満たしています")
    else:
        print("-> まだ進めません:")
        for b in blockers:
            print(f"     - {b}")

    print("\n--- 全段階の基準 ---")
    for s in STAGES:
        slip = (f"スリッページ<={s.max_slippage_bps}bps"
                if s.max_slippage_bps else "-")
        print(f"  {s.name:<14} {s.min_days:>3}日 {s.min_trades:>4}回  {slip}")