orderLinkIdで注文を冪等化する
bot再起動・二重発注事故を防ぐ
Python設計

# この実装は、いつか必ず二重発注する
for attempt in range(3):
    try:
        res = http.place_order(category="linear", symbol="BTCUSDT",
                               side="Buy", orderType="Market", qty="0.1")
        break
    except requests.exceptions.Timeout:
        continue      # <- ここが致命的
#
# 起きること:
#   1) place_order のリクエストは取引所に届いた
#   2) 取引所は注文を受理して約定させた
#   3) しかしレスポンスが返る前にネットワークがタイムアウトした
#   4) botは「失敗した」と判断してリトライ
#   5) 2つ目の注文が通り、意図の2倍のポジションを持つ

"""order_id.py — 決定論的なorderLinkIdの生成
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import hashlib
import re
import time
from dataclasses import dataclass
from decimal import Decimal

MAX_LEN = 36
# 使用可能なのは英数字・ハイフン・アンダースコアのみ
VALID_RE = re.compile(r"^[A-Za-z0-9_-]{1,36}$")


class InvalidOrderLinkId(ValueError):
    pass


def sanitize_tag(tag: str, max_len: int = 8) -> str:
    """戦略名を安全な文字に落とし込む（日本語・記号を除去）。"""
    cleaned = re.sub(r"[^A-Za-z0-9]", "", tag)
    if not cleaned:
        cleaned = "bot"
    return cleaned[:max_len].lower()


def validate(link_id: str) -> str:
    if not VALID_RE.match(link_id):
        raise InvalidOrderLinkId(
            f"orderLinkIdが不正です: {link_id!r} "
            f"(長さ{len(link_id)}, 英数字/-/_ のみ・36文字以内)"
        )
    return link_id


@dataclass(frozen=True)
class OrderIntent:
    """「何を発注したいか」を表す。同じ意図なら同じIDになる。"""

    strategy: str
    symbol: str
    side: str
    qty: Decimal
    price: Decimal | None = None
    category: str = "linear"
    purpose: str = "entry"          # entry / exit / grid / hedge など
    slot: str = ""                  # グリッドの段番号など任意の識別子

    def fingerprint(self) -> str:
        """意図を一意に表す文字列。ここにIDの決定論性が依存する。"""
        px = "mkt" if self.price is None else str(self.price)
        return "|".join([self.category, self.symbol, self.side,
                         str(self.qty), px, self.purpose, self.slot])


def id_by_hash(intent: OrderIntent, salt: str = "") -> str:
    """意図ハッシュ方式。同じ内容の注文は常に同じIDになる。

    用途: グリッドの段・指値の置き直し。
          「同じ価格・同じ数量の注文を二重に置かない」ことを保証する。
    """
    tag = sanitize_tag(intent.strategy)
    digest = hashlib.sha256(
        (intent.fingerprint() + "|" + salt).encode("utf-8")
    ).hexdigest()
    # tag(8) + '-' + hash(27) = 36文字以内
    return validate(f"{tag}-{digest[:MAX_LEN - len(tag) - 1]}")


def id_by_time_bucket(intent: OrderIntent, bucket_sec: int = 60,
                      now: float | None = None) -> str:
    """時刻バケット方式。同じ足の中では1回だけ発注できる。

    用途: 「1分足の確定ごとに1回だけ判断する」戦略。
          同じ足で複数回呼ばれても2つ目は110072で弾かれる。
    """
    tag = sanitize_tag(intent.strategy)
    t = int(now if now is not None else time.time())
    bucket = t - (t % bucket_sec)
    sym = re.sub(r"[^A-Za-z0-9]", "", intent.symbol)[:8]
    side = "b" if intent.side == "Buy" else "s"
    base = f"{tag}-{sym}-{side}{intent.purpose[:1]}-{bucket}"
    return validate(base[:MAX_LEN])


def id_by_sequence(strategy: str, seq: int) -> str:
    """シーケンス方式。台帳が連番を管理する（サンプル2で使う）。

    用途: 高頻度発注。新規注文は常に別IDだが、
          リトライ時は同じseqを再利用することで冪等になる。
    """
    tag = sanitize_tag(strategy)
    return validate(f"{tag}-{seq:012d}")


if __name__ == "__main__":
    intent = OrderIntent(strategy="trend-follow", symbol="BTCUSDT",
                         side="Buy", qty=Decimal("0.01"),
                         price=Decimal("119000"), purpose="entry")

    print("=== 意図ハッシュ方式（同じ意図 -> 同じID） ===")
    a = id_by_hash(intent)
    b = id_by_hash(intent)
    print(f"  1回目: {a}")
    print(f"  2回目: {b}")
    print(f"  一致 : {a == b}   <- リトライしても同じIDになる")

    # 価格を1ティック変えると別のIDになる
    other = OrderIntent(**{**intent.__dict__, "price": Decimal("119000.5")})
    print(f"  価格違い: {id_by_hash(other)}")

    print("\n=== 時刻バケット方式（同じ足では同じID） ===")
    t = 1785000123.0
    print(f"  t+0秒 : {id_by_time_bucket(intent, 60, now=t)}")
    print(f"  t+30秒: {id_by_time_bucket(intent, 60, now=t + 30)}  <- 同じ")
    print(f"  t+90秒: {id_by_time_bucket(intent, 60, now=t + 90)}  <- 次の足")

    print("\n=== シーケンス方式 ===")
    for seq in (1, 2, 1000000):
        print(f"  seq={seq:>8}: {id_by_sequence('grid-bot', seq)}")

    print("\n=== 長さと文字の検証 ===")
    for bad in ("a" * 40, "日本語ID", "id with space", "ok-id_123"):
        try:
            print(f"  {bad[:20]!r:<24} -> {validate(bad)}")
        except InvalidOrderLinkId as e:
            print(f"  {bad[:20]!r:<24} -> 拒否: {str(e)[:50]}")

"""order_ledger.py — 注文の意図と結末を記録する台帳（SQLite）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

