OpenAIのChat Completions API (いわゆるChatGPT)では、関数呼び出し (Function Calling) という機能が使えます。

これを使うと、チャットボットがテキストメッセージだけでなく、指定した関数の仕様に沿ったjson形式のレスポンスを応答してくれるようになるため、不定形なテキストデータなどを加工する手間を抑えてLLMの生成データをその後のプログラム処理に効率的に回すことができるようになります。

この関数呼び出し機能を使って、ChatGPTに仮想通貨の投資判断をしてもらい、その内容に従って実際に売買注文を行う自動トレードボットを作ってみました。

Function callingについて

Chat Completions APIの関数呼び出し (Function calling) は、APIコール時に関数の仕様を指定しておくことで、LLMのモデルが応答時に自動的に関数を呼び出すためのJSONオブジェクトを出力する機能です。
gpt-3.5-turboモデル以降の、gpt-4-turbo、gpt-4oなどがこのFunction callingに対応していますが、関数呼び出しに必要なデータの生成も言語モデルのトレーニングに大きく影響しているとのことで、gpt-4oなどのより新しいモデルの方が適切にデータを生成してくれているように思います。

なお、この関数呼び出しは従来からChat completion APIの function_call というリクエストに含めるようになっていましたが、現在は tools という関数だけでなくいろいろな機能を呼び出すための定義をまとめたリクエストを利用するように移行されているようです。

関数呼び出しの例

例えば、通常のテキストでのやり取りであれば、

no_functions.py

import openai

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
      {"role": "system", "content": "あなたは仮想通貨の自動売買ボットです"},
      {"role": "user", "content": "ビットコイン、買うのと売るのどっちがいい？"}
    ]
)

のような指定になり、このような場合はテキスト形式でGPTからレスポンスが返ってきます。

一方で、以下のようなtoolsパラメーターを追加することで、この関数の呼び出しの必要性をGPTが判断したときに自動的に関数の引数となるJSON形式のデータがレスポンスとして帰ってきます。

with_functions.py

import openai

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
      {"role": "system", "content": "あなたは仮想通貨の自動売買ボットです"},
      {"role": "system", "content": "売買判断を行い、関数を呼び出すこと"}
    ],
    tools = [
        {
            "type": "function",
            "function": {
                "name": "order",
                "description": "仮想通貨の売買を行う関数",
                "parameters": {
                    "side": {
                        "type": "string",
                        "enum": ["BUY", "SELL"],
                        "description": "売買の指定。'BUY'もしくは'SELL'で指定する。"
                    },
                    "price": {"type": "number", "description": "注文金額、日本円"},
                    "size": {"type": "number", "description": "注文量"},
                    },
                    "required": ["side","price","size"]
                }
            }
        }
    ]
)

返ってくるオブジェクトには、message.function_call からアクセスができるので、

tool_calls = response.choices[0].message.tool_calls
tool_call = tool_calls[0]
tool_call.arguments = json.loads(tool_call.function.arguments)

function_name = tool_call.function.name
function_arguments = json.loads(tool_call.function.arguments)

のようにすれば function_arguments に

{"side":"SELL","price":10731703,"size":0.00089664}

のようなJSONデータを取り出すことができます。

生成される引数は type などの定義だけでなく、enum で選択型の値を指定したり、description の内容も適度に読んで判断してくれるので、実際にはもっと仕様を詳しく書くことでより精度を上げることができます。

自動トレードさせるためのプロセス

実際にビットコインのトレードをChatGPTに行わせる方法ですが、基本的には以下のようなステップになります。

マーケット情報を取得
自分のアカウントの資産情報などを取得
取得した情報をChat Completions APIに渡して関数の引数を取得
注文処理

ステップ3はOpenAIのAPIを使用し、ステップ1、2、4の情報取得と発注処理はBitFlyer LitningなどのAPIを使って自動化することができます。

今回はOpenAI APIに加えてこのBitflyer Lightning APIを使い、Pythonで動作するビットコインの売買を行うトレードボットを実装しました。

メインの処理

実際のメイン処理のコードは以下のような感じ。
前述のステップ1から4を順に実行させるような形ではなく、マーケット情報は1分おきに取得して蓄積しておき、5分に1度、投資判断をChatGPTに行わせるようなフローになっています。

main.py

