Bybitでグリッドトレード/DCA botを
Pythonで自作する
等差・等比グリッドと在庫リスク管理

グリッドbotはレンジ相場では機能し、トレンド相場では在庫を抱えて損失が膨らむ——この性質を理解して在庫を制御できるかが分かれ目です。公式のグリッドBotはパラメータが固定されますが、自作すれば在庫スキュー・停止条件・銘柄選定・グリッド幅の自動化を自由に設計できます。この記事では実装に加えて、ナンピン・マーチンゲールが破綻する確率を数値で示します。サンプルコード9本。

確認日
2026-07-30
API バージョン
Bybit V5
Python
3.10 以上
pybit
5.17.0

※ Bybit V5は機能追加のペースが速いAPIです。パラメータの最終的な正解は必ず公式ドキュメントで確認してください。本記事のコードはすべてテストネットで動作確認しています。

目次

  1. グリッドbotの損益構造|何で稼ぎ、何で負けるのか
  2. グリッド価格列の生成|等差と等比
  3. グリッド幅をATRで自動決定する
  4. 初期配置|バッチ発注と注文数の上限
  5. 約定イベント駆動の再発注ループ
  6. 在庫スキュー制御|片寄りを是正する
  7. レンジ逸脱と停止条件
  8. DCA(積立)botの実装
  9. ナンピン・マーチンゲールの破産確率
  10. グリッドのバックテストと注意点
  11. 公式グリッドBotとの比較
  12. よくある疑問Q&A

この記事は記事01・記事13・記事26の部品を組み合わせます。

グリッドbotの損益構造|何で稼ぎ、何で負けるのか

グリッドbotの損益は、はっきり2つに分解できます。この分解を理解していないと、勝っているのか負けているのか判断できません。

構成要素内容符号
グリッド利益 1段分の値幅 × 約定回数。往復するたびに確定する プラス(積み上がる)
在庫損益 片側に溜まった未決済ポジションの含み損益 トレンド時に大きくマイナス
手数料 約定回数 × 手数料率。PostOnlyならメイカー マイナス(回数に比例)
資金調達率 無期限で在庫を持つ間ずっと発生 在庫の方向次第
グリッド利益は「実現」、在庫損は「含み」

グリッドbotが危険なのは、利益が実現損益として目に見えて積み上がる一方、損失が含み損として隠れる点です。「今月は+5%のグリッド利益」と喜んでいる裏で、在庫の含み損が-15%になっていることがあります。

したがって評価は必ず「グリッド利益 + 在庫の含み損益」の合計(純資産)で行ってください。実現損益だけを見る運用は、破綻するまで問題に気づけません。サンプル6の在庫監視はこのために書いています。

どういう相場で成立するのか

つまりグリッドbotの本質は「ボラティリティの売り」です。小さく何度も稼ぎ、稀に大きく負ける損益分布を持ちます。だからこそ停止条件(サンプル7)が戦略の中心になります。

グリッド価格列の生成|等差と等比

方式間隔特徴向く相場
等差(算術) 固定の価格差
(例: 100 USDTごと)
価格が下がると相対的な利幅が拡大する 狭いレンジ・価格変動が小さい銘柄
等比(幾何) 固定の比率
(例: 0.5%ごと)
どの価格帯でも利幅の割合が一定 広いレンジ・ボラティリティの高い銘柄
grid.pyサンプル1
"""grid.py — グリッド価格列の生成(等差・等比)と利幅の計算
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum

from symbol_rules import SymbolRules, floor_to_step, round_to_step


class GridMode(str, Enum):
    ARITHMETIC = "arithmetic"   # 等差
    GEOMETRIC = "geometric"     # 等比


@dataclass
class GridLevel:
    price: Decimal
    qty: Decimal
    side: str                   # この価格で出す注文の向き(Buy / Sell)
    index: int

    def notional(self) -> Decimal:
        return self.price * self.qty


@dataclass
class GridConfig:
    symbol: str
    lower: Decimal
    upper: Decimal
    levels: int                     # グリッドの本数(分割数)
    total_notional: Decimal         # 全段に投入する想定元本の合計(USDT)
    mode: GridMode = GridMode.GEOMETRIC
    category: str = "linear"

    def __post_init__(self) -> None:
        if self.lower >= self.upper:
            raise ValueError("lower < upper でなければなりません")
        if self.levels < 2:
            raise ValueError("levelsは2以上が必要です")
        if self.lower <= 0:
            raise ValueError("lowerは正の値でなければなりません")


@dataclass
class Grid:
    config: GridConfig
    rules: SymbolRules
    prices: list[Decimal] = field(default_factory=list)

    def build(self) -> "Grid":
        """価格列を計算してキャッシュする。"""
        self.prices = self._raw_prices()
        return self

    def _raw_prices(self) -> list[Decimal]:
        cfg = self.config
        n = cfg.levels

        if cfg.mode is GridMode.ARITHMETIC:
            step = (cfg.upper - cfg.lower) / (n - 1)
            raw = [cfg.lower + step * i for i in range(n)]
        else:
            # 等比: 比率 r を n-1 乗すると upper/lower になる
            ratio = (cfg.upper / cfg.lower) ** (Decimal(1) / Decimal(n - 1))
            raw = [cfg.lower * (ratio ** i) for i in range(n)]

        # tickSizeに丸め、重複を除去する(狭いグリッドでは丸めで衝突する)
        seen: set[Decimal] = set()
        out: list[Decimal] = []
        for p in raw:
            q = round_to_step(p, self.rules.tick_size)
            if q not in seen and q > 0:
                seen.add(q)
                out.append(q)
        return sorted(out)

    def levels_for(self, mid: Decimal) -> list[GridLevel]:
        """現在価格を境に、下側は買い・上側は売りの注文を割り当てる。"""
        prices = self.prices or self._raw_prices()
        cfg = self.config

        # 1段あたりの想定元本を均等配分する
        per_level = cfg.total_notional / len(prices)

        out: list[GridLevel] = []
        for i, price in enumerate(prices):
            qty = floor_to_step(per_level / price, self.rules.qty_step)
            if qty < self.rules.min_qty:
                continue                    # 数量が最小未満の段は捨てる
            if self.rules.min_notional > 0 and price * qty < self.rules.min_notional:
                continue
            side = "Buy" if price < mid else "Sell"
            out.append(GridLevel(price=price, qty=qty, side=side, index=i))
        return out

    # ---- 分析 ----------------------------------------------------------
    def step_profit_bps(self) -> list[float]:
        """各段の利幅(bps)。等比なら一定、等差なら価格帯で変わる。"""
        prices = self.prices or self._raw_prices()
        return [float((prices[i + 1] - prices[i]) / prices[i] * 10_000)
                for i in range(len(prices) - 1)]

    def max_inventory_notional(self, mid: Decimal) -> Decimal:
        """最悪ケースの在庫。下限まで落ちたとき全買い注文が約定した想定元本。"""
        return sum((lv.notional() for lv in self.levels_for(mid)
                    if lv.side == "Buy"), Decimal(0))

    def describe(self, mid: Decimal, taker_fee_bps: float = 5.5,
                 maker_fee_bps: float = 2.0) -> str:
        lv = self.levels_for(mid)
        profits = self.step_profit_bps()
        buys = [x for x in lv if x.side == "Buy"]
        sells = [x for x in lv if x.side == "Sell"]
        min_p = min(profits) if profits else 0.0
        # 往復メイカーの手数料
        round_trip = maker_fee_bps * 2

        return (
            f"{self.config.symbol}  {self.config.mode.value}\n"
            f"  レンジ      : {self.config.lower} 〜 {self.config.upper}\n"
            f"  段数        : {len(lv)}(買い{len(buys)} / 売り{len(sells)})\n"
            f"  1段の利幅   : {min_p:.1f} 〜 {max(profits, default=0):.1f} bps\n"
            f"  往復手数料   : {round_trip:.1f} bps(全メイカー想定)\n"
            f"  正味利幅     : {min_p - round_trip:+.1f} bps"
            f"{'  ← 手数料負けします' if min_p - round_trip <= 0 else ''}\n"
            f"  最大在庫     : {self.max_inventory_notional(mid):.0f} USDT"
        )


if __name__ == "__main__":
    from decimal import Decimal as D

    from client import make_client
    from symbol_rules import fetch_rules

    c = make_client()
    rules = fetch_rules(c.http, "BTCUSDT", "linear")
    mid = D(c.http.get_tickers(category="linear",
                               symbol="BTCUSDT")["result"]["list"][0]["lastPrice"])

    for mode in (GridMode.GEOMETRIC, GridMode.ARITHMETIC):
        cfg = GridConfig(symbol="BTCUSDT",
                         lower=mid * D("0.92"), upper=mid * D("1.08"),
                         levels=30, total_notional=D("3000"), mode=mode)
        grid = Grid(config=cfg, rules=rules)
        print(grid.describe(mid))
        print()

    # 段数を増やしすぎると手数料負けすることを確認する
    print("--- 段数を増やしすぎた例 ---")
    cfg = GridConfig(symbol="BTCUSDT", lower=mid * D("0.99"),
                     upper=mid * D("1.01"), levels=40,
                     total_notional=D("3000"))
    print(Grid(config=cfg, rules=rules).describe(mid))
最初に「正味利幅」を確認する

describe() が出力する正味利幅(1段の利幅 − 往復手数料)がプラスでなければ、そのグリッド設定はどんな相場でも損失を出します。段数を増やせば約定回数は増えますが、1段の利幅が縮んで手数料に負けます。グリッドbotの設計は「段数を増やす」ではなく「手数料を上回る利幅を確保する」から始めてください。だからPostOnly(メイカー)が事実上必須です。

グリッド幅をATRで自動決定する

レンジと段数を手で決めると、相場のボラティリティが変わったときに機能しなくなります。ATR(平均的な値動きの幅)を基準にすると、自動で適応します。

grid_sizing.pyサンプル2
"""grid_sizing.py — ボラティリティからグリッドのレンジと段数を決める
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from decimal import Decimal

