Bybit APIで成行・指値注文を出すPython実装
place_order全パラメータと
TIF（GTC/IOC/FOK/PostOnly）の使い分け

# 本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
# 必ず testnet (BYBIT_ENV=testnet) で実行してください。
from client import make_client

c = make_client()

res = c.http.place_order(
    category="linear",      # linear / inverse / spot / option
    symbol="BTCUSDT",
    side="Buy",             # Buy / Sell
    orderType="Market",     # Market / Limit
    qty="0.001",            # 文字列で渡す。数値型はエラーになることがある
)

print(res["retCode"], res["retMsg"])
print("orderId     :", res["result"]["orderId"])
print("orderLinkId :", res["result"]["orderLinkId"])

"""post_only.py — PostOnly指値を、拒否されたら1ティックずらして再試行する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from decimal import Decimal

from client import make_client
from symbol_rules import fetch_rules, round_to_step

# PostOnlyが板を食ってしまう場合の拒否理由（rejectReason）
POST_ONLY_REJECT = "EC_PostOnlyWillTakeLiquidity"


def place_post_only(http, symbol: str, side: str, qty: Decimal,
                    category: str = "linear", max_retry: int = 5) -> dict | None:
    """必ずメイカーになる指値を置く。板に届かない場合は1ティックずつ退避する。"""
    rules = fetch_rules(http, symbol, category)
    tick = rules.tick_size

    for attempt in range(max_retry):
        ob = http.get_orderbook(category=category, symbol=symbol, limit=1)["result"]
        best_bid = Decimal(ob["b"][0][0])
        best_ask = Decimal(ob["a"][0][0])

        # 買いはベストビッド、売りはベストアスクに置く。
        # 再試行ごとに、板を食わない方向へ tick 分だけ退避させる。
        if side == "Buy":
            price = best_bid - tick * attempt
        else:
            price = best_ask + tick * attempt
        price = round_to_step(price, tick)

        res = http.place_order(
            category=category,
            symbol=symbol,
            side=side,
            orderType="Limit",
            qty=str(qty),
            price=str(price),
            timeInForce="PostOnly",
        )

        if res["retCode"] == 0:
            print(f"[OK] PostOnly受理 price={price} (試行{attempt + 1}回目)")
            return res["result"]

        # 拒否理由がPostOnly由来なら価格を変えて再試行、それ以外は諦める
        msg = res.get("retMsg", "")
        if POST_ONLY_REJECT in msg or "post only" in msg.lower():
            print(f"[retry] 板を食うため拒否 price={price} -> ずらして再試行")
            continue
        raise RuntimeError(f"発注失敗: {res['retCode']} {msg}")

    print("[NG] PostOnlyを置けませんでした（板の変動が速すぎる）")
    return None


if __name__ == "__main__":
    c = make_client()
    place_post_only(c.http, "BTCUSDT", "Buy", Decimal("0.001"))

"""order_router.py — 検証・丸め・冪等ID付与をまとめた発注ラッパ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

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

from symbol_rules import SymbolRules, fetch_rules, floor_to_step, round_to_step


class OrderRejected(ValueError):
    """発注前の自前検証で弾いた（APIには送っていない）"""


class OrderFailed(RuntimeError):
    """APIがretCode!=0を返した"""

    def __init__(self, ret_code: int, ret_msg: str):
        super().__init__(f"retCode={ret_code} retMsg={ret_msg}")
        self.ret_code = ret_code
        self.ret_msg = ret_msg


@dataclass
class OrderRequest:
    symbol: str
    side: str                       # Buy / Sell
    qty: Decimal
    order_type: str = "Limit"       # Limit / Market
    price: Decimal | None = None
    category: str = "linear"
    time_in_force: str | None = None
    reduce_only: bool = False
    position_idx: int | None = None
    tag: str = "manual"             # 戦略名。orderLinkIdの先頭に入れて後で集計できる
    extra: dict = field(default_factory=dict)


