my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Hyperliquid DEX Client - Perpetuals and Spot Trading on Hyperliquid L1"""
#2
#3	import json
#4	from typing import Optional, Dict, Any, List, Literal
#5	from dataclasses import dataclass
#6	import httpx
#7	import eth_account
#8	from eth_account.signers.local import LocalAccount
#9
#10	# Import from the Hyperliquid SDK
#11	import sys
#12	import os
#13	sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../hyperliquid-python-sdk-master'))
#14
#15	from hyperliquid.exchange import Exchange
#16	from hyperliquid.info import Info
#17	from hyperliquid.utils import constants
#18
#19
#20	HYPERLIQUID_MAINNET_URL = constants.MAINNET_API_URL
#21	HYPERLIQUID_TESTNET_URL = constants.TESTNET_API_URL
#22
#23	# Asset ID Constants
#24	# Perpetuals: index from meta (e.g., BTC = 0 on mainnet)
#25	# Spot: 10000 + spotInfo["index"]
#26	# Builder-deployed perps: 100000 + perp_dex_index * 10000 + index_in_meta
#27	SPOT_ASSET_OFFSET = 10000
#28	BUILDER_PERP_OFFSET = 100000
#29
#30
#31	@dataclass
#32	class HyperliquidPosition:
#33	"""Position information"""
#34	coin: str
#35	entry_price: float
#36	position_value: float
#37	size: float
#38	unrealized_pnl: float
#39	return_on_equity: float
#40	leverage_type: str
#41	leverage_value: int
#42	liquidation_price: Optional[float]
#43	margin_used: float
#44
#45
#46	@dataclass
#47	class HyperliquidOrder:
#48	"""Order information"""
#49	oid: int
#50	coin: str
#51	side: str
#52	size: float
#53	limit_price: float
#54	timestamp: int
#55	order_type: str
#56
#57
#58	class HyperliquidClient:
#59	"""Client for Hyperliquid DEX Perpetuals and Spot Trading"""
#60
#61	def __init__(
#62	self,
#63	wallet_address: str,
#64	private_key: str,
#65	use_testnet: bool = False,
#66	):
#67	"""
#68	Initialize Hyperliquid DEX client.
#69
#70	Args:
#71	wallet_address: Main account wallet address (0x...)
#72	private_key: Private key for signing (0x...)
#73	use_testnet: Whether to use testnet (default: False for mainnet)
#74	"""
#75	self.wallet_address = wallet_address
#76	self.private_key = private_key
#77	self.use_testnet = use_testnet
#78	self.base_url = HYPERLIQUID_TESTNET_URL if use_testnet else HYPERLIQUID_MAINNET_URL
#79
#80	# Initialize the wallet account
#81	self.account: LocalAccount = eth_account.Account.from_key(private_key)
#82
#83	# Initialize Info client (read operations)
#84	self.info = Info(self.base_url, skip_ws=True)
#85
#86	# Initialize Exchange client (write operations)
#87	self.exchange = Exchange(
#88	self.account,
#89	self.base_url,
#90	account_address=wallet_address if wallet_address != self.account.address else None
#91	)
#92
#93	# HTTP client for additional API calls
#94	self._client = httpx.AsyncClient(
#95	base_url=self.base_url,
#96	timeout=30.0
#97	)
#98
#99	# Cached metadata for asset ID lookups
#100	self._meta_cache: Optional[Dict[str, Any]] = None
#101	self._spot_meta_cache: Optional[Dict[str, Any]] = None
#102	self._perp_coin_to_index: Dict[str, int] = {}
#103	self._perp_index_to_coin: Dict[int, str] = {}
#104	self._spot_coin_to_index: Dict[str, int] = {}
#105	self._spot_index_to_coin: Dict[int, str] = {}
#106
#107	# ==================
#108	# Asset ID Management
#109	# ==================
#110	# Perpetuals: Use integer index from meta (e.g., BTC = 0 on mainnet)
#111	# Spot: Use 10000 + spotInfo["index"]
#112	# Builder-deployed perps: Use 100000 + perp_dex_index * 10000 + index_in_meta
#113
#114	def _refresh_meta_cache(self) -> Dict[str, Any]:
#115	"""Refresh perpetuals metadata cache."""
#116	self._meta_cache = self.info.meta()
#117	self._perp_coin_to_index.clear()
#118	self._perp_index_to_coin.clear()
#119
#120	for idx, asset in enumerate(self._meta_cache.get("universe", [])):
#121	name = asset.get("name", "")
#122	self._perp_coin_to_index[name] = idx
#123	self._perp_index_to_coin[idx] = name
#124
#125	return self._meta_cache
#126
#127	def _refresh_spot_meta_cache(self) -> Dict[str, Any]:
#128	"""Refresh spot metadata cache."""
#129	self._spot_meta_cache = self.info.spot_meta()
#130	self._spot_coin_to_index.clear()
#131	self._spot_index_to_coin.clear()
#132
#133	tokens = {t["index"]: t["name"] for t in self._spot_meta_cache.get("tokens", [])}
#134
#135	for idx, pair in enumerate(self._spot_meta_cache.get("universe", [])):
#136	base_idx, quote_idx = pair.get("tokens", [0, 0])
#137	base_name = tokens.get(base_idx, "")
#138	quote_name = tokens.get(quote_idx, "USDC")
#139	pair_name = f"{base_name}/{quote_name}"
#140
#141	# Also store by just the base name with @ prefix for API compatibility
#142	self._spot_coin_to_index[pair_name] = idx
#143	self._spot_coin_to_index[f"@{idx}"] = idx
#144	self._spot_index_to_coin[idx] = pair_name
#145
#146	return self._spot_meta_cache
#147
#148	def get_perp_asset_id(self, coin: str) -> int:
#149	"""
#150	Get perpetual asset ID for a coin.
#151
#152	Args:
#153	coin: Coin name (e.g., 'BTC', 'ETH', 'SOL')
#154
#155	Returns:
#156	Integer asset ID for perpetual trading
#157	"""
#158	# Check for builder-deployed perps (format: dex:COIN)
#159	if ":" in coin:
#160	dex_name, coin_name = coin.split(":", 1)
#161	# Builder perps use: 100000 + perp_dex_index * 10000 + index_in_meta
#162	# For now, return the coin name and let the SDK handle it
#163	return -1 # SDK handles named coins
#164
#165	if not self._perp_coin_to_index:
#166	self._refresh_meta_cache()
#167
#168	if coin not in self._perp_coin_to_index:
#169	# Refresh cache in case new assets were added
#170	self._refresh_meta_cache()
#171
#172	return self._perp_coin_to_index.get(coin, -1)
#173
#174	def get_spot_asset_id(self, coin: str) -> int:
#175	"""
#176	Get spot asset ID for a coin.
#177
#178	Args:
#179	coin: Coin name or pair (e.g., 'HYPE', 'PURR/USDC', '@107')
#180
#181	Returns:
#182	Integer asset ID for spot trading (10000 + index)
#183	"""
#184	if not self._spot_coin_to_index:
#185	self._refresh_spot_meta_cache()
#186
#187	# If already in @index format, parse it
#188	if coin.startswith("@"):
#189	try:
#190	idx = int(coin[1:])
#191	return SPOT_ASSET_OFFSET + idx
#192	except ValueError:
#193	pass
#194
#195	# Try to find by pair name
#196	if coin in self._spot_coin_to_index:
#197	return SPOT_ASSET_OFFSET + self._spot_coin_to_index[coin]
#198
#199	# Try to find by base token name with /USDC suffix
#200	pair_name = f"{coin}/USDC"
#201	if pair_name in self._spot_coin_to_index:
#202	return SPOT_ASSET_OFFSET + self._spot_coin_to_index[pair_name]
#203
#204	# Refresh and try again
#205	self._refresh_spot_meta_cache()
#206
#207	if coin in self._spot_coin_to_index:
#208	return SPOT_ASSET_OFFSET + self._spot_coin_to_index[coin]
#209
#210	if pair_name in self._spot_coin_to_index:
#211	return SPOT_ASSET_OFFSET + self._spot_coin_to_index[pair_name]
#212
#213	return -1
#214
#215	def get_coin_from_perp_asset_id(self, asset_id: int) -> str:
#216	"""Convert perpetual asset ID back to coin name."""
#217	if not self._perp_index_to_coin:
#218	self._refresh_meta_cache()
#219	return self._perp_index_to_coin.get(asset_id, "")
#220
#221	def get_coin_from_spot_asset_id(self, asset_id: int) -> str:
#222	"""Convert spot asset ID back to coin/pair name."""
#223	if asset_id >= SPOT_ASSET_OFFSET:
#224	idx = asset_id - SPOT_ASSET_OFFSET
#225	if not self._spot_index_to_coin:
#226	self._refresh_spot_meta_cache()
#227	return self._spot_index_to_coin.get(idx, f"@{idx}")
#228	return ""
#229
#230	def get_all_perp_asset_ids(self) -> Dict[str, int]:
#231	"""Get mapping of all perpetual coins to their asset IDs."""
#232	if not self._perp_coin_to_index:
#233	self._refresh_meta_cache()
#234	return dict(self._perp_coin_to_index)
#235
#236	def get_all_spot_asset_ids(self) -> Dict[str, int]:
#237	"""Get mapping of all spot pairs to their asset IDs (with 10000 offset)."""
#238	if not self._spot_coin_to_index:
#239	self._refresh_spot_meta_cache()
#240	return {k: SPOT_ASSET_OFFSET + v for k, v in self._spot_coin_to_index.items()}
#241
#242	# ==================
#243	# Account Information
#244	# ==================
#245
#246	def get_user_state(self) -> Dict[str, Any]:
#247	"""Get user account state including positions and margin summary."""
#248	return self.info.user_state(self.wallet_address)
#249
#250	def get_spot_user_state(self) -> Dict[str, Any]:
#251	"""Get user spot account state."""
#252	return self.info.spot_user_state(self.wallet_address)
#253
#254	def get_positions(self) -> List[HyperliquidPosition]:
#255	"""Get all open positions."""
#256	user_state = self.get_user_state()
#257	positions = []
#258
#259	for asset_position in user_state.get("assetPositions", []):
#260	pos = asset_position.get("position", {})
#261	if float(pos.get("szi", 0)) != 0:
#262	leverage = pos.get("leverage", {})
#263	positions.append(HyperliquidPosition(
#264	coin=pos.get("coin", ""),
#265	entry_price=float(pos.get("entryPx", 0)) if pos.get("entryPx") else 0,
#266	position_value=float(pos.get("positionValue", 0)),
#267	size=float(pos.get("szi", 0)),
#268	unrealized_pnl=float(pos.get("unrealizedPnl", 0)),
#269	return_on_equity=float(pos.get("returnOnEquity", 0)),
#270	leverage_type=leverage.get("type", "cross"),
#271	leverage_value=int(leverage.get("value", 1)),
#272	liquidation_price=float(pos.get("liquidationPx")) if pos.get("liquidationPx") else None,
#273	margin_used=float(pos.get("marginUsed", 0)),
#274	))
#275
#276	return positions
#277
#278	def get_margin_summary(self) -> Dict[str, Any]:
#279	"""Get margin summary for the account."""
#280	user_state = self.get_user_state()
#281	return user_state.get("marginSummary", {})
#282
#283	def get_account_value(self) -> float:
#284	"""Get total account value in USD."""
#285	margin_summary = self.get_margin_summary()
#286	return float(margin_summary.get("accountValue", 0))
#287
#288	def get_withdrawable(self) -> float:
#289	"""Get withdrawable balance."""
#290	user_state = self.get_user_state()
#291	return float(user_state.get("withdrawable", 0))
#292
#293	# ==================
#294	# Market Data
#295	# ==================
#296
#297	def get_all_mids(self) -> Dict[str, float]:
#298	"""Get mid prices for all assets."""
#299	mids = self.info.all_mids()
#300	return {k: float(v) for k, v in mids.items()}
#301
#302	def get_price(self, coin: str) -> float:
#303	"""Get current mid price for a coin."""
#304	mids = self.get_all_mids()
#305	return mids.get(coin, 0)
#306
#307	def get_l2_book(self, coin: str) -> Dict[str, Any]:
#308	"""Get L2 order book snapshot."""
#309	return self.info.l2_snapshot(coin)
#310
#311	def get_meta(self) -> Dict[str, Any]:
#312	"""Get exchange metadata (available assets, decimals, etc.)."""
#313	return self.info.meta()
#314
#315	def get_spot_meta(self) -> Dict[str, Any]:
#316	"""Get spot exchange metadata."""
#317	return self.info.spot_meta()
#318
#319	def get_meta_and_asset_ctxs(self) -> Any:
#320	"""Get metadata and asset contexts including funding rates."""
#321	return self.info.meta_and_asset_ctxs()
#322
#323	def get_funding_history(self, coin: str, start_time: int, end_time: Optional[int] = None) -> List[Dict]:
#324	"""Get funding history for a coin."""
#325	return self.info.funding_history(coin, start_time, end_time)
#326
#327	def get_candles(self, coin: str, interval: str, start_time: int, end_time: int) -> List[Dict]:
#328	"""Get OHLCV candle data."""
#329	return self.info.candles_snapshot(coin, interval, start_time, end_time)
#330
#331	# ==================
#332	# Orders
#333	# ==================
#334
#335	def get_open_orders(self) -> List[HyperliquidOrder]:
#336	"""Get all open orders."""
#337	orders = self.info.open_orders(self.wallet_address)
#338	return [
#339	HyperliquidOrder(
#340	oid=order.get("oid", 0),
#341	coin=order.get("coin", ""),
#342	side=order.get("side", ""),
#343	size=float(order.get("sz", 0)),
#344	limit_price=float(order.get("limitPx", 0)),
#345	timestamp=order.get("timestamp", 0),
#346	order_type="limit"
#347	)
#348	for order in orders
#349	]
#350
#351	def get_user_fills(self) -> List[Dict]:
#352	"""Get user's trade history."""
#353	return self.info.user_fills(self.wallet_address)
#354
#355	def get_user_fills_by_time(self, start_time: int, end_time: Optional[int] = None) -> List[Dict]:
#356	"""Get user's trade history by time range."""
#357	return self.info.user_fills_by_time(self.wallet_address, start_time, end_time)
#358
#359	# ==================
#360	# Trading - Perpetuals
#361	# ==================
#362
#363	def place_order(
#364	self,
#365	coin: str,
#366	is_buy: bool,
#367	size: float,
#368	limit_price: float,
#369	order_type: Literal["limit", "ioc", "alo"] = "limit",
#370	reduce_only: bool = False,
#371	cloid: Optional[str] = None,
#372	) -> Dict[str, Any]:
#373	"""
#374	Place a limit order.
#375
#376	Args:
#377	coin: Trading pair (e.g., 'BTC', 'ETH', 'SOL')
#378	is_buy: True for buy (long), False for sell (short)
#379	size: Order size
#380	limit_price: Limit price
#381	order_type: "limit" (GTC), "ioc" (Immediate or Cancel), "alo" (Add Liquidity Only)
#382	reduce_only: Whether this order can only reduce position
#383	cloid: Optional client order ID
#384
#385	Returns:
#386	Order response
#387	"""
#388	tif_map = {
#389	"limit": "Gtc",
#390	"ioc": "Ioc",
#391	"alo": "Alo"
#392	}
#393
#394	order_type_param = {"limit": {"tif": tif_map.get(order_type, "Gtc")}}
#395
#396	return self.exchange.order(
#397	name=coin,
#398	is_buy=is_buy,
#399	sz=size,
#400	limit_px=limit_price,
#401	order_type=order_type_param,
#402	reduce_only=reduce_only,
#403	cloid=cloid,
#404	)
#405
#406	def market_open(
#407	self,
#408	coin: str,
#409	is_buy: bool,
#410	size: float,
#411	slippage: float = 0.05,
#412	) -> Dict[str, Any]:
#413	"""
#414	Place a market order to open a position.
#415
#416	Args:
#417	coin: Trading pair (e.g., 'BTC', 'ETH', 'SOL')
#418	is_buy: True for long, False for short
#419	size: Position size
#420	slippage: Max slippage (default 5%)
#421
#422	Returns:
#423	Order response
#424	"""
#425	return self.exchange.market_open(coin, is_buy, size, slippage=slippage)
#426
#427	def market_close(
#428	self,
#429	coin: str,
#430	size: Optional[float] = None,
#431	slippage: float = 0.05,
#432	) -> Dict[str, Any]:
#433	"""
#434	Place a market order to close a position.
#435
#436	Args:
#437	coin: Trading pair (e.g., 'BTC', 'ETH', 'SOL')
#438	size: Size to close (None = close entire position)
#439	slippage: Max slippage (default 5%)
#440
#441	Returns:
#442	Order response
#443	"""
#444	return self.exchange.market_close(coin, sz=size, slippage=slippage)
#445
#446	def cancel_order(self, coin: str, oid: int) -> Dict[str, Any]:
#447	"""Cancel an order by order ID."""
#448	return self.exchange.cancel(coin, oid)
#449
#450	def cancel_all_orders(self) -> List[Dict[str, Any]]:
#451	"""Cancel all open orders."""
#452	open_orders = self.get_open_orders()
#453	results = []
#454	for order in open_orders:
#455	try:
#456	result = self.cancel_order(order.coin, order.oid)
#457	results.append(result)
#458	except Exception as e:
#459	results.append({"error": str(e), "oid": order.oid})
#460	return results
#461
#462	# ==================
#463	# Leverage & Margin
#464	# ==================
#465
#466	def update_leverage(self, coin: str, leverage: int, is_cross: bool = True) -> Dict[str, Any]:
#467	"""
#468	Update leverage for a coin.
#469
#470	Args:
#471	coin: Asset name (e.g., 'BTC', 'ETH')
#472	leverage: Leverage value (1-100+)
#473	is_cross: True for cross margin, False for isolated
#474
#475	Returns:
#476	Response
#477	"""
#478	return self.exchange.update_leverage(leverage, coin, is_cross)
#479
#480	def update_isolated_margin(self, coin: str, amount: float) -> Dict[str, Any]:
#481	"""
#482	Update isolated margin for a position.
#483
#484	Args:
#485	coin: Asset name
#486	amount: Amount to add (positive) or remove (negative)
#487
#488	Returns:
#489	Response
#490	"""
#491	return self.exchange.update_isolated_margin(amount, coin)
#492
#493	# ==================
#494	# Transfers
#495	# ==================
#496
#497	def usd_transfer(self, amount: float, destination: str) -> Dict[str, Any]:
#498	"""
#499	Transfer USD to another address.
#500
#501	Args:
#502	amount: Amount to transfer
#503	destination: Destination address
#504
#505	Returns:
#506	Response
#507	"""
#508	return self.exchange.usd_transfer(amount, destination)
#509
#510	def spot_transfer(self, amount: float, destination: str, token: str) -> Dict[str, Any]:
#511	"""
#512	Transfer spot token to another address.
#513
#514	Args:
#515	amount: Amount to transfer
#516	destination: Destination address
#517	token: Token name
#518
#519	Returns:
#520	Response
#521	"""
#522	return self.exchange.spot_transfer(amount, destination, token)
#523
#524	def usd_class_transfer(self, amount: float, to_perp: bool) -> Dict[str, Any]:
#525	"""
#526	Transfer between spot and perp.
#527
#528	Args:
#529	amount: Amount to transfer
#530	to_perp: True to transfer to perp, False to transfer to spot
#531
#532	Returns:
#533	Response
#534	"""
#535	return self.exchange.usd_class_transfer(amount, to_perp)
#536
#537	# ==================
#538	# Spot Trading
#539	# ==================
#540
#541	def spot_order(
#542	self,
#543	coin: str,
#544	is_buy: bool,
#545	size: float,
#546	limit_price: float,
#547	order_type: Literal["limit", "ioc"] = "limit",
#548	) -> Dict[str, Any]:
#549	"""
#550	Place a spot order.
#551
#552	Args:
#553	coin: Spot pair name (e.g., 'PURR/USDC')
#554	is_buy: True for buy, False for sell
#555	size: Order size
#556	limit_price: Limit price
#557	order_type: "limit" or "ioc"
#558
#559	Returns:
#560	Order response
#561	"""
#562	tif_map = {
#563	"limit": "Gtc",
#564	"ioc": "Ioc",
#565	}
#566
#567	order_type_param = {"limit": {"tif": tif_map.get(order_type, "Gtc")}}
#568
#569	return self.exchange.order(
#570	name=coin,
#571	is_buy=is_buy,
#572	sz=size,
#573	limit_px=limit_price,
#574	order_type=order_type_param,
#575	reduce_only=False,
#576	)
#577
#578	# ==================
#579	# Staking
#580	# ==================
#581
#582	def get_staking_summary(self) -> Dict[str, Any]:
#583	"""Get staking summary."""
#584	return self.info.user_staking_summary(self.wallet_address)
#585
#586	def get_staking_delegations(self) -> List[Dict]:
#587	"""Get user's staking delegations."""
#588	return self.info.user_staking_delegations(self.wallet_address)
#589
#590	# ==================
#591	# Account Fees
#592	# ==================
#593
#594	def get_user_fees(self) -> Dict[str, Any]:
#595	"""Get user fee schedule and volume."""
#596	return self.info.user_fees(self.wallet_address)
#597
#598	# ==================
#599	# Utilities
#600	# ==================
#601
#602	def get_available_coins(self) -> List[str]:
#603	"""Get list of available trading coins."""
#604	meta = self.get_meta()
#605	return [asset["name"] for asset in meta.get("universe", [])]
#606
#607	def get_available_spot_pairs(self) -> List[str]:
#608	"""Get list of available spot trading pairs."""
#609	spot_meta = self.get_spot_meta()
#610	pairs = []
#611	tokens = {t["index"]: t["name"] for t in spot_meta.get("tokens", [])}
#612	for pair in spot_meta.get("universe", []):
#613	base_idx, quote_idx = pair["tokens"]
#614	base = tokens.get(base_idx, "")
#615	quote = tokens.get(quote_idx, "")
#616	pairs.append(f"{base}/{quote}")
#617	return pairs
#618
#619	async def close(self):
#620	"""Close the HTTP client."""
#621	await self._client.aclose()
#622
#623	async def __aenter__(self):
#624	return self
#625
#626	async def __aexit__(self, exc_type, exc_val, exc_tb):
#627	await self.close()
#628

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public