import pandas as pd

from grid import GridConfig, GridMode

KLINE_COLS = ["ts", "open", "high", "low", "close", "volume", "turnover"]


def _klines(http, symbol: str, interval: str, limit: int,
            category: str = "linear") -> pd.DataFrame:
    res = http.get_kline(category=category, symbol=symbol,
                         interval=interval, limit=limit)
    rows = list(reversed(res["result"]["list"]))
    df = pd.DataFrame(rows, columns=KLINE_COLS).astype(float)
    df["ts"] = pd.to_datetime(df["ts"].astype("int64"), unit="ms", utc=True)
    return df.set_index("ts")


def atr_pct(http, symbol: str, period: int = 14, interval: str = "60",
            category: str = "linear") -> Decimal:
    """ATRを現在価格に対する割合(%)で返す。"""
    df = _klines(http, symbol, interval, period * 4, category)
    prev = df["close"].shift(1)
    tr = pd.concat([df["high"] - df["low"],
                    (df["high"] - prev).abs(),
                    (df["low"] - prev).abs()], axis=1).max(axis=1)
    atr = tr.rolling(period).mean().iloc[-1]
    return Decimal(str(round(atr / df["close"].iloc[-1] * 100, 6)))


def realized_range_pct(http, symbol: str, days: int = 14,
                       category: str = "linear") -> tuple[Decimal, Decimal]:
    """直近N日の実際の高値・安値を、現在価格に対する割合で返す。"""
    df = _klines(http, symbol, "240", days * 6, category)
    last = Decimal(str(df["close"].iloc[-1]))
    hi = Decimal(str(df["high"].max()))
    lo = Decimal(str(df["low"].min()))
    return (lo / last - 1) * 100, (hi / last - 1) * 100


def auto_grid_config(http, symbol: str, total_notional: Decimal,
                     range_atr_mult: Decimal = Decimal("6"),
                     step_atr_mult: Decimal = Decimal("0.4"),
                     max_levels: int = 40,
                     maker_fee_bps: Decimal = Decimal("2.0"),
                     category: str = "linear") -> GridConfig:
    """ATRからレンジ幅と段数を決める。

    range_atr_mult : レンジの半幅を ATR の何倍にするか
    step_atr_mult  : 1段の幅を ATR の何倍にするか(利幅を決める)
    """
    a = atr_pct(http, symbol, category=category)
    last = Decimal(
        http.get_tickers(category=category, symbol=symbol)["result"]["list"][0]["lastPrice"]
    )

    half_range_pct = a * range_atr_mult
    step_pct = a * step_atr_mult

    # 往復手数料を上回る利幅を確保する(下回るなら段幅を広げる)
    min_step_pct = maker_fee_bps * 2 / 100 * Decimal("1.5")   # 1.5倍の余裕
    if step_pct < min_step_pct:
        print(f"  [調整] 段幅 {step_pct:.3f}% は手数料に対して狭いため "
              f"{min_step_pct:.3f}% に広げます")
        step_pct = min_step_pct

    levels = min(max_levels, int(half_range_pct * 2 / step_pct) + 1)
    levels = max(levels, 4)

    lower = last * (1 - half_range_pct / 100)
    upper = last * (1 + half_range_pct / 100)

    lo_real, hi_real = realized_range_pct(http, symbol, category=category)
    print(f"  ATR(1h)         : {a:.3f}%")
    print(f"  設定レンジ       : {-half_range_pct:.2f}% 〜 {+half_range_pct:.2f}%")
    print(f"  直近14日の実績   : {lo_real:.2f}% 〜 {hi_real:.2f}%")
    if lo_real < -half_range_pct or hi_real > half_range_pct:
        print("  [警告] 直近の実績値動きが設定レンジを超えています。"
              "レンジ逸脱が起きやすい設定です")
    print(f"  段数            : {levels}  1段の幅: {step_pct:.3f}%")

    return GridConfig(symbol=symbol, lower=lower, upper=upper, levels=levels,
                      total_notional=total_notional,
                      mode=GridMode.GEOMETRIC, category=category)


