Bot

開発記録#162(2025/4/1)「論文ベースのbot開発フローpart.24」

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

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

参考

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

✅ 1. 運用ルールの確立

  • Botの稼働スケジュール設計
  • 障害発生時の対応フロー策定
  • 異常検出やリスク回避の優先順位決定

✅ 2. 最終パフォーマンスチェック←今ここ

  • シミュレーション結果と実運用結果の比較
  • 取引成功率、利益率、エラー率の再評価
  • 調整が必要なパラメータの特定

✅ 3. デプロイ後の安定性確認

  • Podの状態、リソース使用率の監視
  • APIエラーやデータ欠損の確認
  • Slack通知やログ記録の精度確認

✅ 1.運用ルールの確立 (Botの稼働スケジュールや障害対応フローの策定)
✅ 2.最終パフォーマンスチェック (シミュレーションと実運用の比較検証)
✅ 3.デプロイ後の安定性確認 (サーバー監視とログの確認)

Yodaka

次のステップとして、2.最終パフォーマンスチェック（シミュレーションと実運用の比較検証）に進みます。

✅ 2. 最終パフォーマンスチェック：概要

🎯 目的

  • シミュレーション（バックテスト）と実運用の結果を比較し、Botの整合性と戦略の信頼性を検証
  • 誤差やズレの要因を特定し、必要であればパラメータやロジックの最終調整を行う

📊 チェック指標一覧

指標シミュレーション実運用差分の考察
総トレード数backtest_result.csvlive_trade_log.csv実行頻度の一致
勝率同上同上実行精度の確認
最終残高initial + PnL初期証拠金 + 実際損益実益と理論値の差
最大ドローダウンbacktest記録live drawdown計測リスク評価の実差
スリッページ-実取引のfill価格で評価オーダー執行遅延の影響

📁 想定ファイル構成

results/
├── backtest_result.csv         ← バックテストログ
├── live_trade_log.csv          ← 実運用ログ
├── performance_comparison.py   ← 比較スクリプト

🧪 比較スクリプト: performance_comparison.py

import pandas as pd

def load_data():
    backtest = pd.read_csv('./results/backtest_result.csv')
    live = pd.read_csv('./results/live_trade_log.csv')
    return backtest, live

def calculate_metrics(df):
    total_trades = len(df)
    wins = df[df['pnl'] > 0]
    losses = df[df['pnl'] <= 0]
    win_rate = len(wins) / total_trades * 100 if total_trades > 0 else 0
    pnl_total = df['pnl'].sum()
    max_drawdown = (df['balance'].cummax() - df['balance']).max()
    return {
        'trades': total_trades,
        'win_rate': round(win_rate, 2),
        'pnl_total': round(pnl_total, 2),
        'max_drawdown': round(max_drawdown, 2)
    }

def compare():
    backtest, live = load_data()

    bt_metrics = calculate_metrics(backtest)
    live_metrics = calculate_metrics(live)

    print("\n📈 最終パフォーマンス比較")
    print(f"{'指標':<20} | {'シミュレーション':<15} | {'実運用':<15}")
    print("-" * 55)
    for key in bt_metrics:
        print(f"{key:<20} | {bt_metrics[key]:<15} | {live_metrics[key]:<15}")

if __name__ == "__main__":
    compare()

✅ 実行方法

python performance_comparison.py

📌 差分が大きい場合の対応例

差分が大きい指標原因候補改善アクション
トレード数が少ないモード切替 / エラー停止ログとスケジューラの再確認
勝率が低い市場の実変動 / 遅延スリッページの補正導入
PnLが悪化高手数料 / 成行注文発注ロジック調整（指値＋監視）
ドローダウンが大きい不正確な損切りATRベースの動的SL再評価

🔜 次のステップ

  • 比較結果を元に戦略ロジックやパラメータの最終調整
  • 本番リリースに向けて「安定稼働のモニタリング設計」へ移行
Yodaka

以下にperformance_comparison.py のコードの構造と役割を分かりやすく解説します。

📄 スクリプト全体の目的performance_comparison.py

このスクリプトは、シミュレーション結果と実運用の結果を比較して、Botの戦略のパフォーマンスが一致しているか、ズレがあるかを評価するためのものです。

🧩 構成の分解と解説

### ① load_data()：データの読み込み

def load_data():
    backtest = pd.read_csv('./results/backtest_result.csv')
    live = pd.read_csv('./results/live_trade_log.csv')
    return backtest, live
  • backtest_result.csv と live_trade_log.csv の2つのCSVファイルを読み込みます。
  • これらは、バックテスト時のトレードログと実運用時のログがそれぞれ格納されている前提です。

### ② calculate_metrics(df)：各種指標の計算

def calculate_metrics(df):
    total_trades = len(df)
    wins = df[df['pnl'] > 0]
    losses = df[df['pnl'] <= 0]
    win_rate = len(wins) / total_trades * 100 if total_trades > 0 else 0
    pnl_total = df['pnl'].sum()
    max_drawdown = (df['balance'].cummax() - df['balance']).max()
    return {
        'trades': total_trades,
        'win_rate': round(win_rate, 2),
        'pnl_total': round(pnl_total, 2),
        'max_drawdown': round(max_drawdown, 2)
    }