状態遷移:
  INTENT  -> 発注しようと決めた（まだ送っていない）
  SENT    -> APIに送った（結果は未確定）
  ACKED   -> 受理された（orderIdを取得）
  FILLED  -> 約定した
  PARTIAL -> 部分約定
  CANCELLED / REJECTED -> 終了
  UNKNOWN -> 送ったが結果が分からない（照会が必要）
"""
from __future__ import annotations

import json
import sqlite3
import time
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
from pathlib import Path

DB_PATH = Path("data/orders.sqlite")

SCHEMA = """
CREATE TABLE IF NOT EXISTS orders (
    link_id     TEXT PRIMARY KEY,
    strategy    TEXT NOT NULL,
    category    TEXT NOT NULL,
    symbol      TEXT NOT NULL,
    side        TEXT NOT NULL,
    qty         TEXT NOT NULL,
    price       TEXT,
    purpose     TEXT,
    state       TEXT NOT NULL,
    order_id    TEXT,
    filled_qty  TEXT DEFAULT '0',
    avg_price   TEXT DEFAULT '0',
    params      TEXT,
    error       TEXT,
    created_at  REAL NOT NULL,
    updated_at  REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_orders_state ON orders(state);
CREATE INDEX IF NOT EXISTS idx_orders_strategy ON orders(strategy);
CREATE INDEX IF NOT EXISTS idx_orders_symbol ON orders(symbol);

CREATE TABLE IF NOT EXISTS seq (
    strategy TEXT PRIMARY KEY,
    value    INTEGER NOT NULL DEFAULT 0
);
"""


class State(str, Enum):
    INTENT = "INTENT"
    SENT = "SENT"
    ACKED = "ACKED"
    PARTIAL = "PARTIAL"
    FILLED = "FILLED"
    CANCELLED = "CANCELLED"
    REJECTED = "REJECTED"
    UNKNOWN = "UNKNOWN"

    @property
    def is_terminal(self) -> bool:
        return self in (State.FILLED, State.CANCELLED, State.REJECTED)

    @property
    def needs_resolution(self) -> bool:
        """結末が確定していない＝起動時に照会すべき状態。"""
        return self in (State.INTENT, State.SENT, State.UNKNOWN)


@dataclass
class LedgerRow:
    link_id: str
    strategy: str
    category: str
    symbol: str
    side: str
    qty: Decimal
    price: Decimal | None
    purpose: str
    state: State
    order_id: str | None
    filled_qty: Decimal
    avg_price: Decimal
    params: dict
    error: str | None
    created_at: float
    updated_at: float

    @property
    def age_sec(self) -> float:
        return time.time() - self.created_at

    def line(self) -> str:
        px = str(self.price) if self.price else "market"
        return (f"{self.link_id:<38}{self.state.value:<10}"
                f"{self.symbol:<11}{self.side:<5}{self.qty:>12}"
                f"{px:>12}  {self.age_sec / 60:.0f}分前")


class OrderLedger:
    def __init__(self, path: Path = DB_PATH):
        path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(str(path), isolation_level=None,
                                    check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA journal_mode=WAL")   # クラッシュ耐性
        self.conn.execute("PRAGMA synchronous=FULL")   # 書き込みを確実にする
        self.conn.executescript(SCHEMA)

    # ---- 連番 -----------------------------------------------------------
    def next_seq(self, strategy: str) -> int:
        cur = self.conn.execute(
            "INSERT INTO seq (strategy, value) VALUES (?, 1) "
            "ON CONFLICT(strategy) DO UPDATE SET value = value + 1 "
            "RETURNING value",
            (strategy,),
        )
        return int(cur.fetchone()[0])

    # ---- 記録 -----------------------------------------------------------
    def record_intent(self, link_id: str, intent, params: dict) -> bool:
        """発注前に意図を記録する。既に存在すればFalse（冪等）。"""
        now = time.time()
        try:
            self.conn.execute(
                "INSERT INTO orders (link_id, strategy, category, symbol,"
                " side, qty, price, purpose, state, params, created_at,"
                " updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
                (link_id, intent.strategy, intent.category, intent.symbol,
                 intent.side, str(intent.qty),
                 str(intent.price) if intent.price is not None else None,
                 intent.purpose, State.INTENT.value,
                 json.dumps(params, ensure_ascii=False, default=str),
                 now, now),
            )
            return True
        except sqlite3.IntegrityError:
            return False        # 同じlink_idが既にある = 既に発注済み

    def set_state(self, link_id: str, state: State, *,
                  order_id: str | None = None,
                  filled_qty: Decimal | None = None,
                  avg_price: Decimal | None = None,
                  error: str | None = None) -> None:
        sets = ["state = ?", "updated_at = ?"]
        vals: list = [state.value, time.time()]
        if order_id is not None:
            sets.append("order_id = ?")
            vals.append(order_id)
        if filled_qty is not None:
            sets.append("filled_qty = ?")
            vals.append(str(filled_qty))
        if avg_price is not None:
            sets.append("avg_price = ?")
            vals.append(str(avg_price))
        if error is not None:
            sets.append("error = ?")
            vals.append(error[:500])
        vals.append(link_id)
        self.conn.execute(
            f"UPDATE orders SET {', '.join(sets)} WHERE link_id = ?", vals
        )

    # ---- 参照 -----------------------------------------------------------
    def _row(self, r: sqlite3.Row) -> LedgerRow:
        return LedgerRow(
            link_id=r["link_id"], strategy=r["strategy"],
            category=r["category"], symbol=r["symbol"], side=r["side"],
            qty=Decimal(r["qty"]),
            price=Decimal(r["price"]) if r["price"] else None,
            purpose=r["purpose"] or "", state=State(r["state"]),
            order_id=r["order_id"],
            filled_qty=Decimal(r["filled_qty"] or "0"),
            avg_price=Decimal(r["avg_price"] or "0"),
            params=json.loads(r["params"] or "{}"),
            error=r["error"], created_at=r["created_at"],
            updated_at=r["updated_at"],
        )

    def get(self, link_id: str) -> LedgerRow | None:
        r = self.conn.execute(
            "SELECT * FROM orders WHERE link_id = ?", (link_id,)
        ).fetchone()
        return self._row(r) if r else None

    def unresolved(self) -> list[LedgerRow]:
        """結末が確定していない注文。起動時にこれを照会する。"""
        states = (State.INTENT.value, State.SENT.value, State.UNKNOWN.value)
        rows = self.conn.execute(
            f"SELECT * FROM orders WHERE state IN "
            f"({','.join('?' * len(states))}) ORDER BY created_at",
            states,
        ).fetchall()
        return [self._row(r) for r in rows]

    def open_orders(self) -> list[LedgerRow]:
        """板にあると台帳が考えている注文。"""
        states = (State.ACKED.value, State.PARTIAL.value)
        rows = self.conn.execute(
            f"SELECT * FROM orders WHERE state IN "
            f"({','.join('?' * len(states))})", states
        ).fetchall()
        return [self._row(r) for r in rows]

    def stats(self) -> dict[str, int]:
        rows = self.conn.execute(
            "SELECT state, COUNT(*) FROM orders GROUP BY state"
        ).fetchall()
        return {r[0]: r[1] for r in rows}

    def purge_old(self, days: float = 30.0) -> int:
        """終了状態の古い記録を削除する（台帳の肥大化を防ぐ）。"""
        cutoff = time.time() - days * 86400
        cur = self.conn.execute(
            "DELETE FROM orders WHERE created_at < ? AND state IN (?,?,?)",
            (cutoff, State.FILLED.value, State.CANCELLED.value,
             State.REJECTED.value),
        )
        return cur.rowcount


if __name__ == "__main__":
    from order_id import OrderIntent, id_by_hash

    ledger = OrderLedger(Path("data/orders_demo.sqlite"))
    intent = OrderIntent(strategy="demo", symbol="BTCUSDT", side="Buy",
                         qty=Decimal("0.001"), price=Decimal("100000"))
    link_id = id_by_hash(intent)

    print(f"link_id: {link_id}")
    print(f"1回目の記録: {ledger.record_intent(link_id, intent, {})}")
    print(f"2回目の記録: {ledger.record_intent(link_id, intent, {})}"
          f"  <- 既にあるのでFalse（冪等）")

    ledger.set_state(link_id, State.ACKED, order_id="mock-000001")
    row = ledger.get(link_id)
    print(f"\n{row.line()}")
    print(f"\n台帳の状態: {ledger.stats()}")

"""idempotent_router.py — 冪等な発注ルータ（write-ahead + 照会）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

発注の手順:
  1) orderLinkIdを決定論的に生成
  2) 台帳にINTENTとして記録（ここでクラッシュしても後で照会できる）
  3) APIに送信 -> SENT
  4) 結果を台帳に反映（ACKED / REJECTED / UNKNOWN）
  5) UNKNOWNなら照会して確定させる