if __name__ == "__main__":
    from client import make_client
    from grid import Grid
    from symbol_rules import fetch_rules

    c = make_client()
    for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
        print(f"=== {sym} ===")
        cfg = auto_grid_config(c.http, sym, total_notional=Decimal("3000"))
        rules = fetch_rules(c.http, sym, "linear")
        mid = Decimal(
            c.http.get_tickers(category="linear", symbol=sym)["result"]["list"][0]["lastPrice"]
        )
        print(Grid(config=cfg, rules=rules).describe(mid))
        print()

初期配置|バッチ発注と注文数の上限

注文数の上限とレート消費を先に計算する

グリッドbotは2つの制限に必ず当たります。

「段数を増やせば利益が増える」と考えて100段・200段にすると、張り直しに10秒以上かかり、その間に相場が動いて意味をなさなくなります。実用的な段数は20〜50程度です。

grid_placer.pyサンプル3
"""grid_placer.py — グリッドをバッチ発注し、amendで価格をスライドさせる
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

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

from grid import Grid, GridLevel

MAX_PER_BATCH = {"linear": 20, "inverse": 20, "spot": 10}
DONE_CODES = {110001, 110008, 110010}      # 既に終了している注文(無視可)


@dataclass
class PlacedOrder:
    link_id: str
    price: Decimal
    qty: Decimal
    side: str
    index: int
    order_id: str | None = None


@dataclass
class GridPlacer:
    http: object
    grid: Grid
    tag: str = "grid"
    placed: dict[str, PlacedOrder] = field(default_factory=dict)

    @property
    def category(self) -> str:
        return self.grid.config.category

    @property
    def symbol(self) -> str:
        return self.grid.config.symbol

    def _link_id(self, level: GridLevel) -> str:
        # 段番号を含めておくと、約定通知から「どの段か」を逆引きできる
        return f"{self.tag}-{level.index:03d}-{int(time.time())}"[:36]

    def _to_request(self, level: GridLevel, link_id: str) -> dict:
        return {
            "symbol": self.symbol,
            "side": level.side,
            "orderType": "Limit",
            "qty": str(level.qty),
            "price": str(level.price),
            "timeInForce": "PostOnly",     # メイカー確定。グリッドでは必須
            "orderLinkId": link_id,
        }

    # ---- 初期配置 ------------------------------------------------------
    def place_all(self, mid: Decimal, sleep: float = 0.35) -> int:
        levels = self.grid.levels_for(mid)
        if not levels:
            raise RuntimeError("発注可能な段がありません(数量が最小未満)")

        size = MAX_PER_BATCH[self.category]
        chunks = [levels[i:i + size] for i in range(0, len(levels), size)]
        placed_count = 0

        print(f"{len(levels)}段を{len(chunks)}リクエストで発注します")

        for ci, chunk in enumerate(chunks, 1):
            requests_ = []
            link_ids = []
            for lv in chunk:
                lid = self._link_id(lv)
                link_ids.append((lid, lv))
                requests_.append(self._to_request(lv, lid))

            res = self.http.place_batch_order(category=self.category,
                                              request=requests_)
            # バッチは「部分成功」があるので、注文単位で結果を確認する
            results = res.get("result", {}).get("list") or []
            ext = res.get("retExtInfo", {}).get("list") or []

            for i, (lid, lv) in enumerate(link_ids):
                code = ext[i].get("code") if i < len(ext) else res.get("retCode")
                if code == 0:
                    order_id = results[i].get("orderId") if i < len(results) else None
                    self.placed[lid] = PlacedOrder(
                        link_id=lid, price=lv.price, qty=lv.qty,
                        side=lv.side, index=lv.index, order_id=order_id,
                    )
                    placed_count += 1
                else:
                    msg = ext[i].get("msg") if i < len(ext) else res.get("retMsg")
                    print(f"  [NG] 段{lv.index} {lv.side} {lv.price}: "
                          f"code={code} {msg}")

            print(f"  {ci}/{len(chunks)} 完了(累計 {placed_count}件)")
            # バッチは本数分レートを消費するので、間隔を空ける(記事05)
            time.sleep(sleep)

        return placed_count

    # ---- 価格スライド(キャンセル+再発注を避ける) ----------------------
    def slide(self, new_grid: Grid, mid: Decimal, sleep: float = 0.35) -> int:
        """既存の注文をamendで新しい価格へ移す。

        キャンセル+再発注より速く、レート消費も同等。
        段の本数が同じ場合にのみ使える。
        """
        new_levels = {lv.index: lv for lv in new_grid.levels_for(mid)}
        amends: list[dict] = []

        for lid, po in self.placed.items():
            target = new_levels.get(po.index)
            if target is None or target.price == po.price:
                continue
            amends.append({
                "symbol": self.symbol,
                "orderLinkId": lid,
                "price": str(target.price),
                "qty": str(target.qty),
            })

        if not amends:
            return 0

        size = MAX_PER_BATCH[self.category]
        moved = 0
        for i in range(0, len(amends), size):
            chunk = amends[i:i + size]
            res = self.http.amend_batch_order(category=self.category,
                                              request=chunk)
            ext = res.get("retExtInfo", {}).get("list") or []
            for j, req in enumerate(chunk):
                code = ext[j].get("code") if j < len(ext) else res.get("retCode")
                if code == 0:
                    lid = req["orderLinkId"]
                    self.placed[lid].price = Decimal(req["price"])
                    moved += 1
                elif code in DONE_CODES:
                    self.placed.pop(req["orderLinkId"], None)  # 約定済み
            time.sleep(sleep)
        return moved

    # ---- 全撤去 --------------------------------------------------------
    def cancel_all(self) -> None:
        res = self.http.cancel_all_orders(category=self.category,
                                          symbol=self.symbol)
        print(f"全キャンセル: retCode={res['retCode']} {res['retMsg']}")
        self.placed.clear()


if __name__ == "__main__":
    from client import make_client
    from grid_sizing import auto_grid_config
    from symbol_rules import fetch_rules

    c = make_client()
    SYMBOL = "BTCUSDT"

    cfg = auto_grid_config(c.http, SYMBOL, total_notional=Decimal("1000"),
                           max_levels=20)
    rules = fetch_rules(c.http, SYMBOL, "linear")
    mid = Decimal(
        c.http.get_tickers(category="linear", symbol=SYMBOL)["result"]["list"][0]["lastPrice"]
    )

    placer = GridPlacer(http=c.http, grid=Grid(config=cfg, rules=rules))
    placer.place_all(mid)
    # placer.cancel_all()

約定イベント駆動の再発注ループ

グリッドの本体は「約定したら反対側に置き直す」という1行のルールです。RESTポーリングでは遅れるため、private WebSocketの execution トピックで受け取ります。

grid_bot.pyサンプル4
"""grid_bot.py — 約定イベント駆動のグリッドbot本体
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
必ず testnet で十分に検証してから使用してください。
"""
from __future__ import annotations

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

from pybit.unified_trading import WebSocket

from grid import Grid, GridLevel
from grid_placer import GridPlacer, PlacedOrder
from symbol_rules import round_to_step


@dataclass
class GridStats:
    fills: int = 0
    grid_profit: Decimal = Decimal(0)     # 確定したグリッド利益(手数料込み)
    fees: Decimal = Decimal(0)
    buy_fills: int = 0
    sell_fills: int = 0

    def summary(self) -> str:
        return (f"約定 {self.fills}回(買{self.buy_fills}/売{self.sell_fills})  "
                f"グリッド利益 {self.grid_profit:+.4f}  手数料 {self.fees:.4f}")


class GridBot:
    """1銘柄のグリッドを、約定イベントで維持し続ける。"""

    def __init__(self, client, grid: Grid, tag: str = "grid"):
        self.c = client
        self.grid = grid
        self.placer = GridPlacer(http=client.http, grid=grid, tag=tag)
        self.tag = tag
        self.stats = GridStats()
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self.step_ratio = self._step_ratio()

    def _step_ratio(self) -> Decimal:
        """1段あたりの価格比率。反対側の価格を計算するのに使う。"""
        prices = self.grid.prices or self.grid._raw_prices()
        if len(prices) < 2:
            return Decimal("1.005")
        return prices[1] / prices[0]

    # ---- 約定イベント ---------------------------------------------------
    def on_execution(self, msg: dict) -> None:
        for e in msg.get("data", []):
            if e.get("symbol") != self.grid.config.symbol:
                continue
            if e.get("execType") != "Trade":
                continue
            link_id = e.get("orderLinkId", "")
            if not link_id.startswith(self.tag):
                continue        # 他の戦略の約定は無視する
            try:
                self._handle_fill(e)
            except Exception as exc:
                print(f"[error] 約定処理に失敗: {type(exc).__name__}: {exc}")

    def _handle_fill(self, e: dict) -> None:
        with self._lock:
            link_id = e["orderLinkId"]
            side = e["side"]
            price = Decimal(e["execPrice"])
            qty = Decimal(e["execQty"])
            fee = Decimal(e.get("execFee") or "0")

            po = self.placer.placed.pop(link_id, None)
            self.stats.fills += 1
            self.stats.fees += fee
            if side == "Buy":
                self.stats.buy_fills += 1
            else:
                self.stats.sell_fills += 1

            print(f"[fill] {side} {qty} @ {price}  fee={fee}")

            # 約定した段の「反対側」に置き直す。
            #   買いが約定 -> 1段上に売りを置く(利確)
            #   売りが約定 -> 1段下に買いを置く
            if side == "Buy":
                new_price = price * self.step_ratio
                new_side = "Sell"
            else:
                new_price = price / self.step_ratio
                new_side = "Buy"

            new_price = round_to_step(new_price, self.grid.rules.tick_size)

            # レンジ外には置かない(際限なく追いかけないための歯止め)
            cfg = self.grid.config
            if not (cfg.lower <= new_price <= cfg.upper):
                print(f"  [skip] {new_price} はレンジ外なので置き直しません")
                return

            index = po.index if po else -1
            level = GridLevel(price=new_price, qty=qty, side=new_side,
                              index=index)
            new_link = f"{self.tag}-{index:03d}-{int(time.time() * 1000)}"[:36]

            res = self.c.http.place_order(
                category=cfg.category, symbol=cfg.symbol, side=new_side,
                orderType="Limit", qty=str(qty), price=str(new_price),
                timeInForce="PostOnly", orderLinkId=new_link,
            )
            if res["retCode"] == 0:
                self.placer.placed[new_link] = PlacedOrder(
                    link_id=new_link, price=new_price, qty=qty,
                    side=new_side, index=index,
                    order_id=res["result"].get("orderId"),
                )
                print(f"  -> {new_side} {qty} @ {new_price} を設置")
            else:
                print(f"  [NG] 置き直し失敗: {res['retCode']} {res['retMsg']}")

    # ---- 起動・停止 -----------------------------------------------------
    def start(self, mid: Decimal) -> None:
        self.placer.place_all(mid)

        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.execution_stream(callback=self.on_execution)
        print("約定イベントの購読を開始しました")

    def stop(self, cancel: bool = True) -> None:
        self._stop.set()
        try:
            self.ws.exit()
        except Exception:
            pass
        if cancel:
            self.placer.cancel_all()
        print(self.stats.summary())


if __name__ == "__main__":
    from client import make_client
    from grid_sizing import auto_grid_config
    from symbol_rules import fetch_rules

    c = make_client()
    SYMBOL = "BTCUSDT"

    cfg = auto_grid_config(c.http, SYMBOL, total_notional=Decimal("1000"),
                           max_levels=20)
    rules = fetch_rules(c.http, SYMBOL, "linear")
    grid = Grid(config=cfg, rules=rules)
    mid = Decimal(
        c.http.get_tickers(category="linear", symbol=SYMBOL)["result"]["list"][0]["lastPrice"]
    )

    bot = GridBot(c, grid)
    bot.start(mid)
    try:
        while True:
            time.sleep(30)
            print(f"[status] {bot.stats.summary()}")
    except KeyboardInterrupt:
        bot.stop()

在庫スキュー制御|片寄りを是正する

価格が下がり続けると、買い注文だけが約定してロングの在庫が積み上がります。このまま等間隔で買い続けるのは、実質的にナンピンです。在庫が増えたら買いを絞り、売りを厚くするのが在庫スキュー制御です。

inventory.pyサンプル5
"""inventory.py — 在庫(片寄ったポジション)の監視とスキュー制御
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal


