はじめに
自動売買Botを作ると、最初はこう思います。
「ロジックを改善すれば利益が出る」
でも実際に運用すると、Botはロジックより先にシステムの問題で壊れます。
この記事では、実際にBotを開発・運用する中で遭遇した「壊れるパターン」と、その対策をまとめます。
環境
- Python 3.12
- FastAPI 0.111
- ccxt 4.x（取引所接続）
- APScheduler 3.10（定期実行）
- PostgreSQL（状態管理）
パターン① 注文成功チェックなし
問題
# ❌ 悪い例
await exchange.create_order(symbol, "buy", amount, price)
# 成功したと仮定して次の処理へ
bot.open_quantity += amount
APIエラー・タイムアウト・注文拒否が起きても検知できません。
ツールは「持っている」と思っているのに、実際には何も持っていない状態になります。
対策
# ✅ 良い例
try:
result = await exchange.create_order(symbol, "buy", amount, price)
order_id = result["id"]
# 注文状態を SUBMITTED に更新
trade.order_status = "SUBMITTED"
db.commit()
# 約定確認（二重チェック）
await asyncio.sleep(2)
order = await exchange.fetch_order(order_id, symbol)
if order["status"] == "closed":
trade.order_status = "FILLED"
elif order["status"] == "open":
trade.order_status = "ACCEPTED"
db.commit()
except ccxt.NetworkError as e:
# リトライ処理
trade.order_status = "FAILED"
db.commit()
注文の状態遷移を管理します。
NONE → SUBMITTED → ACCEPTED → FILLED
↘ FAILED（エラー時）
パターン② 部分約定の未対応
問題
1BTC注文して0.3BTCしか約定しなかった場合、未対応だと1BTC保有と誤認します。
# ❌ 悪い例
bot.open_quantity = order["amount"] # 注文数量をそのまま使う
対策
# ✅ 良い例
filled = order["filled"] # 実際に約定した数量
remaining = order["remaining"] # 残注文数量
bot.open_quantity += filled
if remaining > 0:
trade.order_status = "PARTIAL"
trade.remaining_order_id = order["id"]
# 次のreconcileサイクルで残注文を確認する
パターン③ ポジション同期なし
問題
通信遅延・API遅延でツール内部の状態と取引所の実際のポジションがズレます。
放置すると二重エントリーや逆方向の注文が出ます。
対策
APSchedulerで30秒ごとに突合処理を実行します。
class PositionReconciler:
async def reconcile(self, bot: Bot, exchange):
# 取引所の実際のポジションを取得
positions = await exchange.fetch_positions([bot.symbol])
exchange_qty = sum(p["contracts"] for p in positions)
# 内部状態と比較
if abs(exchange_qty - bot.open_quantity) > 0.001:
logger.warning(f"POSITION_MISMATCH: bot={bot.open_quantity}, exchange={exchange_qty}")
# アラートを発行してBotをerror状態に遷移
bot.status = "error"
db.commit()
パターン④ 重複注文
問題
シグナルが連続で発生したとき、同じ条件で複数の注文が出ます。
ポジションが想定より増えて資金管理が崩れます。
対策
未完了注文がある場合は新規注文をスキップします。
async def run_bot(bot: Bot):
# 未完了注文チェック
pending = db.query(Trade).filter(
Trade.bot_id == bot.id,
Trade.order_status.in_(["SUBMITTED", "ACCEPTED", "PARTIAL"])
).first()
if pending:
logger.info(f"Bot {bot.id}: 未完了注文あり、スキップ")
return
# シグナル判定・注文実行へ
パターン⑤ 価格取得エラー
問題
APIが失敗すると価格が None・0・古い値になります。
そのまま使うと意味不明なトレードが発生します。
# ❌ 悪い例
ticker = await exchange.fetch_ticker(symbol)
price = ticker["last"] # Noneの可能性
order_amount = budget / price # ZeroDivisionError
対策
# ✅ 良い例
def validate_price(price, prev_price=None) -> bool:
if price is None or price <= 0:
return False
# 前回価格から±20%以上の乖離はスパイクとして弾く
if prev_price and abs(price - prev_price) / prev_price > 0.2:
return False
return True
ticker = await exchange.fetch_ticker(symbol)
price = ticker["last"]
if not validate_price(price, bot.last_price):
logger.warning(f"価格バリデーション失敗: {price}")
return # スキップ
パターン⑥ 注文ループ
問題
ロジックのバグで同じ条件を繰り返し検出し、無限に注文が出続けます。
手数料で資金が削られます。
対策
単位時間あたりの注文数に上限を設けます。
class RiskManager:
def check_order_rate_limit(self, bot: Bot) -> bool:
# 直近1分間の注文数を確認
one_minute_ago = datetime.now(timezone.utc) - timedelta(minutes=1)
recent_orders = db.query(Trade).filter(
Trade.bot_id == bot.id,
Trade.created_at >= one_minute_ago
).count()
if recent_orders >= bot.config.get("max_orders_per_minute", 5):
logger.error(f"注文数上限超過: {recent_orders}件/分")
bot.status = "error" # 緊急停止
db.commit()
return False
return True
パターン⑦ タイムゾーンのズレ
問題
サーバー・取引所・DBの時刻が異なると、時間足の区切りが合わずロジックが誤動作します。
対策
全タイムスタンプはUTCで統一します。
# ❌ 悪い例
from datetime import datetime
created_at = datetime.now() # ローカルタイム（環境依存）
# ✅ 良い例
from datetime import datetime, timezone
created_at = datetime.now(timezone.utc) # UTC固定
.env にタイムゾーン設定も追加します。
TZ=UTC
パターン⑧ 例外の素通し
問題
APIエラーをキャッチして pass だけ書くと、ツールが無言で止まります。
ログがないと原因を特定できません。
# ❌ 悪い例
try:
await exchange.create_order(...)
except Exception:
pass # 無言で失敗
対策
# ✅ 良い例
try:
await exchange.create_order(...)
except ccxt.NetworkError as e:
logger.error(f"NetworkError: {e}")
# リトライ処理
except ccxt.AuthenticationError as e:
logger.error(f"APIキー認証失敗: {e}")
bot.status = "error"
db.commit()
except Exception as e:
logger.error(f"予期せぬエラー: {e}", exc_info=True)
bot.status = "error"
db.commit()
全ての例外に対して最低限ログを残すことをルールにします。
パターン⑨ 緊急停止機能がない
問題
Botが暴走したときに止める仕組みがないと、最悪資金が消えます。
対策
複数の停止条件を実装します。
class RiskManager:
def should_emergency_stop(self, bot: Bot) -> bool:
# 連敗停止
if bot.consecutive_losses >= bot.config.get("max_losses", 5):
return True
# 日次損失上限
if bot.daily_pnl <= -bot.config.get("daily_loss_limit", 10000):
return True
# ドローダウン制限
if bot.peak_pnl > 0:
drawdown = (bot.peak_pnl - bot.pnl) / bot.peak_pnl
if drawdown >= bot.config.get("max_drawdown", 0.2):
return True
return False
まとめ
自動売買Botが壊れるのはロジックではなくシステムです。
| パターン | 対策 |
|---|---|
| 注文成功チェックなし | 状態遷移で管理・二重確認 |
| 部分約定未対応 | filled数量で管理・残注文を追跡 |
| ポジション同期なし | 30秒ごとの突合処理 |
| 重複注文 | 未完了注文チェック |
| 価格取得エラー | バリデーション関数 |
| 注文ループ | 注文数レートリミット |
| タイムゾーンズレ | UTC統一 |
| 例外素通し | 全例外をログ記録 |
| 緊急停止なし | 複数の停止条件 |
「動く」より「壊れない」を先に作ることが、自動売買Botを本番運用するための基本です。
関連記事
この記事で紹介した安全設計を実装したAutoTraderのソースコードを公開しています。
FastAPI + React Native（Expo）でスマホからBotを管理できる構成です。