本文へ移動
Fabrkntfor HIP-3
レビューを依頼

vault 開発 · keeper の構成 · 4/7

keeper ボットを作る

keeper の構成

keeper ボットは vault の頭脳です。相場を見張り続け、戦略のロジックに沿って売買します。よくある構成は次のとおりです。

┌─────────────────────────────────────────┐
│                Keeper Bot                │
│                                          │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │  Market   │  │ Strategy │  │ Risk   │ │
│  │  Data     │→ │ Engine   │→ │ Check  │ │
│  │  (WS)     │  │          │  │        │ │
│  └──────────┘  └──────────┘  └────────┘ │
│                                    │     │
│                              ┌─────▼───┐ │
│                              │ Order   │ │
│                              │ Manager │ │
│                              └─────────┘ │
└──────────────────┬──────────────────────┘
                   │
                   ▼
            Hyperliquid API

メインループの形

import logging
import time

from hyperliquid.info import Info
from hyperliquid.exchange import Exchange

logger = logging.getLogger(__name__)

class VaultKeeper:
    # `exchange` must be created with vault_address set, e.g.
    # Exchange(agent_wallet, base_url, vault_address=vault_address),
    # so every order is signed for the vault.
    def __init__(self, exchange: Exchange, info: Info, vault_address: str):
        self.exchange = exchange
        self.info = info
        self.vault_address = vault_address
        self.running = False

    def get_market_data(self) -> dict:
        """Fetch current market data."""
        mid_prices = {}
        all_mids = self.info.all_mids()
        for coin, price in all_mids.items():
            mid_prices[coin] = float(price)
        return mid_prices

    def get_vault_state(self) -> dict:
        """Get current vault positions and margin."""
        return self.info.user_state(self.vault_address)

    def evaluate_strategy(self, market_data: dict, vault_state: dict) -> list:
        """Strategy logic — returns list of actions to take."""
        raise NotImplementedError("Subclasses must implement strategy logic")

    def check_risk(self, actions: list, vault_state: dict) -> list:
        """Filter actions through risk checks."""
        filtered = []
        margin_used = float(vault_state["marginSummary"]["totalMarginUsed"])
        account_value = float(vault_state["marginSummary"]["accountValue"])
        margin_ratio = margin_used / account_value if account_value > 0 else 1.0

        for action in actions:
            if margin_ratio > 0.8:
                logger.warning("Margin ratio too high, skipping new positions")
                continue
            filtered.append(action)
        return filtered

    def execute(self, actions: list):
        """Submit orders to the exchange."""
        for action in actions:
            try:
                result = self.exchange.order(
                    action["coin"],
                    action["is_buy"],
                    action["sz"],
                    action["price"],
                    action["order_type"],
                )
                logger.info(f"Order result: {result}")
            except Exception as e:
                logger.error(f"Order failed: {e}")

    def run(self, interval_seconds: float = 60.0):
        """Main keeper loop."""
        self.running = True
        logger.info(f"Keeper started for vault {self.vault_address}")

        while self.running:
            try:
                market_data = self.get_market_data()
                vault_state = self.get_vault_state()
                actions = self.evaluate_strategy(market_data, vault_state)
                safe_actions = self.check_risk(actions, vault_state)
                if safe_actions:
                    self.execute(safe_actions)
            except Exception as e:
                logger.error(f"Keeper loop error: {e}")

            time.sleep(interval_seconds)

動かす場所

keeper は、ふつうクラウドのサーバーで動かします。