😊

Pythonで仮想通貨の自動売買Botを作ってみた — EMAクロスオーバーで放置トレード

に公開
Python
データ分析
アルゴリズム
crypto
トレード
tech

仮想通貨の自動売買Botってなんか難しそうだなと思ってたんだけど、実際にやってみたら意外となんとかなった。Pythonが書ければ、週末の空き時間で動くものは作れる。

ということで、僕が実際に作ったBotの中身を全部公開してみる。戦略はEMAクロスオーバーっていう、たぶん一番シンプルなやつ。3年分のバックテストではBTC/USDTの日足でシャープレシオ1.30、年率77%だった。（もちろん過去の成績なので将来どうなるかはわからない。）

なんで自動売買なのか

仮想通貨って24時間365日動いてる。寝てる間にチャンス逃すのが嫌で、じゃあBotに任せればいいじゃんと。

手動だと感情でやらかすんですよね。「もうちょっと上がるかも」とか「怖いから早めに売ろう」とか。Botならルール通りに淡々とやってくれる。

環境構築

必要なものはこれだけ。

  • Python 3.11以上
  • pip
  • 取引所のアカウントとAPIキー

取引所のアカウント

MEXCを使ってる。手数料が安いのと、APIが素直なのが決め手だった。

↓ ここから登録すると手数料の割引が効きます

MEXC公式 — 口座開設

登録したらAPI管理画面でAPIキーを発行する。設定はこんな感じ。

  • 読み取り: ON
  • 取引: ON
  • 出金: OFF（これは絶対OFFにしておく。万が一キーが漏れても出金されない）

Pythonのセットアップ

mkdir crypto-bot && cd crypto-bot
python3 -m venv .venv
source .venv/bin/activate
pip install ccxt pandas python-dotenv

ccxtってライブラリが神で、100以上の取引所を同じインターフェースで叩ける。取引所を乗り換えるときもコードをほぼ変えなくていい。

ちなみに最初pip installしたとき、なぜか=0.2.1みたいな謎の空ファイルがプロジェクトのルートに大量生成されたことがある。なんだこれってなって、結局venv作り直したら直った。（あの空ファイルの正体は未だにわからん）

APIキーの管理

APIキーをコードに直書きするのは論外なので、.envファイルで管理する。

# .env（.gitignoreに追加すること！）
MEXC_API_KEY=your_api_key
MEXC_SECRET=your_secret
DRY_RUN=true
TRADING_AMOUNT_USDT=10
TRADING_SYMBOL=BTC/USDT

設定管理 — config.py

dataclass(frozen=True)でイミュータブルにしてる。一度読み込んだ設定を途中で書き換えられないようにするため。

import os
from dataclasses import dataclass
from dotenv import load_dotenv

_REQUIRED = ("MEXC_API_KEY", "MEXC_SECRET")

@dataclass(frozen=True)
class Config:
    api_key: str
    secret: str
    trading_symbol: str
    trading_amount_usdt: float
    dry_run: bool

    @classmethod
    def from_env(cls, env_path: str | None = None) -> "Config":
        load_dotenv(env_path)

        missing = [v for v in _REQUIRED if not os.environ.get(v)]
        if missing:
            raise ValueError(f"Missing: {', '.join(missing)}")

        dry_run = os.environ.get("DRY_RUN", "true").lower() in ("true", "1", "yes")
        amount = float(os.environ.get("TRADING_AMOUNT_USDT", "10"))

        return cls(
            api_key=os.environ["MEXC_API_KEY"],
            secret=os.environ["MEXC_SECRET"],
            trading_symbol=os.environ.get("TRADING_SYMBOL", "BTC/USDT"),
            trading_amount_usdt=amount,
            dry_run=dry_run,
        )

frozen=Trueにしておくとconfig.dry_run = Falseみたいなことをしようとするとエラーになる。バグの温床をひとつ潰せるので地味に便利。

取引所接続 — exchange.py

ccxtでMEXCに接続するラッパー。

import ccxt
import pandas as pd
from config import Config

class ExchangeClient:
    def __init__(self, config: Config) -> None:
        self._config = config
        self._exchange = ccxt.mexc({
            "apiKey": config.api_key,
            "secret": config.secret,
            "enableRateLimit": True,
        })

    def test_connection(self) -> dict:
        ticker = self._exchange.fetch_ticker(self._config.trading_symbol)
        print(f"接続OK: {self._config.trading_symbol} = ${ticker['last']:.2f}")
        return ticker

    def get_usdt_balance(self) -> float:
        balance = self._exchange.fetch_balance()
        return float(balance.get("free", {}).get("USDT", 0))

    def fetch_ohlcv(self, timeframe: str = "1d", limit: int = 50) -> pd.DataFrame:
        raw = self._exchange.fetch_ohlcv(
            self._config.trading_symbol,
            timeframe=timeframe,
            limit=limit,
        )
        df = pd.DataFrame(raw, columns=["timestamp", "open", "high", "low", "close", "volume"])
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
        return df.set_index("timestamp")