@dataclass
class InventoryState:
    qty: Decimal              # 正=ロング、負=ショート
    entry: Decimal
    mark: Decimal
    max_notional: Decimal     # 許容する最大在庫(想定元本)

    @property
    def notional(self) -> Decimal:
        return abs(self.qty) * self.mark

    @property
    def utilization(self) -> Decimal:
        """在庫の使用率。1.0で上限。"""
        if self.max_notional <= 0:
            return Decimal(0)
        return self.notional / self.max_notional

    @property
    def unrealised(self) -> Decimal:
        return (self.mark - self.entry) * self.qty

    @property
    def signed_utilization(self) -> Decimal:
        """符号付きの在庫使用率。+1がロング上限、-1がショート上限。"""
        u = min(self.utilization, Decimal(2))
        return u if self.qty >= 0 else -u


def fetch_inventory(http, symbol: str, max_notional: Decimal,
                    category: str = "linear") -> InventoryState:
    res = http.get_positions(category=category, symbol=symbol)
    rows = [p for p in res["result"]["list"] if Decimal(p["size"]) > 0]

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

    if not rows:
        return InventoryState(qty=Decimal(0), entry=mark, mark=mark,
                              max_notional=max_notional)
    p = rows[0]
    size = Decimal(p["size"])
    qty = size if p["side"] == "Buy" else -size
    return InventoryState(qty=qty, entry=Decimal(p["avgPrice"]), mark=mark,
                          max_notional=max_notional)


def skew_factors(inv: InventoryState, aggressiveness: Decimal = Decimal("1.0")
                 ) -> tuple[Decimal, Decimal]:
    """在庫に応じた (買い数量の倍率, 売り数量の倍率) を返す。

    在庫がロングに偏っていれば買いを縮小し、売りを拡大する。
    在庫がゼロなら (1, 1)。
    """
    u = inv.signed_utilization * aggressiveness

    if u >= 0:
        # ロング在庫あり -> 買いを絞る
        buy_mult = max(Decimal(0), Decimal(1) - u)
        sell_mult = Decimal(1) + u * Decimal("0.5")
    else:
        # ショート在庫あり -> 売りを絞る
        sell_mult = max(Decimal(0), Decimal(1) + u)
        buy_mult = Decimal(1) - u * Decimal("0.5")

    return buy_mult, sell_mult


