Bot トレードロジック

仮想通貨botの開発記録#59(2024/2/24)「ビットコインの積立Botを作ってみる」

2024年2月24日

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。

今回はChatGPT4を使って「ビットコインの積立Bot」のコードを書きました。

参考記事

ビットコイン積立bot（国内取引所用）

初心者の仮想通貨botterが最初に作ってみればいいbot

Yodaka

実装するロジックは以下の通り。

  • 毎日1万円分のビットコインを買う
  • ドルコスト平均法に基づいて運用する
  • scheduleモジュールでBotの定期実行を管理する
  • ccxtライブラリを使う

成り行き注文ver.

Yodaka

まずは成り行き注文で素早く約定させる設定です。

import ccxt
import schedule
import time

# bitFlyerのAPIキーとシークレット
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使ってbitFlyerに接続
exchange = ccxt.bitflyer({
    'apiKey': api_key,
    'secret': api_secret,
})

# 積立購入を行う関数
def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア

    # JPYの価格でBTCを購入するための金額を計算
    price = exchange.fetch_ticker(symbol)['last']  # 最終取引価格を取得
    amount_btc = amount_yen / price  # 購入するBTCの量を計算

    # マーケットオーダーでBTCを購入
    order = exchange.create_market_buy_order(symbol, amount_btc)
    print(f"Purchased {amount_btc} BTC for {amount_yen} JPY")

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

指し値注文ver.

Yodaka

次に、指し値注文を出して大きな損失を抑える方法です。
ここでは、市場価格よりも10円だけ高く指値を置くことで素早く約定できるようにしています。

import ccxt
import schedule
import time

# bitFlyerのAPIキーとシークレット
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使ってbitFlyerに接続
exchange = ccxt.bitflyer({
    'apiKey': api_key,
    'secret': api_secret,
})

# 積立購入を行う関数
def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア

    # JPYの価格でBTCを購入するための金額を計算
    ticker = exchange.fetch_ticker(symbol)
    last_price = ticker['last']  # 最終取引価格を取得
    limit_price = last_price + 10  # マーケット価格から10円だけ高い価格を設定
    
    # 購入するBTCの量を計算（リミット価格を使用）
    amount_btc = amount_yen / limit_price

    # リミットオーダーでBTCを購入
    order = exchange.create_limit_buy_order(symbol, amount_btc, limit_price)
    print(f"Placed limit order for {amount_btc} BTC at {limit_price} JPY")

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

指値注文ver.(改)

Yodaka

指し値注文が約定しなかった場合は、しばらく待機してから注文をキャンセルしてから、再び指値注文を出し直すタイプです。

import ccxt
import schedule
import time

# bitFlyerのAPIキーとシークレット
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使ってbitFlyerに接続
exchange = ccxt.bitflyer({
    'apiKey': api_key,
    'secret': api_secret,
})

# 積立購入を行う関数
def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア

    while True:
        try:
            # JPYの価格でBTCを購入するための金額を計算
            ticker = exchange.fetch_ticker(symbol)
            last_price = ticker['last']  # 最終取引価格を取得
            limit_price = last_price + 10  # マーケット価格から10円だけ高い価格を設定
            
            # 購入するBTCの量を計算（リミット価格を使用）
            amount_btc = amount_yen / limit_price

            # リミットオーダーでBTCを購入
            order = exchange.create_limit_buy_order(symbol, amount_btc, limit_price)
            print(f"Placed limit order for {amount_btc} BTC at {limit_price} JPY")

            # 注文が成立するか30秒待機
            time.sleep(30)

            # 注文の状態を確認
            order_status = exchange.fetch_order(order['id'], symbol)

            if order_status['status'] == 'closed':  # 注文が成立した場合
                print("Order filled.")
                break  # ループを抜ける
            else:  # 注文が成立しなかった場合
                print("Order not filled. Cancelling and retrying...")
                exchange.cancel_order(order['id'], symbol)  # 注文をキャンセル
        except Exception as e:
            print(f"An error occurred: {e}")
            break  # エラーが発生した場合はループを抜ける

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

