创建Hyperliquid交易机器人

一键发币： Aptos | X Layer | SUI | SOL | BNB | ETH | BASE | ARB | OP | Polygon | Avalanche | 用AI学区块链开发

你有一个成功的交易想法，但没有时间全天候盯着图表吗？你想自动化你的策略，但被复杂的API和测试时丢失真实资金的风险所困扰吗？我明白。这就是Hyperliquid的作用。它是一个高性能的去中心化交易所，具有一个对开发者来说强大的功能：一个专为实际交易设计的、开源的API，以及一个完整的测试网。

是的，你没听错。你可以零风险地构建、测试和完善你的交易机器人，然后再部署到主网。让我们一起建立一个简单的机器人。

在我们开始之前，你需要一个Hyperliquid账户。我们将在这篇教程中使用测试网，但有了账户，当你准备好的时候，你就可以准备好进行真正的交易了。

1、开发环境设置

首先，我们需要设置我们的工作区。这是后续所有工作的基础。

先决条件：

• 在你的机器上安装Python 3.8+。
• 安装Git用于克隆仓库。
• 安装一个web3钱包，如MetaMask，安装在你的浏览器中。

1.1 连接到Hyperliquid测试网

导航到测试网网站。当提示时连接你的MetaMask钱包。这个钱包是你在DEX上的身份。通常，你可以从该网站的水龙头获取免费的测试网USDC来资助你的测试交易。

1.2 获取官方Python SDK

Hyperliquid团队在GitHub上提供了一个出色的Python SDK（软件开发工具包）。打开你的终端并运行以下命令来克隆它：

git clone https://github.com/hyperliquid-dex/hyperliquid-python-sdk.git

现在，进入你刚刚创建的目录：

cd hyperliquid-python-sdk

1.3 安装依赖项

SDK有一些Python库作为依赖项。使用pip轻松安装它们：

pip install -r requirements.txt

你还需要pandas来进行RSI计算和requests来获取价格数据。也安装它们：

pip install pandas requests

1.4 配置你的凭证

为了让你的机器人代表你进行交易，它需要API密钥。在Hyperliquid测试网网站上，前往“设置”页面并生成一组新的API密钥。

关键: 你会得到一个密钥。这个密钥只显示一次。请安全保存。SDK需要这个密钥来签署交易。虽然存储密钥的方法可以有所不同，但对于本教程，你可以按照SDK的示例来设置你的凭证，通常涉及将它们放在config.json文件中或直接在脚本中进行测试。

记住，这些是测试网密钥。它们没有实际价值。当你准备好在主网上交易时，必须在主网网站上生成新的密钥，并更新你的机器人的配置。

2、像猎人一样准备：策略与数据

一个好的猎人在追逐前会做好准备。对于算法交易者来说，这种准备就是策略和数据。糟糕的数据或策略会导致失败，无论你的代码有多好。

2.1 策略：相对强弱指数（RSI）

我们将使用基于RSI指标的简单但有效的策略。

• RSI是一个动量指标，衡量价格变动的速度和变化。
• 读数高于70表示资产“超买”，价格可能下跌（做空信号）。
• 读数低于30表示资产“超卖”，价格可能上涨（做多信号）。

2.2 数据：CoinGecko API

我们需要一个可靠的比特币价格来源。免费的CoinGecko API非常适合这一点。它速度快，公开且对于简单的价格查询不需要API密钥。

3、代码：你的第一个BTC交易机器人

这里是完整的Python脚本。将其保存为rsi_bot.py，放在你之前克隆的hyperliquid-python-sdk文件夹中。阅读注释以了解每个部分的功能。

import time  
import requests  
import pandas as pd  
from eth_account import Account  
from eth_account.signers.local import LocalAccount

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

# --- CONFIGURATION ---
ASSET = "BTC"
TRADE_SIZE_USD = 100  # Desired trade size in USD
RSI_PERIOD = 14
RSI_OVERBOUGHT = 70
RSI_OVERSOLD = 30
LOOP_INTERVAL_SECONDS = 300  # 5 minutes

def get_b_token_price_from_coingecko():
    """Fetches the current price of Bitcoin from CoinGecko."""
    
    try:
        url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd"
        response = requests.get(url)
        response.raise_for_status()
        price = response.json()['bitcoin']['usd']
        print(f"LOG: Fetched BTC price: ${price}")
        return float(price)
    except requests.exceptions.RequestException as e:
        print(f"ERROR: Could not fetch price from CoinGecko: {e}")
        return None
        