def inventory_report(inv: InventoryState) -> str:
    buy_m, sell_m = skew_factors(inv)
    side = "ロング" if inv.qty > 0 else ("ショート" if inv.qty < 0 else "フラット")
    warn = ""
    if inv.utilization >= Decimal("1.0"):
        warn = "  ← 在庫上限に到達。新規の同方向は停止すべき"
    elif inv.utilization >= Decimal("0.7"):
        warn = "  ← 在庫が重い"

    return (
        f"在庫: {side} {abs(inv.qty)} ({inv.notional:.0f} USDT)\n"
        f"  使用率     : {inv.utilization:.0%} / 上限 {inv.max_notional:.0f} USDT{warn}\n"
        f"  建値/現在   : {inv.entry:.2f} / {inv.mark:.2f}\n"
        f"  含み損益    : {inv.unrealised:+.2f} USDT\n"
        f"  数量倍率    : 買い ×{buy_m:.2f} / 売り ×{sell_m:.2f}"
    )


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

    c = make_client()
    inv = fetch_inventory(c.http, "BTCUSDT", max_notional=Decimal("2000"))
    print(inventory_report(inv))

    print("\n--- スキュー倍率のシミュレーション ---")
    for util in ("0.0", "0.3", "0.6", "0.9", "1.2"):
        sim = InventoryState(qty=Decimal(util), entry=Decimal("100"),
                             mark=Decimal("100"),
                             max_notional=Decimal("100"))
        b, s = skew_factors(sim)
        print(f"  在庫使用率 {float(util):>5.0%} -> 買い×{b:.2f}  売り×{s:.2f}")

レンジ逸脱と停止条件

グリッドbotで最も重要なコードは、発注ではなく停止です。

停止条件判定アクション
レンジ逸脱価格がレンジの外に出た新規を止める。逸脱幅が大きければ全撤去
在庫上限在庫使用率が100%超同方向の新規を停止
ボラティリティ急拡大ATRが基準の2倍以上グリッドを撤去して再構築
含み損上限在庫の含み損が資金のX%超全撤去+全決済
口座維持率accountMMRateが閾値超即座に縮退(記事22・24)
データ停止WS受信が一定時間途絶新規を止める(記事26)
grid_guard.pyサンプル6
"""grid_guard.py — グリッドbotの停止条件を判定する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal
from enum import IntEnum


class GuardAction(IntEnum):
    """縮退の段階。数値が大きいほど深刻。"""

    CONTINUE = 0        # 通常運転
    NO_NEW_SAME = 1     # 在庫と同方向の新規のみ停止
    NO_NEW = 2          # 新規をすべて停止(既存の利確は継続)
    WITHDRAW = 3        # グリッドを全撤去(ポジションは保持)
    FLATTEN = 4         # 全撤去+全決済


@dataclass
class GuardConfig:
    range_exit_buffer_pct: Decimal = Decimal("1.0")   # レンジ外何%まで許容
    max_inventory_util: Decimal = Decimal("1.0")
    atr_spike_mult: Decimal = Decimal("2.0")          # 基準ATRの何倍で異常
    max_unrealised_loss_pct: Decimal = Decimal("5.0") # 資金に対する含み損上限
    max_account_mmr: Decimal = Decimal("0.35")        # 口座維持率の上限
    max_data_silence_sec: float = 120.0


@dataclass
class GuardDecision:
    action: GuardAction
    reasons: list[str]

    def __bool__(self) -> bool:
        return self.action is not GuardAction.CONTINUE

    def describe(self) -> str:
        if not self.reasons:
            return "通常運転"
        return f"{self.action.name}: " + " / ".join(self.reasons)


def evaluate(price: Decimal, grid_lower: Decimal, grid_upper: Decimal,
             inv_util: Decimal, unrealised: Decimal, equity: Decimal,
             atr_now: Decimal, atr_base: Decimal, account_mmr: Decimal,
             data_silence_sec: float,
             cfg: GuardConfig | None = None) -> GuardDecision:
    """全条件を評価し、最も深刻なアクションを返す。"""
    cfg = cfg or GuardConfig()
    reasons: list[str] = []
    action = GuardAction.CONTINUE

    def escalate(new: GuardAction, why: str) -> None:
        nonlocal action
        reasons.append(why)
        if new > action:
            action = new

    # 1) レンジ逸脱
    buffer_ = cfg.range_exit_buffer_pct / 100
    if price > grid_upper * (1 + buffer_):
        escalate(GuardAction.WITHDRAW,
                 f"レンジ上限を{(price / grid_upper - 1) * 100:.1f}%超過")
    elif price < grid_lower * (1 - buffer_):
        escalate(GuardAction.WITHDRAW,
                 f"レンジ下限を{(1 - price / grid_lower) * 100:.1f}%下回る")
    elif price > grid_upper or price < grid_lower:
        escalate(GuardAction.NO_NEW, "レンジ外(バッファ内)")

    # 2) 在庫上限
    if inv_util >= cfg.max_inventory_util:
        escalate(GuardAction.NO_NEW_SAME, f"在庫使用率{inv_util:.0%}")

    # 3) ボラティリティ急拡大
    if atr_base > 0 and atr_now / atr_base >= cfg.atr_spike_mult:
        escalate(GuardAction.WITHDRAW,
                 f"ATRが基準の{atr_now / atr_base:.1f}倍")

    # 4) 含み損の上限
    if equity > 0:
        loss_pct = -unrealised / equity * 100
        if loss_pct >= cfg.max_unrealised_loss_pct:
            escalate(GuardAction.FLATTEN, f"含み損が資金の{loss_pct:.1f}%")

    # 5) 口座維持率
    if account_mmr >= cfg.max_account_mmr:
        escalate(GuardAction.FLATTEN, f"口座維持率{account_mmr:.0%}")

    # 6) データ停止
    if data_silence_sec >= cfg.max_data_silence_sec:
        escalate(GuardAction.NO_NEW,
                 f"データ受信が{data_silence_sec:.0f}秒停止")

    return GuardDecision(action=action, reasons=reasons)


if __name__ == "__main__":
    from decimal import Decimal as D

    scenarios = [
        ("通常", D("100"), D("0.3"), D("-10"), D("1.0"), D("0.1"), 5.0),
        ("レンジ上抜け", D("112"), D("0.2"), D("50"), D("1.1"), D("0.1"), 5.0),
        ("在庫過多", D("95"), D("1.1"), D("-80"), D("1.0"), D("0.2"), 5.0),
        ("ボラ急拡大", D("98"), D("0.5"), D("-40"), D("2.5"), D("0.2"), 5.0),
        ("含み損超過", D("88"), D("1.0"), D("-600"), D("1.4"), D("0.3"), 5.0),
        ("データ断", D("100"), D("0.2"), D("0"), D("1.0"), D("0.1"), 300.0),
    ]
    for name, price, util, upl, atr_ratio, mmr, silence in scenarios:
        d = evaluate(price=price, grid_lower=D("92"), grid_upper=D("108"),
                     inv_util=util, unrealised=upl, equity=D("10000"),
                     atr_now=atr_ratio, atr_base=D("1.0"),
                     account_mmr=mmr, data_silence_sec=silence)
        print(f"{name:<14} -> {d.describe()}")

DCA(積立)botの実装

グリッドと対照的に、DCA(Dollar Cost Averaging)は定期的に一定金額を買うだけの単純な戦略です。現物の marketUnit="quoteCoin" を使えば「毎週100 USDT分買う」がそのまま書けます。

dca_bot.pyサンプル7
"""dca_bot.py — 現物の定期積立(DCA)bot
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