enableRateLimit: Trueにしておくと、ccxtが自動でAPI制限を守ってくれる。最初これ書き忘れてOHLCVをガンガン叩いたら、APIから429が返ってきてしばらくアク禁食らった。30分くらいで復活したけど焦った。これは絶対つけておいたほうがいい。

EMAクロスオーバー戦略

短期EMA（12日）と長期EMA（26日）の2本線が交差するタイミングで売買する。

  • 短期が長期を上抜け → 買い（ゴールデンクロス）
  • 短期が長期を下抜け → 売り（デッドクロス）
from dataclasses import dataclass

@dataclass(frozen=True)
class SignalResult:
    signal: int       # 1=買い, -1=売り, 0=様子見
    fast_ema: float
    slow_ema: float

class EmaStrategy:
    def __init__(self, fast_period: int = 12, slow_period: int = 26) -> None:
        self._fast = fast_period
        self._slow = slow_period

    def generate_signal(self, ohlcv: pd.DataFrame) -> SignalResult:
        close = ohlcv["close"]
        fast_ema = close.ewm(span=self._fast, adjust=False).mean()
        slow_ema = close.ewm(span=self._slow, adjust=False).mean()

        curr_fast = float(fast_ema.iloc[-1])
        curr_slow = float(slow_ema.iloc[-1])
        prev_fast = float(fast_ema.iloc[-2])
        prev_slow = float(slow_ema.iloc[-2])

        if curr_fast > curr_slow and prev_fast <= prev_slow:
            signal = 1   # ゴールデンクロス
        elif curr_fast < curr_slow and prev_fast >= prev_slow:
            signal = -1  # デッドクロス
        else:
            signal = 0

        return SignalResult(signal=signal, fast_ema=curr_fast, slow_ema=curr_slow)

ewm(span=N, adjust=False).mean()がpandasのEMA計算。自分で実装するより正確だし速い。

で、なんでこの戦略を選んだかというと、3年間のバックテストで50戦略を比較した結果、実用的にはこれが一番よかったから。勝率35%って低く見えるけど、勝ちトレードの平均利益が負けの3.2倍あって、トータルでプラスになる構造。

ポジション管理 — bridge.py

シグナルが出たからって無条件に注文を出すわけじゃない。「今ポジション持ってるか」で行動が変わる。

from enum import Enum

class OrderAction(Enum):
    BUY = "buy"
    SELL = "sell"
    HOLD = "hold"

class PositionState(Enum):
    FLAT = "flat"   # ノーポジ
    LONG = "long"   # 買いポジ保有中

class SignalBridge:
    def __init__(self) -> None:
        self._position = PositionState.FLAT

    def determine_action(self, signal: int) -> OrderAction:
        if signal == 0:
            return OrderAction.HOLD

        if self._position == PositionState.FLAT and signal == 1:
            return OrderAction.BUY
        if self._position == PositionState.LONG and signal == -1:
            return OrderAction.SELL

        return OrderAction.HOLD

    def update_position(self, new_state: PositionState) -> None:
        self._position = new_state

ショートは最初はやらない。ロングだけ。このへんのコードは見ればわかると思うので細かい説明は省くけど、要はシグナルが来ても今のポジション次第で無視するってだけ。

あと再起動でポジション消えると困るのでJSONに保存してる。

# position_state.json
{"position": "long"}

リスク管理 — サーキットブレーカー

ここが一番大事かもしれない。自動売買で怖いのは「Botが暴走して全財産溶かす」パターン。

class CircuitBreaker:
    def __init__(self, initial_balance: float) -> None:
        self._balance = initial_balance
        self._daily_threshold = 0.03    # 日次-3%で停止
        self._weekly_threshold = 0.07   # 週次-7%で停止
        self._monthly_threshold = 0.15  # 月次-15%で完全停止

    def is_trading_allowed(self) -> bool:
        # 日次/週次/月次の損失をチェック
        # 閾値を超えたらFalseを返す
        ...
レベル 閾値 回復
日次 -3% 翌日UTC 0:00
週次 -7% 翌週月曜
月次 -15% 手動リセットのみ

月次のキルスイッチは手動リセットしか効かないようにしてある。-15%やられたら一回冷静になったほうがいい。（経験則）