各トレードログから以下の指標を算出します：

指標名説明
trades総トレード数（行数）
win_rate勝率：pnlがプラスのトレードの割合
pnl_total総損益
max_drawdown最大ドローダウン（口座残高がどれだけ減ったかの最大値）

※ balance 列は、各トレード後の残高を記録した列と仮定しています。

### ③ compare()：比較して出力

def compare():
    backtest, live = load_data()
    bt_metrics = calculate_metrics(backtest)
    live_metrics = calculate_metrics(live)

    print("\n📈 最終パフォーマンス比較")
    print(f"{'指標':<20} | {'シミュレーション':<15} | {'実運用':<15}")
    print("-" * 55)
    for key in bt_metrics:
        print(f"{key:<20} | {bt_metrics[key]:<15} | {live_metrics[key]:<15}")
  • load_data() で読み込んだデータを calculate_metrics() に渡し、バックテストと実運用それぞれの指標を計算します。
  • 表形式で各指標の数値を並べて比較します。
  • 目視でズレを把握しやすくするための出力形式です。

### ④ main() 実行部

if __name__ == "__main__":
    compare()
  • スクリプトとして直接実行した場合に compare() を呼び出します。
  • 他のファイルからインポートしたときは実行されません。

📝 補足：期待されるCSVファイルのフォーマット

backtest_result.csv / live_trade_log.csv の必要列

timestamp,symbol,side,size,price,pnl,balance
2025-03-30 10:01:00,BTCUSDT,buy,0.01,70500,50,10050
2025-03-30 11:05:00,BTCUSDT,sell,0.01,71000,-20,10030
...

✅ このスクリプトでできること

  • シンプルなCSVファイルからBotの実力を検証
  • 実運用がどれだけバックテストと乖離しているか可視化
  • 次の改善アクション（スリッページ考慮、パラメータ再調整など）に繋げる

🔜 次のアクション候補

  • 実際に比較した結果の数値を確認し、ズレがある場合はその要因を深掘り
  • 結果に基づいて、戦略の調整やパラメータ最適化に移る
Yodaka

ここからは、比較用のファイル（シミュレーションと実運用）を正しく作成・整備する方法を紹介します。

🗂️ 必要なファイル一覧と構成

ファイル名目的備考
backtest_result.csvシミュレーションで得た取引ログPythonバックテスト出力など
live_trade_log.csv実運用で実際に行われた取引ログBotのログ or APIから取得

✅ CSVフォーマットの共通仕様

両ファイルで同じ列構成に揃えると比較がスムーズになります。以下のような構造が理想です。

timestamp,symbol,side,size,price,pnl,balance
2025-03-30 10:00:00,BTCUSDT,buy,0.01,70500,50,10050
2025-03-30 11:00:00,BTCUSDT,sell,0.01,71000,-20,10030
列名説明
timestampトレードの日時
symbol通貨ペア（例: BTCUSDT）
side売買（buy or sell）
size注文サイズ（例: 0.01）
price約定価格
pnl損益（例: 50 = +$50）
balanceその時点の口座残高

🧪 1. バックテスト結果のCSVを作成する

例：バックテストログからCSV保存（backtester.py）

import pandas as pd
from datetime import datetime

# ダミー取引ログ
data = [
    {"timestamp": "2025-03-30 10:00:00", "symbol": "BTCUSDT", "side": "buy", "size": 0.01, "price": 70500, "pnl": 50, "balance": 10050},
    {"timestamp": "2025-03-30 11:00:00", "symbol": "BTCUSDT", "side": "sell", "size": 0.01, "price": 71000, "pnl": -20, "balance": 10030}
]

df = pd.DataFrame(data)
df.to_csv("./results/backtest_result.csv", index=False)
print("✅ バックテスト結果を保存しました")

🛰️ 2. 実運用ログのCSVを作成する

方法1：Botの取引ログから整形する（ログファイル → CSV）

# log/live_trade.log の内容：
# [2025-03-30 10:00] Trade executed: BTCUSDT buy 0.01 @ 70500, PnL: 50, Balance: 10050

import re
import pandas as pd

pattern = r"\[(.*?)\] Trade executed: (\w+) (\w+) ([\d.]+) @ ([\d.]+), PnL: (-?\d+), Balance: (\d+)"

rows = []
with open("./logs/live_trade.log", "r") as f:
    for line in f:
        match = re.search(pattern, line)
        if match:
            timestamp, symbol, side, size, price, pnl, balance = match.groups()
            rows.append({
                "timestamp": timestamp,
                "symbol": symbol,
                "side": side,
                "size": float(size),
                "price": float(price),
                "pnl": float(pnl),
                "balance": float(balance)
            })

df = pd.DataFrame(rows)
df.to_csv("./results/live_trade_log.csv", index=False)
print("✅ 実運用ログをCSVに変換しました")

🔧 3. 整備したCSVファイルの確認チェックリスト