# main.py
import time
import os
import traceback
import json
from dotenv import load_dotenv
from bitflyer_client import BitflyerClient
from price_history import PriceHistory
from trading import Trading
from portfolio import Portfolio
from order_book import OrderBook
from trading_decision import TradingDecision
from execution_history import ExecutionHistory
from logger_setup import get_logger

# ロガーの取得
logger = get_logger(__name__)

# .envファイルから環境変数をロード
load_dotenv()

# APIキーとシークレットキーを環境変数から取得
bitflyer_api_key = os.getenv('BITFLYER_API_KEY')
bitflyer_api_secret = os.getenv('BITFLYER_API_SECRET')
openai_api_key = os.getenv('OPENAI_API_KEY')

# クライアントの初期化
client = BitflyerClient(bitflyer_api_key, bitflyer_api_secret)

# 各モジュールの初期化
price_history = PriceHistory(client)
trading = Trading(client)
portfolio = Portfolio(client)
execution_history = ExecutionHistory(client)
order_book = OrderBook(client)
decision_maker = TradingDecision(openai_api_key)


# 注文の実行
def execute_order(function_call):
    try:
        params = function_call.arguments
        side = params['side']
        price = params['price']
        size = params['size']
        order_type = params['order_type']
        trigger_price = params.get('trigger_price')

        # 引数のチェック
        if size < 0.0001:
            logger.info(f"Order size {size} is below the minimum threshold. Order not executed.")
            return

        # 数値を通常の小数点表記に変換
        price = float(f"{price:.8f}")
        size = float(f"{size:.8f}")

        if order_type == 'LIMIT':
            trading.send_order('BTC_JPY', side, price, size, order_type)
        elif order_type == 'STOP':
            trading.send_order('BTC_JPY', side, price, size, order_type, trigger_price)
    except Exception as e:
        logger.error(f"Failed to execute order: {e}")
        raise



# 注文のキャンセル
def cancel_order(function_call):
    try:
        params = function_call.arguments
        order_id = params['order_id']

        trading.cancel_order(order_id)
    except Exception as e:
        logger.error(f"Failed to cancel order: {e}")
        raise


# エラーカウントの初期化
error_count = 0
max_errors = 3


# メインループ
ticker_interval = 60  # ティッカーの取得間隔（秒）
trading_interval = 300  # 売買判断の間隔（秒）
last_trading_time = 0

while True:
    try:
        current_time = time.time()
        
        # 価格履歴の取得（ティッカー）
        market_data = price_history.get_price_history()
        
        # 1分ごとにティッカーを取得
        logger.info(f"Market Data Updated")

        # 最新の板のステータスをチェック
        latest_state = market_data[0].get('state') if market_data else None
        if latest_state != 'RUNNING':
            logger.info(f"Market state is {latest_state}. Skipping trading decision.")
        else:
            # 5分ごとに売買判断を実行
            if current_time - last_trading_time >= trading_interval:
                last_trading_time = current_time

                # ポートフォリオの取得
                portfolio_info = portfolio.get_balance()
                portfolio_data = portfolio_info["current_balance"]
                portfolio_history = portfolio_info["history"]

                # 約定履歴の取得
                execution_data = execution_history.get_execution_history()

                # 注文データの取得
                open_orders_pre = trading.get_open_orders()
                order_data = {"open_orders": open_orders_pre}

                # 板情報の取得
                order_book_data = order_book.get_order_book()

                # OpenAIを使った売買判断
                decision = decision_maker.get_trading_decision(market_data, portfolio_data, order_data, execution_data, order_book_data)

                # ログに出力
                logger.info(f"Balance: JPY {portfolio_data[0]['available']} / BTC {portfolio_data[1]['available']}")
                if open_orders_pre:
                    order_ids = [order['child_order_id'] for order in open_orders_pre]
                    logger.info(f"Open Orders: {order_ids}")

                if decision is None:
                    logger.info("Trading decision: HOLD / WAIT")
                else:
                    # 関数呼び出しが含まれている場合の処理
                    if hasattr(decision, 'type') and decision.type == 'function':
                        function_name = decision.function.name
                        arguments = json.loads(decision.function.arguments)

                        if function_name == 'order_method':
                            logger.info(f"Trading decision: {arguments['side']} / price: {arguments['price']:.8f} / size: {arguments['size']:.8f} / order_type: {arguments['order_type']}")
                            execute_order(decision)
                        elif function_name == 'order_cancel':
                            logger.info(f"Cancel order: {arguments['order_id']}")
                            cancel_order(decision)

                # 注文状況の確認
                open_orders_after = trading.get_open_orders()
                if open_orders_after != open_orders_pre:
                    logger.info(f"Open Orders: {open_orders_after}")

                # エラーカウントをリセット
                error_count = 0

        # 1分間スリープ
        time.sleep(ticker_interval)
    
    except Exception as e:
        logger.error(f"An error occurred: {e}")
        logger.error(traceback.format_exc())
        error_count += 1
        
        if error_count >= max_errors:
            logger.error("Maximum number of consecutive errors reached. Exiting program.")
            break
        else:
            # エラーが発生した場合でも次の試行まで待機
            time.sleep(ticker_interval)

