lang: vi slug: hyperliquid-perp-dex-trading title: ‘Siêu thanh khoản 2026: DEX vĩnh viễn trên chuỗi xử lý 2 tỷ đô la + khối lượng hàng ngày - Tích hợp Bot giao dịch’ description: ‘Hướng dẫn toàn diện về Hyperliquid, DEX vĩnh viễn hoàn toàn trên chuỗi xử lý khối lượng hàng ngày hơn 2 tỷ USD với hơn 100 cặp giao dịch, đòn bẩy lên tới 50 lần, hợp đồng thông minh HyperEVM và Python SDK để tích hợp bot.’ tags: [“ai”, “crypto”, “open-source”, “quantitative”, “trading”] date: 2026-05-20 00:00:00+08:00 lastmod: 2026-05-20 00:00:00+08:00 tech_stack: [] application_domain: Ai Trading source_version: ’' licensing_model: Open Source license_type: Proprietary file_size: ’' file_md5: ’' download_url: ’' backup_url: ’' github_repo: ‘https://github.com/hyperliquid-dex' last_maintained: ‘2026-05-20’ draft: false categories: [‘ai-trading’] aliases:
- /posts/hyperliquid-perp-dex-trading/
faqs:
- q: ‘What makes Hyperliquid different from dYdX and GMX?’ a: ‘Hyperliquid uses a fully on-chain Central Limit Order Book (CLOB) on its own L1 chain with sub-second block times and 10,000+ TPS, whereas dYdX v4 runs on a Cosmos appchain and GMX uses an AMM model. Its HyperEVM layer also adds smart contract composability that the others lack, and it consistently maintains $2B+ daily volume.’
- q: ‘How do I deposit funds onto Hyperliquid?’ a: ‘Deposits require USDC on Arbitrum: connect your wallet to the Hyperliquid interface, click Deposit, specify the USDC amount, and confirm the Arbitrum transaction. Funds typically arrive in your Hyperliquid account within about 5 minutes after Arbitrum confirmation.’
- q: ‘What are the trading fees on Hyperliquid?’ a: ‘Hyperliquid uses a maker-taker model with a -0.002% maker fee (a rebate paid to liquidity providers) and a 0.035% taker fee. High-volume traders can qualify for additional discounts through the Hyperliquid VIP program.’
- q: ‘Do I need an API key to trade on Hyperliquid programmatically?’ a: ‘No. Hyperliquid uses wallet-based authentication via EIP-712 signatures, so no API key registration is needed. Your Ethereum wallet and private key serve as your identity and authentication mechanism. To get started you need Python 3.10+, an Ethereum wallet, and USDC on Arbitrum.’
- q: ‘How much leverage and how many markets does Hyperliquid support?’ a: ‘Hyperliquid supports 100+ trading pairs across major cryptocurrencies with leverage up to 50x on select markets. It also offers both isolated and cross-margin modes, configurable per market through the Python SDK.’
featureImage: /images/articles/hyperliquid-2026-the-on-chain-perpetual.jpg #
{{< resource-info >}}
Date: 2026-05-19
Category: AI Trading
Tags: Hyperliquid, perpetual DEX, on-chain trading, leverage trading, trading bot, DeFi, HyperEVM
Read Time: 18 minutes
Giới thiệu: Tại sao siêu thanh khoản thống trị bối cảnh Perp DEXBối cảnh giao dịch hợp đồng tương lai vĩnh viễn phi tập trung đã trải qua một cơn địa chấn kể từ năm 2023 và tâm điểm của sự chuyển đổi này là Hyperliquid — DEX vĩnh viễn sổ đặt hàng hoàn toàn trên chuỗi đã xử lý liên tục hơn 2 tỷ đô la khối lượng giao dịch hàng ngày trong suốt năm 2026. Không giống như các DEX dựa trên nhà tạo lập thị trường tự động truyền thống (AMM) dựa vào nhóm thanh khoản và chịu sự trượt giá và thua lỗ tạm thời, Hyperliquid mang trải nghiệm Sổ lệnh giới hạn trung tâm (CLOB) quen thuộc đến cho blockchain, kết hợp chất lượng thực thi của các sàn giao dịch tập trung với lợi ích tự lưu ký và minh bạch của DeFi.Ra mắt với sứ mệnh loại bỏ những thỏa hiệp mà các nhà giao dịch gặp phải khi lựa chọn giữa địa điểm tập trung và phi tập trung, Hyperliquid đã phát triển để hỗ trợ 100+ cặp giao dịch trên các loại tiền điện tử lớn, cung cấp đòn bẩy lên tới 50x trên một số thị trường chọn lọc. Sự tích hợp gốc của nền tảng với HyperEVM — lớp thực thi tương thích với EVM được xây dựng có mục đích — đã mở ra cánh cửa cho các chiến lược giao dịch phức tạp, thực thi kháng MEV và khả năng kết hợp hợp đồng thông minh liền mạch mà trước đây không thể thực hiện được trên các DEX vĩnh viễn truyền thống.Đối với các nhà giao dịch thuật toán và nhà phát triển bot, Hyperliquid đại diện cho một sự thay đổi mô hình. Python SDK toàn diện của nền tảng, kết hợp với REST và WebSocket API hiệu suất cao, cho phép đặt hàng dưới giây, quản lý vị trí theo thời gian thực và thực thi chiến lược hoàn toàn tự động. Cho dù bạn đang xây dựng bot tạo thị trường, hệ thống theo xu hướng hay chiến lược chênh lệch giá đa sàn phức tạp, Hyperliquid cung cấp nền tảng cơ sở hạ tầng có thể xử lý khối lượng giao dịch cấp tổ chức mà không phải hy sinh tính phân cấp.Trong hướng dẫn toàn diện này, chúng ta sẽ khám phá mọi khía cạnh của giao dịch trên Hyperliquid vào năm 2026 — từ tìm hiểu kiến trúc cốt lõi đến xây dựng các bot giao dịch sẵn sàng sản xuất. Cuối cùng, bạn sẽ có kiến thức và mẫu mã cần thiết để tích hợp Hyperliquid vào hệ thống giao dịch thuật toán của mình.—## Tìm hiểu kiến trúc cốt lõi của Hyperliquid### Ưu điểm của CLOB: Tại sao sổ đặt hàng lại quan trọng đối với PerpetualsCác DEX vĩnh viễn truyền thống được xây dựng trên các mô hình AMM như vAMM (AMM ảo) gặp phải những hạn chế cố hữu: thanh khoản tập trung xung quanh mức giá hiện tại, độ trượt giá đáng kể đối với các đơn đặt hàng lớn và khả năng chịu lỗ tạm thời đối với các nhà cung cấp thanh khoản. Kiến trúc Sổ lệnh giới hạn trung tâm (CLOB) của Hyperliquid loại bỏ hoàn toàn những vấn đề này.Trong hệ thống CLOB, nhà giao dịch đặt lệnh giới hạn ở các mức giá cụ thể, tạo ra biểu đồ độ sâu rõ ràng cho thấy chính xác mức độ thanh khoản tồn tại ở mỗi mức giá. Điều này cho phép:- Chênh lệch giá thấp hơn — Các nhà tạo lập thị trường cạnh tranh trực tiếp, thu hẹp chênh lệch giá thầu-yêu cầu xuống mức tương đương với Binance và Bybit #
- Lệnh giới hạn trượt giá bằng 0 — Lệnh lớn được thực thi chính xác ở mức giá được chỉ định (hoặc tốt hơn) khi khớp
- Khám phá giá minh bạch — Tất cả các đơn đặt hàng đều hiển thị trên chuỗi, ngăn chặn thao tác ẩn
- Sử dụng vốn hiệu quả — Không cần khóa vốn trong các nhóm thanh khoản; vốn chỉ được sử dụng khi đơn đặt hàng được thực hiệnSổ đặt hàng của Hyperliquid được giải quyết hoàn toàn trên chuỗi L1 độc quyền của nó, đạt được thời gian tạo khối dưới 1 giây và hỗ trợ hơn 10.000 giao dịch mỗi giây. Hiệu suất này giúp loại bỏ các hình phạt về độ trễ thường gây khó khăn cho các địa điểm giao dịch trên chuỗi.### HyperEVM: Hợp đồng thông minh đáp ứng giao dịch tần suất caoSự ra đời của HyperEVM vào cuối năm 2024 là một bước ngoặt đối với Hyperliquid. Lớp thực thi tương thích EVM này cho phép:- Ví hợp đồng thông minh — Logic tài khoản có thể lập trình để kiểm soát truy cập nâng cao và thực thi tự động
- Chiến lược DeFi có thể tổng hợp — Tích hợp với các giao thức cho vay, trình tối ưu hóa lợi nhuận và các tính năng nguyên gốc trên chuỗi khác
- Các loại lệnh tùy chỉnh — Lệnh có điều kiện, lệnh dừng treo và thực thi TWAP thông qua tự động hóa hợp đồng thông minh
- Giao dịch không cần gas — Hỗ trợ giao dịch meta trong đó phí có thể được thanh toán bằng bất kỳ mã thông báo nào hoặc được tài trợ bởi bên thứ baĐối với các nhà phát triển bot, HyperEVM có nghĩa là bạn có thể triển khai các hợp đồng thông minh tương tác trực tiếp với cơ sở hạ tầng sàn giao dịch, thực hiện các chiến lược không thể thực hiện được trên các sàn giao dịch tập trung hoặc DEX truyền thống.—## Thiết lập môi trường giao dịch siêu thanh khoản của bạn### Điều kiện tiên quyết và đăng ký khóa APITrước khi viết bất kỳ mã nào, bạn sẽ cần thiết lập môi trường giao dịch của mình. Hyperliquid sử dụng xác thực dựa trên ví - không cần đăng ký khóa API. Ví Ethereum của bạn (thông qua chữ ký EIP-712) đóng vai trò là cơ chế nhận dạng và xác thực của bạn.Dụng cụ cần thiết:
- Python 3.10 trở lên
- Ví Ethereum có khóa riêng
- USDC trên Arbitrum cho tiền gửi (Lớp tiền gửi của Hyperliquid)### Cài đặt SDK Python siêu lỏngSDK Python chính thức cung cấp quyền truy cập toàn diện vào tất cả các chức năng trao đổi. Việc cài đặt rất đơn giản:``` bash
python -m venv siêu lỏng-env nguồn hyperliquid-env/bin/kích hoạt
Cài đặt SDK chính thức #
pip cài đặt hyperliquid-python-sdk
Cài đặt các phụ thuộc bổ sung để phát triển bot #
pip cài đặt websockets aiohttp pandas python-dotenv
ạo
tệp `.env` để lưu trữ cấu hình của bạn một cách an toàn:``` bash
# .env - KHÔNG BAO GIỜ cam kết điều này với kiểm soát phiên bản
PRIVATE_KEY=your_ethereum_private_key_tại đây
WALLET_ADDRESS=0x_your_wallet_address
TESTNET=true
```### Kết nối và xác thực cơ bảnĐây là mã cơ bản để thiết lập kết nối được xác thực tới Hyperliquid```
bas
h
# .env - KHÔNG BAO GIỜ cam kết điều này với kiểm soát phiên bản
PRIVATE_KEY=your_ethereum_private_key_tại đây
WALLET_ADDRESS=0x_your_wallet_address
TESTNET=true
``` hằng số nhập yperliquid.utils# Tải các biến môi trường
tải_dotenv()lớp HyperliquidTrader:
"""Khách hàng giao dịch siêu thanh khoản sẵn sàng sản xuất."""
def __init__(self, use_testnet=True):
self.private_key = os.getenv('PRIVATE_KEY')
self.wal```
pytho
n
hệ điều hành nhập khẩu
nhập asyncio
từ dotenv nhập tải_dotenv
từ nhập khẩu hyperliquid.exchange Trao đổi
từ thông tin nhập hyperliquid.info
từ hằng số nhập hyperliquid.utils
# Tải các biến môi trường
tải_dotenv()
lớp HyperliquidTrader:
"""Khách hàng giao dịch siêu thanh khoản sẵn sàng sản xuất."""
def __init__(self, use_testnet=True):
self.private_key = os.getenv('PRIVATE_KEY')
self.wallet_address = os.getenv('WALLET_ADDRESS')
# Chọn điểm cuối dựa trên môi trường
nếu use_testnet:
self.base_url = hằng số.TESTNET_API_URL
khác:
self.base_url = hằng số.MAINNET_API_URL
# Khởi tạo trao đổi và thông tin khách hàng
self.exchange = Trao đổi(
tự.wallet_address,
self.private_key,
tự.base_url
)
self.info = Thông tin(self.base_url)
print(f"Đã kết nối với Hyperliquid {'Testnet' if use_testnet else 'Mainnet'}")
print(f"Ví: {self.wallet_address}")
def get_account_summary(tự):
"""Lấy thông tin tài khoản toàn diện."""
user_state = self.info.user_state(self.wallet_address)
account_value = float(user_state['marginSummary']['accountValue'])
Total_margin_used = float(user_state['marginSummary']['totalMarginUsed'])
có thể rút được = float(user_state['withdrawable'])
print(f"Giá trị tài khoản: ${account_value:,.2f}")
print(f"Ký quỹ được sử dụng: ${total_margin_used:,.2f}")
print(f"Có thể rút: ${withdrawable:,.2f}")
trả về user_state
# Khởi tạo nhà giao dịch
nhà giao dịch = HyperliquidTrader(use_testnet=True)
Trader.get_account_summary()
```'name': nội dung['name'],
'max_leverage': nội dung['maxLeverage'],
'sz_decimals': nội dung['szDecimals'],
'only_isolat': assets.get('onlyIsolated', Sai)
})
print(f"\nThị trường có sẵn: {len(assets)}")
cho nội dung trong assets[:10]: # Hiển thị 10 đầu tiên
print(f" {asset['name']}: {asset['max_leverage']}x đòn bẩy tối đa")
trả lại tài sản# Cách sử dụng
tài sản = Trader.get_all_assets()
```---## Xây dựng Bot giao dịch sẵn sàng sản xuất### Dữ liệu thị trường theo thời gian thực qua WebSocketĐối với giao dịch thuật toán, dữ liệu thị trường có độ trễ thấp là điều cần thiết. API WebSocket của Hyperliquid cung cấp các cập nhật sổ đặt hàng, giao dịch và điền thông tin dành riêng cho người dùng theo thời gian thực:``` con trăn
nhập json
nhập ổ cắm weblớp HyperliquidWebSocketFeed:
"""Nguồn cấp dữ liệu WebSocket hiệu suất cao cho Hyperliquid."""
định nghĩa __init__(tự):
self.ws_url = "wss://api.hyperliquid.xyz/ws"
self.subscriptions = {}
self.orderbook_cache = {}
self.running = Sai
kết nối không đồng bộ (tự):
"""Thiết lập kết nối WebSocket bằng tính năng tự động kết nối lại."""
trong khi Đúng:
thử:
không đồng bộ với websockets.connect(self.ws_url) dưới dạng ws:
print("Đã kết nối WebSocket")
self.ws = ws
self.running = Đúng
# Đăng ký lại các kênh trước khi kết nối lại
dành cho phụ trong self.subscriptions.values():
đang chờ ws.send(json.dumps(sub))
chờ đợi self._listen()
ngoại trừ Ngoại lệ là e:
print(f"Lỗi WebSocket: {e}. Sẽ kết nối lại sau 5 giây...")
đang chờ asyncio.sleep(5)
async def _listen(self):
"""Xử lý tin nhắn đến."""
không đồng bộ f```
pytho
n
def get_all_assets(tự):
"""Truy xuất tất cả các thị trường vĩnh viễn có sẵn."""
meta = self.info.meta()
vũ trụ = meta['vũ trụ']
tài sản = []
đối với tài sản trong vũ trụ:
nội dung.append({
'tên': tài sản['tên'],
'max_leverage': nội dung['maxLeverage'],
'sz_decimals': nội dung['szDecimals'],
'only_isolat': assets.get('onlyIsolated', Sai)
})
print(f"\nThị trường có sẵn: {len(assets)}")
cho nội dung trong assets[:10]: # Hiển thị 10 đầu tiên
print(f" {asset['name']}: {asset['max_leverage']}x đòn bẩy tối đa")
trả lại tài sản
# Cách sử dụng
tài sản = Trader.get_all_assets()
``` ở cấp độ [1]],
'dấu thời gian': data.get('time', 0)
}
async def subscribe_orderbook(self, coin):
"""Đăng ký sổ đặt hàng theo thời gian thực cho một thị trường cụ thể."""
phụ = {
"phương thức": "đăng ký",
"đăng ký": {"loại": "l2Book", "coin": coin}
}
self.subscriptions[f"book_{coin}"] = sub
nếu self.running:
đang chờ self.ws.send(json.dumps(sub))
print(f"Đã đăng ký sổ đặt hàng {coin}")
async def subscribe_trades(self, coin):
"""Đăng ký nguồn cấp dữ liệu giao dịch theo thời gian thực."""
phụ = {
"phương thức": "đăng ký",
"đăng ký": {"loại": "giao dịch", "coin": coin}
}
self.subscriptions[f"trades_{coin}"] = sub
nếu self.running:
đang chờ self.ws.send(json.dumps(sub))# Cách sử dụng
async def main():
nguồn cấp dữ liệu = HyperliquidWebSocketFeed()
đang chờ nguồn cấp dữ liệu.subscribe_orderbook("BTC")
đang chờ nguồn cấp dữ liệu.subscribe_trades("BTC")
đang chờ nguồn cấp dữ liệu.```
pytho
n
nhập json
nhập ổ cắm web
lớp HyperliquidWebSocketFeed:
"""Nguồn cấp dữ liệu WebSocket hiệu suất cao cho Hyperliquid."""
định nghĩa __init__(tự):
self.ws_url = "wss://api.hyperliquid.xyz/ws"
self.subscriptions = {}
self.orderbook_cache = {}
self.running = Sai
kết nối không đồng bộ (tự):
"""Thiết lập kết nối WebSocket bằng tính năng tự động kết nối lại."""
trong khi Đúng:
thử:
không đồng bộ với websockets.connect(self.ws_url) dưới dạng ws:
print("Đã kết nối WebSocket")
self.ws = ws
self.running = Đúng
# Đăng ký lại các kênh trước khi kết nối lại
dành cho phụ trong self.subscriptions.values():
đang chờ ws.send(json.dumps(sub))
chờ đợi self._listen()
ngoại trừ Ngoại lệ là e:
print(f"Lỗi WebSocket: {e}. Sẽ kết nối lại sau 5 giây...")
đang chờ asyncio.sleep(5)
async def _listen(self):
"""Xử lý tin nhắn đến."""
không đồng bộ cho tin nhắn trong self.ws:
msg = json.loads(tin nhắn)
nếu msg.get("channel") == "l2Book":
đang chờ self._handle_orderbook(msg['data'])
Elif msg.get("channel") == "giao dịch":
đang chờ self._handle_trades(msg['data'])
tin nhắn Elif.get("channel") == "userFills":
đang chờ self._handle_fills(msg['data'])
async def _handle_orderbook(tự, dữ liệu):
"""Xử lý cập nhật sổ đặt hàng L2."""
xu = dữ liệu['coin']
cấp độ = dữ liệu ['cấp độ']
self.orderbook_cache[coin] = {
'giá thầu': [{'px': float(b['px']), 'sz': float(b['sz'])} cho b ở cấp độ[0]],
'asks': [{'px': float(a['px']), 'sz': float(a['sz'])} cho a ở các cấp độ[1]],
'dấu thời gian': data.get('time', 0)
}
async def subscribe_orderbook(self, coin):
"""Đăng ký sổ đặt hàng theo thời gian thực cho một thị trường cụ thể."""
phụ = {
"phương thức": "đăng ký",
"đăng ký": {"loại": "l2Book", "coin": coin}
}
self.subscriptions[f"book_{coin}"] = sub
nếu self.running:
đang chờ self.ws.send(json.dumps(sub))
print(f"Đã đăng ký sổ đặt hàng {coin}")
async def subscribe_trades(self, coin):
"""Đăng ký nguồn cấp dữ liệu giao dịch theo thời gian thực."""
phụ = {
"phương thức": "đăng ký",
"đăng ký": {"loại": "giao dịch", "coin": coin}
}
self.subscriptions[f"trades_{coin}"] = sub
nếu self.running:
đang chờ self.ws.send(json.dumps(sub))
# Cách sử dụng
async def main():
nguồn cấp dữ liệu = HyperliquidWebSocketFeed()
đang chờ nguồn cấp dữ liệu.subscribe_orderbook("BTC")
đang chờ nguồn cấp dữ liệu.subscribe_trades("BTC")
đang chờ nguồn cấp dữ liệu.connect ()
# asyncio.run(chính())
```p
x
= float(vị trí['entryPx'])
current_px = float(vị trí['markPx'])
kích thước = float(vị trí['szi'])
pnl = (current_px - entry_px) * kích thước nếu kích thước > 0 khác (entry_px - current_px) * abs(size)
pnl_pct = ((current_px / entry_px) - 1) * 100 * (1 nếu kích thước > 0 khác -1)
active_positions.append({
'đồng xu': vị trí['đồng xu'],
'size': kích thước,
'entry_price': entry_px,
'mark_price': current_px,
'unrealized_pnl': pnl,
'pnl_percent': pnl_pct,
'đòn bẩy': float(vị trí['đòn bẩy']['giá trị']),
'margin_used': float(vị trí['marginUsed']),
'liquidation_price': float(position.get('liquidationPx', 0))
})
trả về vị trí hoạt động
def set_leverage(self, coin: str, đòn bẩy: int):
"""Đặt đòn bẩy cho một thị trường cụ thể."""
kết quả = self.exchange.update_leverage(đòn bẩy, coin)
print(f"Đòn bẩy cho {coin} được đặt thành { đòn bẩy}x")
kết quả trả về
def close_position(self, coin: str):
"""Đóng toàn bộ vị thế cho một thị trường cụ thể."""
vị trí = self.get_positions()
cho vị trí ở các vị trí:
if pos['coin'] == coin và pos['size'] != 0:
is_buy = pos['size'] < 0 # Mua để đóng lệnh bán, bán để đóng lệnh mua
sz = abs(pos['size'])
trả về self.place_market_order(coin, is_buy, sz)
print(f"Không có vị trí mở cho {coin}")
trả về Không có# Ví dụ sử dụng
# Trader.set_leverage("BTC", 10)
# Trader.place_limit_order("BTC", Đúng, 0,1, 65000,0, "Gtc")
# vị thế = Trader.get_positions()
```### Ví dụ hoàn chỉnh về Bot theo xu hướngĐây là một bot hoàn chỉnh, theo xu hướng, sẵn sàng sản xuất, kết hợp tất cả các thành phần:``` con trăn
thời gian nhập khẩu
nhập gấu trúc dưới dạng pd
nhập numpy dưới dạng np
từ datetime nhập datetime, timedeltalớp TrendFollowingBot:
"""Bot theo xu hướng chéo EMA cho Hyperliquid."""
def __init__(self, nhà giao dịch: HyperliquidTrader, coin: str = "BTC"):
self.trader = nhà giao dịch
self.coin = đồng xu
self.fast_ema_thời gian = 9
self.slow_ema_thời gian = 21
self.risk_per_trade = 0,02 # 2% tài khoản
self.in_position = Sai
self.position_side = Không
def tính_ema(self,prices: pd.Series, Period: int) -> float:
"""Tính trung bình động hàm mũ."""
giá trả lại.ewm(span=thời gian, adjustment=False).mean().iloc[-1]
def getfetch_recent_price(self, lookback: int = 100) -> pd.Series:
"""Tìm nạp giá thị trường gần đây từ dữ liệu nến."""
# Sử dụng API REST cho nến lịch sử
nến = self.trader.info.candles(
xu=self.coin,
khoảng cách="5m",
startTime=int((datetime.now() - timedelta(hours=12)).timestamp() * 1000),
endTime=int(datetime.now().timestamp() * 1000)
)
``` con trăn
def place_market_order(self, coin: str, is_buy: bool, sz: float):
"""Thực hiện lệnh thị trường với khả năng chống trượt giá."""
order_type = {"limit": {"tif": "Ioc"}} # Ngay lập tức hoặc Hủy
kết quả = self.exchange.order(
đồng xu,
là_mua,
sz,
0, # Giá 0 cho thị trường IOC
thứ tự_loại,
less_only=Sai
)
print(f"Thị trường {'MUA' nếu is_buy khác 'BÁN'} {sz} {coin}")
print(f"Trạng thái: {kết quả['trạng thái']}")
nếu 'phản hồi' trong kết quả và 'dữ liệu' trong kết quả['response']:
trạng thái = kết quả['phản hồi']['dữ liệu']['trạng thái']
cho trạng thái trong trạng thái:
nếu 'điền' ở trạng thái:
điền = trạng thái ['đã điền']
print(f"Đã điền: {fill['totalSz']} @ avg {fill['avgPx']}")
kết quả trả về
def place_limit_order(self, coin: str, is_buy: bool, sz: float,
px: float, tif: str = "Gtc"):
"""Đặt lệnh giới hạn với thời gian có hiệu lực được chỉ định.
Tùy chọn TIF:
- Gtc: Tốt cho đến khi bị hủy
- Ioc: Ngay lập tức hoặc Hủy bỏ
- Fok: Điền hoặc Giết
"""
order_type = {"giới hạn": {"tif": tif}}
kết quả = self.exchange.order(
đồng xu,
là_mua,
sz,
nhãn hiệu,
thứ tự_loại,
less_only=Sai
)
print(f"Giới hạn {'MUA' nếu is_buy khác 'BÁN'} {sz} {coin} @ {px}")
kết quả trả về
def place_stop_loss_order(self, coin: str, is_buy: bool, sz: float,
trigger_px: float, limit_px: float):
"""Đặt lệnh dừng lỗ với giá kích hoạt."""
thứ tự_type = {
"kích hoạt": {
"triggerPx": str(trigger_px),
"isMarket": Đúng,
"tpsl": "sl"
}
}
kết quả = self.exchange.order(
đồng xu,
là_mua,
sz,
giới hạn_px,
thứ tự_loại,
less_only=Đúng
)
print(f"Dừng lỗ {'MUA' nếu is_buy khác 'BÁN'} {sz} {coin}")
print(f"Kích hoạt: {trigger_px}, Giới hạn: {limit_px}")
kết quả trả về
```e
n
t
_px
= float(mids.get(self.coin, 0))
nếu current_px == 0:
print("Không thể lấy giá hiện tại")
trở về
# Tính kích thước vị thế
sz = self.calcate_position_size(current_px)
nếu có tín hiệu == "MUA":
self.trader.place_market_order(self.coin, True, sz)
self.position_side = "DÀI"
tín hiệu Elif == "BÁN":
self.trader.place_market_order(self.coin, Sai, sz)
self.position_side = "NGẮN"
self.in_position = Đúng
def run(self, check_interval: int = 60):
"""Vòng lặp bot chính có kiểm tra tín hiệu."""
print(f"Bot xu hướng bắt đầu cho {self.coin}")
print(f"EMA nhanh: {self.fast_ema_ Period}, EMA chậm: {self.slow_ema_ Period}")
trong khi Đúng:
thử:
tín hiệu = self.check_signals()
print(f"[{datetime.now()}] Tín hiệu: {signal}")
self.execute_signal(tín hiệu)
# In vị trí hiện tại
vị trí = self.trader.get_positions()
cho vị trí ở các vị trí:
print(f" {pos['coin']}: {pos['size']} | "
f"PnL: ${pos['unrealized_pnl']:+.2f} ({pos['pnl_percent']:+.2f}%)")
time.sleep(check_interval)
ngoại trừ Ngoại lệ là e:
print(f"Lỗi trong vòng lặp bot: {e}")
thời gian.ngủ(10)# Khởi chạy bot
# bot = TrendFollowingBot(người giao dịch, "BTC")
# bot.run()
```---## Tích hợp API siêu lỏng nâng cao### REST API: Hoạt động hàng loạt và dữ liệu lịch sửĐối với các chiến lược yêu cầu phân tích lịch sử, API REST cung cấp quyền truy cập dữ liệu toàn diện:``` con trăn
def get_funding_rates(self, coin: str = None):
"""Tìm nạp tỷ lệ tài trợ hiện tại cho tất cả hoặc các thị trường cụ thể."""
meta = self.info.meta()
tài sản = meta['vũ trụ']
tài trợ_data = []
đối với nội dung trong assets[:20]: # Kiểm tra top 20
tên = tài sản['name']
ctx = self.info.funding_history(tên, 1)
nếu ctx:
tài trợ_data.append({
'đồng xu': tên,
'funding_rate': float(ctx[0].get('fundingRate', 0)),
'predicted_rate': float(ctx[0].get('predictedFundingRate'```
pytho
n
def get_positions(tự):
"""Tìm nạp tất cả các vị trí mở với chi tiết P&L."""
user_state = self.info.user_state(self.wallet_address)
vị trí = user_state.get('assetPositions', [])
active_positions = []
cho vị trí ở các vị trí:
vị trí = vị trí ['vị trí']
entry_px = float(vị trí['entryPx'])
current_px = float(vị trí['markPx'])
kích thước = float(vị trí['szi'])
pnl = (current_px - entry_px) * kích thước nếu kích thước > 0 khác (entry_px - current_px) * abs(size)
pnl_pct = ((current_px / entry_px) - 1) * 100 * (1 nếu kích thước > 0 khác -1)
active_positions.append({
'đồng xu': vị trí['đồng xu'],
'size': kích thước,
'entry_price': entry_px,
'mark_price': current_px,
'unrealized_pnl': pnl,
'pnl_percent': pnl_pct,
'đòn bẩy': float(vị trí['đòn bẩy']['giá trị']),
'margin_used': float(vị trí['marginUsed']),
'liquidation_price': float(position.get('liquidationPx', 0))
})
trả về vị trí hoạt động
def set_leverage(self, coin: str, đòn bẩy: int):
"""Đặt đòn bẩy cho một thị trường cụ thể."""
kết quả = self.exchange.update_leverage(đòn bẩy, coin)
print(f"Đòn bẩy cho {coin} được đặt thành { đòn bẩy}x")
kết quả trả về
def close_position(self, coin: str):
"""Đóng toàn bộ vị thế cho một thị trường cụ thể."""
vị trí = self.get_positions()
cho vị trí ở các vị trí:
if pos['coin'] == coin và pos['size'] != 0:
is_buy = pos['size'] < 0 # Mua để đóng lệnh bán, bán để đóng lệnh mua
sz = abs(pos['size'])
trả về self.place_market_order(coin, is_buy, sz)
print(f"Không có vị trí mở cho {coin}")
trả về Không có
# Ví dụ sử dụng
# Trader.set_leverage("BTC", 10)
# Trader.place_limit_order("BTC", Đúng, 0,1, 65000,0, "Gtc")
# vị thế = Trader.get_positions()
``` tài sản:
phụ = {
"phương thức": "đăng ký",
"đăng ký": {"loại": "allMids"}
}
đang chờ ws.send(json.dumps(sub))
phụ = {
"phương thức": "đăng ký",
"đăng ký": {"loại": "l2Book", "coin": nội dung}
}
đang chờ ws.send(json.dumps(sub))
print(f"Đã đăng ký {asset}")
chạy không đồng bộ (tự):
"""Vòng lặp chính với quản lý đa nội dung."""
không đồng bộ với websockets.connect(self.ws_url) dưới dạng ws:
đang chờ self.subscribe_all(ws)
không đồng bộ cho tin nhắn trong ws:
msg = json.loads(tin nhắn)
kênh = msg.get("kênh")
nếu kênh == "allMids":
đối với nội dung, giá tính bằng msg['data'].items():
nếu nội dung trong self.data_cache:
self.data_cache[asset]['mid'] = float(price)
kênh Elif == "l2Book":
xu = tin nhắn['data']['coin']
nếu có xu trong self.data_cache:
giá thầu = tin nhắn['data']['levels'][0]
hỏi = tin nhắn['data']['levels'][1]
nếu trả giá và hỏi:
best_bid = float(bids[0]['px'])
best_ask = float(yêu cầu[0]['px'])
self.data_cache[coin]['lây lan'] = best_ask - best_bid
# Gọi trình xử lý đã đăng ký
nếu kênh trong self.handlers:
đang chờ self.handlers[channel](msg)# Cách sử dụng
# nội dung = ["BTC", "ETH", "SOL", "AVAX", "ARB"]
# người quản lý = MultiAssetWebSocketManager(tài sản)
# asyncio.run(manager.run())
```---## Quản lý rủi ro trên siêu thanh khoản### Tách biệt so với Ký quỹ chéoHyperliquid hỗ trợ cả chế độ biệt lập và biên chéo:```
pytho
n
def set_cross_margin(self, coin: str):
"""Enable cross-margin mode for a market."""
result = self.exchange.update_isolated_margin(coin, False, None)
print(f"Cross margin enabled for {coin}")
return result
def set_isolated_margin(self, coin: str, leverag```
pytho
n
import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class TrendFollowingBot:
"""EMA crossover trend-following bot for Hyperliquid."""
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% of account
self.in_position = False
self.position_side = None
def calculate_ema(self, prices: pd.Series, period: int) -> float:
"""Calculate Exponential Moving Average."""
return prices.ewm(span=period, adjust=False).mean().iloc[-1]
def fetch_recent_prices(self, lookback: int = 100) -> pd.Series:
"""Fetch recent market prices from candle data."""
# Using REST API for historical candles
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)
)
closes = [float(c['c']) for c in candles if 'c' in c]
return pd.Series(closes)
def calculate_position_size(self, entry_px: float) -> float:
"""Calculate position size based on risk parameters."""
account = self.trader.get_account_summary()
account_value = float(account['marginSummary']['accountValue'])
risk_amount = account_value * self.risk_per_trade
# Risk-based sizing: size = risk_amount / (estimated stop distance)
stop_distance = entry_px * 0.015 # 1.5% stop
position_size = risk_amount / stop_distance
# Round to appropriate decimal places
return round(position_size, 4)
def check_signals(self) -> str:
"""Check for EMA crossover signals."""
prices = self.fetch_recent_prices()
if len(prices) < self.slow_ema_period + 5:
return "HOLD"
fast_ema = self.calculate_ema(prices, self.fast_ema_period)
slow_ema = self.calculate_ema(prices, self.slow_ema_period)
# Previous values for crossover detection
prev_fast = prices.ewm(span=self.fast_ema_period).mean().iloc[-2]
prev_slow = prices.ewm(span=self.slow_ema_period).mean().iloc[-2]
# Bullish crossover: fast crosses above slow
if prev_fast <= prev_slow and fast_ema > slow_ema:
return "BUY"
# Bearish crossover: fast crosses below slow
if prev_fast >= prev_slow and fast_ema < slow_ema:
return "SELL"
return "HOLD"
def execute_signal(self, signal: str):
"""Execute trading signal with proper risk management."""
if signal == "HOLD":
return
# Close existing position if opposite signal
if self.in_position:
if (signal == "BUY" and self.position_side == "SHORT") or \
(signal == "SELL" and self.position_side == "LONG"):
self.trader.close_position(self.coin)
self.in_position = False
time.sleep(1) # Wait for fill
else:
return # Same direction, hold
# Get current price for sizing
mids = self.trader.info.all_mids()
current_px = float(mids.get(self.coin, 0))
if current_px == 0:
print("Could not fetch current price")
return
# Calculate position size
sz = self.calculate_position_size(current_px)
if signal == "BUY":
self.trader.place_market_order(self.coin, True, sz)
self.position_side = "LONG"
elif signal == "SELL":
self.trader.place_market_order(self.coin, False, sz)
self.position_side = "SHORT"
self.in_position = True
def run(self, check_interval: int = 60):
"""Main bot loop with signal checking."""
print(f"Starting trend bot for {self.coin}")
print(f"Fast EMA: {self.fast_ema_period}, Slow EMA: {self.slow_ema_period}")
while True:
try:
signal = self.check_signals()
print(f"[{datetime.now()}] Signal: {signal}")
self.execute_signal(signal)
# Print current positions
positions = self.trader.get_positions()
for pos in positions:
print(f" {pos['coin']}: {pos['size']} | "
f"PnL: ${pos['unrealized_pnl']:+.2f} ({pos['pnl_percent']:+.2f}%)")
time.sleep(check_interval)
except Exception as e:
print(f"Error in bot loop: {e}")
time.sleep(10)
# Launch bot
# bot = TrendFollowingBot(trader, "BTC")
# bot.run()
```r
Arbitrum confirmation (typically under 5 minutes)Đối với khoản tiền gửi có lập trình, hãy sử dụng hợp đồng bắc cầu trực tiếp:``` con trăn
def Deposit_usdc(self, money_usdc: float):
"""Gửi USDC từ Arbitrum vào Hyperliquid."""
# Số tiền tính theo 6 chữ số thập phân (USDC)
lượng_wei = int(số tiền_usdc * 1_000_000)
kết quả = self.exchange.spot_transfer(
to=self.wallet_address,
số tiền=str(số_wei),
tài sản="USDC"
)
print(f"Đã gửi {số tiền_usdc} USDC")
kết quả trả về
```### Phí giao dịch trên Hyperliquid là bao nhiêu?Hyperliquid sử dụng **mô hình phí của người sản xuất**:
- **Phí nhà sản xuất**: -0,002% (âm — bạn được giảm giá khi cung cấp tính thanh khoản)
- **Phí người nhận**: 0,035%Các khoản phí này thuộc loại cạnh tranh nhất trong không gian DEX vĩnh viễn. Các nhà giao dịch khối lượng lớn có thể đủ điều kiện để được giảm phí bổ sung thông qua chương trình Hyperliquid VIP.###Siêu thanh khoản có an toàn không? Những rủi ro là gì?Hyperliquid hoạt động như một **sàn giao dịch không giám sát** — tiền của bạn vẫn nằm trong một hợp đồng thông minh được kiểm soát bởi các khóa riêng tư của bạn. Tuy nhiên, giống như bất kỳ giao thức DeFi nào, vẫn tồn tại rủi ro:1. **Rủi ro hợp đồng thông minh**: Trong khi được kiểm toán, các lỗi có thể ảnh hưởng đến an toàn quỹ
2. **Rủi ro thanh lý**: Các vị thế có đòn bẩy cao có thể bị thanh lý khi có biến động mạnh
3. **Rủi ro của Oracle**: Giá đánh dấu dựa vào nguồn cấp dữ liệu của Oracle có thể sai lệch về mặt lý thuyết
4. **Rủi ro cầu**: Tiền gửi chuyển qua cơ sở hạ tầng cầu ArbitrumNền tảng này đã xử lý khối lượng tích lũy hơn 500 tỷ USD kể từ khi thành lập mà không có sự cố bảo mật lớn nào. Quỹ bảo hiểm được duy trì để bảo vệ chống lại tổn thất xã hội hóa.### Làm cách nào để kiểm tra lại các chiến lược trước khi đi vào hoạt động?Hyperliquid cung cấp **dữ liệu lịch sử miễn phí** thông qua API của nó. Đây là một khung kiểm tra lại:```
pytho
n
def fetch_historical_candles(self, coin: str, interval: str = "1h",
days: int = 30) -> pd.DataFrame:
"""Fetch historical OHLCV data for backtesting."""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
candles = self.info.candles(
coin=coin,
interval=interval,
startTime=start_time,
endTime=end_time
)
df = pd.DataFrame(candles)
df['timestamp'] = pd.to_datetime(df['t'], unit='ms')
df['open'] = df['o'].astype(float)
df['high'] = df['h'].astype(float)
df['low'] = df['l'].astype(float)
df['close'] = df['c'].astype(float)
df['volume'] = df['v'].astype(float)
return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']]
pytho n class BacktestEngine: “““Simple backtesting engine for Hyperliquid strategies.”””
def __init__(self, data: pd.DataFrame, initial_capital: float = 10000):
self.data = data
self.capital = initial_capital
self.position = 0
self.trades = []
def run_ema_strategy(self, fast: int = 9, slow: int = 21):
"""Run EMA crossover backtest."""
self.data['fast_ema'] = self.data['close'].ewm(span=fast).mean()
self.data['slow_ema'] = self.data['close'].ewm(span=slow).mean()
for i in range(slow + 1, len(self.data)):
prev_fast = self.data['fast_ema'].iloc[i-1]
prev_slow = self.data['slow_ema'].iloc[i-1]
curr_fast = self.data['fast_ema'].iloc[i]
curr_slow = self.data['slow_ema'].iloc[i]
price = self.data['close'].iloc[i]
# Bullish crossover
if prev_fast <= prev_slow and curr_fast > curr_slow and self.position <= 0:
if self.position < 0:
self.capital += abs(self.position) * price # Close short
self.position = self.capital / price # Go long
self.capital = 0
self.trades.append({'time': self.data['timestamp'].iloc[i],
'action': 'BUY', 'price': price})
# Bearish crossover
elif prev_fast >= prev_slow and curr_fast < curr_slow and self.position >= 0:
if self.position > 0:
self.capital = self.position * price # Close long
self.position = -self.capital / price # Go short
self.capital = 0
self.trades.append({'time': self.data['timestamp'].iloc[i],
'action': 'SELL', 'price': price})
# Calculate final P&L
final_price = self.data['close'].iloc[-1]
final_value = self.capital + abs(self.position) * final_price
total_return = (final_value / 10000 - 1) * 100
print(f"Backtest Results:")
print(f"Total Return: {total_return:.2f}%")
print(f"Total Trades: {len(self.trades)}")
print(f"Final Portfolio Value: ${final_value:,.2f}")
return self.trades# Usage
data = trader.fetch_historical_candles(“BTC”, “1h”, 90) #
bt = BacktestEngine(data) #
bt.run_ema_strategy(9, 21) #
### Can
pytho
n
def get_funding_rates(self, coin: str = None):
“““Fetch current funding rates for all or specific markets.”””
meta = self.info.meta()
assets = meta[‘universe’]
funding_data = []
for asset in assets[:20]: # Check top 20
name = asset['name']
ctx = self.info.funding_history(name, 1)
if ctx:
funding_data.append({
'coin': name,
'funding_rate': float(ctx[0].get('fundingRate', 0)),
'predicted_rate': float(ctx[0].get('predictedFundingRate', 0))
})
# Sort by absolute funding rate
funding_data.sort(key=lambda x: abs(x['funding_rate']), reverse=True)
print("\nTop Funding Rates:")
for f in funding_data[:10]:
print(f" {f['coin']}: {f['funding_rate']*100:+.4f}%")
return funding_data
def get_recent_trades(self, coin: str, limit: int = 100):
"""Fetch recent public trades for volume analysis."""
trades = self.info.recent_trades(coin)
trade_list = []
for t in trades[:limit]:
trade_list.append({
'price': float(t['px']),
'size': float(t['sz']),
'side': 'buy' if t['side'] == 'B' else 'sell',
'time': t['time']
})
df = pd.DataFrame(trade_list)
# Calculate buy/sell ratio
buy_vol = df[df['side'] == 'buy']['size'].sum()
sell_vol = df[df['side'] == 'sell']['size'].sum()
ratio = buy_vol / sell_vol if sell_vol > 0 else float('inf')
print(f"\n{coin} Trade Analysis (last {limit} trades)")
print(f"Buy/Sell Ratio: {ratio:.2f} | Buy Vol: {buy_vol:.4f} | Sell Vol: {sell_vol:.4f}")
return df
Usage #
funding = trader.get_funding_rates() #
trades = trader.get_recent_trades(“ETH”, 200) #
your own creativity and market insight.Start building today and experience why billions in daily volume flow through Hyperliquid's orderbooks.---*Disclaimer: Cryptocurrency trading carries significant risk. This article is for educational purposes only and does not constitute financial advice. Always conduct your own research and never trade with funds you cannot afford to lose. Past performance does not guarantee future results.*---**Related Articles:**
- Minara AI Trading Bot
— AI-powered automated trading
- Binance Exchange
— World's leading crypto exchange**Resources:**
- [Hyperliquid Documentation](https://hyperliquid.gitbook.io/hyperliquid-docs)
- [GitHub: hyperliquid-dex](https://github.com/hyperliquid-dex)
- [Hyperliquid Python SDK](https://github.com/hyperliquid-dex/hyperliquid-python-sdk)
```pyt
h
o
n
class MultiAssetWebSocketManager:
"""Manage WebSocket connections for multiple assets simultaneously."""
def __init__(self, assets: list):
self.assets = assets
self.ws_url = "wss://api.hyperliquid.xyz/ws"
self.handlers = {}
self.data_cache = {asset: {'mid': 0, 'spread': 0} for asset in assets}
def register_handler(self, channel: str, handler):
"""Register custom handler for a channel type."""
self.handlers[channel] = handler
async def subscribe_all(self, ws):
"""Subscribe to all assets in batch."""
for asset in self.assets:
sub = {
"method": "subscribe",
"subscription": {"type": "allMids"}
}
await ws.send(json.dumps(sub))
sub = {
"method": "subscribe",
"subscription": {"type": "l2Book", "coin": asset}
}
await ws.send(json.dumps(sub))
print(f"Subscribed to {asset}")
async def run(self):
"""Main loop with multi-asset management."""
async with websockets.connect(self.ws_url) as ws:
await self.subscribe_all(ws)
async for message in ws:
msg = json.loads(message)
channel = msg.get("channel")
if channel == "allMids":
for asset, price in msg['data'].items():
if asset in self.data_cache:
self.data_cache[asset]['mid'] = float(price)
elif channel == "l2Book":
coin = msg['data']['coin']
if coin in self.data_cache:
bids = msg['data']['levels'][0]
asks = msg['data']['levels'][1]
if bids and asks:
best_bid = float(bids[0]['px'])
best_ask = float(asks[0]['px'])
self.data_cache[coin]['spread'] = best_ask - best_bid
# Call registered handlers
if channel in self.handlers:
await self.handlers[channel](msg)
# Usage
# assets = ["BTC", "ETH", "SOL", "AVAX", "ARB"]
# manager = MultiAssetWebSocketManager(assets)
# asyncio.run(manager.run())
pytho n def set_cross_margin(self, coin: str): “““Enable cross-margin mode for a market.””” result = self.exchange.update_isolated_margin(coin, False, None) print(f"Cross margin enabled for {coin}") return result
def set_isolated_margin(self, coin: str, leverage: int):
"""Enable isolated margin with specific leverage."""
result = self.exchange.update_isolated_margin(coin, True, leverage)
print(f"Isolated margin enabled for {coin} at {leverage}x")
return result
pytho
n
class RiskManager:
"""Comprehensive risk management system for Hyperliquid trading."""
def __init__(self, trader: HyperliquidTrader):
self.trader = trader
self.max_daily_loss = 0.05 # 5% max daily loss
self.max_position_size = 0.50 # 50% of account max
self.max_leverage = 25
self.daily_pnl = 0
self.last_reset = datetime.now().date()
def check_daily_limit(self) -> bool:
"""Check if daily loss limit has been reached."""
today = datetime.now().date()
if today != self.last_reset:
self.daily_pnl = 0
self.last_reset = today
account = self.trader.get_account_summary()
account_value = float(account['marginSummary']['accountValue'])
loss_pct = abs(min(self.daily_pnl, 0)) / account_value if account_value > 0 else 0
if loss_pct >= self.max_daily_loss:
print(f"DAILY LOSS LIMIT REACHED: {loss_pct*100:.2f}%")
return False
return True
def validate_order(self, coin: str, size: float, leverage: int) -> bool:
"""Validate order against risk parameters."""
# Check leverage limit
if leverage > self.max_leverage:
print(f"Leverage {leverage}x exceeds maximum {self.max_leverage}x")
return False
# Check position size limit
account = self.trader.get_account_summary()
account_value = float(account['marginSummary']['accountValue'])
mids = self.trader.info.all_mids()
price = float(mids.get(coin, 0))
notional = size * price
if notional / account_value > self.max_position_size:
print(f"Position size {notional/account_value*100:.1f}% exceeds limit")
return False
return True
def emergency_close_all(self):
"""Emergency close all positions immediately."""
print("EMERGENCY CLOSE ALL POSITIONS")
positions = self.trader.get_positions()
for pos in positions:
if pos['size'] != 0:
self.trader.close_position(pos['coin'])
time.sleep(0.5)
# Integrate with bot
# risk = RiskManager(trader)
# if risk.check_daily_limit() and risk.validate_order("BTC", 0.1, 10):
# trader.place_market_order("BTC", True, 0.1)
pytho n import aiohttp import asyncio
class AsyncHyperliquidClient: “““Async Hyperliquid client for maximum throughput.”””
def __init__(self):
self.base_url = "https://api.hyperliquid.xyz"
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit=100)
)
return self
async def __aexit__(self, *args):
await self.session.close()
async def batch_requests(self, requests: list) -> list:
"""Execute multiple requests concurrently."""
tasks = [self._make_request(req) for req in requests]
return await asyncio.gather(*tasks, return_exceptions=True)
async def _make_request(self, request: dict):
"""Single async request."""
async with self.session.post(
f"{self.base_url}/info",
json=request
) as resp:
return await resp.json()
Usage #
async with AsyncHyperliquidClient() as client: #
reqs = [{“type”: “meta”}, {“type”: “allMids”}] #
results = await client.batch_requests(reqs) #
pytho
n
def deposit_usdc(self, amount_usdc: float):
"""Deposit USDC from Arbitrum to Hyperliquid."""
# Amount in 6 decimal places (USDC)
amount_wei = int(amount_usdc * 1_000_000)
result = self.exchange.spot_transfer(
to=self.wallet_address,
amount=str(amount_wei),
asset="USDC"
)
print(f"Deposited {amount_usdc} USDC")
return result
pytho n def fetch_historical_candles(self, coin: str, interval: str = “1h”, days: int = 30) -> pd.DataFrame: “““Fetch historical OHLCV data for backtesting.””” end_time = int(datetime.now().timestamp() * 1000) start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
candles = self.info.candles(
coin=coin,
interval=interval,
startTime=start_time,
endTime=end_time
)
df = pd.DataFrame(candles)
df['timestamp'] = pd.to_datetime(df['t'], unit='ms')
df['open'] = df['o'].astype(float)
df['high'] = df['h'].astype(float)
df['low'] = df['l'].astype(float)
df['close'] = df['c'].astype(float)
df['volume'] = df['v'].astype(float)
return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']]
pytho
n
class BacktestEngine:
"""Simple backtesting engine for Hyperliquid strategies."""
def __init__(self, data: pd.DataFrame, initial_capital: float = 10000):
self.data = data
self.capital = initial_capital
self.position = 0
self.trades = []
def run_ema_strategy(self, fast: int = 9, slow: int = 21):
"""Run EMA crossover backtest."""
self.data['fast_ema'] = self.data['close'].ewm(span=fast).mean()
self.data['slow_ema'] = self.data['close'].ewm(span=slow).mean()
for i in range(slow + 1, len(self.data)):
prev_fast = self.data['fast_ema'].iloc[i-1]
prev_slow = self.data['slow_ema'].iloc[i-1]
curr_fast = self.data['fast_ema'].iloc[i]
curr_slow = self.data['slow_ema'].iloc[i]
price = self.data['close'].iloc[i]
# Bullish crossover
if prev_fast <= prev_slow and curr_fast > curr_slow and self.position <= 0:
if self.position < 0:
self.capital += abs(self.position) * price # Close short
self.position = self.capital / price # Go long
self.capital = 0
self.trades.append({'time': self.data['timestamp'].iloc[i],
'action': 'BUY', 'price': price})
# Bearish crossover
elif prev_fast >= prev_slow and curr_fast < curr_slow and self.position >= 0:
if self.position > 0:
self.capital = self.position * price # Close long
self.position = -self.capital / price # Go short
self.capital = 0
self.trades.append({'time': self.data['timestamp'].iloc[i],
'action': 'SELL', 'price': price})
# Calculate final P&L
final_price = self.data['close'].iloc[-1]
final_value = self.capital + abs(self.position) * final_price
total_return = (final_value / 10000 - 1) * 100
print(f"Backtest Results:")
print(f"Total Return: {total_return:.2f}%")
print(f"Total Trades: {len(self.trades)}")
print(f"Final Portfolio Value: ${final_value:,.2f}")
return self.trades
# Usage
# data = trader.fetch_historical_candles("BTC", "1h", 90)
# bt = BacktestEngine(data)
# bt.run_ema_strategy(9, 21)
```
💬 Bình luận & Thảo luận