Vault Development · Keeper Architecture · 4/7
Building a Keeper Bot
Keeper Architecture
A keeper bot is the brain of a vault. It continuously monitors market conditions and executes trades according to the strategy logic. Here's the typical architecture:
┌─────────────────────────────────────────┐
│ Keeper Bot │
│ │
│ ┌──────────┐ ┌──────────┐ ┌────────┐ │
│ │ Market │ │ Strategy │ │ Risk │ │
│ │ Data │→ │ Engine │→ │ Check │ │
│ │ (WS) │ │ │ │ │ │
│ └──────────┘ └──────────┘ └────────┘ │
│ │ │
│ ┌─────▼───┐ │
│ │ Order │ │
│ │ Manager │ │
│ └─────────┘ │
└──────────────────┬──────────────────────┘
│
▼
Hyperliquid API
Main Loop Pattern
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)
Deployment
Keepers are typically deployed on cloud servers:
- AWS EC2 — Most common, use a small instance (t3.micro is sufficient)
- Always-on — The keeper must run 24/7
- Monitoring — Set up alerts for crashes, disconnections, and errors
- Auto-restart — Use systemd or supervisord to restart on crash
- Testnet first — Run the keeper against
constants.TESTNET_API_URLbefore pointing it at a real vault