Bot

開発記録#167(2025/4/3)「論文ベースのbot開発フローpart.29」

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

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

参考

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

Bot Ops Full Stack

# ✅ Kubernetes Deployment Template for Multi-Bot
apiVersion: apps/v1
kind: Deployment
metadata:
  name: bot-{{PAIR}}
spec:
  replicas: 1
  selector:
    matchLabels:
      app: bot-{{PAIR}}
  template:
    metadata:
      labels:
        app: bot-{{PAIR}}
    spec:
      restartPolicy: Always
      containers:
      - name: bot
        image: your_dockerhub_user/crypto-bot:latest
        args: ["bots/{{PAIR}}/config.yaml"]
        env:
        - name: PAIR
          value: "{{PAIR}}"
        resources:
          limits:
            memory: "1Gi"
            cpu: "500m"
---
# ✅ Slack Alert Script (send_slack.py)
import requests

def send_slack_alert(message):
    url = "https://hooks.slack.com/services/XXXX/XXXX/XXXX"
    payload = {"text": f"\ud83d\udea8 {message}"}
    requests.post(url, json=payload)
---
# ✅ Logger Utility (logger.py)
import logging
from datetime import datetime

def get_logger(name, log_file=None):
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    if not logger.handlers:
        formatter = logging.Formatter("[%(asctime)s] %(levelname)s - %(message)s", "%Y-%m-%d %H:%M:%S")
        log_file = log_file or f"./logs/{name}_{datetime.now().strftime('%Y%m%d')}.log"
        file_handler = logging.FileHandler(log_file)
        console_handler = logging.StreamHandler()
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)
    return logger
---
# ✅ Retry With Alert (retry_with_alert.py)
import time
from send_slack import send_slack_alert

def retry_with_alert(task_func, retries=3, delay=5, task_name=""):
    for attempt in range(1, retries + 1):
        try:
            return task_func()
        except Exception as e:
            if attempt == retries:
                send_slack_alert(f"{task_name} \u306f {retries} \u56de\u9023\u7d9a\u5931\u6557\u3057\u307e\u3057\u305f: {e}")
            time.sleep(delay)
---
# ✅ Backup Script (backup_script.py)
import boto3
import os
from datetime import datetime

AWS_ACCESS_KEY = "YOUR_ACCESS"
AWS_SECRET_KEY = "YOUR_SECRET"
BUCKET_NAME = "crypto-bot-backups"

def backup_to_s3():
    s3 = boto3.client('s3', aws_access_key_id=AWS_ACCESS_KEY, aws_secret_access_key=AWS_SECRET_KEY)
    for folder in ["./logs", "./models"]:
        for root, _, files in os.walk(folder):
            for f in files:
                local_path = os.path.join(root, f)
                s3_key = f"{datetime.now().strftime('%Y-%m-%d')}/{f}"
                s3.upload_file(local_path, BUCKET_NAME, s3_key)
                print(f"\u2705 {f} \u3092 S3 \u306b\u30d0\u30c3\u30af\u30a2\u30c3\u30d7")

if __name__ == "__main__":
    backup_to_s3()
---
# ✅ Monitor & Backup Executor (monitor.py)
from retry_with_alert import retry_with_alert
from backup_script import backup_to_s3
from send_slack import send_slack_alert

# placeholder for resource checks

def monitor():
    try:
        # ここでPrometheus値の監視などに置換可能
        print("\U0001f9e0 モニタリング中...")
    except Exception as e:
        send_slack_alert(f"\u30e2\u30cb\u30bf\u30ea\u30f3\u30b0\u30a8\u30e9\u30fc: {e}")

if __name__ == "__main__":
    monitor()
    backup_to_s3()

Bot運用の安定化・監視・通知・バックアップすべてを網羅したテンプレート一式です ✅
このファイルは以下の内容を含んでいます：

📦 セット内容

ファイル名内容
deployment-template.yaml通貨ペアごとのKubernetes Deployment定義（テンプレート形式）
send_slack.pySlack通知機能（Webhook連携）
logger.py共通ロガーユーティリティ
retry_with_alert.py異常時リトライ＋Slack通知
backup_script.pyS3バックアップ用スクリプト（logs, models）
monitor.pyモニタリング & バックアップの自動実行用

コードの解説

上記のコードを複数通貨ペア対応の自動取引BotをKubernetes上で安定的に運用するためのテンプレート一式として、それぞれのコードの役割と実装意図を解説します。

🧱 構成の全体像

このテンプレートは、次の機能を網羅しています：

  • ✅ 各通貨ペアBotを個別にKubernetesでデプロイ可能（テンプレート対応）
  • ✅ エラー検知時の Slack通知連携
  • ✅ ログのローカル保存と S3へのバックアップ
  • ✅ monitor.py による簡易 モニタリング＆バックアップ自動実行
  • ✅ 共通ユーティリティ：ログ記録・リトライ機構