購入ログをcsvファイルで出力する

Yodaka

購入ログをcsvファイルに出力することで、購入履歴を保管しておくこともできます。

import ccxt
import schedule
import time
import csv
from datetime import datetime

# bitFlyerのAPIキーとシークレット
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使ってbitFlyerに接続
exchange = ccxt.bitflyer({
    'apiKey': api_key,
    'secret': api_secret,
})

# CSVファイル名の設定
csv_filename = 'purchase_log.csv'

# 積立購入を行う関数
def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア

    while True:
        try:
            # JPYの価格でBTCを購入するための金額を計算
            ticker = exchange.fetch_ticker(symbol)
            last_price = ticker['last']  # 最終取引価格を取得
            limit_price = last_price + 10  # マーケット価格から10円だけ高い価格を設定
            
            # 購入するBTCの量を計算（リミット価格を使用）
            amount_btc = amount_yen / limit_price

            # リミットオーダーでBTCを購入
            order = exchange.create_limit_buy_order(symbol, amount_btc, limit_price)
            print(f"Placed limit order for {amount_btc} BTC at {limit_price} JPY")

            # 注文が成立するか30秒待機
            time.sleep(30)

            # 注文の状態を確認
            order_status = exchange.fetch_order(order['id'], symbol)

            if order_status['status'] == 'closed':  # 注文が成立した場合
                print("Order filled.")
                # CSVファイルに記録
                with open(csv_filename, 'a', newline='') as file:
                    writer = csv.writer(file)
                    # ヘッダーの書き込み（ファイルが空の場合のみ）
                    if file.tell() == 0:
                        writer.writerow(['Timestamp', 'BTC Amount', 'Limit Price (JPY)'])
                    writer.writerow([datetime.now(), amount_btc, limit_price])
                break  # ループを抜ける
            else:  # 注文が成立しなかった場合
                print("Order not filled. Cancelling and retrying...")
                exchange.cancel_order(order['id'], symbol)  # 注文をキャンセル
        except Exception as e:
            print(f"An error occurred: {e}")
            break  # エラーが発生した場合はループを抜ける

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

bitbank用

import ccxt
import schedule
import time
import csv
from datetime import datetime

# bitbankのAPIキーとシークレット
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使ってbitbankに接続
exchange = ccxt.bitbank({
    'apiKey': api_key,
    'secret': api_secret,
})

# CSVファイル名の設定
csv_filename = 'purchase_log_bitbank.csv'

# 積立購入を行う関数
def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア

    while True:
        try:
            # JPYの価格でBTCを購入するための金額を計算
            ticker = exchange.fetch_ticker(symbol)
            last_price = ticker['last']  # 最終取引価格を取得
            limit_price = last_price + 10  # マーケット価格から10円だけ高い価格を設定
            
            # 購入するBTCの量を計算（リミット価格を使用）
            amount_btc = amount_yen / limit_price

            # リミットオーダーでBTCを購入
            order = exchange.create_limit_buy_order(symbol, amount_btc, limit_price)
            print(f"Placed limit order for {amount_btc} BTC at {limit_price} JPY on bitbank")

            # 注文が成立するか30秒待機
            time.sleep(30)

            # 注文の状態を確認
            order_status = exchange.fetch_order(order['id'], symbol)

            if order_status['status'] == 'closed':  # 注文が成立した場合
                print("Order filled on bitbank.")
                # CSVファイルに記録
                with open(csv_filename, 'a', newline='') as file:
                    writer = csv.writer(file)
                    # ヘッダーの書き込み（ファイルが空の場合のみ）
                    if file.tell() == 0:
                        writer.writerow(['Timestamp', 'BTC Amount', 'Limit Price (JPY)'])
                    writer.writerow([datetime.now(), amount_btc, limit_price])
                break  # ループを抜ける
            else:  # 注文が成立しなかった場合
                print("Order not filled on bitbank. Cancelling and retrying...")
                exchange.cancel_order(order['id'], symbol)  # 注文をキャンセル
        except Exception as e:
            print(f"An error occurred on bitbank: {e}")
            break  # エラーが発生した場合はループを抜ける

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

