Bot

開発記録#148(2025/3/25)「論文ベースのbot開発フローpart.10 システム全体の統合テスト」

2025年3月25日

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

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

参考

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

✅ システム全体の統合テスト

まずは、システム全体の動作確認とデバッグに取り組みます。

1. 統合テストの目的

🔍 システム全体の一貫した動作を確認し、各コンポーネントの連携が正しく機能するかをテストします。
以下の項目に焦点を当てます：

✅ データの流れ確認：データ収集 → 異常検出 → 取引執行 → 監視の連携
✅ エラーハンドリングの確認：API接続エラー、データ欠損時の動作
✅ リカバリー機能の検証：異常終了後の自動再起動
✅ 通知システムの確認：Slack通知が正しく送信されるか

2. 統合テストのチェックリスト

チェック項目内容ステータス
データ収集Bybit APIから正しくOHLCVデータが取得できるか✅
異常検出P&Dイベントが検出されるか✅
取引執行Botが指定の条件で売買を行うか✅
監視システムSlack通知とダッシュボード表示が正確か✅
再接続機能ネットワーク障害後に自動再接続できるか✅
データ保存データの正確な記録とファイル出力ができるか✅

3. 統合テスト用のスクリプト (integration_test.py)

以下のスクリプトは、システム全体の動作を順次テストし、エラーが発生した場合に詳細情報を出力します。

import subprocess
import time

# 各コンポーネントのテストコマンド
TEST_COMMANDS = [
    "python data_collector.py",
    "python pnd_detector.py",
    "python trade_executor.py",
    "python monitoring_system.py"
]

# ログ出力
def log_message(message):
    print(f"[INFO] {message}")

# 各コンポーネントのテスト
def run_component_test(command):
    try:
        log_message(f"{command} を開始しています...")
        process = subprocess.Popen(command, shell=True)
        time.sleep(5)  # 起動確認のため一時停止
        if process.poll() is None:  # 正常に起動しているか確認
            log_message(f"{command} が正常に稼働しています ✅")
            process.terminate()  # 一時停止中のプロセスを停止
        else:
            log_message(f"❌ {command} の起動に失敗しました")
    except Exception as e:
        log_message(f"❗ エラー: {e}")

# テストの実行
def run_integration_test():
    log_message("🚀 統合テストを開始します")

    for command in TEST_COMMANDS:
        run_component_test(command)

    log_message("✅ 統合テストが完了しました")

if __name__ == "__main__":
    run_integration_test()

このスクリプトがどのように機能するか、ポイントごとに解説していきます。

スクリプトの概要

このスクリプトは、データ収集、P&D（パンプアンドダンプ）検出、取引実行、システム監視の4つの異なるコンポーネントをテストするためのものです。各コンポーネントは、独自のPythonファイルで実装されており、このスクリプトではそれぞれを起動し、正しく動作するかどうかをチェックします。

主要な関数とその役割

  1. log_message関数
    • この関数は、ログメッセージを出力するために使われます。メッセージには情報レベルが含まれ、どのアクションが現在行われているかをユーザーに知らせます。
  2. run_component_test関数
    • 指定されたコマンド（各コンポーネントのテストスクリプト）を実行します。コマンドはsubprocess.Popenを使用して非同期に実行され、5秒間待機してそのプロセスが正常に起動しているかを確認します。問題がなければプロセスが終了されますが、起動に失敗した場合はその旨がログに記録されます。
  3. run_integration_test関数
    • これは全てのテストコマンドを順に実行し、システムの各コンポーネントが連携して正しく動作するかを確認するための関数です。全コンポーネントのテストが終了した後に、統合テストの完了がログに記録されます。

実行フロー

プログラムはif __name__ == "__main__":ブロックから実行を開始し、run_integration_test関数を呼び出して全コンポーネントのテストを行います。これにより、システム全体が一貫して正しく機能するかどうかを一度に確認できます。

4. 実行方法

4.1 統合テストの実行

このスクリプトは、システムの全コンポーネントが一体として正しく動作するかを検証するためのものです。統合テストを実行するには、以下のコマンドを使用します。

python integration_test.py

このコマンドにより、スクリプトが自動的に各コンポーネントのPythonファイルを起動し、システムの機能を一つずつ確認します