def calculate_rsi(prices, period=14):
    """Calculates the RSI for a given list of prices."""
    
    if len(prices) < period:
        return None  # Not enough data
    price_series = pd.Series(prices)
    delta = price_series.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi.iloc[-1]
    
def run_bot():
    # IMPORTANT: Replace with your Testnet private key.
    # NEVER expose this key in a public repository.
    # For real bots, use a secure method like environment variables or a key management service.
    
    account: LocalAccount = Account.from_key("0x" + "YOUR_TESTNET_PRIVATE_KEY_HERE")
    # Initialize the Info and Exchange clients for Testnet
    info = Info(constants.TESTNET_API_URL, skip_ws=True)
    exchange = Exchange(account, constants.TESTNET_API_URL)
    # Login to the exchange to enable trading
    user_state = info.user_state(account.address)
    if not user_state.get("is_logged_in", False):
        print("LOG: User is not logged in. Sending login signature...")
        exchange.login_if_needed()
    print("Bot starting... Press Ctrl+C to stop.")
    price_history = []
    while True:
        try:
            current_price = get_b_token_price_from_coingecko()
            if current_price is None:
                time.sleep(60)
                continue
            price_history.append(current_price)
            # Keep the price history from getting too large
            if len(price_history) > RSI_PERIOD * 10:
                price_history.pop(0)
            rsi_value = calculate_rsi(price_history)
            if rsi_value is None:
                print(f"LOG: Need more data points to calculate RSI. Have {len(price_history)}, need at least {RSI_PERIOD + 1}.")
                time.sleep(LOOP_INTERVAL_SECONDS)
                continue
            print(f"LOG: Current RSI is {rsi_value:.2f}")
            user_state = info.user_state(account.address)
            positions = user_state.get("assetPositions", [])
            # Find if we have an existing position for our asset
            current_position = next((p for p in positions if p.get("position", {}).get("coin") == ASSET), None)
            in_long_position = current_position and float(current_position["position"]["szi"]) > 0
            in_short_position = current_position and float(current_position["position"]["szi"]) < 0
            # --- TRADING LOGIC ---
            if rsi_value < RSI_OVERSOLD and not in_long_position:
                print(f"SIGNAL: RSI is oversold ({rsi_value:.2f}). Placing LONG order.")
                trade_size_in_asset = TRADE_SIZE_USD / current_price
                order_result = exchange.order(ASSET, True, trade_size_in_asset, current_price, {"limit": {"tif": "Gtc"}})
                print(f"ACTION: Placed LONG order. Result: {order_result}")
            elif rsi_value > RSI_OVERBOUGHT and not in_short_position:
                print(f"SIGNAL: RSI is overbought ({rsi_value:.2f}). Placing SHORT order.")
                trade_size_in_asset = TRADE_SIZE_USD / current_price
                order_result = exchange.order(ASSET, False, trade_size_in_asset, current_price, {"limit": {"tif": "Gtc"}})
                print(f"ACTION: Placed SHORT order. Result: {order_result}")
            else:
                print(f"LOG: RSI is neutral or position already open. No action taken.")
        except Exception as e:
            print(f"An error occurred: {e}")
        print(f"--- Waiting for {LOOP_INTERVAL_SECONDS} seconds ---")
        time.sleep(LOOP_INTERVAL_SECONDS)
if __name__ == "__main__":
    run_bot()

4、让这个机器人运行起来！

兴奋吗？让我们让它活起来。

4.1 运行脚本

确保你已将"YOUR_TESTNET_PRIVATE_KEY_HERE"替换为你连接到测试网的钱包的实际私钥。从终端中，进入hyperliquid-python-sdk目录，运行：

python rsi_bot.py

4.2 观看它的运行

你应该在终端中看到日志出现：机器人获取价格，计算RSI，并最终在条件满足时下订单。

4.3 在Hyperliquid上验证

这是最棒的部分。回到你的浏览器窗口，打开Hyperliquid测试网。你将看到你的机器人的订单出现在订单簿中，并且一旦成交，你的仓位就会显示出来。这是一个由你的代码执行的实时自动化策略！

就是这样。你刚刚在实时（测试）网络上构建并启动了你的第一个交易机器人。这是你的基础。从这里，你可以调整RSI设置，尝试不同的资产，或者编写全新的策略。SDK的官方示例是一个很好的灵感来源。

原文链接：How to Create a Trading Bot with Hyperliquid

DefiPlot翻译整理，转载请标明出处

免责声明：本站资源仅用于学习目的，也不应被视为投资建议，读者在采取任何行动之前应自行研究并对自己的决定承担全部责任。