グリッドトレードの仕組みを理解したとして、「実際にコードでどう実装するのか」という疑問を持つ方は多いでしょう。
Pythonとccxtを使えば、比較的少ないコード量でグリッドボットの骨格を作ることができます。ただし実用的なボットには、約定を検知して次の注文を出し直す仕組み、ネットワークエラーへの対応、ログの記録、緊急停止の仕組みなど、多くの要素が必要です。
本記事では、ccxtを使ったグリッドボットの実装を段階的に解説します。完全な動作コードではなく、学習と理解を目的とした設計の解説です。実際の資金を使った運用は十分なテストの後、自己責任で行ってください。
1. 実装の全体設計と要件整理
1-1. グリッドボットに必要な機能一覧
- 初期化処理: グリッドパラメータの設定、取引所への接続確認、残高チェック
- グリッド注文の一括発注: 設定した価格帯に買い・売り指値を配置
- 約定監視ループ: 一定間隔でオープン注文を確認し、約定済み注文を検出
- グリッド再配置: 約定した注文の反対側に新しい注文を発注
- エラーハンドリング: ネットワークエラー・残高不足・APIエラーへの対応
- ログ管理: 取引履歴・損益・エラーをファイルに記録
- 緊急停止: 価格が下限を大きく割り込んだ場合の全注文キャンセル
1-2. テストネットでの検証を必ず先行する
import ccxt
exchange = ccxt.binance({
'apiKey': 'テストネット用APIキー',
'secret': 'テストネット用シークレット',
})
exchange.set_sandbox_mode(True)2. グリッドボットのクラス設計
2-1. 設定データクラスと初期化
import ccxt, os, time, logging
from dataclasses import dataclass
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
handlers=[logging.FileHandler('grid_bot.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
@dataclass
class GridConfig:
symbol: str # 例: 'BTC/USDT'
lower_price: float # グリッド下限
upper_price: float # グリッド上限
grid_count: int # グリッド数
total_budget: float # 総投入資金（USDT）
stop_loss_pct: float # ストップロス %
class GridBot:
def __init__(self, exchange, config):
self.exchange = exchange
self.config = config
self.active_orders = {}
self.grid_prices = []
self.is_running = False2-2. グリッド価格計算と残高確認
def calculate_grid_prices(self):
# 等比グリッドの価格一覧を計算
ratio = (self.config.upper_price / self.config.lower_price) ** (1 / self.config.grid_count)
prices = [self.config.lower_price * (ratio ** i) for i in range(self.config.grid_count + 1)]
self.grid_prices = [round(p, 2) for p in prices]
logger.info(f"グリッド計算完了: {self.grid_prices[0]:.2f} 〜 {self.grid_prices[-1]:.2f}")
return self.grid_prices
def check_balance(self):
balance = self.exchange.fetch_balance()
quote = self.config.symbol.split('/')[1]
available = balance[quote]['free']
logger.info(f"利用可能残高: {available:.2f} {quote}")
if available < self.config.total_budget:
raise ValueError(f"残高不足: 必要 {self.config.total_budget}, 利用可能 {available:.2f}")
return available3. 初期注文発注の実装
3-1. 現在価格を基準にした初期グリッド発注
def place_initial_orders(self):
ticker = self.exchange.fetch_ticker(self.config.symbol)
current_price = ticker['last']
qty_per_grid = self.config.total_budget / self.config.grid_count / current_price
for i, price in enumerate(self.grid_prices[:-1]):
try:
if price < current_price:
order = self.exchange.create_limit_buy_order(
self.config.symbol, round(qty_per_grid, 6), price)
else:
order = self.exchange.create_limit_sell_order(
self.config.symbol, round(qty_per_grid, 6), price)
self.active_orders[order['id']] = {'order': order, 'grid_index': i}
logger.info(f"{order['side']} 注文発注: {price:.2f}")
time.sleep(0.2)
except ccxt.InsufficientFunds as e:
logger.error(f"残高不足: {e}"); break
except ccxt.InvalidOrder as e:
logger.warning(f"注文パラメータ不正（スキップ）: {e}")3-2. 最小注文量・価格精度への対応
各取引所・銘柄には最小注文量（lot size）と最小価格単位（tick size）が設定されています。ccxtでは exchange.amount_to_precision() や exchange.price_to_precision() を使うことで、取引所の要求する精度に合わせた注文量・価格を生成できます。この処理を省くとInvalidOrderエラーで注文が拒否されます。
4. 約定監視ループの実装
4-1. メインループと例外処理
def run(self):
self.is_running = True
try:
self.check_balance()
self.calculate_grid_prices()
self.place_initial_orders()
while self.is_running:
try:
self._check_and_replace_orders()
self._check_stop_loss()
time.sleep(5)
except ccxt.NetworkError as e:
logger.warning(f"ネットワークエラー（30秒後に再試行）: {e}")
time.sleep(30)
except ccxt.RateLimitExceeded:
logger.warning("レート制限: 60秒待機")
time.sleep(60)
except KeyboardInterrupt:
logger.info("Ctrl+C 検出: 終了処理を開始")
finally:
self.stop()
def stop(self):
self.is_running = False
try:
for order in self.exchange.fetch_open_orders(self.config.symbol):
self.exchange.cancel_order(order['id'], self.config.symbol)
logger.info(f"注文キャンセル: {order['id']}")
except Exception as e:
logger.error(f"キャンセル中にエラー: {e}")4-2. グリッド再配置ロジック
def _check_and_replace_orders(self):
for order_id in list(self.active_orders.keys()):
try:
info = self.active_orders[order_id]
order = self.exchange.fetch_order(order_id, self.config.symbol)
if order['status'] != 'closed':
continue
idx = info['grid_index']
qty = order['amount']
if order['side'] == 'buy' and idx + 1 < len(self.grid_prices):
new_price = self.grid_prices[idx + 1]
new_order = self.exchange.create_limit_sell_order(
self.config.symbol, qty, new_price)
self.active_orders[new_order['id']] = {'order': new_order, 'grid_index': idx + 1}
logger.info(f"買い約定 → 売り注文: {new_price:.2f}")
elif order['side'] == 'sell' and idx - 1 >= 0:
new_price = self.grid_prices[idx - 1]
new_order = self.exchange.create_limit_buy_order(
self.config.symbol, qty, new_price)
self.active_orders[new_order['id']] = {'order': new_order, 'grid_index': idx - 1}
logger.info(f"売り約定 → 買い注文: {new_price:.2f}")
del self.active_orders[order_id]
except ccxt.OrderNotFound:
logger.warning(f"注文が見つかりません: {order_id}")
del self.active_orders[order_id]5. ストップロスとリスク管理
5-1. ストップロス判定の実装
def _check_stop_loss(self):
ticker = self.exchange.fetch_ticker(self.config.symbol)
current_price = ticker['last']
stop_price = self.config.lower_price * (1 - self.config.stop_loss_pct)
if current_price < stop_price:
logger.warning(f"ストップロス発動: {current_price:.2f} < {stop_price:.2f}")
self.stop()
raise SystemExit("ストップロスで停止しました")5-2. 損益計算の考え方
fetch_my_trades() で約定履歴を取得し、売買の差益と手数料を集計することで損益を計算できます。この数値を定期的にログに出力する習慣をつけることが、戦略の継続評価につながります。
6. VPSでの実運用と監視体制
6-1. systemdサービスとして登録する
VPSでボットを24時間稼働させる場合、systemdにサービスとして登録しておくと、プロセスが異常終了した際に自動再起動されます。
[Service]
Type=simple
ExecStart=/path/to/venv/bin/python grid_bot.py
Restart=always
RestartSec=106-2. Telegram通知による監視
ストップロス発動・大きな約定・エラーが発生した際にTelegramで通知を受け取る仕組みを組み込むことで、スマートフォンからリアルタイムで状況を把握できます。Telegram Bot APIは無料で利用でき、PythonからのHTTPリクエスト送信も数行で実装できます。
まとめ
- GridConfigデータクラスでパラメータを一元管理する設計が保守しやすい
- 等比グリッドの価格計算は対数比率（幾何数列）で実装できる
- 約定監視ループはfetch_orderを定期的に呼び出す方式が安全
- ストップロスは最初から組み込んでおくことが必須
- 最小注文量・価格精度への対応を忘れないこと
- systemdとTelegram通知で24時間監視体制を整える
よくある質問（FAQ）
Q: fetch_open_ordersを頻繁に呼ぶとレート制限に引っかかりますか？
A: 取引所ごとにレート制限が異なります。Binanceの場合1分あたり1,200リクエストの制限があります。5秒間隔のチェックなら通常問題ありませんが、監視する注文数が増えるとリクエスト数も増えるため注意が必要です。
Q: ポジションが増えすぎた場合の対処法は？
A: 最大ポジション量を事前に設定し、上限を超えた場合は新規買いを止める処理を追加することを推奨します。
Q: 自作とFreqtradeはどちらがおすすめですか？
A: 学習目的なら自作が理解を深めます。実用目的ならFreqtradeの方が安定性・機能性で優れており、バックテストも充実しているため、入門にはFreqtradeの方がリスクが低いと考えられます。
※本記事は情報提供を目的としており、投資を推奨するものではありません。暗号資産への投資は元本割れのリスクがあります。投資判断はご自身の責任で行ってください。