"""
from __future__ import annotations

import time
from dataclasses import dataclass
from decimal import Decimal

import requests

from order_id import OrderIntent, id_by_hash
from order_ledger import LedgerRow, OrderLedger, State

# 「既に同じ注文がある」ことを示すretCode。冪等性が効いた証拠
DUPLICATE_CODES = {110072, 170141, 110030}
# 送ったが結果が不明になりうるretCode
AMBIGUOUS_CODES = {10000, 10016, 170007, 170146}
NETWORK_ERRORS = (requests.exceptions.Timeout,
                  requests.exceptions.ConnectionError,
                  requests.exceptions.ChunkedEncodingError)


@dataclass
class SubmitResult:
    link_id: str
    state: State
    order_id: str | None
    was_duplicate: bool = False        # 既存注文だった
    was_recovered: bool = False        # 照会で確定させた

    @property
    def accepted(self) -> bool:
        return self.state in (State.ACKED, State.PARTIAL, State.FILLED)

    def line(self) -> str:
        flags = []
        if self.was_duplicate:
            flags.append("重複検出")
        if self.was_recovered:
            flags.append("照会で確定")
        suffix = f"  ({' / '.join(flags)})" if flags else ""
        return f"{self.link_id} -> {self.state.value}{suffix}"


class IdempotentRouter:
    """二重発注が構造的に起きない発注ルータ。"""

    def __init__(self, http, ledger: OrderLedger,
                 id_fn=id_by_hash, resolve_attempts: int = 4):
        self.http = http
        self.ledger = ledger
        self.id_fn = id_fn
        self.resolve_attempts = resolve_attempts

    # ---- 発注 -----------------------------------------------------------
    def submit(self, intent: OrderIntent, order_type: str = "Limit",
               **extra) -> SubmitResult:
        link_id = self.id_fn(intent)

        params: dict = {
            "category": intent.category,
            "symbol": intent.symbol,
            "side": intent.side,
            "orderType": order_type,
            "qty": str(intent.qty),
            "orderLinkId": link_id,
        }
        if order_type == "Limit":
            if intent.price is None:
                raise ValueError("Limit注文にpriceがありません")
            params["price"] = str(intent.price)
        params.update(extra)

        # ---- (2) write-ahead: 送る前に記録する -------------------------
        is_new = self.ledger.record_intent(link_id, intent, params)
        if not is_new:
            existing = self.ledger.get(link_id)
            # 既に終わっている注文なら、それを返す（再送しない）
            if existing and existing.state.is_terminal:
                return SubmitResult(link_id, existing.state,
                                    existing.order_id, was_duplicate=True)
            # 結末が不明なら照会して確定させる
            if existing and existing.state.needs_resolution:
                return self._resolve_and_record(link_id, intent)
            if existing:
                return SubmitResult(link_id, existing.state,
                                    existing.order_id, was_duplicate=True)

        # ---- (3) 送信 ---------------------------------------------------
        self.ledger.set_state(link_id, State.SENT)

        try:
            res = self.http.place_order(**params)
        except NETWORK_ERRORS as e:
            # 結果不明。リトライせず照会する
            self.ledger.set_state(link_id, State.UNKNOWN,
                                  error=f"{type(e).__name__}: {e}")
            return self._resolve_and_record(link_id, intent)

        code = res.get("retCode")

        # ---- (4) 結果を反映 ----------------------------------------------
        if code == 0:
            order_id = res["result"].get("orderId")
            self.ledger.set_state(link_id, State.ACKED, order_id=order_id)
            return SubmitResult(link_id, State.ACKED, order_id)

        if code in DUPLICATE_CODES:
            # 取引所が二重発注を止めた。既存注文を照会する
            r = self._resolve_and_record(link_id, intent)
            r.was_duplicate = True
            return r

        if code in AMBIGUOUS_CODES:
            self.ledger.set_state(link_id, State.UNKNOWN,
                                  error=f"{code} {res.get('retMsg')}")
            return self._resolve_and_record(link_id, intent)

        # 明確な拒否（パラメータ誤り・残高不足など）
        self.ledger.set_state(link_id, State.REJECTED,
                              error=f"{code} {res.get('retMsg')}")
        return SubmitResult(link_id, State.REJECTED, None)

    # ---- 照会 -----------------------------------------------------------
    def resolve(self, link_id: str, symbol: str,
                category: str = "linear") -> tuple[State, str | None,
                                                   Decimal, Decimal]:
        """orderLinkIdから実際の状態を確定させる。

        NOT_FOUNDを1回で確定させないのが要点。取引所側の反映に
        遅延があり、直後の照会では見つからないことがある。
        """
        status_map = {
            "New": State.ACKED, "Untriggered": State.ACKED,
            "Triggered": State.ACKED,
            "PartiallyFilled": State.PARTIAL,
            "PartiallyFilledCanceled": State.PARTIAL,
            "Filled": State.FILLED,
            "Cancelled": State.CANCELLED, "Deactivated": State.CANCELLED,
            "Rejected": State.REJECTED,
        }

        delay = 0.5
        for attempt in range(1, self.resolve_attempts + 1):
            for fetch in (self.http.get_open_orders,
                          self.http.get_order_history):
                try:
                    res = fetch(category=category, symbol=symbol,
                                orderLinkId=link_id)
                except Exception:
                    continue
                rows = (res.get("result") or {}).get("list") or []
                if rows:
                    o = rows[0]
                    st = status_map.get(o.get("orderStatus", ""), State.ACKED)
                    return (st, o.get("orderId"),
                            Decimal(o.get("cumExecQty") or "0"),
                            Decimal(o.get("avgPrice") or "0"))

            if attempt < self.resolve_attempts:
                time.sleep(delay)
                delay *= 2

        # 何度確認しても存在しない = 届いていなかった
        return State.CANCELLED, None, Decimal(0), Decimal(0)

    def _resolve_and_record(self, link_id: str,
                            intent: OrderIntent) -> SubmitResult:
        st, order_id, filled, avg = self.resolve(
            link_id, intent.symbol, intent.category)
        self.ledger.set_state(link_id, st, order_id=order_id,
                              filled_qty=filled, avg_price=avg)
        return SubmitResult(link_id, st, order_id, was_recovered=True)


if __name__ == "__main__":
    from mock_exchange import MockExchange

    ledger = OrderLedger()
    ex = MockExchange(price=Decimal("100000"))
    router = IdempotentRouter(ex, ledger)

    intent = OrderIntent(strategy="idem-demo", symbol="BTCUSDT", side="Buy",
                         qty=Decimal("0.001"), price=Decimal("99000"))

    print("--- 同じ意図で3回発注してみる ---")
    for i in range(3):
        r = router.submit(intent, timeInForce="PostOnly")
        print(f"  {i + 1}回目: {r.line()}")

    n = len([o for o in ex.orders.values()])
    print(f"\n取引所に作られた注文数: {n}  <- 1件だけであるべき")
    print(f"台帳の状態: {ledger.stats()}")

"""reconciler.py — 起動時に台帳と取引所を突き合わせる
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