メインループ

全部つなげて1時間ごとに回すループ。

import signal
import time

def run_bot() -> None:
    config = Config.from_env()
    client = ExchangeClient(config)
    strategy = EmaStrategy()
    bridge = SignalBridge()
    breaker = CircuitBreaker(initial_balance=client.get_usdt_balance())

    client.test_connection()
    running = True

    def on_shutdown(signum, frame):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, on_shutdown)
    signal.signal(signal.SIGTERM, on_shutdown)

    while running:
        if not breaker.is_trading_allowed():
            print("サーキットブレーカー発動中")
        else:
            ohlcv = client.fetch_ohlcv(timeframe="1d", limit=50)
            sig = strategy.generate_signal(ohlcv)
            action = bridge.determine_action(sig.signal)

            if action == OrderAction.BUY:
                # 買い注文 + ストップロス設定
                pass
            elif action == OrderAction.SELL:
                # 売り注文
                pass

        for _ in range(3600):
            if not running:
                break
            time.sleep(1)

time.sleep(3600)を1回呼ぶんじゃなくて、1秒×3600回のループにしてるのがポイント。こうするとCtrl+Cですぐ止まる。3600秒のsleepだとSIGINT来ても3600秒待たないと止まらない。（最初やらかした）

DRY_RUNで動作確認

本番の金を使う前に、DRY_RUNモードでサクッとテストする。

# .envを確認
DRY_RUN=true

# 起動
python main.py

DRY_RUNだと注文は送信されず、ログにこういうのが出る。

[DRY_RUN] Market BUY BTC/USDT: amount=0.00015000 (not executed)

確認すること:

  • API接続が通るか
  • シグナルが出るか
  • ログが出てるか
  • サーキットブレーカーが効くか

全部OKだったらDRY_RUNを外す。

DRY_RUN=false
TRADING_AMOUNT_USDT=10  # まずは少額から

バックテスト結果（参考）

指標 値
期間 2023/1 〜 2026/2（3年間）
通貨ペア BTC/USDT 日足
年率リターン 77.2%
シャープレシオ 1.30
最大ドローダウン -34.0%
取引回数 34回（月0.9回）
勝率 35.3%
プロフィットファクター 1.73

50戦略を比較して、EMAクロスオーバーが実用的にベストだった。1位のmulti_timeframeはシャープ1.50だけど取引2回で統計的に信頼できない。詳しくは別記事に書いた。

ただしこの期間はBTCの上昇トレンドが強かったので、トレンドフォロー系に有利な環境だったのは間違いない。

ちなみに50戦略の中にはML（機械学習）系も6個くらい入れてたんだけど、全部取引回数ゼロだった。データ足りなくて学習がうまくいかなかったっぽい。MLでBot作ろうとしてた時期もあったけど、結局シンプルなテクニカル指標のほうが実用的だったという。

本番で気をつけること

バックテストと実運用は別物。実際にやってみて気づいたこと。

  • スリッページ: 成行注文だと思った価格で約定しないことがある。日足ならほぼ影響ないけど
  • API遅延: シグナル検出から注文まで数秒かかる。これも日足なら問題ない
  • メンタル: 5連敗くらい普通にある。35%の勝率ってそういうこと。事前にわかってても実際に食らうとキツい
  • 少額スタート: 僕は$1から始めた。いきなり全力はやめたほうがいい

ファイル構成

ファイルはこんな感じで分けてる。config.py、exchange.py、ema_strategy.py、bridge.py、circuit_breaker.py、main.py。それぞれ200行くらい。1ファイルに全部書くと見通し悪くなるので、機能ごとにバラした。正直もっと雑に1ファイルでもいいんだけど、後から戦略追加するとき分けてあると楽。

取引所の選び方

ccxtを使えば取引所の乗り換えは楽。ccxt.mexcをccxt.binanceに変えるだけ。

僕はMEXCを使ってる。手数料が安いのとAPIの挙動が素直なのが理由。あと日本語対応してるので口座開設が楽だった。

MEXCで口座開設する

↑のリンクから登録すると手数料の割引が適用されます。

次にやりたいこと

  • 複数通貨ペアの同時運用（SOL/USDTがバックテストでシャープ1.35出てて、ジワジワ気になってる）
  • MACD戦略の追加（EMAと並行運用でリスク分散）
  • Telegram通知（売買のたびにスマホに通知が来るようにしたい）

自動売買は作って終わりじゃなくて、ちょこちょこ改善していくのが面白い。

※投資は自己責任でお願いします。バックテスト結果は将来の利益を保証するものではありません。記事内のリンクにはアフィリエイトが含まれます。

Discussion