Bot

開発記録#157(2025/3/25)「論文ベースのbot開発フローpart.19　運用後の改善提案と最適化計画」

2025年3月25日

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

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

参考

Detecting Crypto Pump-and-Dump Schemes: A Thresholding-Based Approach to Handling Market Noise

🚀 運用後の改善提案と最適化計画

次のステップとして、以下の改善提案に基づきシステムの強化に取り組みます。

1. トレードロジックの強化

🔎 強化ポイント

  • 📈 エントリーポイントの最適化
    • ボラティリティ指標 (ATR) や相対強度指数 (RSI) を活用
    • サポート＆レジスタンスラインの組み込み
  • 🔄 P&Dイベントの予測モデル追加
    • LSTM / Transformerベースの予測モデルを組み込み
    • 過去のP&Dイベントデータを活用した学習モデルの設計
  • 🛡️ リスク管理の高度化
    • ダイナミックな損切り (Stop Loss) 設定
    • トレードサイズの自動調整

🛠 1. トレードロジックの強化

Botの収益性をさらに高めるために、エントリーポイントの最適化、P&Dイベント予測モデルの導入、そしてリスク管理の高度化という３つの強化ポイントを設定しました。

  1. エントリーポイントの最適化
    ボラティリティ指標（ATR）や相対強度指数（RSI）を使って、市場の「勢い」を定量的に判断します。加えて、チャート上のサポート＆レジスタンスラインを条件に組み込むことで、より精緻なタイミングでエントリーを狙います。
  2. P&Dイベントの予測モデル追加
    過去のパンプ＆ダンプ発生データを元に、LSTMやTransformerといった時系列予測モデルを構築。これにより、異常な価格変動パターンを事前に捉え、検出ロジックを強化します。
  3. リスク管理の高度化
    相場状況に応じて自動で損切りラインを引き直す「ダイナミックStop Loss」、およびポジションサイズをリアルタイムで調整する仕組みを導入。ポジションの過大化や突然の逆行リスクを抑えつつ、安定したパフォーマンスを目指します。

これらの強化により、単なる売買ロジックの改善にとどまらず、事前予測と損失抑制を組み合わせた**「攻めと守りの両立」**を実現します。

2. セキュリティ強化

🔒 強化ポイント

  • 🔐 APIキーの定期ローテーション
    • APIキーを1ヶ月ごとに自動更新
    • 既存のキーが漏洩した際のリスク最小化
  • 📂 データバックアップの自動化
    • AWS S3 / Google Cloud Storage を利用したデータの自動保存
    • 主要ログ、取引データ、モデルファイルのバックアップスクリプトを作成

✅ バックアップスクリプト (backup_script.py)

import boto3
import os
from datetime import datetime

# AWS S3の設定
AWS_ACCESS_KEY = "YOUR_AWS_ACCESS_KEY"
AWS_SECRET_KEY = "YOUR_AWS_SECRET_KEY"
BUCKET_NAME = "crypto-bot-backup"

# バックアップ対象ファイル
BACKUP_PATHS = ["./data", "./logs", "./models"]

def backup_to_s3():
    s3 = boto3.client('s3', aws_access_key_id=AWS_ACCESS_KEY, aws_secret_access_key=AWS_SECRET_KEY)

    for path in BACKUP_PATHS:
        for root, _, files in os.walk(path):
            for file in files:
                full_path = os.path.join(root, file)
                backup_key = f"{datetime.now().strftime('%Y-%m-%d')}/{full_path}"
                s3.upload_file(full_path, BUCKET_NAME, backup_key)
                print(f"✅ {full_path} をバックアップしました")

if __name__ == "__main__":
    backup_to_s3()

✅ スクリプトの実行

python backup_script.py

🛡️ 2. セキュリティ強化の解説

Botを長期・安定的に運用するには、「キーが漏れた」「データが消えた」といった致命的なリスクにも備えておく必要があります。ここでは、APIキーの定期ローテーションとデータバックアップの自動化という２つの柱でセキュリティを強化します。