CoinCheck用

import ccxt
import schedule
import time
import csv
from datetime import datetime

# CoinCheckのAPIキーとシークレット
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使ってCoinCheckに接続（サポートされている場合）
exchange = ccxt.coincheck({
    'apiKey': api_key,
    'secret': api_secret,
})

# CSVファイル名の設定
csv_filename = 'purchase_log_coincheck.csv'

# 積立購入を行う関数
def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア

    while True:
        try:
            # JPYの価格でBTCを購入するための金額を計算
            ticker = exchange.fetch_ticker(symbol)
            last_price = ticker['last']  # 最終取引価格を取得
            limit_price = last_price + 10  # マーケット価格から10円だけ高い価格を設定
            
            # 購入するBTCの量を計算（リミット価格を使用）
            amount_btc = amount_yen / limit_price

            # リミットオーダーでBTCを購入
            order = exchange.create_limit_buy_order(symbol, amount_btc, limit_price)
            print(f"Placed limit order for {amount_btc} BTC at {limit_price} JPY on bitbank")

            # 注文が成立するか30秒待機
            time.sleep(30)

            # 注文の状態を確認
            order_status = exchange.fetch_order(order['id'], symbol)

            if order_status['status'] == 'closed':  # 注文が成立した場合
                print("Order filled on bitbank.")
                # CSVファイルに記録
                with open(csv_filename, 'a', newline='') as file:
                    writer = csv.writer(file)
                    # ヘッダーの書き込み（ファイルが空の場合のみ）
                    if file.tell() == 0:
                        writer.writerow(['Timestamp', 'BTC Amount', 'Limit Price (JPY)'])
                    writer.writerow([datetime.now(), amount_btc, limit_price])
                break  # ループを抜ける
            else:  # 注文が成立しなかった場合
                print("Order not filled on bitbank. Cancelling and retrying...")
                exchange.cancel_order(order['id'], symbol)  # 注文をキャンセル
        except Exception as e:
            print(f"An error occurred on bitbank: {e}")
            break  # エラーが発生した場合はループを抜ける

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

おまけ：ChatGPTに問題点を指摘させる

以下のような質問をして、ChatGPTにコード修正案を提案させてみました。

Yodaka

上記の積立botの構造にはリスク管理の点で何か問題はあるだろうか？長期にわたって運用することが目的なので、購入にあたって損失をできる限り抑えたい。改善案を教えて。

提供された積立Botの構造には、リスク管理の観点からいくつかの考慮すべき点があります。長期運用を目指し、損失を最小限に抑えるためには、以下のような改善が必要です。

1. マーケットの変動に対する考慮の欠如

  • 問題点: 現在の構造では、マーケット価格に基づいて一律に購入を行いますが、市場の変動性やトレンドに応じて購入戦略を調整する仕組みがありません。
  • 改善案: テクニカル分析や市場のトレンド分析を取り入れ、市場が下降トレンドの時は購入額を減らし、上昇トレンドでは積極的に購入するなど、動的な購入戦略を採用する。

2. 価格スリッページへの対策不足

  • 問題点: 特にリミットオーダーが成立しなかった場合、再注文時に市場価格が大きく変動している可能性があります。
  • 改善案: 注文の成立を待つ時間を調整するか、成立しない場合のリトライロジックに価格変動のチェックを入れ、一定範囲以上の価格変動があった場合は注文を見送るなどの対策を講じる。

3. 注文量の固定

  • 問題点: 一定額の購入を続けることで、価格が高騰している時には少ない量しか購入できず、価格が下落している時には多くの量を購入することになります。
  • 改善案: ドルコスト平均法（DCA）を応用し、市場価格に応じて購入量を調整する。または、バリューアベレージング戦略を取り入れ、ポートフォリオのバランスを考慮して購入額を調整する。