原則: 取引所の状態を常に正とする。台帳はそれに合わせて訂正する。
      差分が解消できないなら、botを起動させない。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal

from order_ledger import LedgerRow, OrderLedger, State


@dataclass
class ReconcileReport:
    resolved: list[str] = field(default_factory=list)      # 結末を確定した
    corrected: list[str] = field(default_factory=list)     # 状態を訂正した
    orphans: list[str] = field(default_factory=list)       # 台帳に無い注文
    stale: list[str] = field(default_factory=list)         # 台帳のみに存在
    errors: list[str] = field(default_factory=list)

    @property
    def safe_to_start(self) -> bool:
        """botを起動してよいか。エラーが残っていれば起動しない。"""
        return not self.errors

    def report(self) -> str:
        lines = [
            f"照会で確定 : {len(self.resolved)}件",
            f"状態を訂正 : {len(self.corrected)}件",
            f"孤児注文   : {len(self.orphans)}件",
            f"台帳のみ   : {len(self.stale)}件",
            f"エラー     : {len(self.errors)}件",
        ]
        for label, items in (("確定", self.resolved), ("訂正", self.corrected),
                             ("孤児", self.orphans), ("台帳のみ", self.stale),
                             ("エラー", self.errors)):
            for it in items[:10]:
                lines.append(f"  [{label}] {it}")
            if len(items) > 10:
                lines.append(f"  [{label}] ... 他 {len(items) - 10}件")
        return "\n".join(lines)