🔐 APIキーの定期ローテーション

  • なぜ必要？
    APIキーが第三者に盗まれると、不正アクセスや大量注文による大損失につながります。
  • 対策
    毎月１回、自動的に新しいキーを発行・Botに配信する仕組みを入れることで、もし古いキーが漏洩しても被害範囲を最小化できます。
  • 運用例
    スクリプトやクラウド機能を使って「月初に古いキーを無効化 → 新キーを発行 → 環境変数を更新」という流れを自動化しておくと安心です。

📂 データバックアップの自動化

  • なぜ必要？
    取引ログや市場データ、学習モデルなどが失われると、再現環境の復旧やトラブルシュートに大幅な手間と時間がかかります。
  • 対策
    AWS S3 や Google Cloud Storage などのクラウドストレージを活用し、主要フォルダ（データ／ログ／モデル）を定期的に丸ごとバックアップするスクリプトを用意します。
  • バックアップ例 bashコピーするpython backup_script.py 上記スクリプトでは、./data、./logs、./modelsディレクトリ内のすべてのファイルを、
    当日の日付フォルダ付きでS3バケットにアップロード。
    これをCronやスケジューラーで毎日・毎時間実行すれば、万一のデータ損失にも即座に対応できます。

まとめ

  • APIキーは「期限付きの鍵」と考え、定期ローテーションで常に新鮮な状態を保つ
  • データは「バックアップ第一」にし、クラウドへ自動保存でいつでも復旧可能に

この２つを組み込むだけで、Bot運用の“安心度”が格段にアップします。ぜひ導入を検討してみてください！

3. パフォーマンス改善

⚡ 改善ポイント

  • 🚀 非同期処理 (Asyncio) の最適化
    • データ取得や異常検出を非同期処理で高速化
  • 🧠 モデル軽量化と最適化
    • TensorRTやONNXを利用したモデルの高速化
    • P&D予測モデルの推論速度を向上

✅ Asyncioの最適化例 (async_collector.py)

import asyncio
import aiohttp
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()
            print(f"✅ {symbol} データ取得成功")
            return data

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

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

✅ 非同期処理の実行

python async_collector.py

🚀 3. パフォーマンス改善の解説

Botの処理速度を高め、よりリアルタイム性を追求するために、次の２つのポイントでパフォーマンスを改善します。

  • 非同期処理 (Asyncio) の最適化
    従来はシンジンクで順番にデータを取得していたところを、aiohttp + asyncio.gather を使って複数通貨ペアを同時取得する非同期処理に置き換えます。
    例えば以下のサンプルでは、BTCUSDT、ETHUSDT、XRPUSDT の3つを並列でフェッチし、全完了までの時間を秒単位で計測。
python async_collector.py

と実行するだけで、ブロッキングなしに高速にデータ取得が完了します。

  • モデル軽量化と推論最適化
    P&D検出や価格予測に使う機械学習モデルは、TensorRT や ONNX Runtime を活用して軽量化・量子化し、推論速度を数倍高速化します。
    • モデルをONNXフォーマットにエクスポート→TensorRTでエンジン化
    • バッチサイズやワークスペースをチューニング
      などにより、リアルタイムのストリーミング予測でもレイテンシを数十ミリ秒以下に抑えられます。

これらの改善により、**「より多くのシンボルを」「より高い頻度で」「より低い遅延で」**モニタリング・予測できるようになり、Botの立ち回り精度と収益機会の最大化につながります。

1. シンジンク（同期処理）

カンタン説明

「順番どおりに一つずつ処理するやり方」です。
例えば A→B→C の３つのAPI呼び出しを「Aが終わるまで待って」「次にB」「最後にC」という流れで実行します。

厳密な定義

同期処理（synchronous processing）とは、呼び出し元のスレッドが関数呼び出しの完了を待ってから次の処理に進む方式を指します。

  • 関数呼び出しはブロッキング（呼び出し元が待機状態）を伴い、並列実行や割り込みを行わない。
  • 実装上は、呼び出し側コードが戻り値を受け取るまで次の行へ進まない。
  • 長時間かかるI/Oや計算処理を含むと、全体の応答性が低下しやすい。

2. TensorRT

カンタン説明