グリッドと違い在庫リスクの制御が不要で、実装も検証も簡単。
「まず何か動かしたい」なら、こちらから始めるのが現実的。
"""
from __future__ import annotations

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

STATE_PATH = Path("data/dca_state.json")


@dataclass
class DcaConfig:
    symbol: str = "BTCUSDT"
    amount_usdt: Decimal = Decimal("100")
    interval_hours: float = 168.0            # 週1回
    max_executions: int | None = None
    # 価格が移動平均を下回っているときだけ買う(任意のフィルタ)
    only_below_sma: int | None = None        # 例: 200 で200期間SMA


class DcaBot:
    def __init__(self, client, cfg: DcaConfig):
        self.c = client
        self.cfg = cfg
        self.state = self._load()

    # ---- 状態の永続化(プロセス再起動でも二重実行しない) --------------
    def _load(self) -> dict:
        if STATE_PATH.exists():
            return json.loads(STATE_PATH.read_text(encoding="utf-8"))
        return {"executions": 0, "last_ts": 0, "total_usdt": "0",
                "total_base": "0"}

    def _save(self) -> None:
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        tmp = STATE_PATH.with_suffix(".tmp")
        tmp.write_text(json.dumps(self.state, ensure_ascii=False),
                       encoding="utf-8")
        tmp.replace(STATE_PATH)

    # ---- 条件判定 -------------------------------------------------------
    def _sma_filter_ok(self) -> bool:
        n = self.cfg.only_below_sma
        if not n:
            return True
        res = self.c.http.get_kline(category="spot", symbol=self.cfg.symbol,
                                    interval="D", limit=n)
        closes = [float(r[4]) for r in res["result"]["list"]]
        if len(closes) < n:
            return True
        sma = sum(closes) / len(closes)
        last = closes[0]      # 降順なので先頭が最新
        ok = last < sma
        print(f"  SMA{n}フィルタ: 価格{last:.2f} vs SMA{sma:.2f} -> "
              f"{'買う' if ok else '見送る'}")
        return ok

    def _due(self) -> bool:
        elapsed_h = (time.time() - self.state["last_ts"]) / 3600
        return elapsed_h >= self.cfg.interval_hours

    # ---- 実行 -----------------------------------------------------------
    def execute_once(self, force: bool = False) -> bool:
        if not force and not self._due():
            remain = self.cfg.interval_hours - (time.time() - self.state["last_ts"]) / 3600
            print(f"次回まで {remain:.1f} 時間")
            return False

        if (self.cfg.max_executions is not None
                and self.state["executions"] >= self.cfg.max_executions):
            print("実行回数の上限に達しました")
            return False

        if not self._sma_filter_ok():
            # 見送った場合も時刻を進める(次の周期まで待つ)
            self.state["last_ts"] = time.time()
            self._save()
            return False

        # 冪等ID: 同じ周期で二重に買わないよう時刻バケットを含める
        bucket = int(time.time() // (self.cfg.interval_hours * 3600))
        link_id = f"dca-{self.cfg.symbol[:6]}-{bucket}"[:36]

        res = self.c.http.place_order(
            category="spot",
            symbol=self.cfg.symbol,
            side="Buy",
            orderType="Market",
            qty=str(self.cfg.amount_usdt),
            marketUnit="quoteCoin",        # qtyをUSDT金額として解釈させる
            orderLinkId=link_id,
        )

        if res["retCode"] == 0:
            print(f"[DCA] {self.cfg.amount_usdt} USDT分の"
                  f"{self.cfg.symbol}を購入 (orderId={res['result']['orderId']})")
        elif res["retCode"] in (170141, 110072):
            print(f"[DCA] この周期は既に実行済みです(冪等IDで防止)")
            return False
        else:
            print(f"[DCA] 失敗: {res['retCode']} {res['retMsg']}")
            return False

        self.state["executions"] += 1
        self.state["last_ts"] = time.time()
        self.state["total_usdt"] = str(
            Decimal(self.state["total_usdt"]) + self.cfg.amount_usdt
        )
        self._save()
        return True

    def report(self) -> None:
        total = Decimal(self.state["total_usdt"])
        n = self.state["executions"]
        last = (datetime.fromtimestamp(self.state["last_ts"], timezone.utc)
                if self.state["last_ts"] else None)
        print(f"積立回数 {n}回 / 投入額 {total} USDT / "
              f"平均 {total / n if n else 0:.2f} USDT/回")
        if last:
            print(f"最終実行 {last:%Y-%m-%d %H:%M} UTC")


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

    c = make_client()
    bot = DcaBot(c, DcaConfig(symbol="BTCUSDT", amount_usdt=Decimal("100"),
                              interval_hours=168, only_below_sma=None))
    bot.execute_once(force=False)
    bot.report()

    # 実運用ではcron/systemd timerで定期実行する(記事38)
    #   0 9 * * 1  cd /opt/bot && .venv/bin/python dca_bot.py

ナンピン・マーチンゲールの破産確率

グリッドbotの派生として「下がったらロットを倍にする」マーチンゲール型が紹介されることがあります。これは数学的に破綻が確定している手法です。感覚論ではなく、シミュレーションで示します。

martingale_ruin.pyサンプル8
"""martingale_ruin.py — ナンピン・マーチンゲールの必要資金と破産確率
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