class OrderRouter:
    """発注の単一窓口。ルールキャッシュを持ち、検証を通してから送る。"""

    def __init__(self, http, price_limit_check: bool = True):
        self.http = http
        self._rules: dict[tuple[str, str], SymbolRules] = {}
        self.price_limit_check = price_limit_check

    # ---- 銘柄仕様のキャッシュ ------------------------------------------
    def rules(self, symbol: str, category: str) -> SymbolRules:
        key = (category, symbol)
        if key not in self._rules:
            self._rules[key] = fetch_rules(self.http, symbol, category)
        return self._rules[key]

    # ---- 冪等ID --------------------------------------------------------
    @staticmethod
    def make_link_id(tag: str) -> str:
        """orderLinkIdは36文字以内・ユニーク。先頭に戦略タグを入れて後で集計する。
        完全な決定論的IDの作り方は記事17で扱う。
        """
        return f"{tag[:8]}-{int(time.time() * 1000)}-{uuid.uuid4().hex[:6]}"[:36]

    # ---- 検証 ----------------------------------------------------------
    def _validate(self, req: OrderRequest) -> tuple[Decimal, Decimal | None]:
        r = self.rules(req.symbol, req.category)

        qty = floor_to_step(req.qty, r.qty_step)
        if qty <= 0:
            raise OrderRejected(
                f"{req.symbol}: 数量が刻み幅{r.qty_step}で丸めるとゼロになります"
                f"（元の数量 {req.qty}）"
            )
        if qty < r.min_qty:
            raise OrderRejected(f"{req.symbol}: 数量{qty} < 最小{r.min_qty}")
        if r.max_qty > 0 and qty > r.max_qty:
            raise OrderRejected(f"{req.symbol}: 数量{qty} > 最大{r.max_qty}")

        price = None
        if req.order_type == "Limit":
            if req.price is None:
                raise OrderRejected("Limit注文にpriceがありません")
            price = round_to_step(req.price, r.tick_size)
            if price <= 0:
                raise OrderRejected(f"価格が不正です: {price}")

            if r.min_notional > 0 and price * qty < r.min_notional:
                raise OrderRejected(
                    f"{req.symbol}: 想定元本{price * qty} < 最小{r.min_notional}"
                )
            if self.price_limit_check:
                self._check_price_limit(req, price)
        return qty, price

    def _check_price_limit(self, req: OrderRequest, price: Decimal) -> None:
        """現在価格から大きく外れた指値を、APIに送る前に弾く（桁ミス対策）。"""
        if req.category not in ("linear", "inverse"):
            return
        res = self.http.get_price_limit(category=req.category, symbol=req.symbol)
        if res.get("retCode") != 0:
            return   # 取得できない場合は検証をスキップ（発注自体は止めない）
        lim = res["result"]
        lo, hi = Decimal(lim["buyLmt"]), Decimal(lim["sellLmt"])
        if not (min(lo, hi) <= price <= max(lo, hi)):
            raise OrderRejected(
                f"{req.symbol}: 指値{price}が許容レンジ[{lo}, {hi}]の外です"
            )

    # ---- 発注 ----------------------------------------------------------
    def submit(self, req: OrderRequest) -> dict:
        qty, price = self._validate(req)

        params: dict = {
            "category": req.category,
            "symbol": req.symbol,
            "side": req.side,
            "orderType": req.order_type,
            "qty": str(qty),
            "orderLinkId": self.make_link_id(req.tag),
        }
        if price is not None:
            params["price"] = str(price)
        if req.time_in_force:
            params["timeInForce"] = req.time_in_force
        if req.reduce_only:
            params["reduceOnly"] = True
        if req.position_idx is not None:
            params["positionIdx"] = req.position_idx
        params.update(req.extra)

        res = self.http.place_order(**params)
        if res["retCode"] != 0:
            raise OrderFailed(res["retCode"], res["retMsg"])

        out = res["result"]
        out["_sent"] = params      # 何を送ったかを残す（障害調査で必須）
        return out


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

    c = make_client()
    router = OrderRouter(c.http)

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

    # 現在価格の2%下に、メイカー確定の買い指値
    result = router.submit(OrderRequest(
        symbol="BTCUSDT",
        side="Buy",
        qty=Decimal("0.001"),
        order_type="Limit",
        price=last * Decimal("0.98"),
        time_in_force="PostOnly",
        tag="demo",
    ))
    print(result["orderId"], result["orderLinkId"])

    # 検証で弾かれる例（小さすぎる数量）
    try:
        router.submit(OrderRequest(symbol="BTCUSDT", side="Buy",
                                   qty=Decimal("0.0000001"), order_type="Market"))
    except OrderRejected as e:
        print("想定どおり弾かれました:", e)