class Reconciler:
    def __init__(self, http, ledger: OrderLedger, router,
                 strategy_prefixes: set[str] | None = None):
        self.http = http
        self.ledger = ledger
        self.router = router
        # 自分の戦略が出した注文だけを対象にする（他botの注文を触らない）
        self.prefixes = strategy_prefixes or set()

    def _is_mine(self, link_id: str) -> bool:
        if not self.prefixes:
            return True
        return any(link_id.startswith(p) for p in self.prefixes)

    def _exchange_open_orders(self) -> dict[str, dict]:
        """取引所側の未約定注文を link_id -> 生データ で返す。"""
        out: dict[str, dict] = {}
        for category, extra in (("linear", {"settleCoin": "USDT"}),
                                ("linear", {"settleCoin": "USDC"}),
                                ("inverse", {"settleCoin": "BTC"}),
                                ("spot", {})):
            cursor = None
            while True:
                kw: dict = {"category": category, "limit": 50, **extra}
                if cursor:
                    kw["cursor"] = cursor
                res = self.http.get_open_orders(**kw)
                if res.get("retCode") != 0:
                    break
                for o in res["result"]["list"]:
                    lid = o.get("orderLinkId") or ""
                    o["_category"] = category
                    out[lid] = o
                cursor = res["result"].get("nextPageCursor")
                if not cursor:
                    break
        return out

    def run(self, cancel_orphans: bool = False) -> ReconcileReport:
        rep = ReconcileReport()

        # ---- 1) 結末不明の注文を照会して確定させる ----------------------
        for row in self.ledger.unresolved():
            if not self._is_mine(row.link_id):
                continue
            try:
                st, order_id, filled, avg = self.router.resolve(
                    row.link_id, row.symbol, row.category)
                self.ledger.set_state(row.link_id, st, order_id=order_id,
                                      filled_qty=filled, avg_price=avg)
                rep.resolved.append(
                    f"{row.link_id} {row.state.value} -> {st.value}"
                    + (f" (約定 {filled})" if filled > 0 else "")
                )
            except Exception as e:
                rep.errors.append(f"{row.link_id}: 照会失敗 {e}")

        # ---- 2) 取引所の未約定注文と台帳を突き合わせる ------------------
        exchange = self._exchange_open_orders()
        ledger_open = {r.link_id: r for r in self.ledger.open_orders()}

        # 取引所にあるが台帳に無い = 孤児注文
        for lid, o in exchange.items():
            if lid in ledger_open:
                continue
            if not self._is_mine(lid):
                continue        # 他botや手動の注文は触らない
            existing = self.ledger.get(lid)
            if existing and existing.state.is_terminal:
                rep.corrected.append(
                    f"{lid}: 台帳は{existing.state.value}だが板に存在 -> ACKED"
                )
                self.ledger.set_state(lid, State.ACKED,
                                      order_id=o.get("orderId"))
                continue

            rep.orphans.append(
                f"{lid} {o.get('symbol')} {o.get('side')} "
                f"{o.get('qty')}@{o.get('price')}"
            )
            if cancel_orphans:
                r = self.http.cancel_order(
                    category=o["_category"], symbol=o["symbol"],
                    orderLinkId=lid)
                if r.get("retCode") in (0, 110001, 110008, 110010):
                    rep.corrected.append(f"{lid}: 孤児注文をキャンセル")

        # 台帳にあるが取引所に無い = 既に終わっている
        for lid, row in ledger_open.items():
            if lid in exchange:
                continue
            try:
                st, order_id, filled, avg = self.router.resolve(
                    lid, row.symbol, row.category)
                self.ledger.set_state(lid, st, order_id=order_id,
                                      filled_qty=filled, avg_price=avg)
                rep.stale.append(f"{lid}: 板に無い -> {st.value}")
            except Exception as e:
                rep.errors.append(f"{lid}: 照会失敗 {e}")

        return rep


def startup_reconcile(client, ledger: OrderLedger, router,
                      strategy_prefixes: set[str] | None = None,
                      cancel_orphans: bool = False,
                      require_clean: bool = True) -> ReconcileReport:
    """bot起動シーケンスの最初に必ず呼ぶ。"""
    print("=" * 60)
    print("起動時リコンサイルを実行します")
    print("=" * 60)

    rec = Reconciler(client.http, ledger, router, strategy_prefixes)
    rep = rec.run(cancel_orphans=cancel_orphans)
    print(rep.report())

    if require_clean and not rep.safe_to_start:
        raise RuntimeError(
            "リコンサイルでエラーが残りました。botを起動しません。\n"
            "手動で状態を確認してください。"
        )
    if rep.orphans and not cancel_orphans:
        print("\n[警告] 孤児注文があります。"
              "cancel_orphans=True で自動キャンセルできます")

    print("\nリコンサイル完了。稼働を開始できます")
    return rep


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

    c = make_client()
    ledger = OrderLedger()
    router = IdempotentRouter(c.http, ledger)

    startup_reconcile(c, ledger, router,
                      strategy_prefixes={"trendfol", "gridbot"},
                      cancel_orphans=False, require_clean=True)