4. リスク分散の不足

  • 問題点: 単一の暗号通貨に対する積立のみでは、その暗号通貨の市場動向に大きく左右されるリスクがあります。
  • 改善案: 複数の暗号通貨に対して積立を行うことでリスク分散を図る。また、暗号通貨以外の資産（株式、債券など）にも分散投資を行うことを検討する。

5. 情報セキュリティとプライバシー

  • 問題点: APIキーとシークレットは、不正アクセスによる損失のリスクがあります。
  • 改善案: APIキーとシークレットの安全な管理（例：環境変数の利用、秘密管理ツールの利用）、二要素認証の有効化など、セキュリティ対策を強化する。

6. 自動取引ルールの監視と評価

  • 問題点: 市場状況は常に変動しているため、一度設定したルールが常に最適とは限りません。
  • 改善案: 定期的に取引ルールのパフォーマンスをレビューし、必要に応じて戦略を調整する。また、機械学習を活用して最適な取引戦略を学習させるアプローチも考えられます。

長期にわたって積立を行い、損失を抑えつつ安定したリターンを目指すには、上記のようなリスク管理戦略を総合的に検討し、適切に実装することが重要です。

Yodaka

上記の2,3を加味して、コード修正をさせました。

import ccxt
import schedule
import time
import csv
from datetime import datetime

# APIキーとシークレットを設定
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使って取引所に接続
exchange = ccxt.bitbank({
    'apiKey': api_key,
    'secret': api_secret,
})

# CSVファイル名の設定
csv_filename = 'purchase_log.csv'

def accumulate_bitcoin():
    amount_yen = 10000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア
    max_price_diff = 10  # 許容する価格差（円）

    try:
        ticker = exchange.fetch_ticker(symbol)
        last_price = ticker['last']
        order_price = last_price + 10  # 10円高い指し値

        # 注文量を計算
        amount_btc = amount_yen / order_price
        
        # 注文を出す
        order = exchange.create_limit_buy_order(symbol, amount_btc, order_price)
        print(f"Placed limit order for {amount_btc} BTC at {order_price} JPY")

        # 30秒待機
        time.sleep(30)

        order_status = exchange.fetch_order(order['id'], symbol)
        if order_status['status'] == 'closed':
            print("Order filled.")
        else:
            print("Order not filled within 30 seconds. Checking price difference...")
            new_ticker = exchange.fetch_ticker(symbol)
            new_price = new_ticker['last']
            price_diff = new_price - last_price

            if abs(price_diff) <= max_price_diff:
                print("Price difference within range, cancelling and retrying...")
                exchange.cancel_order(order['id'], symbol)
            else:
                print("Price difference too large, not retrying.")
                exchange.cancel_order(order['id'], symbol)

    except Exception as e:
        print(f"An error occurred: {e}")

# 毎日特定の時間に積立購入を実行するスケジュール設定
schedule.every().day.at("10:00").do(accumulate_bitcoin)

# スケジューラを起動し、設定されたタスクを実行
while True:
    schedule.run_pending()
    time.sleep(1)

追記：ジョブスケジューラCron Jobsを使う方法

Yodaka

本記事の公開後に@daniel_tanaka_さんからアイデアを頂きました。便利なツールの情報共有に感謝です。

Cron Jobsとは？

Cron Jobs（クロンジョブ）は、UNIX系オペレーティングシステムで使用される時間ベースのジョブスケジューラです。これを使用して、特定の時間や定期的なスケジュールで自動的にスクリプト、コマンド、またはソフトウェアを実行することができます。Cronは、システムメンテナンスタスク、自動バックアップ、ウェブサイトのスクレイピングなど、さまざまな自動化タスクに使用されます。

Cron Jobsは「crontab（クロンテーブル）」と呼ばれる設定ファイルを通じて設定されます。このファイルには、実行するコマンドとその実行時間を指定するためのスケジューリング情報が含まれています。スケジュールは、分、時間、日、月、曜日の5つのフィールドを使用して指定されます。