4.2 確認ポイント

  • コンポーネントの起動確認: スクリプトは各コンポーネントが起動し、指定された時間（このスクリプトでは5秒後）内にプロセスが正常に動作しているかを確認します。正常に動作していれば、その旨のメッセージが表示されます。
  • エラー発生時の対応: もしコンポーネントが正常に起動しなかった場合、エラーメッセージが出力されます。この情報をもとに、何が問題だったのかを特定しやすくなります。

5. テスト完了後の修正点

🟩 成功した場合

  • システムが正常に機能しているため、次のステップに進みます。

テストが全て成功し、各コンポーネントからエラーが報告されなかった場合、システムは正常に機能していると判断されます。その結果、開発者は次の開発ステップに進むことができます。これには新しい機能の追加や他のシステムとの統合などが含まれるかもしれません。

🟥 失敗した場合

  • エラーログをもとにデバッグを行い、問題点を修正します。
  • 具体的な修正内容が発生した際は、コードを調整し再テストします。
  • デバッグと問題の特定: 具体的なエラーメッセージやコンポーネントの情報をもとに、問題が発生した原因を特定します。
  • コードの修正: 問題の原因が判明したら、該当するコードを修正します。これには変数の誤用、構文エラー、ロジックの誤りなどが含まれることがあります。
  • 再テスト: 修正後、再び統合テストを実行して、修正が正しく機能するかを確認します。これにより、修正が適切であったかどうかを検証できます。

6. 具体的な修正ポイント

このコードにおいてエラー修正を行う際に意識すべきポイントはいくつかあります。それらを適切に把握し、効果的なデバッグを行うことが重要です。

1. エラーハンドリングの強化

  • 例外処理の詳細化: 現在のコードでは、try-exceptブロックを用いているものの、エラーの種類によって対応を変えることができます。特定のエラータイプ（例えば、FileNotFoundErrorやValueError）に対してカスタマイズされた例外処理を追加すると、より詳細なエラー情報を提供できます。

2. ロギングの改善

  • ログレベルの活用: 現在はすべてのメッセージを単一の形式で出力していますが、loggingモジュールを使用して、エラー、警告、情報、デバッグメッセージに異なるログレベルを設定することが有効です。これにより、デバッグ時にログの重要度に応じて適切な情報をフィルタリングできます。

3. プロセスの監視と終了処理の改善

  • プロセスの状態確認: 現在のスクリプトでは、プロセスがまだ動作しているかどうかを5秒後に一度だけチェックしています。プロセスが予期せず終了した場合やハングアップした場合の対応を強化するため、状態確認を定期的に行い、タイムアウトを設定することが考えられます。
  • リソースのクリーンアップ: process.terminate()を呼び出した後、process.wait()を使用してプロセスが確実に終了するのを待つことで、システムリソースの適切な解放を確実に行うことが重要です。

4. セキュリティと安定性の向上

  • コマンドインジェクションのリスク: subprocess.Popenにshell=Trueを使用している場合、コマンドインジェクションの脆弱性が生じる可能性があります。可能であれば、shell=Trueを避け、より安全な方法でプロセスを起動することが望ましいです。

5. パフォーマンスの最適化

  • 並列実行の検討: 複数のコンポーネントを順番にテストする代わりに、マルチスレッディングやマルチプロセッシングを使用して、テストを並列に実行することで全体のテスト時間を短縮することができます。

これらの改善ポイントを意識することで、コードの信頼性、効率、安全性を向上させることができます。

7. 具体的な修正フロー

以下は、提案された改善ポイントを具体的に実装するための修正フローです。これに従ってコードを段階的に更新し、全体の品質と安全性を向上させることができます。

ステップ1: エラーハンドリングの強化

  1. 例外処理の追加: 既存のtry-exceptブロックを拡張して、特定のエラータイプごとにカスタマイズされたエラーハンドリングを行うようにします。これには、ファイル操作やネットワークアクセスに関連する特定の例外（FileNotFoundError, OSErrorなど）を捕捉するコードを追加します。
  2. エラー情報の詳細化: エラー発生時により多くのコンテキスト情報をログに記録するようにします。これにより、デバッグが容易になります。 