"""ledger_audit.py — 台帳の健全性を点検する（定期実行を推奨）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time
from dataclasses import dataclass

from order_ledger import OrderLedger, State

# この時間を超えて未確定のままなら異常
STUCK_THRESHOLD_SEC = 300.0


@dataclass
class AuditFinding:
    level: str
    link_id: str
    message: str

    def line(self) -> str:
        icon = {"info": "  ", "warn": "! ", "error": "!!"}.get(self.level, "  ")
        return f"{icon} {self.link_id:<38} {self.message}"


def audit_ledger(ledger: OrderLedger) -> list[AuditFinding]:
    findings: list[AuditFinding] = []
    now = time.time()

    # 1) 長時間 INTENT / SENT / UNKNOWN のまま = 照会が回っていない
    for row in ledger.unresolved():
        age = now - row.created_at
        if age > STUCK_THRESHOLD_SEC:
            findings.append(AuditFinding(
                "error", row.link_id,
                f"{row.state.value} のまま {age / 60:.0f}分経過。"
                f"照会処理が動いていない可能性"
            ))

    # 2) 長時間 ACKED のまま = 板に置きっぱなし
    for row in ledger.open_orders():
        age = now - row.created_at
        if age > 3600 * 6:
            findings.append(AuditFinding(
                "warn", row.link_id,
                f"{age / 3600:.1f}時間 板に残っています。"
                f"注文の寿命管理を検討（記事13）"
            ))

    # 3) 部分約定のまま放置
    for row in ledger.open_orders():
        if row.state is State.PARTIAL:
            remaining = row.qty - row.filled_qty
            findings.append(AuditFinding(
                "warn", row.link_id,
                f"部分約定（{row.filled_qty}/{row.qty}、残 {remaining}）"
            ))

    # 4) エラーが記録されている注文
    stats = ledger.stats()
    rejected = stats.get(State.REJECTED.value, 0)
    if rejected > 0:
        findings.append(AuditFinding(
            "info", "(集計)",
            f"拒否された注文が通算 {rejected}件。"
            f"多い場合は発注前検証を見直す（記事13）"
        ))

    return findings


def print_audit(ledger: OrderLedger) -> None:
    print(f"台帳の状態: {ledger.stats()}\n")
    findings = audit_ledger(ledger)
    if not findings:
        print("問題は検出されませんでした")
        return
    for f in findings:
        print(f.line())

    errors = [f for f in findings if f.level == "error"]
    if errors:
        print(f"\n[!!] error が {len(errors)}件あります。"
              f"リコンサイルを実行してください")


if __name__ == "__main__":
    print_audit(OrderLedger())

"""test_idempotency.py — 二重発注が起きないことを検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

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

from decimal import Decimal
from pathlib import Path

import pytest
import requests

from idempotent_router import IdempotentRouter
from mock_exchange import MockExchange, err
from order_id import OrderIntent, id_by_hash, id_by_time_bucket
from order_ledger import OrderLedger, State


@pytest.fixture
def env(tmp_path: Path):
    ledger = OrderLedger(tmp_path / "orders.sqlite")
    ex = MockExchange(price=Decimal("100000"))
    router = IdempotentRouter(ex, ledger, resolve_attempts=2)
    return ex, ledger, router


@pytest.fixture
def intent() -> OrderIntent:
    return OrderIntent(strategy="test", symbol="BTCUSDT", side="Buy",
                       qty=Decimal("0.001"), price=Decimal("99000"))


# ---- ID生成 -------------------------------------------------------------
def test_same_intent_same_id(intent):
    assert id_by_hash(intent) == id_by_hash(intent)


def test_different_price_different_id(intent):
    other = OrderIntent(strategy="test", symbol="BTCUSDT", side="Buy",
                        qty=Decimal("0.001"), price=Decimal("99001"))
    assert id_by_hash(intent) != id_by_hash(other)


def test_id_length_and_charset(intent):
    lid = id_by_hash(intent)
    assert len(lid) <= 36
    assert all(c.isalnum() or c in "-_" for c in lid)


def test_time_bucket_stable_within_window(intent):
    a = id_by_time_bucket(intent, 60, now=1785000000.0)
    b = id_by_time_bucket(intent, 60, now=1785000059.0)
    c = id_by_time_bucket(intent, 60, now=1785000061.0)
    assert a == b            # 同じ足なら同じ
    assert a != c            # 次の足では変わる


# ---- 二重発注の防止 -----------------------------------------------------
def test_repeated_submit_creates_one_order(env, intent):
    """同じ意図で何度submitしても、取引所には1件しか作られないこと。"""
    ex, ledger, router = env
    for _ in range(5):
        router.submit(intent, timeInForce="PostOnly")
    assert len(ex.orders) == 1


def test_duplicate_flagged(env, intent):
    ex, ledger, router = env
    first = router.submit(intent)
    second = router.submit(intent)
    assert first.was_duplicate is False
    assert second.was_duplicate is True
    assert second.link_id == first.link_id