1️⃣ Kubernetes Deployment テンプレート

apiVersion: apps/v1
kind: Deployment
metadata:
  name: bot-{{PAIR}}
spec:
  replicas: 1
  selector:
    matchLabels:
      app: bot-{{PAIR}}
  template:
    metadata:
      labels:
        app: bot-{{PAIR}}
    spec:
      restartPolicy: Always
      containers:
      - name: bot
        image: your_dockerhub_user/crypto-bot:latest
        args: ["bots/{{PAIR}}/config.yaml"]
        env:
        - name: PAIR
          value: "{{PAIR}}"
        resources:
          limits:
            memory: "1Gi"
            cpu: "500m"

🔍 解説

  • {{PAIR}} の部分に btcusdt や ethusdt を入れることで、複数通貨のBotをテンプレートから自動展開可能。
  • restartPolicy: Always によりBotが異常終了しても自動復旧。
  • コンテナのリソース上限（CPU/メモリ）も制限済みで安全。

2️⃣ Slack通知スクリプト：send_slack.py

import requests

def send_slack_alert(message):
    url = "https://hooks.slack.com/services/XXXX/XXXX/XXXX"
    payload = {"text": f"🚨 {message}"}
    requests.post(url, json=payload)

🔍 解説

  • 異常検出時にSlackへリアルタイム通知を送る仕組み。
  • send_slack_alert() はどのモジュールからも呼び出せる汎用関数です。

3️⃣ ロガーユーティリティ：logger.py

import logging
from datetime import datetime

def get_logger(name, log_file=None):
    ...

🔍 解説

  • 標準出力（コンソール）とログファイルの両方に同時出力。
  • ログファイルは自動的に日付付きのファイル名で保存されます。

4️⃣ エラー時のリトライ＋通知：retry_with_alert.py

from send_slack import send_slack_alert

def retry_with_alert(task_func, retries=3, delay=5, task_name=""):
    ...

🔍 解説

  • 指定関数を最大retries回まで実行。
  • 最終失敗時にSlack通知を飛ばすため、重要なAPIコールやファイル操作時に安全。

5️⃣ S3へのバックアップスクリプト：backup_script.py

import boto3
import os
from datetime import datetime

def backup_to_s3():
    ...

🔍 解説

  • ./logs と ./models ディレクトリ配下のファイルを AWS S3 に日付フォルダ付きでアップロード。
  • ログ保持・モデル保存に最適で、バックアップ漏れを防ぎます。

6️⃣ 簡易モニタリング＆定期バックアップ実行：monitor.py

from retry_with_alert import retry_with_alert
from backup_script import backup_to_s3
from send_slack import send_slack_alert

def monitor():
    print("🧠 モニタリング中...")

🔍 解説

  • monitor()：将来的には CPU/メモリ監視やPrometheusとの連携も追加可能。
  • backup_to_s3()：毎回呼び出されるため、Cronから実行すれば自動化も可能。

⏰ Cron への組み込み例

# 毎日3:00にmonitor.pyを実行し、S3にバックアップ
0 3 * * * /usr/bin/python3 /app/monitor.py >> /var/log/monitor.log 2>&1

✅ まとめ：このテンプレートが解決する課題

課題解決方法
Botの異常停止に気づけないSlack通知 + 自動再起動
ログがローカルにしか残らないS3へのバックアップ導入
デプロイが手動で面倒{{PAIR}} テンプレート化で自動展開
管理が分散する共通ユーティリティで一元化

🔜 次にやりたいこと

  • 📈 Grafana用のPrometheus連携テンプレート
  • 🧠 ML戦略の追加と切り替えロジック
  • 🐳 Helmによる複数Botの一括管理化
  • 🌐 Slack通知テンプレートのカスタマイズ（チャートURLや絵文字など）

🔜 次にできること

  • 実運用に合わせて config.yaml や PAIR を指定して起動
  • モニタリング対象を Prometheus に拡張（ご希望あれば対応可）
  • Slack通知テンプレートのカスタマイズ（チャートURLやハイライト付き）
  • 各通貨ペアごとのPodを Helm で一括展開する仕組みへ拡張

必要に応じて、自動デプロイ化 / ダッシュボード連携 / 複数戦略統合などもサポートできます。

Yodaka

次回は「自動デプロイ化 / ダッシュボード連携 / 複数戦略統合のための方法」を詳しく説明します。

関連
開発記録#168(2025/4/3)「論文ベースのbot開発フローpart.30」

続きを見る

-Bot