目的: 「勝率が高い」戦略が、なぜ最終的に資金を失うのかを数値で示す。
"""
from __future__ import annotations

import numpy as np


def required_capital(base_lot: float, multiplier: float, steps: int,
                     step_pct: float, entry_price: float = 100.0) -> dict:
    """N段ナンピンした場合の必要資金と平均建値を計算する。"""
    total_qty = 0.0
    total_cost = 0.0
    lot = base_lot
    prices = []

    for i in range(steps):
        price = entry_price * (1 - step_pct / 100 * i)
        prices.append(price)
        total_qty += lot
        total_cost += lot * price
        lot *= multiplier

    avg = total_cost / total_qty if total_qty else 0.0
    last = prices[-1]
    # 最終段の価格での含み損
    unrealised = (last - avg) * total_qty

    return {
        "steps": steps,
        "total_qty": total_qty,
        "total_cost": total_cost,
        "avg_price": avg,
        "last_price": last,
        "drawdown_pct": (last / entry_price - 1) * 100,
        "unrealised": unrealised,
        "final_lot": base_lot * multiplier ** (steps - 1),
    }


def ruin_probability(win_prob_per_cycle: float, cycles: int,
                     trials: int = 20_000, seed: int = 42) -> float:
    """1サイクルあたりの成功確率が p のとき、N サイクル全て生き残る確率。

    マーチンゲールは「1サイクルの勝率」が非常に高い一方、
    1回の失敗で資金を失う。生存確率は p^N で指数的に減衰する。
    """
    rng = np.random.default_rng(seed)
    survived = 0
    for _ in range(trials):
        if np.all(rng.random(cycles) < win_prob_per_cycle):
            survived += 1
    return 1.0 - survived / trials


if __name__ == "__main__":
    print("=== 必要資金(倍率別・1段2%下落) ===")
    print(f"{'倍率':>6}{'段数':>6}{'最終ロット':>12}{'必要資金':>14}"
          f"{'到達下落率':>12}")
    print("-" * 54)
    for mult in (1.0, 1.5, 2.0):
        for steps in (5, 8, 10):
            r = required_capital(base_lot=1.0, multiplier=mult, steps=steps,
                                 step_pct=2.0)
            print(f"{mult:>6.1f}{steps:>6}{r['final_lot']:>12.1f}"
                  f"{r['total_cost']:>14.0f}{r['drawdown_pct']:>11.1f}%")

    print("\n=== 生存確率(1サイクルの勝率が高くても…) ===")
    print(f"{'サイクル勝率':>12}{'10回':>10}{'50回':>10}{'200回':>10}"
          f"{'1000回':>10}")
    print("-" * 54)
    for p in (0.99, 0.98, 0.95, 0.90):
        row = [f"{(1 - ruin_probability(p, n)) * 100:>9.1f}%"
               for n in (10, 50, 200, 1000)]
        print(f"{p:>11.0%}" + "".join(row))

    print("\n--- 解釈 ---")
    print("1サイクルの勝率99%でも、200サイクル続けて生き残る確率は約13%。")
    print("マーチンゲールは『ほぼ毎回勝つ』が『いつか必ず全部失う』構造で、")
    print("勝率の高さは安全性をまったく意味しない。")
    print("必要資金も倍率2.0では指数的に増え、10段で現実的な額を超える。")
「勝率が高い」は安全性を意味しない

上のシミュレーションが示すとおり、1サイクルの勝率が99%でも、200サイクル続けると生存率は約13%です。マーチンゲールの損益分布は「小さな利益を何度も、稀に全損」という形で、期待値がプラスであっても破産確率が1に収束します(Gambler's ruin)。

グリッドbotを作る場合はロット倍率を1.0(等ロット)に固定し、在庫上限(サンプル5)と停止条件(サンプル6)で守ってください。「倍率を上げれば回復が早い」という発想は、破産までの時間を短くするだけです。

グリッドのバックテストと注意点

記事31のバックテスト基盤でグリッドを検証できますが、グリッド特有の落とし穴が3つあります。

grid_strategy_bt.pyサンプル9
"""grid_strategy_bt.py — 記事31のバックテスト基盤でグリッドを検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd

from bt.strategy import Strategy
from bt.types import Order, OrderType, Position, Side


class GridStrategy(Strategy):
    """等比グリッドをバックテストで再現する簡易版。

    エンジンは1足ごとに注文を処理するため、実運用の
    「約定と同時に反対側へ置き直す」挙動を完全には再現できない。
    傾向を掴む目的で使い、最終判断はtestnetでの実運用検証で行う。
    """

    def __init__(self, lower_pct: float = -8.0, upper_pct: float = 8.0,
                 levels: int = 20, notional_per_level: float = 100.0,
                 anchor_bars: int = 24):
        self.lower_pct = lower_pct
        self.upper_pct = upper_pct
        self.levels = levels
        self.notional = notional_per_level
        self.anchor_bars = anchor_bars
        self.anchor: float | None = None
        self.grid_prices: list[float] = []

    def _rebuild(self, price: float) -> None:
        self.anchor = price
        lo = price * (1 + self.lower_pct / 100)
        hi = price * (1 + self.upper_pct / 100)
        ratio = (hi / lo) ** (1 / (self.levels - 1))
        self.grid_prices = [lo * ratio ** i for i in range(self.levels)]

    def on_bar(self, history: pd.DataFrame, position: Position,
               equity: float) -> list[Order] | None:
        if len(history) < self.anchor_bars:
            return None

        price = history["close"].iloc[-1]

        if self.anchor is None:
            self._rebuild(price)
            return None

        # レンジ逸脱で撤去+再構築(実運用の停止条件に相当)
        lo, hi = self.grid_prices[0], self.grid_prices[-1]
        if price < lo * 0.99 or price > hi * 1.01:
            self._rebuild(price)
            if not position.is_flat:
                return [Order(side=position.side.opposite,
                              qty=abs(position.qty),
                              order_type=OrderType.MARKET,
                              reduce_only=True, tag="grid-range-exit")]
            return None

        # 現在価格の直下に買い、直上に売りを1本ずつ置く(簡略化)
        below = [p for p in self.grid_prices if p < price]
        above = [p for p in self.grid_prices if p > price]
        orders: list[Order] = []

        # 在庫上限: 想定元本の合計を超えたら買いを出さない
        inventory_notional = abs(position.qty) * price
        max_inventory = self.notional * self.levels / 2

        if below and inventory_notional < max_inventory:
            px = max(below)
            orders.append(Order(side=Side.BUY,
                                qty=round(self.notional / px, 6),
                                order_type=OrderType.POST_ONLY, price=px,
                                ttl_bars=6, tag="grid-buy"))
        if above and position.qty > 0:
            px = min(above)
            qty = min(round(self.notional / px, 6), position.qty)
            if qty > 0:
                orders.append(Order(side=Side.SELL, qty=qty,
                                    order_type=OrderType.POST_ONLY, price=px,
                                    reduce_only=True, ttl_bars=6,
                                    tag="grid-sell"))
        return orders or None


if __name__ == "__main__":
    from bt.engine import BacktestConfig, Backtester
    from bt.fill import FillModel, PessimisticFillModel
    from bt.report import print_report
    from client import make_client
    from costs import fetch_funding_history, fetch_fee_model
    from kline_store import KlineStore

    c = make_client()
    bars = KlineStore().load("BTCUSDT", "60")
    if bars.empty:
        raise SystemExit("先に記事07のKlineStore.update()を実行してください")

    fees = fetch_fee_model(c.http, "BTCUSDT")
    start_ms = int(bars.index[0].timestamp() * 1000)
    funding = fetch_funding_history(c.http, "BTCUSDT", start_ms)["funding_rate"]

    # レンジ期間とトレンド期間の両方で検証する(これが重要)
    for label, model in (("標準", FillModel(maker_fill_prob=0.7)),
                         ("悲観", PessimisticFillModel())):
        print(f"\n########## {label}モデル ##########")
        bt = Backtester(bars, fees, fill_model=model, funding=funding,
                        config=BacktestConfig(initial_cash=10_000))
        print_report(bt.run(GridStrategy()), bars_per_year=8_760)

公式グリッドBotとの比較

