Bot

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

2025年3月25日

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

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

参考

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

🔒 セキュリティ強化 (APIキーのローテーションとデータバックアップの導入)

次のステップとして、以下の2つのセキュリティ強化策に取り組みます。

✅ APIキーのローテーション (自動更新システムの実装)
✅ データバックアップの自動化 (重要データの安全な保存)

1. APIキーのローテーション

🔎 なぜAPIキーのローテーションが重要か？

  • APIキーが漏洩した際のリスクを最小化
  • 定期的にAPIキーを自動更新し、セキュリティの確保

✅ APIキーのローテーション設計

  1. 新しいAPIキーの発行 (Bybit API)
  2. Kubernetes Secretsの更新
  3. Botの再デプロイ

✅ ローテーションスクリプト (api_key_rotation.py)

import requests
import subprocess
import json
from datetime import datetime

# Bybit API情報
BYBIT_API_KEY = "YOUR_OLD_API_KEY"
BYBIT_API_SECRET = "YOUR_OLD_API_SECRET"
API_URL = "https://api.bybit.com/v2/private/api_key"

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

# 新しいAPIキーの取得
def generate_new_api_key():
    headers = {"Content-Type": "application/json"}
    payload = {
        "api_key": BYBIT_API_KEY,
        "secret": BYBIT_API_SECRET,
        "endpoint": "generate_new_key"
    }

    response = requests.post(API_URL, headers=headers, json=payload)
    if response.status_code == 200:
        new_key_data = response.json()
        return new_key_data["api_key"], new_key_data["api_secret"]
    else:
        print("❌ APIキーの発行に失敗しました")
        return None, None

# Kubernetes Secretsの更新
def update_k8s_secrets(api_key, api_secret):
    subprocess.run([
        "kubectl", "create", "secret", "generic", "api-keys",
        f"--from-literal=BYBIT_API_KEY={api_key}",
        f"--from-literal=BYBIT_API_SECRET={api_secret}",
        "--dry-run=client", "-o", "yaml",
        "|", "kubectl", "apply", "-f", "-"
    ], shell=True)
    print("✅ APIキーの更新が完了しました")

# Slack通知
def send_slack_notification(message):
    payload = {"text": message}
    requests.post(SLACK_WEBHOOK_URL, json=payload)

# メイン処理
def main():
    print("🚀 APIキーのローテーションを開始します...")
    new_api_key, new_api_secret = generate_new_api_key()
    
    if new_api_key and new_api_secret:
        update_k8s_secrets(new_api_key, new_api_secret)
        send_slack_notification(f"✅ 新しいAPIキーが正常に更新されました (更新日時: {datetime.now()})")
        print("✅ APIキーのローテーションが完了しました")
    else:
        send_slack_notification(f"❌ APIキーの更新に失敗しました (日時: {datetime.now()})")

if __name__ == "__main__":
    main()

✅ スクリプトの実行

python api_key_rotation.py

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

🔎 バックアップ対象データ

✅ 市場データ (収集した過去データ)
✅ P&Dイベントの検出結果
✅ 取引履歴とログデータ
✅ トレードモデルと関連パラメータ

✅ バックアップスクリプト (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

3. 自動化のためのスケジュール設定 (cron)

APIキーのローテーションとデータバックアップを自動化するために、cronジョブを設定します。

✅ cron設定ファイル

# 毎月1日の午前2時にAPIキーを更新
0 2 1 * * /usr/bin/python3 /app/api_key_rotation.py

# 毎日午前3時にデータバックアップを実行
0 3 * * * /usr/bin/python3 /app/backup_script.py

✅ cronジョブの設定

crontab -e

4. 次のステップ

Yodaka

次の記事では、パフォーマンス最適化 (非同期処理とモデルの高速化)についてまとめます。

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

続きを見る

-Bot