def test_timeout_does_not_duplicate(env, intent, monkeypatch):
    """タイムアウト後も再送されず、照会で確定すること。"""
    ex, ledger, router = env

    calls = {"n": 0}
    original = ex.place_order

    def flaky(**kw):
        calls["n"] += 1
        # 1回目: 取引所には届くが、レスポンスでタイムアウトする
        result = original(**kw)
        if calls["n"] == 1:
            raise requests.exceptions.Timeout("simulated")
        return result

    monkeypatch.setattr(ex, "place_order", flaky)

    result = router.submit(intent, timeInForce="PostOnly")

    # 取引所には1件だけ存在し、照会で状態が確定していること
    assert len(ex.orders) == 1
    assert result.was_recovered is True
    assert result.state in (State.ACKED, State.PARTIAL, State.FILLED)


def test_ledger_records_before_send(env, intent, monkeypatch):
    """送信前に台帳へ記録されていること（write-ahead）。"""
    ex, ledger, router = env

    def boom(**kw):
        # 送信時に必ず例外 -> 台帳にはINTENT/SENTが残っているはず
        raise requests.exceptions.ConnectionError("network down")

    monkeypatch.setattr(ex, "place_order", boom)
    link_id = id_by_hash(intent)

    router.submit(intent)
    row = ledger.get(link_id)
    assert row is not None                      # 記録が残っている
    assert row.params.get("symbol") == "BTCUSDT"   # 何を送ろうとしたか分かる


def test_rejected_order_not_retried(env, intent):
    """明確な拒否（残高不足など）は再送しないこと。"""
    ex, ledger, router = env
    ex.inject_error["place_order"] = err(110007,
                                         "Available balance insufficient")
    r = router.submit(intent)
    assert r.state is State.REJECTED
    assert r.accepted is False
    assert len(ex.orders) == 0


def test_terminal_state_not_resubmitted(env, intent):
    """既に約定済みの意図を再submitしても発注されないこと。"""
    ex, ledger, router = env
    link_id = id_by_hash(intent)
    router.submit(intent)
    ledger.set_state(link_id, State.FILLED, filled_qty=intent.qty)

    before = len(ex.orders)
    r = router.submit(intent)
    assert len(ex.orders) == before          # 増えていない
    assert r.state is State.FILLED
    assert r.was_duplicate is True


# ---- リコンサイル -------------------------------------------------------
def test_reconcile_resolves_unknown(env, intent):
    ex, ledger, router = env
    from reconciler import Reconciler

    # 送信済みだが結末不明の状態を作る
    link_id = id_by_hash(intent)
    ledger.record_intent(link_id, intent, {"symbol": "BTCUSDT"})
    ledger.set_state(link_id, State.UNKNOWN)
    ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                   orderType="Limit", qty="0.001", price="99000",
                   orderLinkId=link_id)

    rep = Reconciler(ex, ledger, router).run()
    assert len(rep.resolved) == 1
    assert ledger.get(link_id).state is State.ACKED


def test_reconcile_detects_orphan(env):
    ex, ledger, router = env
    from reconciler import Reconciler

    # 台帳に無い注文を取引所に作る
    ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                   orderType="Limit", qty="0.001", price="98000",
                   orderLinkId="test-orphan-001")

    rep = Reconciler(ex, ledger, router, {"test-"}).run()
    assert len(rep.orphans) == 1


def test_reconcile_ignores_other_strategies(env):
    """他botの注文は孤児として扱わないこと。"""
    ex, ledger, router = env
    from reconciler import Reconciler

    ex.place_order(category="linear", symbol="BTCUSDT", side="Buy",
                   orderType="Limit", qty="0.001", price="98000",
                   orderLinkId="otherbot-999")

    rep = Reconciler(ex, ledger, router, {"mybot-"}).run()
    assert len(rep.orphans) == 0