チェック項目内容
✅ 列名と構成が両ファイルで一致しているか？timestamp, symbol, side, size, price, pnl, balance
✅ 日時のフォーマットは同一か？%Y-%m-%d %H:%M:%S
✅ 欠損値や空欄はないか？df.isnull().sum() で確認
✅ ファイルパスが ./results/ にあるか？performance_comparison.py で読まれる

✅ 整備が終わったら次にできること

  • performance_comparison.py を実行して比較
  • ズレがあれば分析 → パラメータ調整へ進行
  • モデルや戦略ロジックの改善につなげる
Yodaka

必要であれば、あなたのログファイルやバックテスト出力に合わせて自動変換スクリプトを作成するのも良いでしょう。

パフォーマンス比較

✅ ステップ 1：前提ファイルがこのように整備されていることを確認

results/
├── backtest_result.csv      ✅ バックテスト結果
├── live_trade_log.csv       ✅ 実運用ログ
performance_comparison.py    ✅ 比較スクリプト

✅ ステップ 2：スクリプトを実行

以下のコマンドをターミナルやPython環境で実行してください：

python performance_comparison.py

💡 出力される内容（例）

📈 最終パフォーマンス比較
指標                  | シミュレーション     | 実運用           
-------------------------------------------------------
trades               | 128               | 116             
win_rate             | 58.59             | 51.72           
pnl_total            | 2650.50           | 1780.10         
max_drawdown         | 420.25            | 690.80

✅ 出力の読み解き方

指標解釈コメント例
trades実行トレード数実運用がやや少なければ、通信障害・Bot停止の可能性あり
win_rate勝率実運用が下がっているなら、マーケットのズレやスリッページの影響あり
pnl_total総利益シミュレーションに比べて少ない場合、手数料や不発注が要因かも
max_drawdown最大損失幅実運用が大きければ、ロスカット条件や取引間隔の見直しが必要

🔧 差分が大きかったら？

次のような対応ができます：

差分改善方法
勝率低下戦略条件の緩和、RSIやATRの再調整
実行数不足スケジューラ（時間帯）やクラッシュ対応の見直し
ドローダウン大ダイナミックなStop Lossやロット縮小の導入

🔜 次のステップ

  1. 🔁 パフォーマンス改善のためのロジックやパラメータの再調整
  2. ✅ デプロイ後の安定性確認（サーバー監視とログのチェック）
Yodaka

次に、パフォーマンス改善のためのロジックやパラメータの再調整を行います。

✅ 現時点で見直すべきポイント（例）

以下のような問題があれば、対応策を具体的に打っていきます：

問題原因候補改善アプローチ
勝率がシミュより低い売買判断の過剰 or トリガー遅れRSIやATR閾値の再評価、トレンドフィルター導入
PnLが大幅に下がっている手数料/スリッページ影響指値注文＆キャンセル再試行による調整
トレード数が少ないモード制限、誤検出抑制が過剰稼働スケジュールまたはイベント検出条件の緩和
ドローダウンが大きい損切り条件が甘い or 無制限エントリATR/ボラティリティによるStopLoss調整導入

🛠️ 1. パラメータ見直し例（config.yaml）

# パラメータ調整例（元）
RSI_OVERSOLD: 30
RSI_OVERBOUGHT: 70
ATR_THRESHOLD: 100
STOP_LOSS_PERCENT: 4.2
TAKE_PROFIT_PERCENT: 6.8

# ↓↓↓ 調整後（例）
RSI_OVERSOLD: 35                # 早めの買い判断
RSI_OVERBOUGHT: 65              # 早めの売り判断
ATR_THRESHOLD: 80               # 小さいボラでも反応
STOP_LOSS_PERCENT: 3.0          # 小さく切る
TAKE_PROFIT_PERCENT: 5.5        # 早め利確

🔄 2. ロジック強化：トレンドフィルターを加える（移動平均で方向性確認）

# 移動平均線を加える
df['ma_fast'] = df['close'].rolling(window=5).mean()
df['ma_slow'] = df['close'].rolling(window=20).mean()

# トレンド方向でフィルタ
if df['ma_fast'].iloc[i] > df['ma_slow'].iloc[i]:  # 上昇トレンド中
    if df['rsi'].iloc[i] < RSI_OVERSOLD:
        # 買いエントリ検討

📊 3. ロジック検証のための追加ログ出力（勝率・P&L改善の検証）

logger.info(f"取引判定：RSI={rsi:.2f}, ATR={atr:.2f}, Trend={trend_direction}")
logger.info(f"約定結果：Price={price}, PnL={pnl}, Balance={balance}")

🧪 4. 調整後のバックテスト実行

  • 調整した戦略で backtest_result.csv を更新
  • 再度 performance_comparison.py を実行して、改善効果を数値で確認

🔜 次の選択肢

  1. ⚙️ このまま ロジック再調整コードを具体的に修正したい場合 → AIなどでコーディングする場合は元のコードがあるとベスト
  2. ✅ 調整後の 再テスト実行へ進む→このプロセスを繰り返して、堅牢なプログラムを書いていく
  3. 📈 改善効果を検証後、デプロイと運用安定化へ進む
Yodaka

次回は、デプロイ後の安定性確認（サーバー監視とログのチェック）について解説します。

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

続きを見る

-Bot