観点公式グリッドBot自作(本記事)
導入数クリック実装が必要
在庫スキュー制御不可自由に設計できる
停止条件用意されたものだけ維持率・ATR・データ断など任意
グリッド幅の自動化限定的ATR連動などが可能
銘柄選定手動スクリーニングと連動できる
バックテストできない記事31の基盤で検証できる
稼働の安定性取引所側で動くVPSの運用が必要(記事38)
障害時botは動き続ける自分のプロセスが落ちれば止まる
自作の最大の価値は「止められること」

自作の利点はパラメータの自由度より、自分が決めた条件で確実に止められることにあります。グリッドbotは停止条件が損益を決める戦略なので、ここを制御できるかどうかが本質的な差になります。

ただし自作botはプロセスが落ちれば在庫を持ったまま放置されます。取引所側の広めの損切(記事14)を保険として併せて置き、VPSでの自動再起動(記事38)とリコンサイル(記事17)を必ず用意してください。

参考リンク・関連文献
  • Batch Place Order / Amend: bybit-exchange.github.io/docs/v5/order/batch-place
  • WebSocket Execution: /websocket/private/execution
  • Rate Limit(バッチは本数分消費): /v5/rate-limit
  • 在庫リスクの理論的背景: Avellaneda & Stoikov (2008) High-frequency Trading in a Limit Order Book(在庫を持つことのコストを定式化した論文。記事37で扱います)
  • 破産確率: Gambler's ruin problem(マーチンゲールが破綻する数学的根拠)

よくある疑問Q&A

段数は多いほうが有利ですか?
いいえ。段数を増やすと約定回数は増えますが、1段の利幅が縮んで手数料に負けます。サンプル1の describe() が出力する「正味利幅(1段の利幅 − 往復手数料)」がプラスであることが最低条件です。加えて注文数の上限(500件)とレート消費(本数分)の制約もあります。実用的には20〜50段で、利幅が往復手数料の1.5倍以上になるよう設計してください。
現物と無期限先物、どちらでやるべきですか?
現物のほうが安全です。清算がなく、資金調達も発生しません。最悪のケースは「買った銘柄が下がったまま塩漬け」で済みます。無期限先物はレバレッジをかけられる代わりに、在庫が積み上がったところで清算されると全損します。グリッドbotの損益分布(稀に大きく負ける)とレバレッジの相性は非常に悪いです。無期限で行うなら在庫上限を厳しく設定し、レバレッジは低く(2倍以下)してください。
銘柄はどう選べばいいですか?
グリッドに向くのは「ボラティリティが高く、スプレッドが狭く、レンジ回帰しやすい」銘柄です。記事01のスクリーナーで spread_bps と24h売買代金でフィルタし、記事07のDuckDB集計で年率ボラを見るのが実務的です。スプレッドが利幅を食う薄い銘柄は、ボラが高くても不適です。なお「レンジ回帰しやすいか」は過去データから判定できても将来には保証がないため、停止条件で守る前提を崩さないでください。
レンジを抜けたら追いかけて再構築すべきですか?
これが最も判断が難しい点です。追いかければ「トレンドに沿ってナンピンし続ける」危険があり、追いかけなければ「機会を逃す」。安全側の設計は「抜けたら在庫を決済してから、新しい価格帯で再構築する」です。在庫を持ったまま追いかけるのが最悪のパターンです。サンプル6の WITHDRAW と FLATTEN を区別しているのはこのためで、含み損が大きい場合は決済(FLATTEN)を選ぶようにしています。
PostOnlyだと約定しないことがあるので、成行にすべきですか?
してはいけません。グリッドの利益源は「メイカー手数料での往復」です。テイカーにすると往復コストが数倍になり、1段の利幅では回収できなくなります(サンプル1で確認できます)。PostOnlyで約定しない段があるのは正常で、それでも成立するように利幅を設計してください。「約定しないから成行に変える」は、戦略の前提を壊す変更です。
botが落ちたときに在庫が放置されるのが怖いです
正しい懸念です。対策は3段構えです。(1) 取引所側に広めの損切を置く(記事14の set_trading_stop。botが落ちても取引所側で動きます)。(2) VPSでプロセスの自動再起動を設定する(記事38のsystemd / Docker restart policy)。(3) 再起動時にリコンサイルする(記事17。取引所側の未約定注文とポジションを取得してから再開する)。この3つを入れるまで、実資金での運用は避けてください。
DCAとグリッド、どちらから始めるべきですか?
DCA(サンプル7)です。在庫リスクの制御が不要で、実装も検証も単純で、失敗しても「買った価格が高かった」で済みます。グリッドbotは在庫管理・停止条件・レート制限・注文数上限のすべてを扱う必要があり、初めて作るbotとしては難易度が高すぎます。DCAで「定期実行・冪等性・状態の永続化・通知」といった運用の基礎を固めてから、グリッドに進むのが安全な順序です。

注意事項・免責事項

投資は自己判断・自己責任でお願いいたします。本ブログの情報・ツール・ソフトウェアを参考・使用して、取引をされて何らかの損失を被られても、当サイトでは責任は一切負いません(間接的・付随的な障害・損失も含め一切の責任を負いません)。自己責任でご利用ください。

当サイトからダウンロードしたプログラム・EAは本人利用に限り完全無料でご利用いただけます。プログラムの全てまたは一部の複製・コピー・リバースエンジニアリング・改変・2次配布・商用利用・転売・販売等は一切禁止いたします。違反を発見した場合は法的処置を行います。

当サイト上に存在するソフトウェア・ツール・文章・画像・動画等の著作物を無断転載することを禁止します。引用の範囲を超えるものについては、法的処置を行います。

当サイトで配布するツール・ソフトウェアは無料提供ですのでサポートなどは行っておりません(逆に丁寧にサポートしているサイトはほとんど違法サイトですのでご注意ください。投資助言・代理業登録なしでサポート・助言することは法律で禁止されています)。万が一、ツールやソフトウェアに不具合や問題があっても当サイトでは修正・対応などができない場合があります。ご利用の環境でトラブルが発生しても当サイトでは助言することはできませんので自己責任でご利用ください。検索するなどして自分で解決してください。

当サイトでは取引所・証券会社へのリンクを貼っていますが、これは情報として提供するものであり、勧誘行為などは一切行っておりません。そのため、取引所・証券会社の登録方法・使い方・オススメ金融機関・オススメ金融商品などを説明した記事はサイト内にありませんので、ご自身で調べた上で判断ください(本当はサイト内で丁寧に説明したいのですが、仲介・取次ぎにあたる可能性があり第一種金融商品取引業登録を行っていない場合、違法となる可能性があるため出来ません。ご容赦ください)。

当サイトからリンクやバナーなどによって他のサイトに移動された場合、移動先サイトで提供される情報、サービス等について一切の責任を負いません。当サイト掲載情報やリンク先の掲載情報の合法性、正確性、道徳性、最新性、適切性、著作権の許諾や有無など、その内容については一切の保証を致しかねます。

当サイトは、ご訪問いただいた時間、訪問者のアクセス元のホスト名、IPアドレス、OSとブラウザのバージョン、リンク元のURL、モニターの情報等のパソコン環境情報を収集し、アクセス解析をしております。この目的は訪問者が検索サイトからご訪問いただいた場合、どのような検索で当サイトに来訪いただいたかの解析にのみ使用しており、第3者に公開はいたしません。また、個人を特定するものではありません。