趣旨

Telegramを利用して複雑な取引を行うための基本フレームができたので共有したら面白そうだなと思って共有します。

.envの設定、OSの設定、exchangeInstanceの設定及び取引所ごとのメソッド調整、Telegram botのトークン発行方法など説明しようとすると無限に説明しないといけないことがありそうなので全て割愛します。興味のある方はぜひご自分でご設定ください。

ただし本記事に記載されているライブラリ等の解説は一番下で行います。

telegram_bot.py

import ccxt
import os 
from dotenv import dotenv_values
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

current_directory = os.path.dirname(os.path.realpath(__file__))
env_path = os.path.join(current_directory, '.env')
config = dotenv_values(env_path)

token = config.get('TELEGRAM_TOKEN')
chat_id = config.get('TELEGRAM_CHAT_ID')

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text('Hi! Use /info to get instructions on how to use this bot.')

async def info(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        'To place an order, use the /trade command with the following format:\n'
        '/trade <exchange_id> <symbol> <market_type> <order_type> <amount> [<price>]\n')

async def trade(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    try:
        args = context.args
        if len(args) < 5:
            await update.message.reply_text('Usage: /trade <exchange_id> <symbol> <market_type> <order_type> <amount> [<price>]')
            return
        exchange_id, symbol, market_type, order_type, amount, *optional = args
        price = optional[0] if optional else None
        exchange_class = getattr(ccxt, exchange_id.lower())({
            'apiKey': config.get('API_KEY'),  # Your API Key
            'secret': config.get('API_SECRET'),  # Your API Secret
            'enableRateLimit': True,
        })
        if market_type == 'future':
            exchange_class.options['defaultType'] = 'future'
        else:
            exchange_class.options['defaultType'] = 'spot'
        if order_type == 'market':
            order = exchange_class.create_market_buy_order(symbol, float(amount))
        elif order_type == 'limit':
            if not price:
                await update.message.reply_text('For limit orders, please specify a price.')
                return
            order = exchange_class.create_limit_buy_order(symbol, float(amount), float(price))
        else:
            await update.message.reply_text('Unknown order type. Use "market" or "limit".')
            return
        trade_message = f"Order placed on {exchange_id.upper()}: {order}"
        await update.message.reply_text(trade_message)
    except Exception as e:
        await update.message.reply_text(f"Error processing trade command: {str(e)}")

def main() -> None:
    current_directory = os.path.dirname(os.path.realpath(__file__))
    env_path = os.path.join(current_directory, '.env')
    config = dotenv_values(env_path)
    token = config.get('TELEGRAM_TOKEN')
    application = Application.builder().token(token).build()
    async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        await update.message.reply_text('Hello! I am your bot.')
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("info", info))
    application.add_handler(CommandHandler("trade", trade))
    application.run_polling()

if __name__ == '__main__':
    main()

詳細は

CCXTを最後まで設定していないTelegram用サーバーでテスト運用してみましたが、うまく起動しました。
エラーが返ってきてるのは、CCXTのインスタンスを作れない状況のままだからです。

企業を運営してる場合は、こういったTelegramボットを置いてるサーバーにLLMを置いて顧客に返事をしてもらう体制を準備しても面白そうですね。画像もSDで自動生成してLLMとこれをセットにすればサイバー彼女、サイバー彼氏とかもすぐ作れそうです。

脱獄LLMに価格予想をしてもらって勝手に売買基準を作ってもらって完全自動運用とかもできそうではありますね。

UmbrelにそういうLLM置いてあったのでGitHubページ載せておきます。

以下、ライブラリやモジュールの説明となります。

Telegram Botライブラリの紹介

Pythonのpython-telegram-bot パッケージを使用すると、Telegram のためのボットを作成することができます。以下はその主要な部分の説明です。

from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

from telegram import Update： Update クラスは telegram モジュールの一部で、 Telegram からボットへの入ってくる更新を表します。更新とは、ユーザーからのメッセージ、コールバッククエリ、インラインクエリなど、botが Telegram から受け取るあらゆるものを指します。
from telegram.ext import Application, CommandHandler, ContextTypes：これらはtelegram.extサブモジュールのコンポーネントで、Telegram Bot API を使いやすくするための拡張フレームワークを提供します。

Application：このクラスは、イベント駆動システムを設定してボットアプリケーションを定義するために使用されます。更新を受け取り、適切なハンドラーにディスパッチするなど、botのワークフローを管理します。
CommandHandler： CommandHandler は、ユーザーから送信されたコマンドを処理するために使用されます。特定のコマンドをリッスンし、それらのコマンドが受信されたときに定義されたコールバック関数をトリガーします。
ContextTypes： ContextTypes は、非同期バージョンのライブラリで使用されるコンテキストオブジェクトのカスタムタイプを定義する方法です。コンテキストオブジェクトには、現在のUpdate、Application インスタンスなどのデータが含まれています。

非同期関数の説明

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:

この関数は、python-telegram-bot ライブラリを使用してTelegram bot を作成する際に使用される非同期関数の定義です。非同期関数は、イベントループ内で非同期に実行されることを可能にし、I/Oバウンド操作の応答に適しています。

update: Update - Update クラスのインスタンスで、Telegram から入ってくる更新を表します。
context: ContextTypes.DEFAULT_TYPE - ContextTypes.DEFAULT_TYPE は、ボットが更新を処理するために必要な追加データにアクセスおよび管理する方法を提供します。

try, except, finally

Pythonでの例外処理には try, except, finally 文が使用されます。

try ブロックは、エラーのあるコードブロックのテストに使用されます。
except ブロックは、エラーを処理するために使用されます。異なるタイプの例外を指定して、異なる種類のエラーをキャッチできます。
finally ブロックは、例外の有無に関わらず実行されるコードブロックです。リソースのクリーンアップや必ず実行する必要があるコードに使用されます。

非同期IO操作の例

await exchange_class.close()

この文は、非同期IO操作の一例です。await キーワードは、非同期操作が完了するまで関数の実行を一時停止し、その間にイベントループが他のタスクを実行できるようにします。

Telegram bot の起動

def main() -> None:
    """Start the bot."""
    application = Application.builder().token(token).build()

    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("info", info))
    application.add_handler(CommandHandler("fetch", fetch))

    application.run_polling()

この関数は、特定のコマンド（start, info, fetch）を持つ Telegram bot を設定し、メッセージを受信するためのポーリングを開始します。ユーザーがこれらのコマンドのいずれかをボットに送信すると、対応する関数が呼び出されてコマンドが処理されます。