NVIDIA製の「機械学習モデルをGPU上でめちゃ速く動かすエンジン」です。
すでに学習済みのニューラルネットを最適化して、推論（推測）だけを高速化します。

厳密な定義

TensorRT は NVIDIA が提供する Deep Learning 推論用ライブラリで、学習済みモデルを内部表現（TensorRT ネットワーク）に変換し、以下を行います：

  1. レイヤー統合（layer fusion） や 精度低減（FP32→FP16, INT8）
  2. バッチサイズやワークスペースを最適化
  3. CUDA カーネルを生成
    結果として、GPU 上での推論レイテンシを大幅に低減し、推論スループットを向上させます。

3. ONNX Runtime

カンタン説明

汎用的な「機械学習モデル実行エンジン」です。ONNX形式のモデルを、CPUやGPUなどさまざまなハードで高速に動かせます。

厳密な定義

ONNX Runtime は、マイクロソフト主導で開発されたクロスプラットフォームな推論エンジンで、以下を特徴とします：

  • ONNX（Open Neural Network Exchange）フォーマットで定義されたモデルをロード
  • プラグイン式のバックエンド（CUDA／TensorRT／MKL‑DNN など）を通じて最適化されたカーネルを使い分け
  • グラフ最適化（演算削減、演算融合）や キャッシュ機構 によって、CPU・GPU・FPGA などで安定した高速推論を実現
  • C／C++／Python API を備え、多言語から利用可能

これらを組み合わせることで、

  • 「シンジンク⇒非同期」に置き換えてI/O待ちをなくし、
  • 「ONNX Runtime／TensorRT」でモデル推論を軽量化・高速化……

という、**「Botの全体応答性を飛躍的に向上させる」**施策が可能になります。

4. 次のステップ

  1. ✅ トレードロジックの強化 (ATR/RSI、P&D予測モデル)
  2. ✅ セキュリティ強化 (APIキーのローテーション、自動バックアップ)
  3. ✅ パフォーマンス最適化 (非同期処理の導入、モデルの高速化)

5. 宿題

  • 🚀 まずはトレードロジックの強化から開始し、その後にセキュリティ強化、パフォーマンス最適化に進める。
  • 具体的なコード実装やデプロイのタイミングに合わせて、次のステップを決定する。
Yodaka

次回の記事では、トレードロジックの強化に取り組みます。

関連
開発記録#158(2025/3/25)「論文ベースのbot開発フローpart.20 トレードロジックの強化(ATR×RSI×サポレジ)」

続きを見る

👇ラジオで話したこと

🎙️ こんにちは、Yodakaです。
今日はイベント検知系bot開発フロー、part.19「運用後の改善提案と最適化計画」を、じっくり解説していきます。
前回までにBotをデプロイして、監視・ログ自動化まで整えましたが、今回はいよいよ“攻め”と“守り”を強化しつつ、“速さ”も追い求めるフェーズに入ります。

イントロダクション

「Botが動いているだけでは不十分。
どう強化し、どう守り、どう速く動かすか。
ここからが本当の勝負です」

この一言に尽きます。
単に“動く”から、“市場環境やリスクに適応できる”“誰よりも速く意思決定できる”Botへと昇華させていきましょう。

1. トレードロジックの強化

1.1 エントリーポイントの最適化

コード例

from ta.volatility import AverageTrueRange
from ta.momentum import RSIIndicator

# DataFrame df に high, low, close カラムがある前提
atr = AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()
rsi = RSIIndicator(df['close'], window=14).rsi()

# サポート/レジスタンスの検出
# 単純なピーク/ボトム検出
support = df['low'].rolling(20).min()
resistance = df['high'].rolling(20).max()

# エントリー条件
if rsi.iloc[-1] < 30 and df['close'].iloc[-1] < support.iloc[-1]:
    # RSIが売られすぎ＋価格がサポート下抜け→逆張りLong
    place_order("Buy", df['close'].iloc[-1])
  • ATR で「市場のボラティリティ」を捉え、現在のレンジ幅を把握。
  • RSI で「買われすぎ／売られすぎ」を判断。
  • サポート＆レジスタンス は20期間の高値安値のピーク・ボトムを利用。