"""spot_market.py — 現物成行を「USDT金額指定」で出す（DCA積立の基本形）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from client import make_client

c = make_client()

# 100 USDT分のBTCを成行で買う。
# marketUnit="quoteCoin" にすると qty は「USDTの金額」として解釈される。
res = c.http.place_order(
    category="spot",
    symbol="BTCUSDT",
    side="Buy",
    orderType="Market",
    qty="100",                 # 100 USDT（BTCの数量ではない）
    marketUnit="quoteCoin",
    orderLinkId="dca-20260730-01",
)
print(res["retCode"], res["result"])

# 売りの場合、baseCoin指定なら「保有BTCを何枚売るか」になる
res = c.http.place_order(
    category="spot",
    symbol="BTCUSDT",
    side="Sell",
    orderType="Market",
    qty="0.001",               # 0.001 BTC
    marketUnit="baseCoin",
)
print(res["retCode"], res["result"])

"""market_with_guard.py — スリッページ上限付きの成行と、IOC指値による代替
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from decimal import Decimal

from client import make_client
from symbol_rules import fetch_rules, round_to_step


def market_with_slippage_cap(http, symbol: str, side: str, qty: str,
                             pct: str = "0.1", category: str = "linear") -> dict:
    """許容スリッページを%で指定した成行。超える場合は約定しない。"""
    return http.place_order(
        category=category,
        symbol=symbol,
        side=side,
        orderType="Market",
        qty=qty,
        slippageToleranceType="Percent",   # TickSize も指定可
        slippageTolerance=pct,             # "0.1" = 0.1%
    )


def ioc_limit_as_safe_market(http, symbol: str, side: str, qty: Decimal,
                             max_slip_pct: Decimal = Decimal("0.1"),
                             category: str = "linear") -> dict:
    """IOC指値で「価格上限付きの成行」を自作する。
    どの取引所でも使える汎用手法で、約定価格の上限を自分で完全に制御できる。
    """
    rules = fetch_rules(http, symbol, category)
    ob = http.get_orderbook(category=category, symbol=symbol, limit=1)["result"]

    if side == "Buy":
        ref = Decimal(ob["a"][0][0])                       # ベストアスク
        cap = ref * (1 + max_slip_pct / 100)               # これ以上では買わない
    else:
        ref = Decimal(ob["b"][0][0])                       # ベストビッド
        cap = ref * (1 - max_slip_pct / 100)               # これ以下では売らない

    return http.place_order(
        category=category,
        symbol=symbol,
        side=side,
        orderType="Limit",
        qty=str(qty),
        price=str(round_to_step(cap, rules.tick_size)),
        timeInForce="IOC",     # 上限内で約定できる分だけ約定、残りは即キャンセル
    )


if __name__ == "__main__":
    c = make_client()
    print(market_with_slippage_cap(c.http, "BTCUSDT", "Buy", "0.001", pct="0.1"))
    print(ioc_limit_as_safe_market(c.http, "BTCUSDT", "Buy", Decimal("0.001")))

"""amend.py — 指値をベストビッド/アスクに追従させる（キャンセル＋再発注を避ける）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time
from decimal import Decimal

from client import make_client
from symbol_rules import fetch_rules, round_to_step


def chase_best_price(http, symbol: str, order_link_id: str, side: str,
                     category: str = "linear", interval: float = 1.0,
                     max_loops: int = 30) -> None:
    """自分の指値を、常にベスト気配に張り付かせる。"""
    rules = fetch_rules(http, symbol, category)
    last_price: Decimal | None = None

    for _ in range(max_loops):
        # 1) 自分の注文がまだ生きているか確認
        res = http.get_open_orders(category=category, symbol=symbol,
                                   orderLinkId=order_link_id)
        orders = res["result"]["list"]
        if not orders:
            print("注文が消えました（約定またはキャンセル済み）")
            return

        # 2) 目標価格をベスト気配から計算
        ob = http.get_orderbook(category=category, symbol=symbol, limit=1)["result"]
        target = Decimal(ob["b"][0][0]) if side == "Buy" else Decimal(ob["a"][0][0])
        target = round_to_step(target, rules.tick_size)

        # 3) 変わっていなければ何もしない（無駄なリクエストを打たない）
        if target == last_price:
            time.sleep(interval)
            continue

        r = http.amend_order(
            category=category,
            symbol=symbol,
            orderLinkId=order_link_id,   # orderIdでもよい
            price=str(target),
        )
        if r["retCode"] == 0:
            print(f"[amend] price -> {target}")
            last_price = target
        elif r["retCode"] in (110001, 110008):
            # 110001 注文が存在しない / 110008 完了済みまたはキャンセル済み
            print(f"[end] 注文は終了状態です: {r['retMsg']}")
            return
        else:
            print(f"[warn] amend失敗: {r['retCode']} {r['retMsg']}")

        time.sleep(interval)


if __name__ == "__main__":
    c = make_client()
    chase_best_price(c.http, "BTCUSDT", order_link_id="demo-chase-01", side="Buy")

"""order_ttl.py — 一定時間で未約定注文を自動キャンセルする（注文の寿命管理）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time

