Hyperliquid Perp DEX 2026：全链永续合约去中心化交易所完整指南

# 创建虚拟环境
python -m venv hyperliquid-env
source hyperliquid-env/bin/activate

# 安装官方 SDK
pip install hyperliquid-python-sdk

# 安装机器人开发依赖
pip install websockets aiohttp pandas numpy python-dotenv

# .env - 切勿提交到版本控制
PRIVATE_KEY=your_ethereum_private_key_here
WALLET_ADDRESS=0x_your_wallet_address
TESTNET=true

import os
import asyncio
from dotenv import load_dotenv
from hyperliquid.exchange import Exchange
from hyperliquid.info import Info
from hyperliquid.utils import constants

load_dotenv()

class HyperliquidTrader:
    """生产级 Hyperliquid 交易客户端。"""
    
    def __init__(self, use_testnet=True):
        self.private_key = os.getenv('PRIVATE_KEY')
        self.wallet_address = os.getenv('WALLET_ADDRESS')
        
        # 根据环境选择端点
        if use_testnet:
            self.base_url = constants.TESTNET_API_URL
        else:
            self.base_url = constants.MAINNET_API_URL
        
        # 初始化交易所和信息客户端
        self.exchange = Exchange(
            self.wallet_address,
            self.private_key,
            self.base_url
        )
        self.info = Info(self.base_url)
        
        print(f"已连接到 Hyperliquid {'Testnet' if use_testnet else 'Mainnet'}")
        print(f"钱包: {self.wallet_address}")
    
    def get_account_summary(self):
        """获取综合账户信息。"""
        user_state = self.info.user_state(self.wallet_address)
        
        account_value = float(user_state['marginSummary']['accountValue'])
        total_margin_used = float(user_state['marginSummary']['totalMarginUsed'])
        withdrawable = float(user_state['withdrawable'])
        
        print(f"账户价值: ${account_value:,.2f}")
        print(f"已用保证金: ${total_margin_used:,.2f}")
        print(f"可提现: ${withdrawable:,.2f}")
        
        return user_state

# 初始化交易员
trader = HyperliquidTrader(use_testnet=True)
trader.get_account_summary()

import json
import websockets

class HyperliquidWebSocketFeed:
    """Hyperliquid 高性能 WebSocket 数据流。"""
    
    def __init__(self):
        self.ws_url = "wss://api.hyperliquid.xyz/ws"
        self.subscriptions = {}
        self.orderbook_cache = {}
        self.running = False
    
    async def connect(self):
        """建立带自动重连的 WebSocket 连接。"""
        while True:
            try:
                async with websockets.connect(self.ws_url) as ws:
                    print("WebSocket 已连接")
                    self.ws = ws
                    self.running = True
                    
                    # 重连时重新订阅之前频道
                    for sub in self.subscriptions.values():
                        await ws.send(json.dumps(sub))
                    
                    await self._listen()
            except Exception as e:
                print(f"WebSocket 错误: {e}. 5 秒后重连...")
                await asyncio.sleep(5)
    
    async def _listen(self):
        """处理传入消息。"""
        async for message in self.ws:
            msg = json.loads(message)
            
            if msg.get("channel") == "l2Book":
                await self._handle_orderbook(msg['data'])
            elif msg.get("channel") == "trades":
                await self._handle_trades(msg['data'])
            elif msg.get("channel") == "userFills":
                await self._handle_fills(msg['data'])
    
    async def _handle_orderbook(self, data):
        """处理 L2 订单簿更新。"""
        coin = data['coin']
        levels = data['levels']
        
        self.orderbook_cache[coin] = {
            'bids': [{'px': float(b['px']), 'sz': float(b['sz'])} for b in levels[0]],
            'asks': [{'px': float(a['px']), 'sz': float(a['sz'])} for a in levels[1]],
            'timestamp': data.get('time', 0)
        }
    
    async def subscribe_orderbook(self, coin):
        """订阅特定市场的实时订单簿。"""
        sub = {
            "method": "subscribe",
            "subscription": {"type": "l2Book", "coin": coin}
        }
        self.subscriptions[f"book_{coin}"] = sub
        if self.running:
            await self.ws.send(json.dumps(sub))
        print(f"已订阅 {coin} 订单簿")