これらを組み合わせることで、一過性のノイズではなく本当に勢いが転換するポイントでエントリーできます。

1.2 P&Dイベント予測モデルの追加

  • データ準備
    過去数年分のP&Dトリガー価格や時間帯をCSVで収集。
  • モデル構築 
import torch
import torch.nn as nn

class PnDPredictor(nn.Module):
    def __init__(self, input_size, hidden_size):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, 1)
    def forward(self, x):
        out, _ = self.lstm(x)
        return torch.sigmoid(self.fc(out[:, -1, :]))
  • 学習
    • シーケンス長：60分間のティックデータ
    • 損失関数：バイナリクロスエントロピー
    • 最適化：Adam, lr=1e-4
  • 推論
    ONNXにエクスポート → TensorRTエンジンで秒間数千回の推論が可能。
# ONNX変換
torch.onnx.export(model, dummy_input, "pnd_model.onnx")

これで、**「あと10分でP&Dが起こる確率」**をリアルタイムに算出できるようになります。

1.3 リスク管理の高度化

  • ダイナミックStop Loss
    市場ボラティリティに応じて自動計算。
stop_loss = entry_price - atr.iloc[-1] * 1.5
  • ポジションサイジング
    アカウント残高とATRをもとに、1トレードあたりリスクを総資産の1%に制限。
risk_per_trade = 0.01 * account_balance
position_size = risk_per_trade / (entry_price - stop_loss)

これらを組み合わせれば、損失を一定に抑えつつ、利益機会を逃さない「攻めと守りの両立」が実現できます。

2. セキュリティ強化

2.1 APIキーの定期ローテーション

  • 仕様
    毎月1日の00:00に古いキーを無効化 → 新キー発行 → 環境変数を更新
  • 実装例（Pseudo）
0 0 1 * * /usr/local/bin/rotate_api_key.sh

rotate_api_key.sh は Bybit API経由でキーを再生成、Kubernetes Secretを kubectl patch secret で更新します。

2.2 データバックアップの自動化

Pythonスクリプト

import boto3, os
from datetime import datetime

s3 = boto3.client('s3', aws_access_key_id=os.getenv('AWS_KEY'),
                  aws_secret_access_key=os.getenv('AWS_SECRET'))

for base in ["data", "logs", "models"]:
    for root, _, files in os.walk(base):
        for f in files:
            gkey = f"{datetime.today():%Y-%m-%d}/{root}/{f}"
            s3.upload_file(os.path.join(root, f), "crypto-bot-backup", gkey)
  • CronJob化
    Kubernetesの CronJob リソースで毎日動かすと一元管理しやすいですね。

これでデータ消失リスクがほぼゼロになります。

3. パフォーマンス改善

3.1 非同期処理の導入

import asyncio, aiohttp

async def fetch(pair):
    async with aiohttp.ClientSession() as s:
        r = await s.get(API_URL, params={"symbol": pair})
        return await r.json()

async def main():
    pairs = ["BTCUSDT", "ETHUSDT", "XRPUSDT"]
    results = await asyncio.gather(*(fetch(p) for p in pairs))
    print(results)
  • 結果：3ペア同時取得で、従来の3倍以上のスループット。

3.2 モデル推論の高速化

  • ONNX Runtime
    ort_session = onnxruntime.InferenceSession("pnd_model.onnx")
  • TensorRT
    trtexec --onnx=pnd_model.onnx --saveEngine=pnd_model.trt

いずれも推論レイテンシを数十ミリ秒以下に抑えられます。

クロージング＆次回予告

今日は「トレードロジック強化」「セキュリティ強化」「パフォーマンス改善」の３本柱を、
技術面とコード例を交えながら解説しました。いかがだったでしょうか？

次回はATR（Average True Range）とRSI（Relative Strength Index）、
そしてサポート＆レジスタンスラインを組み込んだ“強化版トレードロジック”を
実際のコードを交えて詳しく解説します。

🎙️ おまけコーナー：LSTMパイプライン＆運用周りの深掘り

よだかです。今回は「LSTM／Transformerモデルを実運用する際に気をつけたい細かいポイント」をまとめてお送りします。主に５つのテーマです。

