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

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

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

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

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

🟢 運用開始後の初日チェック

次のステップでは、運用開始直後の初日チェックを実施し、システムの安定性と正常稼働を確認します。

1. 初日チェックの目的

✅ 稼働状況のリアルタイム確認
✅ 主要コンポーネントの異常検出
✅ Slack通知や監視システムの動作確認
✅ トレードの正確性とログの記録状況の確認

2. 初日チェックのチェックリスト

3. 初日チェック用スクリプト (first_day_check.py)

以下のスクリプトでは、各コンポーネントの状態を自動で確認し、Slack通知の送信とログファイルへの保存を行います。

import subprocess
import requests
import time
from datetime import datetime

# Slack通知設定
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR_WEBHOOK_URL"

# ログファイル設定
LOG_FILE = "./first_day_check.log"

# ログ出力関数
def log_message(message):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"[{timestamp}] {message}\n"
    with open(LOG_FILE, "a") as file:
        file.write(log_entry)
    print(log_entry)

# Slack通知送信
def send_slack_alert(message):
    payload = {"text": f"🚨 {message}"}
    requests.post(SLACK_WEBHOOK_URL, json=payload)

# Podの状態確認
def check_pod_status():
    result = subprocess.getoutput("kubectl get pods")
    if "Running" in result:
        log_message("✅ Pod稼働状態確認 - 正常")
    else:
        log_message("❌ Pod稼働状態確認 - 異常検出")
        send_slack_alert("❌ Pod稼働状態に異常があります")

# データ収集状況確認
def check_data_collection():
    result = subprocess.getoutput("kubectl logs <data_collector_pod> | tail -n 10")
    if "data saved" in result:
        log_message("✅ データ収集確認 - 正常")
    else:
        log_message("❌ データ収集確認 - 異常検出")
        send_slack_alert("❌ データ収集に異常があります")

# APIエラー確認
def check_api_errors():
    result = subprocess.getoutput("kubectl logs <trade_executor_pod> | tail -n 10")
    if "API error" in result:
        log_message("❌ APIエラー検出")
        send_slack_alert("❌ APIエラーが発生しました")
    else:
        log_message("✅ APIエラー確認 - 異常なし")

# 初日チェックの実行
def first_day_check():
    log_message("🚀 初日チェックを開始します")
    check_pod_status()
    check_data_collection()
    check_api_errors()
    log_message("✅ 初日チェックが完了しました")

if __name__ == "__main__":
    first_day_check()

4. 実行方法

1. スクリプトの実行

python first_day_check.py

2. Slack通知の確認
   • Botの状態やエラー発生状況がSlackに通知されることを確認
3. ログの確認

cat first_day_check.log

5. 初日チェック後の対応指針

6. 次のステップ

次は、以下のタスクに進みます：

✅ 初週チェック結果の分析：

• 取引成功率、誤検出の有無、Slack通知の精度確認

✅ 結果に基づいたシステムの調整：

• パラメータ最適化、エラーハンドリングの追加、Botの再デプロイ

Yodaka

次の記事では、初週チェック結果の分析を行い、その結果をもとにシステムの最適化を実施します。