はじめに

第1回では、CCXTの基本概念から環境構築、市場データの取得方法までを解説しました。今回の第2回では、いよいよ実際の取引を実行する方法と、基本的な取引戦略の実装について詳しく解説します。

重要な注意事項: この記事では実際の取引を扱います。必ず少額でのテストから始め、十分に理解してから本格的な運用を行ってください。

残高確認の実装

取引を始める前に、まずは自分のアカウント残高を確認する方法を学びましょう。

基本的な残高確認

import ccxt
import os

# APIキーの設定（環境変数から取得）
exchange = ccxt.coincheck({
    'apiKey': os.getenv('COINCHECK_API_KEY'),
    'secret': os.getenv('COINCHECK_SECRET_KEY'),
    'enableRateLimit': True,
})

try:
    # 残高情報を取得
    balance = exchange.fetch_balance()

    print("=== アカウント残高 ===")
    for currency, amount in balance['total'].items():
        if amount > 0:  # 残高がある通貨のみ表示
            print(f"{currency}: {amount}")

    # 特定通貨の詳細情報
    print(f"\n=== BTC詳細 ===")
    print(f"総残高: {balance['BTC']['total']} BTC")
    print(f"利用可能: {balance['BTC']['free']} BTC")
    print(f"注文中: {balance['BTC']['used']} BTC")

    print(f"\n=== JPY詳細 ===")
    print(f"総残高: ¥{balance['JPY']['total']:,.0f}")
    print(f"利用可能: ¥{balance['JPY']['free']:,.0f}")
    print(f"注文中: ¥{balance['JPY']['used']:,.0f}")

except ccxt.AuthenticationError:
    print("認証エラー: APIキーまたはシークレットキーが正しくありません")
except ccxt.NetworkError as e:
    print(f"ネットワークエラー: {e}")
except Exception as e:
    print(f"エラー: {e}")

残高データの理解

• total: 総残高（利用可能 + 注文中）
• free: 利用可能残高（新規注文に使用可能）
• used: 拘束残高（未約定注文で使用中）

実際のトレード実装

成行注文の実行

成行注文は、現在の市場価格で即座に売買を実行する注文方法です。

def place_market_order(exchange, symbol, side, amount):
    """
    成行注文を実行する関数

    Args:
        exchange: CCXT取引所インスタンス
        symbol: 通貨ペア（例: 'BTC/JPY'）
        side: 'buy' または 'sell'
        amount: 注文数量
    """
    try:
        if side == 'buy':
            order = exchange.create_market_buy_order(symbol, amount)
            print(f"買い注文実行: {amount} {symbol.split('/')[0]} を成行で購入")
        elif side == 'sell':
            order = exchange.create_market_sell_order(symbol, amount)
            print(f"売り注文実行: {amount} {symbol.split('/')[0]} を成行で売却")

        print(f"注文ID: {order['id']}")
        print(f"注文状況: {order['status']}")
        print(f"実行価格: ¥{order['price']:,.0f}" if order['price'] else "市場価格")

        return order

    except ccxt.InsufficientFunds:
        print("エラー: 残高不足です")
    except ccxt.InvalidOrder as e:
        print(f"注文エラー: {e}")
    except Exception as e:
        print(f"予期しないエラー: {e}")

    return None

# 使用例：0.001 BTCを成行で購入
# order = place_market_order(exchange, 'BTC/JPY', 'buy', 0.001)

指値注文の実行

指値注文は、指定した価格で売買を行う注文方法です。

def place_limit_order(exchange, symbol, side, amount, price):
    """
    指値注文を実行する関数

    Args:
        exchange: CCXT取引所インスタンス
        symbol: 通貨ペア（例: 'BTC/JPY'）
        side: 'buy' または 'sell'
        amount: 注文数量
        price: 指値価格
    """
    try:
        if side == 'buy':
            order = exchange.create_limit_buy_order(symbol, amount, price)
            print(f"指値買い注文: {amount} {symbol.split('/')[0]} を ¥{price:,.0f} で注文")
        elif side == 'sell':
            order = exchange.create_limit_sell_order(symbol, amount, price)
            print(f"指値売り注文: {amount} {symbol.split('/')[0]} を ¥{price:,.0f} で注文")

        print(f"注文ID: {order['id']}")
        print(f"注文状況: {order['status']}")

        return order

    except ccxt.InsufficientFunds:
        print("エラー: 残高不足です")
    except ccxt.InvalidOrder as e:
        print(f"注文エラー: {e}")
    except Exception as e:
        print(f"予期しないエラー: {e}")

    return None