try:
    # コマンド実行
    process = subprocess.Popen(command, shell=True)
except FileNotFoundError as e:
    log_message(f"ファイルが見つかりません: {e}")
except OSError as e:
    log_message(f"システムエラーが発生しました: {e}")
except Exception as e:
    log_message(f"予期せぬエラーが発生しました: {e}")

ステップ2: ロギングの改善

  1. loggingモジュールの導入: printステートメントの代わりにloggingモジュールを使用して、異なるログレベル（INFO, ERROR, DEBUG）を設定します。これにより、運用環境でのログ管理がより柔軟に行えるようになります。
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def log_message(message, level=logging.INFO):
    logging.log(level, message)

ステップ3: プロセスの監視と終了処理の改善

  1. プロセス状態の継続的確認: プロセスが終了していないか定期的にチェックし、指定した時間が経過した後にまだ実行中の場合はプロセスを終了します。
  2. リソースのクリーンアップ強化: process.terminate()後にprocess.wait()を呼び出し、プロセスが完全に終了するのを確実にします。
try:
    process = subprocess.Popen(command, shell=True)
    time.sleep(5)
    if process.poll() is None:
        process.terminate()
        process.wait()  # プロセスの完全な終了を確実にする
        log_message(f"{command} が正常に終了しました")
    else:
        log_message(f"{command} が予期せず終了しました")

ステップ4: セキュリティと安定性の向上

  1. コマンドインジェクション対策: 安全なプロセス起動方法に切り替え、shell=Trueの使用を避けます。コマンドをリスト形式で渡す方法に変更します。
process = subprocess.Popen(['python', 'script.py'], shell=False)

ステップ5: パフォーマンスの最適化

  1. 並列処理の導入: concurrent.futuresモジュールを使用してテストを並行して実行し、全体の実行時間を短縮します。
import concurrent.futures

def run_integration_test():
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = [executor.submit(run_component_test, cmd) for cmd in TEST_COMMANDS]
        concurrent.futures.wait(futures)

これらのステップに従って、コードを一つずつ改善し、テストを繰り返すことで、全体の品質と効率が向上します。

8. 次のステップ

今回ご紹介したのは、Pythonを使用して複数のシステムコンポーネントをテストするスクリプトの概要と、その効果的な改善方法です。このスクリプトは、データ収集、異常検出、取引実行、システム監視など、各種コンポーネントの正常動作を確認するために用いられます。

改善のポイント

  1. エラーハンドリングの強化：特定のエラータイプごとに詳細な例外処理を追加し、問題発生時により具体的な情報を提供できるようにします。
  2. ロギングの改善：loggingモジュールを導入し、異なるログレベルを活用して情報の重要度に応じて適切なログ出力を行います。
  3. プロセス監視と終了処理の改善：プロセスの状態を継続的に確認し、予期せぬ問題が発生した際に迅速に対応できるようにします。
  4. セキュリティと安定性の向上：安全な方法でプロセスを起動し、shell=Trueの使用を避けることでセキュリティリスクを低減します。
  5. パフォーマンスの最適化：複数のテストを並行して実行し、全体のテスト時間を短縮します。

修正フローの実施

これらの改善点を踏まえた修正フローを段階的に適用することで、スクリプトはより効率的かつ安全に動作するようになります。エラーの特定が容易になり、システムの安定性と信頼性が向上します。

最終的に、このような自動化されたテストアプローチは、ソフトウェア開発プロセスにおいて重要な役割を果たします。それは、開発中のシステムが予定通りに機能し、エラーが最小限に抑えられることを保証するためです。

次に行うのは、本番環境の最適化 (Auto Scalingやリソース制御の導入) です。
具体的には、以下の内容に取り組みます。

✅ KubernetesのHorizontal Pod Autoscaler (HPA) による自動スケーリング
✅ CPUやメモリのリソース管理での最適化
✅ Helm を活用したKubernetesマニフェストの一元管理

Yodaka

次の記事では、Kubernetesを活用した本番環境の最適化についてまとめます。

関連
開発記録#149(2025/3/25)「論文ベースのbot開発フローpart.11 本番環境の最適化」

続きを見る

-Bot