投資判断をChatGPTに行わせるモジュールのコードは以下のようになっています。

Chat completions APIのリクエストに関数を定義し、また tool_choice="required"とすることで、返答をテキストのメッセージではなく必ずいずれかの関数を選択して返すようにすることができます。

trading_decision.py

import openai
import json
from datetime import datetime
from logger_setup import get_logger
import logging

# ロガーの取得
logger = get_logger(__name__)

class TradingDecision:
    def __init__(self, openai_api_key):
        self.openai_api_key = openai_api_key
        openai.api_key = self.openai_api_key

    def load_messages(self, market_data, portfolio_data, order_data, execution_data, order_book_data, current_time):
        with open('messages.json', 'r', encoding='utf-8') as file:
            messages = json.load(file)
        
        # プレースホルダーを変数で置換
        for message in messages:
            message['content'] = message['content'].format(
                market_data=market_data,
                portfolio_data=portfolio_data,
                order_data=order_data,
                execution_data=execution_data,
                order_book_data=order_book_data,
                current_time=current_time
            )
        return messages

    def get_trading_decision(self, market_data, portfolio_data, order_data, execution_data, order_book_data):
        current_time = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"

        messages = self.load_messages(market_data, portfolio_data, order_data, execution_data, order_book_data, current_time)

        tools = [
            {
                "type": "function",
                "function": {
                    "name": "order_method",
                    "description": "Function calling to place a BTC buy or sell order. The minimum order size is 0.001 BTC.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "side": {
                                "type": "string",
                                "enum": ["BUY", "SELL"],
                                "description": "The direction of the trade. 'BUY' or 'SELL'."
                            },
                            "price": {"type": "number", "description": "The order price in JPY."},
                            "size": {"type": "number", "description": "The order quantity. The minimum order size is 0.001 BTC."},
                            "order_type": {
                                "type": "string",
                                "enum": ["LIMIT", "STOP"],
                                "description": "The type of order. Specify 'LIMIT' for limit order or 'STOP' for stop order."
                            },
                            "trigger_price": {"type": "number", "description": "The trigger price. Only specify for 'STOP' orders."}
                        },
                        "required": ["side", "price", "size", "order_type"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "order_cancel",
                    "description": "Function calling to cancel an open BTC order.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "order_id": {"type": "string", "description": "The child_order_acceptance_id of the order to be canceled."}
                        },
                        "required": ["order_id"]
                    }
                }
            }
        ]

        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="required"
        )

        logger.debug(f"Response: {response}")

        if response.choices[0].message.tool_calls:
            tool_calls = response.choices[0].message.tool_calls
            tool_call = tool_calls[0]
            tool_call.arguments = json.loads(tool_call.function.arguments)
            return tool_call
        else:
            return None

Chat completions APIのリクエストから関数の引数を取得し、関数に渡すまでの流れを一部加工して抜粋すると下記のような感じになります。

関数呼び出し

# 関数の定義
tools = [
    {
        "type": "function",
        "function": {
            "name": "order_method",
            # 関数の定義
            }
        }
    },
]

# 関数呼び出しを含めてAPIをコール
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="required"
)

# 関数呼び出しのデータを取り出す
if response.choices[0].message.tool_calls:
    tool_calls = response.choices[0].message.tool_calls
    tool_call = tool_calls[0]
    tool_call.arguments = json.loads(tool_call.function.arguments)