from client import make_client


def cancel_stale_orders(http, category: str = "linear", settle_coin: str = "USDT",
                        ttl_sec: int = 300, tag_prefix: str | None = None) -> int:
    """createdTimeがttl_secより古い未約定注文をキャンセルする。

    tag_prefix を指定すると、その戦略が出した注文だけを対象にできる
    （orderLinkId の先頭に戦略名を入れておく前提。サンプル3参照）。
    """
    res = http.get_open_orders(category=category, settleCoin=settle_coin, limit=50)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_open_orders失敗: {res['retMsg']}")

    now_ms = int(time.time() * 1000)
    cancelled = 0

    for o in res["result"]["list"]:
        age_sec = (now_ms - int(o["createdTime"])) / 1000
        if age_sec < ttl_sec:
            continue
        if tag_prefix and not o.get("orderLinkId", "").startswith(tag_prefix):
            continue

        r = http.cancel_order(
            category=category, symbol=o["symbol"], orderId=o["orderId"]
        )
        if r["retCode"] == 0:
            print(f"[cancel] {o['symbol']} {o['side']} {o['price']} "
                  f"({age_sec:.0f}秒経過)")
            cancelled += 1
        elif r["retCode"] in (110001, 110008, 110010):
            # 既に約定/キャンセル済み。競合状態なので警告不要
            pass
        else:
            print(f"[warn] cancel失敗 {o['orderId']}: {r['retCode']} {r['retMsg']}")
    return cancelled


def cancel_all(http, category: str = "linear", symbol: str | None = None,
               settle_coin: str = "USDT") -> dict:
    """全キャンセル。緊急時はこちら（1リクエストで済む）。"""
    params = {"category": category}
    if symbol:
        params["symbol"] = symbol
    else:
        params["settleCoin"] = settle_coin
    return http.cancel_all_orders(**params)


if __name__ == "__main__":
    c = make_client()
    n = cancel_stale_orders(c.http, ttl_sec=300, tag_prefix="demo")
    print(f"{n}件キャンセルしました")