# 使用例：0.001 BTCを500万円で指値売り
# order = place_limit_order(exchange, 'BTC/JPY', 'sell', 0.001, 5000000)

注文状況の確認とキャンセル

def check_and_manage_orders(exchange, symbol=None):
    """
    注文状況を確認し、管理する関数
    """
    try:
        # 未約定注文の取得
        if symbol:
            open_orders = exchange.fetch_open_orders(symbol)
        else:
            open_orders = exchange.fetch_open_orders()

        if not open_orders:
            print("未約定注文はありません")
            return

        print("=== 未約定注文一覧 ===")
        for order in open_orders:
            print(f"注文ID: {order['id']}")
            print(f"通貨ペア: {order['symbol']}")
            print(f"種類: {order['side']} / {order['type']}")
            print(f"数量: {order['amount']}")
            print(f"価格: ¥{order['price']:,.0f}" if order['price'] else "成行")
            print(f"状況: {order['status']}")
            print("-" * 30)

        return open_orders

    except Exception as e:
        print(f"注文確認エラー: {e}")
        return []

def cancel_order(exchange, order_id, symbol):
    """
    指定した注文をキャンセルする関数
    """
    try:
        result = exchange.cancel_order(order_id, symbol)
        print(f"注文 {order_id} をキャンセルしました")
        return result
    except Exception as e:
        print(f"注文キャンセルエラー: {e}")
        return None

# 使用例
# orders = check_and_manage_orders(exchange, 'BTC/JPY')
# if orders:
#     cancel_order(exchange, orders[0]['id'], 'BTC/JPY')

トレード記録の確認

