開発記録#160(2025/3/25)「論文ベースのbot開発フローpart.22」

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

本記事では「暗号通貨のパンプ＆ダンプスキームの検出」に関する論文をベースにbot開発の過程をまとめていきます。

Detecting Crypto Pump-and-Dump Schemes: A Thresholding-Based Approach to Handling Market Noisehttps://t.co/ctCJEV1MBs

— よだか(夜鷹/yodaka) (@yodakablog) March 22, 2025

⚡️ パフォーマンス最適化 (非同期処理とモデルの高速化)

次のステップとして、以下の2つの最適化に取り組みます。

✅ 非同期処理 (Asyncio) の導入
✅ モデルの高速化 (ONNX / TensorRT最適化)

1. 非同期処理 (Asyncio) の導入

🔎 非同期処理の目的

• 複数のAPIリクエストやデータ処理を並行処理して高速化
• P&Dイベントの検出やデータ収集の応答速度を改善

✅ 改善版データ収集スクリプト (async_collector.py)

import asyncio
import aiohttp
import pandas as pd
import time

API_URL = "https://api.bybit.com/v2/public/kline/list"

async def fetch_data(symbol, interval="1m"):
    async with aiohttp.ClientSession() as session:
        params = {"symbol": symbol, "interval": interval, "limit": 200}
        async with session.get(API_URL, params=params) as response:
            data = await response.json()
            df = pd.DataFrame(data['result'])
            df['symbol'] = symbol
            print(f"✅ {symbol} データ取得成功")
            return df

async def main():
    symbols = ["BTCUSDT", "ETHUSDT", "XRPUSDT"]
    tasks = [fetch_data(symbol) for symbol in symbols]
    results = await asyncio.gather(*tasks)

    # データ統合と保存
    combined_df = pd.concat(results)
    combined_df.to_csv("./data/market_data.csv", index=False)
    print("✅ データの統合と保存が完了しました")

if __name__ == "__main__":
    start_time = time.time()
    asyncio.run(main())
    print(f"⏱️ 完了までの時間: {time.time() - start_time:.2f}秒")

✅ スクリプトの実行

pip install aiohttp pandas
python async_collector.py

2. モデルの高速化 (ONNX / TensorRT)

🔎 ONNXとは？

• モデルを最適化し、GPUやCPU環境での推論を高速化できるフォーマット

🔎 TensorRTとは？

• NVIDIAが提供するGPU最適化ツールで、ディープラーニングモデルの推論を最大限高速化

✅ PyTorchモデルのONNX変換スクリプト

import torch
import onnx

# モデルの定義
class TradingModel(torch.nn.Module):
    def __init__(self):
        super(TradingModel, self).__init__()
        self.fc = torch.nn.Linear(10, 1)

    def forward(self, x):
        return self.fc(x)

# モデルの保存
model = TradingModel()
dummy_input = torch.randn(1, 10)

# ONNX形式にエクスポート
onnx_path = "trading_model.onnx"
torch.onnx.export(model, dummy_input, onnx_path)

print(f"✅ モデルが {onnx_path} に保存されました")

✅ ONNX Runtimeでの推論 (最適化後の推論スクリプト)

import onnxruntime as ort
import numpy as np

# モデルの読み込み
session = ort.InferenceSession("trading_model.onnx")

# ダミーデータ (サンプルデータ)
input_data = np.random.randn(1, 10).astype(np.float32)

# 推論実行
output = session.run(None, {"input": input_data})

print(f"✅ 推論結果: {output}")

✅ TensorRTによるGPU最適化

1. TensorRTのインストール

pip install nvidia-pyindex
pip install tensorrt

2. TensorRT最適化スクリプト

trtexec --onnx=trading_model.onnx --saveEngine=trading_model.engine

3. TensorRT推論スクリプト

import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np

# モデルの読み込み
with open("trading_model.engine", "rb") as f:
    engine_data = f.read()

# 推論実行
engine = trt.Runtime(trt.Logger(trt.Logger.WARNING)).deserialize_cuda_engine(engine_data)
context = engine.create_execution_context()

input_data = np.random.randn(1, 10).astype(np.float32)
output_data = np.empty([1, 1], dtype=np.float32)

d_input = cuda.mem_alloc(input_data.nbytes)
d_output = cuda.mem_alloc(output_data.nbytes)

cuda.memcpy_htod(d_input, input_data)
context.execute(1, [int(d_input), int(d_output)])
cuda.memcpy_dtoh(output_data, d_output)

print(f"✅ TensorRT推論結果: {output_data}")

3. 効果のまとめ

4. 次のステップ

次のフェーズでは、以下のタスクに取り組みます。

✅ 運用ルールの確立 (Botの稼働スケジュールや障害対応フローの策定)
✅ 最終パフォーマンスチェック (シミュレーションと実運用の比較検証)
✅ デプロイ後の安定性確認 (サーバー監視とログの確認)

Yodaka

次回は「運用ルールの確立 (Botの稼働スケジュールや障害対応フローの策定)」についてまとめます。