1️⃣ シーケンス長：60分間のティックデータ

  • なぜ60分？
    1分ごとのティックを60本まとめると、直近1時間の流れをモデルが把握できます。
  • メリット・デメリット
    • 長すぎるとメモリ負荷・訓練時間が増大
    • 短すぎると “前振り” が足りず、P&Dの前兆を捉え損ねる

→ ポイント：最初は60分で試し、過学習や遅延が出たら30分や90分で再チューニングしましょう。

2️⃣ 損失関数：バイナリクロスエントロピー

  • 何を最適化？
    P&Dが「起こる／起こらない」の二択をモデルに学習させます。
  • 選ぶ理由：確率出力と二値分類の相性が良く、勾配も滑らかです。

3️⃣ 最適化：Adam, lr=1e-4

  • Adamの強み
    モーメンタムとRMSPropを組み合わせ、学習率を自動調整。
  • 学習率 1e-4 の目安
    • 小さすぎると収束が遅い
    • 大きすぎると発散しやすい
  • 実践Tip：
    • 最初は1e-3で動かし、「損失が乱降下する」ようなら1e-4へ。
    • バッチサイズを増やしたら、学習率も微調整しましょう。

4️⃣ CronJob化：モデル訓練＆推論のスケジューリング

  • なぜ必要？
    定期的に新データで再訓練したり、毎時間／毎分で推論タスクを回したりするため。
  • Kubernetes CronJob 例
apiVersion: batch/v1beta1
kind: CronJob
metadata:
  name: pnd-retrain
spec:
  schedule: "0 3 * * *"       # 毎日3時に再訓練
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: trainer
            image: my-bot-trainer:latest
            command: ["python","train_pnd_model.py"]
          restartPolicy: OnFailure
  • ポイント：スケジュールは負荷とデータ更新頻度に合わせて最適化しましょう。

5️⃣ 推論レイテンシ：実運用の“心臓部”を守る

  • どう測る？
    Pythonなら time.perf_counter() で前後を計測。
start = perf_counter()
output = model.predict(input)
print(f"Latency: {perf_counter() - start:.3f}s")
  • 目標値：数百ミリ秒以下、できれば数十ミリ秒。
  • 改善策：
    • バッチ推論 ではなく 単一入力推論 を高速化
    • TensorRT や ONNX Runtime で 量子化（FP16/INT8）
    • GPU ではなく CPU エッジ推論 が必要なら、ONNX Runtime の MKL‑DNN
  • 監視：Prometheus で推論レイテンシをメトリクス化し、Grafana でアラート設定しましょう。

🎙️ おまけまとめ

  1. シーケンス長 は 60 分を基準に、性能と精度のバランスを調整。
  2. 損失関数 は二値分類に最適なバイナリクロスエントロピーを採用。
  3. Adam lr=1e-4 からスタートし、バッチサイズやモデル規模に応じて微調整。
  4. CronJob 化 で再訓練・定期推論を自動化し、運用負荷を軽減。
  5. 推論レイテンシ は必ず計測・監視し、リアルタイム性を担保。

これらを押さえておくだけで、モデルパイプラインの信頼性とパフォーマンスが大きく向上します。
ぜひ次回の開発に活かしてみてください！

あなた:

Adam, lr=1e-4

Adam はオプティマイザ（最適化アルゴリズム）の一種で、

  • 読み方：「アダム」
  • 意味：勾配の１次モーメント（平均）と２次モーメント（分散）を利用して、各パラメータごとに適応的な学習率を調整しながら学習を進める手法です。

lr = 1e-4 は「learning rate（学習率）＝1×10⁻⁴」を表し、

  • 読み方：「エルアール イコール イチ イー マイナス ヨン」
    （あるいは「learning rate イコール 10 の マイナス４乗」）
  • 意味：パラメータ更新の「１ステップあたりの大きさ」を 0.0001 に設定するという指定です。

学習率が大きいほどパラメータの更新幅が大きく、早く収束しやすい一方で不安定になりがち。
1e-4（0.0001）は比較的控えめな値で、「ゆっくり・安定的に学習を進めたい」場合によく使われます。

-Bot