"""ledger_sync.py — WebSocketのorderイベントで台帳を即時更新する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import threading
import time
from decimal import Decimal

from order_ledger import OrderLedger, State

STATUS_MAP = {
    "New": State.ACKED,
    "Untriggered": State.ACKED,
    "Triggered": State.ACKED,
    "PartiallyFilled": State.PARTIAL,
    "PartiallyFilledCanceled": State.PARTIAL,
    "Filled": State.FILLED,
    "Cancelled": State.CANCELLED,
    "Deactivated": State.CANCELLED,
    "Rejected": State.REJECTED,
}


class LedgerSync:
    """orderイベントを台帳に反映し、RESTで定期的に取りこぼしを補う。"""

    def __init__(self, client, ledger: OrderLedger, router,
                 strategy_prefixes: set[str] | None = None,
                 audit_sec: float = 60.0):
        self.c = client
        self.ledger = ledger
        self.router = router
        self.prefixes = strategy_prefixes or set()
        self.audit_sec = audit_sec
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self.events = 0
        self.unknown_ids = 0
        self.ws = None

    def _is_mine(self, link_id: str) -> bool:
        if not self.prefixes:
            return True
        return any(link_id.startswith(p) for p in self.prefixes)

    # ---- WebSocket ------------------------------------------------------
    def on_order(self, msg: dict) -> None:
        with self._lock:
            for o in msg.get("data", []):
                link_id = o.get("orderLinkId") or ""
                if not link_id or not self._is_mine(link_id):
                    continue

                status = o.get("orderStatus", "")
                state = STATUS_MAP.get(status)
                if state is None:
                    continue

                row = self.ledger.get(link_id)
                if row is None:
                    # 台帳に無い = 孤児。記録だけ残して後で調べる
                    self.unknown_ids += 1
                    print(f"[sync] 台帳に無い注文のイベント: {link_id} "
                          f"({status})")
                    continue

                self.ledger.set_state(
                    link_id, state,
                    order_id=o.get("orderId"),
                    filled_qty=Decimal(o.get("cumExecQty") or "0"),
                    avg_price=Decimal(o.get("avgPrice") or "0"),
                )
                self.events += 1

                if state is State.FILLED:
                    print(f"[sync] 約定: {link_id} "
                          f"{o.get('cumExecQty')}@{o.get('avgPrice')}")

    # ---- 定期監査 --------------------------------------------------------
    def _audit_loop(self) -> None:
        from ledger_audit import audit_ledger

        while not self._stop.wait(self.audit_sec):
            try:
                # WSで取りこぼした未確定注文をRESTで確定させる
                for row in self.ledger.unresolved():
                    if not self._is_mine(row.link_id):
                        continue
                    if time.time() - row.created_at < 30:
                        continue        # まだ新しいので待つ
                    st, oid, filled, avg = self.router.resolve(
                        row.link_id, row.symbol, row.category)
                    self.ledger.set_state(row.link_id, st, order_id=oid,
                                          filled_qty=filled, avg_price=avg)
                    print(f"[sync] RESTで確定: {row.link_id} -> {st.value}")

                for f in audit_ledger(self.ledger):
                    if f.level == "error":
                        print(f"[sync] {f.line()}")
            except Exception as e:
                print(f"[sync] 監査に失敗: {type(e).__name__}: {e}")

    # ---- 起動・停止 ------------------------------------------------------
    def start(self) -> None:
        from pybit.unified_trading import WebSocket

        env = self.c.env.value
        self.ws = WebSocket(
            channel_type="private",
            testnet=(env == "testnet"),
            demo=(env == "demo"),
            api_key=self.c.cred.api_key,
            api_secret=self.c.cred.api_secret,
        )
        self.ws.order_stream(callback=self.on_order)

        threading.Thread(target=self._audit_loop, daemon=True,
                         name="ledger-audit").start()
        print("[sync] 台帳同期を開始しました")

    def stop(self) -> None:
        self._stop.set()
        if self.ws:
            try:
                self.ws.exit()
            except Exception:
                pass


if __name__ == "__main__":
    from client import make_client
    from idempotent_router import IdempotentRouter
    from reconciler import startup_reconcile

    c = make_client()
    ledger = OrderLedger()
    router = IdempotentRouter(c.http, ledger)

    # 1) 起動時リコンサイル（これを飛ばしてはいけない）
    startup_reconcile(c, ledger, router, cancel_orphans=False)

    # 2) WebSocket同期を開始
    sync = LedgerSync(c, ledger, router)
    sync.start()

    try:
        while True:
            time.sleep(30)
            print(f"[sync] イベント {sync.events} / "
                  f"台帳 {ledger.stats()}")
    except KeyboardInterrupt:
        sync.stop()

"""bot_startup.py — 冪等性を担保したbot起動シーケンス
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

この順序を守れば、何度再起動しても二重発注は起きない。
"""
from __future__ import annotations

import signal
import sys
import time

from client import make_client
from idempotent_router import IdempotentRouter
from ledger_sync import LedgerSync
from order_ledger import OrderLedger
from position_keeper import PositionKeeper
from reconciler import startup_reconcile

STRATEGY_PREFIXES = {"trendfol", "gridbot"}


def main() -> int:
    c = make_client()
    ledger = OrderLedger()
    router = IdempotentRouter(c.http, ledger)

    # ---- 1) 注文のリコンサイル（記事17） -----------------------------
    #      これを飛ばすと、前回の未確定注文が二重発注の種になる
    try:
        startup_reconcile(c, ledger, router,
                          strategy_prefixes=STRATEGY_PREFIXES,
                          cancel_orphans=False, require_clean=True)
    except RuntimeError as e:
        print(f"[起動中止] {e}")
        return 1

    # ---- 2) ポジションのリコンサイル（記事20） -------------------------
    #      WSは「今から先」しか来ないので、RESTで初期状態を確定させる
    keeper = PositionKeeper(c, reconcile_sec=30.0)
    keeper.start()
    print("\n現在のポジション:")
    for p in keeper.all():
        print(f"  {p.line()}")

    # ---- 3) 台帳のWebSocket同期を開始 --------------------------------
    sync = LedgerSync(c, ledger, router, STRATEGY_PREFIXES)
    sync.start()

    # ---- 4) graceful shutdown の登録（記事30） ------------------------
    stopping = {"flag": False}

    def on_signal(signum, frame) -> None:
        print(f"\nシグナル {signum} を受信。停止します")
        stopping["flag"] = True

    signal.signal(signal.SIGINT, on_signal)
    signal.signal(signal.SIGTERM, on_signal)

    # ---- 5) 戦略の稼働 ------------------------------------------------
    print("\n" + "=" * 60)
    print("稼働開始")
    print("=" * 60)
    try:
        while not stopping["flag"]:
            time.sleep(1)
            # ここに戦略のループを置く（記事30のアーキテクチャ参照）
    finally:
        print("\n停止処理:")
        sync.stop()
        keeper.stop()
        # 未確定注文を最後に確定させておくと、次回起動が楽になる
        for row in ledger.unresolved():
            st, oid, filled, avg = router.resolve(row.link_id, row.symbol,
                                                  row.category)
            ledger.set_state(row.link_id, st, order_id=oid,
                             filled_qty=filled, avg_price=avg)
        print(f"  台帳の最終状態: {ledger.stats()}")
        print("  停止完了")
    return 0


if __name__ == "__main__":
    sys.exit(main())