0 3 * * * /path/to/backup_script.sh

たとえば、毎日午前3時にバックアップスクリプトを実行するには、crontabに次のようなエントリを追加します。

この行は、毎日午前3時0分に/path/to/backup_script.shスクリプトを実行するようCronに指示します。

Cron Jobsは、サーバー管理者や開発者が定期的なタスクを自動化し、システムの維持管理を容易にするための強力なツールです。

Cron Jobを使用してPythonスクリプトを定期的に実行するためには、スクリプト内の無限ループやscheduleライブラリを使用したスケジューリング部分を取り除き、スクリプトが一度実行されたらそのタスクを完了し終了するように変更する必要があります。Cron Job自体がスクリプトを定期的に実行するスケジューラとして機能するためです。

参考

How to Run Python Scripts using CRON jobs (on Mac Crontab) – Automation & Scheduling

Linux/Mac Tutorial: Cron Jobs - How to Schedule Commands with crontab

Cron ジョブを使用して Python スクリプトをスケジュールする方法

積立Botの設定

以下は、Cron Jobで実行するために修正したコードです。

Yodaka

このスクリプトは一度実行されると、一度だけビットコインの積立購入を試み、終了します。

import ccxt
import time
import csv
from datetime import datetime

# APIキーとシークレットを設定
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

# ccxtライブラリを使って取引所に接続
exchange = ccxt.bitbank({
    'apiKey': api_key,
    'secret': api_secret,
})

# CSVファイル名の設定
csv_filename = 'purchase_log.csv'

def accumulate_bitcoin():
    amount_yen = 1000  # 積立金額（円）
    symbol = 'BTC/JPY'  # 取引する通貨ペア
    max_price_diff = 10  # 許容する価格差（円）

    try:
        ticker = exchange.fetch_ticker(symbol)
        last_price = ticker['last']
        order_price = last_price + 10  # 10円高い指し値

        # 注文量を計算
        amount_btc = amount_yen / order_price
        
        # 注文を出す
        order = exchange.create_limit_buy_order(symbol, amount_btc, order_price)
        print(f"Placed limit order for {amount_btc} BTC at {order_price} JPY")

        # 注文の成立を確認するために30秒待機
        time.sleep(30)

        order_status = exchange.fetch_order(order['id'], symbol)
        if order_status['status'] == 'closed':
            print("Order filled.")
        else:
            print("Order not filled within 30 seconds. Cancelling order.")
            exchange.cancel_order(order['id'], symbol)

    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    accumulate_bitcoin()

Cron Jobの設定の例

Cron Jobでこのスクリプトを定期的に実行するには、Cronの設定を行う必要があります。

Cronの設定は通常、コマンドラインからcrontab -eコマンドを使用して行います。以下は、毎日午前10時にこのスクリプトを実行するCron Jobの例です。

0 10 * * * /usr/bin/python3 /path/to/your/script/accumulate_bitcoin.py

この行をCronの設定ファイルに追加します。

/usr/bin/python3はPythonインタープリタのパスであり、/path/to/your/script/accumulate_bitcoin.pyはスクリプトの実際のパスに置き換えてください。Cronの設定が完了すると、指定したスケジュールでスクリプトが自動的に実行されます。

注意

実際に取引を行う前に、APIキーとシークレットを安全に管理し、小額でシステムの動作をテストしてください。また、Cronの設定やPythonスクリプトのパスはシステムによって異なる場合があるため、適宜調整してください。

まとめ

ドルコスト平均法による積立戦略は、長期投資における最適解の一つです。

ビットコインを長期にわたって保有するつもりがあるのなら、検討したい運用方法の一つです。

この戦略では、利用する取引所の選定や指し値注文を通す工夫などを気をつけることで、長期に渡る損失を少しでも抑えて運用することがポイントです。

また、scheduleモジュールを組み合わせることで、Botの定期実行を管理する方法も知ることができました。

今後もこの調子で学習を進めていきます。

-Bot, トレードロジック