# 関数名や引数を判定して次のpython関数に引き渡す
if hasattr(tool_call, 'type') and tool_call.type == 'function':
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)

    if function_name == 'order_method':
        # 注文を実行
        execute_order(decision)
    elif function_name == 'order_cancel':
        #キャンセルを実行
        cancel_order(decision)

なお、ChatGPTに売買判断指示するプロンプトは別ファイル化しています。

一般的な自動トレードツールはプログラミング知識なども必要ですが、このボットであればあらかじめ動作環境さえ作ってしまえば、このファイルのテキストをいじるだけでも自動売買のコントロールができます。

messages.json

[
  {"role": "system", "content": "あなたはJPYとBTCのトレードについて、テクニカル分析の視点から売買判断を行います。トレード方針は、1日（=1440分）での総利益目標を5.0%の投資収益率とすることを目指します。この売買判断サイクルは5分ごとに実行されます。"},
  {"role": "system", "content": "- 以下はビットコインの最新の市場データ履歴です：\n{market_data}"},
  {"role": "system", "content": "- 以下はbitFlyer Lightning APIから取得した現在のあなたの資産残高です：\n{portfolio_data}"},
  {"role": "system", "content": "- 以下はbitFlyer Lightning APIから取得した現在のあなたのオープンオーダーです：\n{order_data}"},
  {"role": "system", "content": "- 以下はbitFlyer Lightning APIから取得したあなたの取引履歴です：\n{execution_data}"},
  {"role": "system", "content": "- 以下はビットコインの現在の注文板データです：\n{order_book_data}"},
  {"role": "system", "content": "- 現在時刻は{current_time}です。"},
  {"role": "system", "content": "これらのデータに基づいて、日々の利益目標を戦略的に達成するための次のアクションを判断し、order_methodまたはorder_cancelの適切なパラメータを使用して関数呼び出しを行ってください。HOLDまたは待機の判断の場合は関数呼び出しを行わないでください。また、損失リスクを考慮し、必要に応じてHOLDの判断でも適切な反対注文を関数呼び出しで行ってください。メッセージでの回答は行わないでください。"}
]

で、これって儲かるの？

仮想通貨をお前ら難しく考えすぎだろ
株や仮想通貨なんて、複雑そうにみえるが、実はすごく単純
誰でも知ってるような原則、安い時に買って高い時に売る
ホントこれを忠実に守ってれば金は増えてくばっかり
みんなもうちょっと待てばもっと高くなる！とか
今は安くなっちゃったけどもうちょっとで戻る！とか
根拠のない妄想してるから損ばっかりしてるんだよ
俺はこの手法で10万円を96,436円にした

というわけで、なけなしの10万円をBitFlyerの口座に入れて、6時間ほどこのボットを実行して放置してみたところ、損益は見事にマイナスでしたとさ。

それどころか、GPTのAPI利用料が$23.91 （約3,750円）で、1日ほっておくと余裕で1万円を超えます。

今回売買のサイクルを5分にしたのですが、10万円の原資だと1回のChatGPTのAPI利用料の方が5分のボラティリティで得られる利益よりも大きいので、毎回のトレードで手数料分（API利用料分）負けてしまいマイナスが積もる、という落ちがついてしまいました。

さすがに原資を増やして毎月の給料が飛ぶほどのお金をChatGPTに払ってすべての判断をゆだねる勇気はないので、このボットにはその後停止してもらいました。
GPT-3.5-TurboモデルであればAPIの利用量は10分の1くらいになりそうですが、3.5はトークン数の上限が低いのでこれはこれでデータの削減などが肝になりそうですね。

ChatGPTの関数呼び出しなどのを活用したボットは従来のプログラムベースのボットと異なり、動作のコアとなる部分をテキストベースで記述でき、だれでも自由に好みに応じて調整などが与えるほか、与えるデータやプロンプトによってさらに制度などを上げることもできるので、幅広い領域でプロセスの自動化などのハードルを下げる可能性が感じられます。

パッケージはGitHubからどうぞ

こちらのボットのソースコードはGitHubに公開しています。
このコードはあくまでChatGPTの関数呼び出しと他APIとの機能連携を学習するためのサンプルツールとして公開しています。
動作の保証のほか、このツールで発生したいかなる事象に対して一切の責任を負いません。