"""close_position.py — reduceOnlyによる安全な決済（全決済・部分決済）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from decimal import Decimal

from client import make_client
from symbol_rules import fetch_rules, floor_to_step


def close_position(http, symbol: str, category: str = "linear",
                   ratio: Decimal = Decimal("1.0"), use_market: bool = True) -> dict | None:
    """保有ポジションを ratio の割合だけ決済する。ratio=1.0 で全決済。"""
    res = http.get_positions(category=category, symbol=symbol)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_positions失敗: {res['retMsg']}")

    positions = [p for p in res["result"]["list"] if Decimal(p["size"]) > 0]
    if not positions:
        print(f"{symbol}: ポジションなし")
        return None

    pos = positions[0]
    size = Decimal(pos["size"])
    rules = fetch_rules(http, symbol, category)

    qty = floor_to_step(size * ratio, rules.qty_step)
    if qty <= 0:
        print(f"{symbol}: 決済数量が刻み幅未満です（size={size} ratio={ratio}）")
        return None

    # 保有と逆方向に出す
    side = "Sell" if pos["side"] == "Buy" else "Buy"

    params = {
        "category": category,
        "symbol": symbol,
        "side": side,
        "orderType": "Market" if use_market else "Limit",
        "qty": str(qty),
        "reduceOnly": True,          # これが安全弁。新規建てにならない
    }
    # ヘッジモードでは決済対象の側を明示する必要がある
    if int(pos.get("positionIdx", 0)) != 0:
        params["positionIdx"] = int(pos["positionIdx"])

    if not use_market:
        params["price"] = pos["markPrice"]
        params["timeInForce"] = "PostOnly"

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

    print(f"[close] {symbol} {side} {qty} (保有 {size} の {ratio:.0%})")
    return r["result"]


if __name__ == "__main__":
    c = make_client()
    close_position(c.http, "BTCUSDT", ratio=Decimal("0.5"))   # 半分決済
    # close_position(c.http, "BTCUSDT")                        # 全決済

"""tif_bench.py — PostOnly / IOC / Market の約定率と実効コストを実測する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
testnetは板が薄く結果が歪むため、本番の最小ロットで測ることを推奨。
"""
from __future__ import annotations

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

from client import make_client
from symbol_rules import fetch_rules, round_to_step


@dataclass
class Trial:
    tif: str
    filled: bool
    mid_at_send: Decimal
    avg_price: Decimal | None = None

    @property
    def slippage_bps(self) -> float | None:
        """発注時ミッドからの乖離(bps)。プラスは不利側。"""
        if not self.filled or self.avg_price is None:
            return None
        diff = (self.avg_price - self.mid_at_send) / self.mid_at_send
        return float(diff) * 10_000


@dataclass
class Summary:
    tif: str
    trials: list[Trial] = field(default_factory=list)

    def report(self) -> str:
        n = len(self.trials)
        fills = [t for t in self.trials if t.filled]
        rate = len(fills) / n * 100 if n else 0
        slips = [t.slippage_bps for t in fills if t.slippage_bps is not None]
        med = statistics.median(slips) if slips else float("nan")
        return (f"{self.tif:<9} 試行{n:3d}  約定率{rate:5.1f}%  "
                f"スリッページ中央値{med:+7.2f} bps")


def mid_price(http, symbol: str, category: str) -> Decimal:
    ob = http.get_orderbook(category=category, symbol=symbol, limit=1)["result"]
    return (Decimal(ob["b"][0][0]) + Decimal(ob["a"][0][0])) / 2


def run_trial(http, symbol: str, tif: str, qty: Decimal,
              category: str = "linear", wait_sec: float = 3.0) -> Trial:
    rules = fetch_rules(http, symbol, category)
    mid = mid_price(http, symbol, category)
    link_id = f"bench-{tif}-{int(time.time() * 1000)}"

    params = {
        "category": category, "symbol": symbol, "side": "Buy",
        "qty": str(qty), "orderLinkId": link_id,
    }
    if tif == "Market":
        params["orderType"] = "Market"
    else:
        params["orderType"] = "Limit"
        params["timeInForce"] = tif
        # PostOnlyはベストビッド、IOCはベストアスク（＝即約定を狙う）
        ob = http.get_orderbook(category=category, symbol=symbol, limit=1)["result"]
        px = Decimal(ob["b"][0][0]) if tif == "PostOnly" else Decimal(ob["a"][0][0])
        params["price"] = str(round_to_step(px, rules.tick_size))

    res = http.place_order(**params)
    if res["retCode"] != 0:
        # PostOnlyが板を食って拒否された場合も「約定せず」として記録する
        return Trial(tif=tif, filled=False, mid_at_send=mid)

    time.sleep(wait_sec)

    hist = http.get_order_history(category=category, orderLinkId=link_id)
    rows = hist["result"]["list"]
    if not rows:
        return Trial(tif=tif, filled=False, mid_at_send=mid)

    o = rows[0]
    filled_qty = Decimal(o.get("cumExecQty") or "0")
    if filled_qty <= 0:
        # 未約定なら片付けてから終了
        http.cancel_order(category=category, symbol=symbol, orderLinkId=link_id)
        return Trial(tif=tif, filled=False, mid_at_send=mid)

    return Trial(tif=tif, filled=True, mid_at_send=mid,
                 avg_price=Decimal(o["avgPrice"]))


if __name__ == "__main__":
    c = make_client()
    QTY = Decimal("0.001")
    summaries = {tif: Summary(tif) for tif in ("PostOnly", "IOC", "Market")}

    for _ in range(10):
        for tif, s in summaries.items():
            s.trials.append(run_trial(c.http, "BTCUSDT", tif, QTY))
            # 建った分は毎回decloseして条件を揃える（記事のclose_positionを使う）
            time.sleep(1)

    print("\n=== TIF別の執行品質 ===")
    for s in summaries.values():
        print(s.report())