def get_trade_history(exchange, symbol, limit=10):
    """
    トレード記録を取得する関数
    """
    try:
        trades = exchange.fetch_my_trades(symbol, limit=limit)

        print(f"=== {symbol} トレード記録（最新{limit}件）===")
        for trade in trades:
            timestamp = pd.to_datetime(trade['timestamp'], unit='ms')
            print(f"日時: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
            print(f"種類: {trade['side']}")
            print(f"数量: {trade['amount']}")
            print(f"価格: ¥{trade['price']:,.0f}")
            print(f"手数料: {trade['fee']['cost']} {trade['fee']['currency']}")
            print(f"注文ID: {trade.get('order', '')}")  # 注文IDを追加
            print(f"状況: {trade.get('status', '')}")   # ステータスを追加（取引所による）
            print("-" * 30)

        # トレード記録のデータ構造例（第4回のTradeRecorderと連携しやすい形）
        # {
        #   'timestamp': ...,
        #   'symbol': ...,
        #   'side': ...,
        #   'amount': ...,
        #   'price': ...,
        #   'fee': ...,
        #   'order_id': ...,
        #   'status': ...
        # }

        return trades

    except Exception as e:
        print(f"トレード記録取得エラー: {e}")
        return []

# 使用例
# trades = get_trade_history(exchange, 'BTC/JPY', 5)

簡単な取引戦略の実装

移動平均を使った戦略

移動平均クロス戦略は、短期移動平均と長期移動平均の交差を売買シグナルとする基本的な戦略です。

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class MovingAverageStrategy:
    def __init__(self, exchange, symbol, short_window=5, long_window=20):
        self.exchange = exchange
        self.symbol = symbol
        self.short_window = short_window
        self.long_window = long_window
        self.position = None  # 'long', 'short', None

    def get_data(self, timeframe='1h', limit=100):
        """過去データを取得してDataFrameに変換"""
        try:
            ohlcv = self.exchange.fetch_ohlcv(self.symbol, timeframe, limit=limit)
            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
            return df
        except Exception as e:
            print(f"データ取得エラー: {e}")
            return None

    def calculate_signals(self, df):
        """移動平均とシグナルを計算"""
        # 移動平均の計算
        df[f'ma_short'] = df['close'].rolling(window=self.short_window).mean()
        df[f'ma_long'] = df['close'].rolling(window=self.long_window).mean()

        # シグナルの生成
        df['signal'] = 0
        df['signal'][self.short_window:] = np.where(
            df[f'ma_short'][self.short_window:] > df[f'ma_long'][self.short_window:], 1, 0
        )

        # シグナルの変化点を検出
        df['position'] = df['signal'].diff()

        return df

    def analyze_market(self):
        """市場分析を実行"""
        df = self.get_data()
        if df is None:
            return None

        df = self.calculate_signals(df)

        # 最新のシグナル情報
        latest = df.iloc[-1]
        prev = df.iloc[-2]

        current_price = latest['close']
        ma_short = latest[f'ma_short']
        ma_long = latest[f'ma_long']

        print(f"=== {self.symbol} 市場分析 ===")
        print(f"現在価格: ¥{current_price:,.0f}")
        print(f"短期MA({self.short_window}): ¥{ma_short:,.0f}")
        print(f"長期MA({self.long_window}): ¥{ma_long:,.0f}")

        # シグナル判定
        signal = None
        if latest['position'] == 1:  # ゴールデンクロス
            signal = 'BUY'
            print("🟢 買いシグナル: ゴールデンクロス発生")
        elif latest['position'] == -1:  # デッドクロス
            signal = 'SELL'
            print("🔴 売りシグナル: デッドクロス発生")
        else:
            print("⚪ ホールド: シグナルなし")

        return {
            'signal': signal,
            'price': current_price,
            'ma_short': ma_short,
            'ma_long': ma_long,
            'dataframe': df
        }

# 使用例
# strategy = MovingAverageStrategy(exchange, 'BTC/JPY', short_window=5, long_window=20)
# analysis = strategy.analyze_market()

RSIを使った戦略

RSI（相対力指数）は、買われすぎ・売られすぎを判定するテクニカル指標です。

class RSIStrategy:
    def __init__(self, exchange, symbol, period=14, oversold=30, overbought=70):
        self.exchange = exchange
        self.symbol = symbol
        self.period = period
        self.oversold = oversold
        self.overbought = overbought

    def calculate_rsi(self, prices, period=14):
        """RSIを計算"""
        deltas = np.diff(prices)
        seed = deltas[:period+1]
        up = seed[seed >= 0].sum() / period
        down = -seed[seed < 0].sum() / period
        rs = up / down
        rsi = np.zeros_like(prices)
        rsi[:period] = 100. - 100. / (1. + rs)

        for i in range(period, len(prices)):
            delta = deltas[i-1]
            if delta > 0:
                upval = delta
                downval = 0.
            else:
                upval = 0.
                downval = -delta

            up = (up * (period - 1) + upval) / period
            down = (down * (period - 1) + downval) / period
            rs = up / down
            rsi[i] = 100. - 100. / (1. + rs)

        return rsi

    def analyze_market(self):
        """RSI分析を実行"""
        try:
            ohlcv = self.exchange.fetch_ohlcv(self.symbol, '1h', limit=100)
            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

            # RSI計算
            df['rsi'] = self.calculate_rsi(df['close'].values, self.period)

            latest_rsi = df['rsi'].iloc[-1]
            current_price = df['close'].iloc[-1]

            print(f"=== {self.symbol} RSI分析 ===")
            print(f"現在価格: ¥{current_price:,.0f}")
            print(f"RSI({self.period}): {latest_rsi:.2f}")

            # シグナル判定
            signal = None
            if latest_rsi <= self.oversold:
                signal = 'BUY'
                print(f"🟢 買いシグナル: RSI {latest_rsi:.2f} <= {self.oversold} (売られすぎ)")
            elif latest_rsi >= self.overbought:
                signal = 'SELL'
                print(f"🔴 売りシグナル: RSI {latest_rsi:.2f} >= {self.overbought} (買われすぎ)")
            else:
                print("⚪ ホールド: 通常範囲内")

            return {
                'signal': signal,
                'rsi': latest_rsi,
                'price': current_price,
                'dataframe': df
            }

        except Exception as e:
            print(f"RSI分析エラー: {e}")
            return None

# 使用例
# rsi_strategy = RSIStrategy(exchange, 'BTC/JPY', period=14, oversold=30, overbought=70)
# rsi_analysis = rsi_strategy.analyze_market()

複合戦略の実装

複数の指標を組み合わせた戦略を実装してみましょう。

class CombinedStrategy:
    def __init__(self, exchange, symbol, trade_amount=0.001):
        self.exchange = exchange
        self.symbol = symbol
        self.trade_amount = trade_amount
        self.ma_strategy = MovingAverageStrategy(exchange, symbol)
        self.rsi_strategy = RSIStrategy(exchange, symbol)
        self.position = None

    def analyze_and_trade(self):
        """分析を実行し、条件が揃えば取引を実行"""
        print(f"=== 複合戦略分析開始: {self.symbol} ===")

        # 各戦略の分析実行
        ma_analysis = self.ma_strategy.analyze_market()
        rsi_analysis = self.rsi_strategy.analyze_market()

        if not ma_analysis or not rsi_analysis:
            print("分析データの取得に失敗しました")
            return

        # 複合判定
        ma_signal = ma_analysis['signal']
        rsi_signal = rsi_analysis['signal']

        print(f"\n=== 総合判定 ===")
        print(f"移動平均戦略: {ma_signal}")
        print(f"RSI戦略: {rsi_signal}")

        # 両方のシグナルが一致した場合のみ取引実行
        if ma_signal == 'BUY' and rsi_signal == 'BUY':
            print("🟢 強い買いシグナル: 取引実行を検討")
            if self.position != 'long':
                self.execute_buy()
        elif ma_signal == 'SELL' and rsi_signal == 'SELL':
            print("🔴 強い売りシグナル: 取引実行を検討")
            if self.position == 'long':
                self.execute_sell()
        else:
            print("⚪ シグナル不一致: ホールド")

    def execute_buy(self):
        """買い注文実行"""
        print(f"\n💰 買い注文実行: {self.trade_amount} {self.symbol.split('/')[0]}")

        # 実際の注文実行（コメントアウトして安全性を確保）
        # order = place_market_order(self.exchange, self.symbol, 'buy', self.trade_amount)
        # if order:
        #     self.position = 'long'
        #     print("ポジション: ロング")

        # デモモード
        print("※ デモモード: 実際の注文は実行されていません")
        self.position = 'long'

    def execute_sell(self):
        """売り注文実行"""
        print(f"\n💸 売り注文実行: {self.trade_amount} {self.symbol.split('/')[0]}")

        # 実際の注文実行（コメントアウトして安全性を確保）
        # order = place_market_order(self.exchange, self.symbol, 'sell', self.trade_amount)
        # if order:
        #     self.position = None
        #     print("ポジション: なし")

        # デモモード
        print("※ デモモード: 実際の注文は実行されていません")
        self.position = None

# 使用例
# combined_strategy = CombinedStrategy(exchange, 'BTC/JPY', trade_amount=0.001)
# combined_strategy.analyze_and_trade()

自動取引ボットの基本構造

定期的に市場を監視し、取引を実行する基本的なボットを作成してみましょう。

import time
import logging
from datetime import datetime

class TradingBot:
    def __init__(self, exchange, symbol, strategy, interval=300):
        self.exchange = exchange
        self.symbol = symbol
        self.strategy = strategy
        self.interval = interval  # 監視間隔（秒）
        self.running = False

        # ログ設定
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('trading_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    def start(self):
        """ボット開始"""
        self.running = True
        self.logger.info(f"取引ボット開始: {self.symbol}")

        while self.running:
            try:
                self.logger.info("市場分析実行中...")
                self.strategy.analyze_and_trade()

                self.logger.info(f"{self.interval}秒待機中...")
                time.sleep(self.interval)

            except KeyboardInterrupt:
                self.logger.info("ユーザーによる停止")
                self.stop()
            except Exception as e:
                self.logger.error(f"予期しないエラー: {e}")
                time.sleep(60)  # エラー時は1分待機

    def stop(self):
        """ボット停止"""
        self.running = False
        self.logger.info("取引ボット停止")

# 使用例（実際に実行する場合は十分注意してください）
# bot = TradingBot(exchange, 'BTC/JPY', combined_strategy, interval=300)
# bot.start()  # 5分間隔で監視開始

まとめ

第2回では、CCXTを使った実際の取引実装から基本的な取引戦略まで幅広く解説しました。重要なポイントを振り返ってみましょう：

1. 注文管理: 成行・指値注文の実行方法と注文状況の確認
2. 戦略実装: 移動平均とRSIを使った基本的な取引戦略
3. 複合判定: 複数の指標を組み合わせた総合的な判断
4. 自動化: 基本的な取引ボットの構造

次回（第3回）では、リスク管理やバックテスト、パフォーマンス評価について詳しく解説する予定です。

実践時の注意事項

1. 少額テスト: 必ず少額から始めて動作を確認
2. ログ記録: 全ての取引を記録し、後で分析できるようにする
3. リスク管理: 損失限度額を事前に決めて厳守する
4. 市場監視: 重要な経済指標発表時などは手動監視を強化
5. 定期見直し: 戦略の有効性を定期的に評価し、必要に応じて修正

---

警告: 実際の取引では大きな損失を被る可能性があります。十分な検証と理解なしに自動取引を行わないでください。投資は自己責任で行ってください。

---

関連記事:

• CCXTを使って仮想通貨のトレードをしてみる（第1回）
• CCXTを使って仮想通貨のトレードをしてみる（第3回）
• CCXTで作る仮想通貨自動取引ボット
• Pythonの環境構築