Hermes Agentをトレーディングパートナーとして、マルチチェーンDEXトレーディングボットをどのように構築したか

これは、への提出です。エルメス・エージェント・チャレンジ: : Hermes Agent（エルメスエージェント）で構築

私が作ったもの

IYOPトレーディングボットは、AI駆動のマルチチェーンDEXトレーディングボットであり、Solana（Jupiter）、Ethereum（Uniswap V3）、BSC（PancakeSwap V2）全体で取引を実行します。MiMo v2を使用して市場分析を行い、複数のテクニカル指標からシグナルを生成し、ポジションサイジングとドローダウン保護によってリスクを管理します。

私が解決しようとしていた問題は、分散型取引が断片化していることです。各チェーンには独自のDEX、独自の特性、独自の流動性プールがあります。それらを手動で監視することは不可能です。チェーン間での取引実行には、異なるSDK、異なるガス管理、異なるウォレット処理が必要です。

統一された層が必要でした。それは次のことが可能なものです：

複数のDEXを同時にスキャンする
テクニカル指標とAI分析からシグナルを生成
適切なリスク管理で取引を実行
チェーン間でポジションを追跡
ストップロスとテイクプロフィットを自動処理

そして、Hermes Agent により開発プロセスを実際に出荷可能な速度にする必要がありました。

デモ

ライブデモ: iyop-trading-terminal.vercel.app

ソース： github.com/iyop666/iyop-trading-bot

ボットはFastAPIバックエンドとウェブダッシュボードを備えて動作します。主な機能：

マルチチェーンDEXスキャン（Jupiter、Uniswap V3、PancakeSwap）
IYOP v2 APIによるAIを活用した市場分析
信頼度スコア付きのシグナル生成
ポジションサイジングとドローダウン制限によるリスク管理
ストップロス/テイクプロフィットによるトレード執行
リアルタイムWebSocket価格フィード

コード

以下は、Hermes Agent(エルメスエージェント)が構築を支援したコアモジュールです：

AI Market Analyzer(AIマーケットアナライザー)

このアナライザーはIYOP v2に接続し、高速な市場分析、シグナル生成、センチメント分析を実現します：

import httpx
from dataclasses import dataclass
from enum import Enum

MIMO_API_URL = "http://127.0.0.1:19911/v1/chat/completions"
MIMO_MODEL = "gitlawb/mimo-v2-flash"

class Signal(Enum):
    BUY = "buy"
    SELL = "sell"
    HOLD = "hold"

@dataclass
class MarketAnalysis:
    signal: Signal
    confidence: float
    reasoning: str
    risk_level: str