def place_market_order(self, coin: str, is_buy: bool, sz: float):
    """执行带滑点保护的市价单。"""
    order_type = {"limit": {"tif": "Ioc"}}  # 立即成交或取消
    
    result = self.exchange.order(
        coin,
        is_buy,
        sz,
        0,  # 市价单价格 0
        order_type,
        reduce_only=False
    )
    
    print(f"市价 {'买入' if is_buy else '卖出'} {sz} {coin}")
    print(f"状态: {result['status']}")
    return result

def place_limit_order(self, coin: str, is_buy: bool, sz: float, 
                     px: float, tif: str = "Gtc"):
    """放置指定时间属性的限价单。
    
    TIF 选项:
    - Gtc: 取消前有效
    - Ioc: 立即成交或取消
    - Fok: 全部成交或取消
    """
    order_type = {"limit": {"tif": tif}}
    
    result = self.exchange.order(
        coin,
        is_buy,
        sz,
        px,
        order_type,
        reduce_only=False
    )
    
    print(f"限价 {'买入' if is_buy else '卖出'} {sz} {coin} @ {px}")
    return result

def place_stop_loss_order(self, coin: str, is_buy: bool, sz: float,
                          trigger_px: float, limit_px: float):
    """放置带触发价格的止损单。"""
    order_type = {
        "trigger": {
            "triggerPx": str(trigger_px),
            "isMarket": True,
            "tpsl": "sl"
        }
    }
    
    result = self.exchange.order(
        coin,
        is_buy,
        sz,
        limit_px,
        order_type,
        reduce_only=True
    )
    
    print(f"止损 {'买入' if is_buy else '卖出'} {sz} {coin}")
    print(f"触发: {trigger_px}, 限价: {limit_px}")
    return result

import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class TrendFollowingBot:
    """Hyperliquid EMA 交叉趋势跟踪机器人。"""
    
    def __init__(self, trader: HyperliquidTrader, coin: str = "BTC"):
        self.trader = trader
        self.coin = coin
        self.fast_ema_period = 9
        self.slow_ema_period = 21
        self.risk_per_trade = 0.02  # 账户 2%
        self.in_position = False
        self.position_side = None
    
    def calculate_ema(self, prices: pd.Series, period: int) -> float:
        """计算指数移动平均线。"""
        return prices.ewm(span=period, adjust=False).mean().iloc[-1]
    
    def fetch_recent_prices(self, lookback: int = 100) -> pd.Series:
        """从蜡烛数据获取近期市场价格。"""
        candles = self.trader.info.candles(
            coin=self.coin,
            interval="5m",
            startTime=int((datetime.now() - timedelta(hours=12)).timestamp() * 1000),
            endTime=int(datetime.now().timestamp() * 1000)
        )
        
        prices = pd.Series([c['close'] for c in candles[-lookback:]])
        return prices
    
    def check_signals(self) -> str:
        """检查 EMA 交叉信号。"""
        prices = self.fetch_recent_prices()
        
        fast_ema = self.calculate_ema(prices, self.fast_ema_period)
        slow_ema = self.calculate_ema(prices, self.slow_ema_period)
        
        if fast_ema > slow_ema:
            return "BUY"
        elif fast_ema < slow_ema:
            return "SELL"
        return "HOLD"
    
    def run(self, check_interval: int = 60):
        """主机器人循环。"""
        print(f"启动 {self.coin} 趋势机器人")
        print(f"快速 EMA: {self.fast_ema_period}, 慢速 EMA: {self.slow_ema_period}")
        
        while True:
            try:
                signal = self.check_signals()
                print(f"[{datetime.now()}] 信号: {signal}")
                
                # 执行信号逻辑...
                time.sleep(check_interval)
                
            except Exception as e:
                print(f"机器人循环错误: {e}")
                time.sleep(10)

# 启动机器人
# bot = TrendFollowingBot(trader, "BTC")
# bot.run()

def calculate_position_size(self, account_value: float, risk_pct: float = 0.02) -> float:
    """根据账户价值和风险百分比计算仓位规模。"""
    risk_amount = account_value * risk_pct
    
    # 基于波动性调整
    volatility = self.fetch_volatility(self.coin, period=24)  # 24 小时波动性
    position_size = risk_amount / (volatility * account_value)
    
    return position_size