async def analyze_market(pair: str, timeframe: str) -> MarketAnalysis:
    """Send market data to MiMo v2 for analysis."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            MIMO_API_URL,
            json={
                "model": MIMO_MODEL,
                "messages": [
                    {"role": "system", "content": "Analyze this trading pair..."},
                    {"role": "user", "content": f"Pair: {pair}, Timeframe: {timeframe}"}
                ],
                "temperature": 0.1
            },
            timeout=30
        )
        result = response.json()
        return parse_analysis(result)

信号生成

複数のテクニカル指標を加重スコアリングで組み合わせます:

from dataclasses import dataclass
from enum import Enum

class SignalType(Enum):
    STRONG_BUY = "strong_buy"
    BUY = "buy"
    WEAK_BUY = "weak_buy"
    HOLD = "hold"
    WEAK_SELL = "weak_sell"
    SELL = "sell"
    STRONG_SELL = "strong_sell"

@dataclass
class TradeSignal:
    signal_type: SignalType
    confidence: float  # 0.0 to 1.0
    pair: str
    price: float
    indicators: dict
    reasoning: str

def generate_signal(indicators: dict, ai_analysis: dict) -> TradeSignal:
    """Combine technical indicators with AI analysis."""
    weights = {
        "rsi": 0.25,
        "macd": 0.20,
        "bollinger": 0.15,
        "volume": 0.15,
        "ai_analysis": 0.25
    }

    score = 0.0
    for indicator, weight in weights.items():
        score += normalize(indicators.get(indicator, 0)) * weight

    signal_type = classify_signal(score)
    return TradeSignal(
        signal_type=signal_type,
        confidence=abs(score),
        pair=indicators["pair"],
        price=indicators["price"],
        indicators=indicators,
        reasoning=build_reasoning(indicators, ai_analysis)
    )

マルチチェーンDEX統合

Jupiter、Uniswap V3、PancakeSwapのための統合インターフェース:

from enum import Enum
import httpx

class DEX(Enum):
    JUPITER = "jupiter"      # Solana
    UNISWAP = "uniswap"      # Ethereum
    PANCAKESWAP = "pancakeswap"  # BSC

class Chain(Enum):
    SOLANA = "solana"
    ETHEREUM = "ethereum"
    BSC = "bsc"

@dataclass
class SwapResult:
    success: bool
    tx_hash: str
    input_amount: float
    output_amount: float
    chain: Chain
    dex: DEX
    gas_used: float

async def execute_swap(
    dex: DEX,
    chain: Chain,
    token_in: str,
    token_out: str,
    amount: float,
    slippage: float = 0.5
) -> SwapResult:
    """Execute a swap on the specified DEX."""
    if dex == DEX.JUPITER:
        return await jupiter_swap(token_in, token_out, amount, slippage)
    elif dex == DEX.UNISWAP:
        return await uniswap_swap(token_in, token_out, amount, slippage)
    elif dex == DEX.PANCAKESWAP:
        return await pancakeswap_swap(token_in, token_out, amount, slippage)

リスク管理

ポジションサイズとドローダウン保護:

@dataclass
class RiskManager:
    max_position_pct: float = 0.1  # 10% max per position
    max_daily_loss: float = 0.05   # 5% max daily loss
    max_drawdown: float = 0.15     # 15% max drawdown
    current_drawdown: float = 0.0
    daily_pnl: float = 0.0

    def can_open_position(self, portfolio_value: float, position_size: float) -> bool:
        """Check if a new position is within risk limits."""
        # Check position size limit
        if position_size / portfolio_value > self.max_position_pct:
            return False

        # Check daily loss limit
        if self.daily_pnl < -(portfolio_value * self.max_daily_loss):
            return False

        # Check drawdown limit
        if self.current_drawdown > self.max_drawdown:
            return False

        return True

    def calculate_position_size(
        self,
        portfolio_value: float,
        entry_price: float,
        stop_loss: float
    ) -> float:
        """Calculate optimal position size based on risk parameters."""
        risk_per_trade = portfolio_value * self.max_position_pct
        price_risk = abs(entry_price - stop_loss) / entry_price
        return risk_per_trade / price_risk if price_risk > 0 else 0

取引執行エンジン

ストップロスとテイクプロフィットで注文ライフサイクルを処理:

class Side(str, Enum):
    BUY = "buy"
    SELL = "sell"

class OrderStatus(str, Enum):
    PENDING = "pending"
    FILLED = "filled"
    CANCELLED = "cancelled"
    SL_HIT = "stop_loss_hit"
    TP_HIT = "take_profit_hit"

@dataclass
class Position:
    id: str
    pair: str
    side: Side
    entry_price: float
    current_price: float
    size: float
    stop_loss: float
    take_profit: float
    pnl: float
    chain: Chain
    dex: DEX

class Trader:
    def __init__(self, risk_manager: RiskManager):
        self.positions: dict[str, Position] = {}
        self.risk_manager = risk_manager
        self.trade_history: list[dict] = []

    async def open_position(
        self,
        pair: str,
        side: Side,
        size: float,
        stop_loss: float,
        take_profit: float,
        chain: Chain,
        dex: DEX
    ) -> Position:
        """Open a new position with risk checks."""
        if not self.risk_manager.can_open_position(self.portfolio_value, size):
            raise RiskLimitExceeded("Position exceeds risk limits")

        # Execute the swap
        result = await execute_swap(dex, chain, pair, size)

        position = Position(
            id=str(uuid.uuid4()),
            pair=pair,
            side=side,
            entry_price=result.price,
            current_price=result.price,
            size=size,
            stop_loss=stop_loss,
            take_profit=take_profit,
            pnl=0.0,
            chain=chain,
            dex=dex
        )

        self.positions[position.id] = position
        return position

私の技術スタック

バックエンド: Python、FastAPI、Pydantic
AIエンジン: MiMo v2（ローカルAPI）
DEX統合: Jupiter（Solana）、Uniswap V3（Ethereum）、PancakeSwap V2（BSC）
データ: WebSocket価格フィード、非同期HTTPのためのhttpx
フロントエンド: HTMLダッシュボード
AIアシスタント: Hermes Agent（開発、デバッグ、デプロイメント用）

Hermes Agentの使い方

1. アーキテクチャ設計

トレーディングボットの構築を始めたとき、要件をHermesに説明しました:

「AI分析、シグナル生成、リスク管理、取引実行が可能なマルチチェーンDEX取引ボットが必要です。Jupiter、Uniswap V3、PancakeSwapをサポートする必要があります。」

Hermesはモジュール構造を設計しました：

backend/ai/ AI分析とシグナル生成用
backend/core/ DEX統合、リスク管理、取引実行用
分析と実行の関心の分離

そのアーキテクチャのおかげで、後でリファクタリングに何時間も費やさずに済んだ。

2. DEX統合コード

各DEXはAPI、トランザクション形式、ガス見積もりがそれぞれ異なる。個々のドキュメントを読む代わりに、私はHermesにこう指示した:

「Jupiter、Uniswap V3、PancakeSwap用の非同期スワップ関数を書け。httpxを使い、エラーを処理し、トランザクションハッシュを返せ。」

Hermesは3つのDEXすべての統合コードを生成した。私はエラーハンドリングを確認し、スリッページパラメータを調整し、テストを行った。ドキュメントを読んでボイラープレートを書くのに丸一日かかるはずだった作業が30分で完了した。

3. リスク管理ロジック

リスクマネージャーは、ポジションサイジング、ドローダウン制限、日次損失上限を処理する必要があった。私はルールを説明した：

"1ポジションあたり最大10%、1日損失最大5%、最大ドローダウン15%。ストップロスの距離に基づいてポジションサイズを計算します。"

Hermesは適切なエッジケースを含むロジックを実装しました。極端な値（100%損失シナリオ）でテストしたところ、トレードを正しく拒否しました。

4. シグナル重み付けシステム

複数のインジケーターを単一のシグナルに結合するには、重み付けシステムが必要でした。私はHermesに尋ねました：

「RSI（25%）、MACD（20%）、ボリンジャー（15%）、出来高（15%）、AI分析（25%）の重み付けを行うシグナルジェネレーターを作成する。それぞれを-1から+1の範囲に正規化する。」

Hermesは設定可能な重みを持つ加重スコアリングシステムを構築した。その後、テストによりAI分析が最も予測力があることが示されたため、AI分析の重みを30%に調整した。

ボットが正しく取引を実行していなかったとき、私はヘルメスに問題を追跡するよう依頼しました：

「取引が実行されていません。スワップ機能、リスクマネージャー、トレードエグゼキューターを確認してください。ボトルネックを特定してください。」

ヘルメスはコードを読み通し、ポートフォリオの価値が各取引後に更新されていなかったためにリスクマネージャーが取引を拒否していることを発見しました。修正は1行でしたが、手動で見つけるには1時間のデバッグが必要だったでしょう。

Hermes Agentが不可欠だった理由

トレーディングボットの構築には以下が必要です：

高速な反復（市場は動くため、コードもそれに追従する必要がある）
モジュール間の認識（リスク管理はトレーダーに影響し、トレーダーはポートフォリオに影響する）
非同期プログラミングパターン（すべてのDEX呼び出しは非同期である）
エラーハンドリングのエッジケース（ネットワーク障害、ガス不足、スリッページ）

Hermesはコードベース全体でコンテキストを維持していたため、これらすべてを処理しました。リスクマネージャーを変更すると、自動的にトレーダーへの更新を提案しました。新しいDEXを追加すると、新しいチェーンを含めるようにシグナルジェネレーターを更新しました。

そのクロスモジュール認識が、プロトタイプと動作するボットの違いを生み出しました。

プロジェクト統計：

約2,000行のPython
3つのDEX統合（Jupiter、Uniswap V3、PancakeSwap）
5つの取引戦略（モメンタム、平均回帰、AI駆動、スキャルピング、ハイブリッド）
3層の安全策を備えたリスク管理
リアルタイムWebSocket価格フィード
MiMo v2によるAI駆動の市場分析

リンク：

おすすめ購読元

DEV Community